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>
56 lines
1.6 KiB
Python
56 lines
1.6 KiB
Python
"""Envelope model unit tests — no DB required."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from kb_mail.envelope import Envelope
|
|
|
|
|
|
def test_envelope_minimal():
|
|
ts = datetime(2024, 1, 15, 10, 30, tzinfo=timezone.utc)
|
|
env = Envelope(
|
|
id="<abc123@fastmail.com>",
|
|
source="fastmail",
|
|
ts=ts,
|
|
raw_ref="fastmail/2024/01/abc123_fastmail.com.eml",
|
|
)
|
|
assert env.id == "<abc123@fastmail.com>"
|
|
assert env.source == "fastmail"
|
|
assert env.ts == ts
|
|
assert env.geo is None
|
|
assert env.entities == []
|
|
|
|
|
|
def test_envelope_rejects_naive_ts():
|
|
with pytest.raises(ValueError, match="timezone-aware"):
|
|
Envelope(
|
|
id="msg",
|
|
source="fastmail",
|
|
ts=datetime(2024, 1, 1), # naive — no tzinfo
|
|
raw_ref="x",
|
|
)
|
|
|
|
|
|
def test_envelope_with_entities():
|
|
ts = datetime(2024, 6, 10, tzinfo=timezone.utc)
|
|
env = Envelope(
|
|
id="msg2",
|
|
source="gmail",
|
|
ts=ts,
|
|
raw_ref="gmail/2024/06/msg2.eml",
|
|
entities=[{"type": "person", "name": "Alice"}],
|
|
)
|
|
assert len(env.entities) == 1
|
|
assert env.entities[0]["name"] == "Alice"
|
|
|
|
|
|
def test_envelope_entities_are_independent():
|
|
"""Mutable default must not be shared between instances."""
|
|
ts = datetime(2024, 1, 1, tzinfo=timezone.utc)
|
|
a = Envelope(id="a", source="fastmail", ts=ts, raw_ref="a.eml")
|
|
b = Envelope(id="b", source="fastmail", ts=ts, raw_ref="b.eml")
|
|
a.entities.append({"type": "org", "name": "ACME"})
|
|
assert b.entities == []
|