"""Unit tests for the Paperless -> envelope adapter — no DB, no real HTTP.""" from __future__ import annotations import json from datetime import datetime, timezone import pytest from documents_ingest.paperless_adapter import ( build_entities, build_registry_index, find_source_mail, fetch_lookup, iter_documents, map_document, parse_document_ts, run, ) BASE_URL = "http://fake-paperless" def _doc( doc_id=1, created="2020-04-21T00:00:00+02:00", content="some ocr text", correspondent=None, tags=None, original_file_name="file.pdf", mime_type="application/pdf", ): return { "id": doc_id, "correspondent": correspondent, "tags": tags or [], "content": content, "created": created, "original_file_name": original_file_name, "mime_type": mime_type, } def _registry_entry(sha256, envelope_id, filename, consume_name): return {sha256: {"envelope_id": envelope_id, "filename": filename, "consume_name": consume_name, "size": 100, "ingested_at": "2026-07-13T17:50:49.281559+00:00"}} class TestParseDocumentTs: def test_parses_offset_to_utc(self): ts = parse_document_ts("2020-04-21T00:00:00+02:00") assert ts == datetime(2020, 4, 20, 22, 0, 0, tzinfo=timezone.utc) def test_result_is_utc_aware(self): assert parse_document_ts("2026-06-09T12:00:00+00:00").tzinfo is not None def test_parses_z_suffix(self): ts = parse_document_ts("2026-06-09T12:00:00Z") assert ts == datetime(2026, 6, 9, 12, 0, 0, tzinfo=timezone.utc) class TestBuildRegistryIndex: def test_inverts_sha256_key_to_consume_name_key(self): registry = { "sha-a": {"envelope_id": "env-a", "filename": "a.pdf", "consume_name": "2026-06-09_a.pdf"}, } index = build_registry_index(registry) assert index == { "2026-06-09_a.pdf": {"envelope_id": "env-a", "sha256": "sha-a", "filename": "a.pdf"}, } class TestFindSourceMail: def test_hit_on_matching_consume_name(self): index = build_registry_index({ "sha-a": {"envelope_id": "env-a", "filename": "a.pdf", "consume_name": "2026-06-09_a.pdf"}, }) match = find_source_mail("2026-06-09_a.pdf", index) assert match == {"envelope_id": "env-a", "sha256": "sha-a", "filename": "a.pdf"} def test_miss_returns_none(self): index = build_registry_index({ "sha-a": {"envelope_id": "env-a", "filename": "a.pdf", "consume_name": "2026-06-09_a.pdf"}, }) assert find_source_mail("polisa 920008969228.pdf", index) is None def test_none_filename_returns_none(self): assert find_source_mail(None, {}) is None class TestBuildEntities: def test_shape_without_correspondent_tags_or_source_mail(self): doc = _doc() entities = build_entities(doc, correspondents={}, tags={}, source_mail=None) assert entities == [ {"type": "content", "text": "some ocr text"}, {"type": "correspondent", "name": None}, {"type": "filename", "value": "file.pdf"}, {"type": "content_type", "value": "application/pdf"}, ] def test_correspondent_name_resolved(self): doc = _doc(correspondent=7) entities = build_entities(doc, correspondents={7: "WARTA"}, tags={}, source_mail=None) assert {"type": "correspondent", "name": "WARTA"} in entities def test_multiple_tags_each_get_own_entity(self): doc = _doc(tags=[1, 2]) entities = build_entities(doc, correspondents={}, tags={1: "faktury", 2: "2026"}, source_mail=None) tag_entities = [e for e in entities if e["type"] == "tag"] assert tag_entities == [{"type": "tag", "name": "faktury"}, {"type": "tag", "name": "2026"}] def test_no_tags_means_no_tag_entities(self): entities = build_entities(_doc(tags=[]), correspondents={}, tags={}, source_mail=None) assert not [e for e in entities if e["type"] == "tag"] def test_source_mail_entity_appended_when_present(self): source_mail = {"envelope_id": "env-a", "sha256": "sha-a", "filename": "a.pdf"} entities = build_entities(_doc(), correspondents={}, tags={}, source_mail=source_mail) assert entities[-1] == { "type": "source_mail", "envelope_id": "env-a", "attachment_sha256": "sha-a", "attachment_filename": "a.pdf", } def test_empty_content_becomes_empty_string_not_none(self): entities = build_entities(_doc(content=None), correspondents={}, tags={}, source_mail=None) assert entities[0] == {"type": "content", "text": ""} class TestMapDocument: def test_id_gets_paperless_prefix(self): env = map_document(_doc(doc_id=42), correspondents={}, tags={}, source_mail=None) assert env.id == "paperless:42" def test_source_is_paperless(self): env = map_document(_doc(), correspondents={}, tags={}, source_mail=None) assert env.source == "paperless" def test_raw_ref_is_document_id_string(self): env = map_document(_doc(doc_id=42), correspondents={}, tags={}, source_mail=None) assert env.raw_ref == "42" def test_geo_is_none(self): env = map_document(_doc(), correspondents={}, tags={}, source_mail=None) assert env.geo is None def test_ts_parsed_from_created(self): env = map_document(_doc(created="2020-04-21T00:00:00+02:00"), correspondents={}, tags={}, source_mail=None) assert env.ts == datetime(2020, 4, 20, 22, 0, 0, tzinfo=timezone.utc) class _FakeResponse: def __init__(self, payload): self._payload = payload async def __aenter__(self): return self async def __aexit__(self, *exc): return False async def json(self): return self._payload def raise_for_status(self): pass class _FakeSession: def __init__(self, pages: dict[str, dict]): self._pages = pages def get(self, url): return _FakeResponse(self._pages[url]) async def __aenter__(self): return self async def __aexit__(self, *exc): return False def _empty_page(): return {"count": 0, "next": None, "previous": None, "results": []} class TestFetchLookup: async def test_single_page(self): session = _FakeSession({ f"{BASE_URL}/api/tags/": {"count": 2, "next": None, "results": [ {"id": 1, "name": "faktury"}, {"id": 2, "name": "2026"}, ]}, }) result = await fetch_lookup(session, BASE_URL, "/api/tags/") assert result == {1: "faktury", 2: "2026"} async def test_follows_pagination(self): session = _FakeSession({ f"{BASE_URL}/api/tags/": { "count": 2, "next": f"{BASE_URL}/api/tags/?page=2", "results": [{"id": 1, "name": "a"}], }, f"{BASE_URL}/api/tags/?page=2": { "count": 2, "next": None, "results": [{"id": 2, "name": "b"}], }, }) result = await fetch_lookup(session, BASE_URL, "/api/tags/") assert result == {1: "a", 2: "b"} async def test_empty_results(self): session = _FakeSession({f"{BASE_URL}/api/correspondents/": _empty_page()}) assert await fetch_lookup(session, BASE_URL, "/api/correspondents/") == {} class TestIterDocuments: async def test_yields_all_docs_across_pages(self): url1 = f"{BASE_URL}/api/documents/?page_size=200&ordering=id" url2 = f"{BASE_URL}/api/documents/?page=2" session = _FakeSession({ url1: {"count": 3, "next": url2, "results": [_doc(1), _doc(2)]}, url2: {"count": 3, "next": None, "results": [_doc(3)]}, }) docs = [d async for d in iter_documents(session, BASE_URL, page_size=200)] assert [d["id"] for d in docs] == [1, 2, 3] async def test_limit_stops_early_across_pages(self): url1 = f"{BASE_URL}/api/documents/?page_size=200&ordering=id" url2 = f"{BASE_URL}/api/documents/?page=2" session = _FakeSession({ url1: {"count": 3, "next": url2, "results": [_doc(1), _doc(2)]}, url2: {"count": 3, "next": None, "results": [_doc(3)]}, }) docs = [d async for d in iter_documents(session, BASE_URL, page_size=200, limit=1)] assert [d["id"] for d in docs] == [1] class _FakeConn: def __init__(self, existing_ids=None): self.existing_ids = list(existing_ids or []) self.execute_calls: list[tuple] = [] async def fetch(self, query, *params): return [{"id": i} for i in self.existing_ids] async def execute(self, query, *params): self.execute_calls.append(params) return "INSERT 0 1" async def close(self): pass def _docs_page(docs, next_url=None): return {"count": len(docs), "next": next_url, "results": docs} class TestRun: def _patch(self, monkeypatch, conn, pages): async def _fake_connect(dsn): return conn monkeypatch.setattr("documents_ingest.paperless_adapter.asyncpg.connect", _fake_connect) def _fake_session_factory(*args, **kwargs): return _FakeSession(pages) monkeypatch.setattr("documents_ingest.paperless_adapter.aiohttp.ClientSession", _fake_session_factory) def _base_pages(self, docs): docs_url = f"{BASE_URL}/api/documents/?page_size=200&ordering=id" return { docs_url: _docs_page(docs), f"{BASE_URL}/api/correspondents/": _empty_page(), f"{BASE_URL}/api/tags/": _empty_page(), } async def test_dry_run_does_not_insert(self, monkeypatch, tmp_path): registry_path = tmp_path / "registry.json" registry_path.write_text("{}") conn = _FakeConn() pages = self._base_pages([_doc(1), _doc(2)]) self._patch(monkeypatch, conn, pages) stats = await run( dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", registry_path=registry_path, apply=False, ) assert stats["fetched"] == 2 assert stats["inserted"] == 2 assert conn.execute_calls == [] async def test_apply_inserts_new_documents(self, monkeypatch, tmp_path): registry_path = tmp_path / "registry.json" registry_path.write_text("{}") conn = _FakeConn() pages = self._base_pages([_doc(1), _doc(2)]) self._patch(monkeypatch, conn, pages) stats = await run( dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", registry_path=registry_path, apply=True, ) assert stats["inserted"] == 2 assert len(conn.execute_calls) == 2 async def test_idempotent_skips_existing_ids(self, monkeypatch, tmp_path): registry_path = tmp_path / "registry.json" registry_path.write_text("{}") conn = _FakeConn(existing_ids=["paperless:1"]) pages = self._base_pages([_doc(1), _doc(2)]) self._patch(monkeypatch, conn, pages) stats = await run( dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", registry_path=registry_path, apply=True, ) assert stats["already_in_db"] == 1 assert stats["inserted"] == 1 assert len(conn.execute_calls) == 1 async def test_rerun_after_apply_inserts_nothing_new(self, monkeypatch, tmp_path): registry_path = tmp_path / "registry.json" registry_path.write_text("{}") pages = self._base_pages([_doc(1), _doc(2)]) conn1 = _FakeConn() self._patch(monkeypatch, conn1, pages) first = await run( dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", registry_path=registry_path, apply=True, ) assert first["inserted"] == 2 # Second run sees both ids already in the DB (simulated via existing_ids). conn2 = _FakeConn(existing_ids=["paperless:1", "paperless:2"]) self._patch(monkeypatch, conn2, pages) second = await run( dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", registry_path=registry_path, apply=True, ) assert second["inserted"] == 0 assert second["already_in_db"] == 2 assert conn2.execute_calls == [] async def test_source_mail_linked_counted_on_registry_hit(self, monkeypatch, tmp_path): registry_path = tmp_path / "registry.json" registry_path.write_text(json.dumps( _registry_entry("sha-a", "env-a", "invoice.pdf", "2026-06-09_invoice.pdf") )) conn = _FakeConn() pages = self._base_pages([_doc(1, original_file_name="2026-06-09_invoice.pdf"), _doc(2)]) self._patch(monkeypatch, conn, pages) stats = await run( dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", registry_path=registry_path, apply=False, ) assert stats["source_mail_linked"] == 1 async def test_empty_content_counted(self, monkeypatch, tmp_path): registry_path = tmp_path / "registry.json" registry_path.write_text("{}") conn = _FakeConn() pages = self._base_pages([_doc(1, content=""), _doc(2, content="text")]) self._patch(monkeypatch, conn, pages) stats = await run( dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", registry_path=registry_path, apply=False, ) assert stats["empty_content"] == 1 async def test_map_error_is_isolated_and_counted(self, monkeypatch, tmp_path): registry_path = tmp_path / "registry.json" registry_path.write_text("{}") conn = _FakeConn() bad_doc = _doc(1, created=None) # parse_document_ts(None) raises pages = self._base_pages([bad_doc, _doc(2)]) self._patch(monkeypatch, conn, pages) stats = await run( dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", registry_path=registry_path, apply=False, ) assert stats["errors"] == 1 assert stats["inserted"] == 1 assert stats["fetched"] == 2 async def test_stats_balance_across_all_outcomes(self, monkeypatch, tmp_path): registry_path = tmp_path / "registry.json" registry_path.write_text("{}") conn = _FakeConn(existing_ids=["paperless:1"]) pages = self._base_pages([_doc(1), _doc(2), _doc(3, created=None)]) self._patch(monkeypatch, conn, pages) stats = await run( dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", registry_path=registry_path, apply=True, ) assert stats["fetched"] == 3 assert stats["fetched"] == stats["already_in_db"] + stats["inserted"] + stats["errors"] assert stats["already_in_db"] == 1 assert stats["inserted"] == 1 assert stats["errors"] == 1 async def test_limit_caps_documents_processed(self, monkeypatch, tmp_path): registry_path = tmp_path / "registry.json" registry_path.write_text("{}") conn = _FakeConn() pages = self._base_pages([_doc(1), _doc(2), _doc(3)]) self._patch(monkeypatch, conn, pages) stats = await run( dsn="postgresql://fake", paperless_url=BASE_URL, paperless_token="tok", registry_path=registry_path, limit=2, apply=False, ) assert stats["fetched"] == 2