87 lines
2.4 KiB
Python
87 lines
2.4 KiB
Python
|
"""DB round-trip integration tests — require a running kb-postgres.
|
||
|
|
|
||
|
|
Run with:
|
||
|
|
KB_TEST_DSN=postgresql://kb:<pw>@solaria:5433/kb pytest -m integration
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from kb_mail.db import get_envelope, insert_envelope
|
||
|
|
from kb_mail.envelope import Envelope
|
||
|
|
|
||
|
|
pytestmark = pytest.mark.integration
|
||
|
|
|
||
|
|
_TS = datetime(2024, 3, 15, 12, 0, 0, tzinfo=timezone.utc)
|
||
|
|
|
||
|
|
|
||
|
|
async def test_envelope_roundtrip(db_conn):
|
||
|
|
env = Envelope(
|
||
|
|
id="<roundtrip@fastmail.com>",
|
||
|
|
source="fastmail",
|
||
|
|
ts=_TS,
|
||
|
|
raw_ref="fastmail/2024/03/roundtrip_fastmail.com.eml",
|
||
|
|
)
|
||
|
|
await insert_envelope(db_conn, env)
|
||
|
|
fetched = await get_envelope(db_conn, env.id)
|
||
|
|
|
||
|
|
assert fetched is not None
|
||
|
|
assert fetched.id == env.id
|
||
|
|
assert fetched.source == env.source
|
||
|
|
assert fetched.ts == env.ts
|
||
|
|
assert fetched.raw_ref == env.raw_ref
|
||
|
|
assert fetched.geo is None
|
||
|
|
assert fetched.entities == []
|
||
|
|
|
||
|
|
|
||
|
|
async def test_envelope_with_entities_roundtrip(db_conn):
|
||
|
|
env = Envelope(
|
||
|
|
id="<entities@gmail.com>",
|
||
|
|
source="gmail",
|
||
|
|
ts=_TS,
|
||
|
|
raw_ref="gmail/2024/03/entities_gmail.com.eml",
|
||
|
|
entities=[{"type": "person", "name": "Alice"}, {"type": "org", "name": "ACME"}],
|
||
|
|
)
|
||
|
|
await insert_envelope(db_conn, env)
|
||
|
|
fetched = await get_envelope(db_conn, env.id)
|
||
|
|
|
||
|
|
assert fetched is not None
|
||
|
|
assert fetched.entities == env.entities
|
||
|
|
|
||
|
|
|
||
|
|
async def test_get_nonexistent_returns_none(db_conn):
|
||
|
|
result = await get_envelope(db_conn, "nonexistent-id-xyz-abc")
|
||
|
|
assert result is None
|
||
|
|
|
||
|
|
|
||
|
|
async def test_duplicate_insert_is_idempotent(db_conn):
|
||
|
|
env = Envelope(
|
||
|
|
id="<dup@fastmail.com>",
|
||
|
|
source="fastmail",
|
||
|
|
ts=_TS,
|
||
|
|
raw_ref="fastmail/2024/03/dup.eml",
|
||
|
|
)
|
||
|
|
await insert_envelope(db_conn, env)
|
||
|
|
await insert_envelope(db_conn, env) # ON CONFLICT DO NOTHING
|
||
|
|
|
||
|
|
fetched = await get_envelope(db_conn, env.id)
|
||
|
|
assert fetched is not None
|
||
|
|
assert fetched.id == env.id
|
||
|
|
|
||
|
|
|
||
|
|
async def test_envelope_with_geo_roundtrip(db_conn):
|
||
|
|
env = Envelope(
|
||
|
|
id="<geo@gmail.com>",
|
||
|
|
source="gmail",
|
||
|
|
ts=_TS,
|
||
|
|
raw_ref="gmail/2024/03/geo_gmail.com.eml",
|
||
|
|
geo={"lat": 52.2297, "lon": 21.0122, "label": "Warsaw"},
|
||
|
|
)
|
||
|
|
await insert_envelope(db_conn, env)
|
||
|
|
fetched = await get_envelope(db_conn, env.id)
|
||
|
|
|
||
|
|
assert fetched is not None
|
||
|
|
assert fetched.geo == env.geo
|