1666511475
- services/kb-postgres: pgvector/pgvector:pg16 na SOLARIA (:5433), named volume, init/001_envelope.sql (CREATE EXTENSION vector + zamrożona tabela envelope: id/source/ts/geo/raw_ref/entities), service.yaml, healthcheck, README z poprawnym mechanizmem deploy (deploy-node.sh składa dwa -f) - hosts/solaria/runtime/kb-postgres/docker-compose.override.yml: mem_limit 4g - inventory/topology.yaml + hosts/solaria/services.yaml: kb-postgres wpisany - packages/kb-mail: nowa konwencja shared lib (pip install /repo/packages/<lib>/) envelope.py — @dataclass Envelope, walidacja tz-aware ts db.py — insert_envelope / get_envelope (asyncpg, ON CONFLICT DO NOTHING) archive.py — save_eml append-only (asyncio.to_thread, FileExistsError na dup) tests: 15 unit pass + 5 integration (@pytest.mark.integration, wymaga KB_TEST_DSN) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
"""Archive helper unit tests — no DB required, uses tmp_path."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from kb_mail.archive import save_eml
|
|
|
|
|
|
_TS = datetime(2024, 6, 10, tzinfo=timezone.utc)
|
|
_RAW = b"From: alice@example.com\r\nSubject: Hello\r\n\r\nBody text"
|
|
|
|
|
|
async def test_save_eml_creates_file(tmp_path):
|
|
raw_ref = await save_eml(tmp_path, "msg1@fastmail.com", "fastmail", _TS, _RAW)
|
|
|
|
dest = tmp_path / raw_ref
|
|
assert dest.exists()
|
|
assert dest.read_bytes() == _RAW
|
|
|
|
|
|
async def test_save_eml_returns_correct_rel_path(tmp_path):
|
|
raw_ref = await save_eml(tmp_path, "msg1@fastmail.com", "fastmail", _TS, _RAW)
|
|
# @ is safe on Linux/macOS filesystems and is preserved
|
|
assert raw_ref == "fastmail/2024/06/msg1@fastmail.com.eml"
|
|
|
|
|
|
async def test_save_eml_creates_subdirs(tmp_path):
|
|
await save_eml(tmp_path, "msgX", "gmail", datetime(2023, 12, 31, tzinfo=timezone.utc), b"data")
|
|
assert (tmp_path / "gmail" / "2023" / "12").is_dir()
|
|
|
|
|
|
async def test_save_eml_append_only_raises_on_duplicate(tmp_path):
|
|
await save_eml(tmp_path, "dup@fastmail.com", "fastmail", _TS, _RAW)
|
|
with pytest.raises(FileExistsError, match="append-only"):
|
|
await save_eml(tmp_path, "dup@fastmail.com", "fastmail", _TS, _RAW)
|
|
|
|
|
|
async def test_save_eml_sanitizes_unsafe_chars(tmp_path):
|
|
raw_ref = await save_eml(tmp_path, "<foo/bar:baz>", "fastmail", _TS, b"x")
|
|
# < and > stripped, / and : replaced with _
|
|
assert "<" not in raw_ref
|
|
assert ">" not in raw_ref
|
|
assert "/" not in raw_ref.split("fastmail/")[1].rsplit("/", 1)[-1]
|