homelab-codex-ws/jobs/gmail-header-backfill/tests/test_backfill.py

"""Unit tests for the gmail header backfill job — no DB, no external services."""
from __future__ import annotations

import json
from datetime import datetime, timezone

import pytest

from gmail_header_backfill.backfill import (
    _decode_jsonb,
    _has_headers,
    fetch_batch,
    parse_headers,
    parse_headers_fallback,
    run,
)


def _eml(headers: dict, body: str = "body") -> bytes:
    lines = [f"{k}: {v}" for k, v in headers.items()]
    return ("\r\n".join(lines) + "\r\n\r\n" + body).encode("utf-8")


class TestParseHeadersBasics:
    def test_shape_has_all_expected_keys(self):
        raw = _eml({
            "From": "alice@example.com",
            "To": "bob@example.com",
            "Subject": "Hello",
            "Date": "Tue, 10 Jun 2025 12:00:00 +0000",
        })
        headers = parse_headers(raw)
        assert headers["type"] == "headers"
        assert {"from", "to", "cc", "delivered_to", "subject", "date_raw"} <= headers.keys()

    def test_from_parsed_to_name_and_address(self):
        raw = _eml({"From": "Alice Example <alice@example.com>", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        headers = parse_headers(raw)
        assert headers["from"] == {"name": "Alice Example", "address": "alice@example.com"}

    def test_from_without_display_name(self):
        raw = _eml({"From": "alice@example.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        headers = parse_headers(raw)
        assert headers["from"] == {"name": None, "address": "alice@example.com"}

    def test_missing_from_is_none(self):
        raw = _eml({"To": "bob@example.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        headers = parse_headers(raw)
        assert headers["from"] is None

    def test_subject_present(self):
        raw = _eml({"From": "a@b.com", "Subject": "Twoja polisa", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        assert parse_headers(raw)["subject"] == "Twoja polisa"

    def test_subject_missing_is_none(self):
        raw = _eml({"From": "a@b.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        assert parse_headers(raw)["subject"] is None

    def test_empty_to_and_cc_yield_empty_lists(self):
        raw = _eml({"From": "a@b.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        headers = parse_headers(raw)
        assert headers["to"] == []
        assert headers["cc"] == []
        assert headers["delivered_to"] == []


class TestParseHeadersMultiTo:
    def test_multiple_to_addresses_across_one_header(self):
        raw = _eml({
            "From": "a@b.com",
            "To": "bob@example.com, Carol Jones <carol@example.com>",
            "Date": "Tue, 10 Jun 2025 12:00:00 +0000",
        })
        headers = parse_headers(raw)
        assert headers["to"] == [
            {"name": None, "address": "bob@example.com"},
            {"name": "Carol Jones", "address": "carol@example.com"},
        ]

    def test_quoted_display_name_with_comma_not_split(self):
        raw = _eml({
            "From": '"Kowalski, Jan" <jan@example.com>',
            "Date": "Tue, 10 Jun 2025 12:00:00 +0000",
        })
        headers = parse_headers(raw)
        assert headers["from"] == {"name": "Kowalski, Jan", "address": "jan@example.com"}


class TestParseHeadersMultiDeliveredTo:
    def test_multiple_delivered_to_headers_all_kept(self):
        lines = (
            "From: a@b.com\r\n"
            "Delivered-To: oskar+alias@gmail.com\r\n"
            "Delivered-To: oskar@gmail.com\r\n"
            "Date: Tue, 10 Jun 2025 12:00:00 +0000\r\n"
            "\r\n"
            "body"
        )
        headers = parse_headers(lines.encode())
        assert headers["delivered_to"] == ["oskar+alias@gmail.com", "oskar@gmail.com"]

    def test_single_delivered_to(self):
        raw = _eml({
            "From": "a@b.com",
            "Delivered-To": "oskar@gmail.com",
            "Date": "Tue, 10 Jun 2025 12:00:00 +0000",
        })
        assert parse_headers(raw)["delivered_to"] == ["oskar@gmail.com"]


class TestParseHeadersRfc2047:
    def test_decodes_encoded_word_subject(self):
        raw = _eml({
            "From": "a@b.com",
            "Subject": "=?UTF-8?Q?Twoja_polisa?=",
            "Date": "Tue, 10 Jun 2025 12:00:00 +0000",
        })
        assert parse_headers(raw)["subject"] == "Twoja polisa"

    def test_decodes_encoded_word_display_name(self):
        raw = _eml({
            "From": "=?UTF-8?B?V2FydGE=?= <no-reply@warta.pl>",
            "Date": "Tue, 10 Jun 2025 12:00:00 +0000",
        })
        headers = parse_headers(raw)
        assert headers["from"] == {"name": "Warta", "address": "no-reply@warta.pl"}

    def test_decodes_encoded_word_with_polish_chars(self):
        raw = _eml({
            "From": "a@b.com",
            "To": "=?UTF-8?Q?Oskar_K=C4=85pa=C5=82a?= <oskar@gmail.com>",
            "Date": "Tue, 10 Jun 2025 12:00:00 +0000",
        })
        headers = parse_headers(raw)
        assert headers["to"] == [{"name": "Oskar Kąpała", "address": "oskar@gmail.com"}]

    def test_malformed_encoded_word_does_not_raise(self):
        raw = _eml({
            "From": "a@b.com",
            "Subject": "=?UTF-8?B?not-valid-base64!!!?=",
            "Date": "Tue, 10 Jun 2025 12:00:00 +0000",
        })
        headers = parse_headers(raw)  # must not raise
        assert isinstance(headers["subject"], str)


class TestParseHeadersMultipleFrom:
    def test_multiple_from_headers_uses_first_and_logs(self, caplog):
        raw = (
            b"From: alice@example.com\r\n"
            b"From: bob@example.com\r\n"
            b"Date: Tue, 10 Jun 2025 12:00:00 +0000\r\n"
            b"\r\nbody"
        )
        headers = parse_headers(raw)
        assert headers["from"] == {"name": None, "address": "alice@example.com"}


class TestParseHeadersDateRaw:
    def test_date_raw_preserves_literal_original_text(self):
        # policy.default's DateHeader reformats (e.g. corrects weekday, zero-pads day) —
        # date_raw must preserve the byte-for-byte original text instead (plan §4.1).
        raw = _eml({"From": "a@b.com", "Date": "Mon, 9 Jun 2026 12:34:56 +0200"})
        assert parse_headers(raw)["date_raw"] == "Mon, 9 Jun 2026 12:34:56 +0200"

    def test_date_raw_none_when_missing(self):
        raw = _eml({"From": "a@b.com"})
        assert parse_headers(raw)["date_raw"] is None

    def test_date_raw_preserved_even_when_unparseable(self):
        raw = _eml({"From": "a@b.com", "Date": "not-a-date-at-all"})
        assert parse_headers(raw)["date_raw"] == "not-a-date-at-all"

    def test_date_raw_with_8bit_bytes_is_json_serializable_str(self):
        # compat32 .get() returns email.header.Header (not str) when the raw
        # value has 8-bit bytes; unguarded, json.dumps then raises TypeError —
        # this exact case crashed the original full run and lost 4999 rows.
        raw = (
            b"From: a@b.com\r\n"
            b"Date: Tue, 19 May 2009 10:27:09 +0200 (Ho\xe9ra)\r\n"
            b"\r\nbody"
        )
        headers = parse_headers(raw)
        assert isinstance(headers["date_raw"], str)
        json.dumps([headers], ensure_ascii=False).encode("utf-8")  # must not raise


# Anonymized real patterns from the 2026-07-14 diagnosis of the 9 parse_errors
# left by the full 225 030-row run (see the job README, "Fallback parse").
#
# Pattern 1 (7 of 9): RFC 2047 encoded-word whose decoded text contains a
# newline (=0A) inside a display name — policy.default raises
# "ValueError: invalid arguments; address parts cannot contain CR or LF".
_CRLF_ENCODED_WORD_EML = (
    b"From: Rekrutacja z =?utf-8?Q?ExampleCorp=0A?= <mailing@example.pl>\r\n"
    b"To: Jan Kowalski <jan@example.com>\r\n"
    b"Subject: Nowe oferty\r\n"
    b"Date: Mon, 15 Sep 2014 13:24:18 +0000\r\n"
    b"\r\nbody"
)

# Pattern 2 (1 of 9): RFC 5322 group syntax in To: — policy.default raises
# "AttributeError: 'Group' object has no attribute 'local_part'".
_GROUP_TO_EML = (
    b'From: "Maria Example" <maria@example.pl>\r\n'
    b"To: unlisted-recipients:; (no To-header on input)\r\n"
    b"Subject: Re: [List] hello\r\n"
    b"Date: Tue, 16 Oct 2007 20:12:41 +0200\r\n"
    b"\r\nbody"
)


class TestParseHeadersFallback:
    def test_typed_parse_raises_on_crlf_encoded_word_from(self):
        with pytest.raises(ValueError):
            parse_headers(_CRLF_ENCODED_WORD_EML)

    def test_fallback_recovers_crlf_encoded_word_from(self):
        headers = parse_headers_fallback(_CRLF_ENCODED_WORD_EML)
        assert headers["type"] == "headers"
        # address is extracted; display name stays a RAW undecoded string
        assert headers["from"]["address"] == "mailing@example.pl"
        assert "=?utf-8?Q?ExampleCorp" in headers["from"]["name"]
        assert headers["to"] == [{"name": "Jan Kowalski", "address": "jan@example.com"}]
        assert headers["subject"] == "Nowe oferty"
        assert headers["date_raw"] == "Mon, 15 Sep 2014 13:24:18 +0000"

    def test_typed_parse_raises_on_group_to(self):
        with pytest.raises(AttributeError):
            parse_headers(_GROUP_TO_EML)

    def test_fallback_recovers_group_to(self):
        headers = parse_headers_fallback(_GROUP_TO_EML)
        assert headers["from"] == {"name": "Maria Example", "address": "maria@example.pl"}
        assert headers["to"] == []  # empty group has no real addresses

    def test_fallback_shape_matches_normal_parse_shape(self):
        # Degradation is parse quality only, never the schema (§4.1).
        normal = parse_headers(_eml({
            "From": "a@b.com", "To": "c@d.com",
            "Subject": "x", "Date": "Tue, 10 Jun 2025 12:00:00 +0000",
        }))
        fallback = parse_headers_fallback(_CRLF_ENCODED_WORD_EML)
        assert set(fallback.keys()) == set(normal.keys())
        assert isinstance(fallback["to"], list)
        assert isinstance(fallback["delivered_to"], list)

    def test_fallback_output_is_json_and_utf8_safe(self):
        # compat32 over raw 8-bit bytes leaves surrogates in header text;
        # postgres jsonb rejects them, so the fallback must sanitize.
        raw = (
            b"From: Rekrutacja z =?utf-8?Q?ExampleCorp=0A?= <mailing@example.pl>\r\n"
            b"Subject: Pr\xe9sent\r\n"
            b"Date: Tue, 19 May 2009 10:27:09 +0200\r\n"
            b"\r\nbody"
        )
        headers = parse_headers_fallback(raw)
        # the undecodable byte degrades to U+FFFD (via compat32's Header) or
        # '?' (via _sanitize) — never a lone surrogate, which jsonb rejects
        assert headers["subject"] in ("Pr<EFBFBD>sent", "Pr?sent")
        json.dumps(headers, ensure_ascii=False).encode("utf-8")  # must not raise


class TestDecodeJsonb:
    def test_passes_through_object(self):
        assert _decode_jsonb([{"type": "attachment"}]) == [{"type": "attachment"}]

    def test_decodes_string(self):
        assert _decode_jsonb('[{"type": "attachment"}]') == [{"type": "attachment"}]

    def test_none_stays_none(self):
        assert _decode_jsonb(None) is None


class TestHasHeaders:
    def test_true_when_present(self):
        assert _has_headers([{"type": "attachment"}, {"type": "headers"}]) is True

    def test_false_when_absent(self):
        assert _has_headers([{"type": "attachment"}]) is False

    def test_false_on_empty_list(self):
        assert _has_headers([]) is False


class _FakeConn:
    def __init__(self, rows):
        self._rows = rows
        self.executemany_calls: list[tuple[str, list]] = []

    async def fetch(self, query, *params):
        return self._rows

    async def executemany(self, query, rows):
        self.executemany_calls.append((query, list(rows)))

    async def close(self):
        pass


def _row(envelope_id, raw_ref, entities):
    return {"id": envelope_id, "raw_ref": raw_ref, "entities": json.dumps(entities)}


def _attachment_entity():
    return {"type": "attachment", "filename": "x.pdf", "content_type": "application/pdf",
            "size": 100, "sha256": "abc"}


class TestRun:
    def _setup_archive(self, tmp_path, raw_ref, raw_bytes):
        archive_root = tmp_path / "archive"
        eml_path = archive_root / raw_ref
        eml_path.parent.mkdir(parents=True, exist_ok=True)
        eml_path.write_bytes(raw_bytes)
        return archive_root

    def _patch_connect(self, monkeypatch, conn):
        async def _fake_connect(dsn):
            return conn
        monkeypatch.setattr("gmail_header_backfill.backfill.asyncpg.connect", _fake_connect)

    async def test_dry_run_does_not_call_executemany(self, tmp_path, monkeypatch):
        raw = _eml({"From": "a@b.com", "Subject": "Hi", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        archive_root = self._setup_archive(tmp_path, "gmail/2025/06/m1.eml", raw)
        conn = _FakeConn([_row("env1", "gmail/2025/06/m1.eml", [_attachment_entity()])])
        self._patch_connect(monkeypatch, conn)

        stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=False)

        assert stats["scanned"] == 1
        assert stats["updated"] == 1
        assert conn.executemany_calls == []

    async def test_apply_calls_executemany_with_headers_entity(self, tmp_path, monkeypatch):
        raw = _eml({"From": "a@b.com", "Subject": "Hi", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        archive_root = self._setup_archive(tmp_path, "gmail/2025/06/m1.eml", raw)
        conn = _FakeConn([_row("env1", "gmail/2025/06/m1.eml", [_attachment_entity()])])
        self._patch_connect(monkeypatch, conn)

        stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True)

        assert stats["updated"] == 1
        assert len(conn.executemany_calls) == 1
        query, rows = conn.executemany_calls[0]
        assert "NOT EXISTS" in query
        assert rows[0][0] == "env1"
        patch = json.loads(rows[0][1])
        assert patch == [{
            "type": "headers",
            "from": {"name": None, "address": "a@b.com"},
            "to": [],
            "cc": [],
            "delivered_to": [],
            "subject": "Hi",
            "date_raw": "Tue, 10 Jun 2025 12:00:00 +0000",
        }]

    async def test_existing_entities_are_not_touched_client_side(self, tmp_path, monkeypatch):
        # The job never rewrites the existing manifest — it only appends a new patch row
        # for the UPDATE (entities || $2::jsonb happens in SQL, not here).
        raw = _eml({"From": "a@b.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        archive_root = self._setup_archive(tmp_path, "gmail/2025/06/m1.eml", raw)
        original_entities = [_attachment_entity()]
        conn = _FakeConn([_row("env1", "gmail/2025/06/m1.eml", original_entities)])
        self._patch_connect(monkeypatch, conn)

        await run(dsn="postgresql://fake", archive_root=archive_root, apply=True)

        query, rows = conn.executemany_calls[0]
        patch = json.loads(rows[0][1])
        assert len(patch) == 1
        assert patch[0]["type"] == "headers"

    async def test_idempotent_skips_rows_already_backfilled(self, tmp_path, monkeypatch):
        archive_root = tmp_path / "archive"
        archive_root.mkdir()
        entities = [_attachment_entity(), {"type": "headers", "from": None, "to": [],
                                            "cc": [], "delivered_to": [], "subject": None,
                                            "date_raw": None}]
        conn = _FakeConn([_row("env1", "gmail/2025/06/m1.eml", entities)])
        self._patch_connect(monkeypatch, conn)

        stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True)

        assert stats["already_has_headers"] == 1
        assert stats["updated"] == 0
        assert conn.executemany_calls == []

    async def test_missing_eml_file_counted_as_missing_file(self, tmp_path, monkeypatch):
        archive_root = tmp_path / "archive"
        archive_root.mkdir()
        conn = _FakeConn([_row("env1", "gmail/2025/06/missing.eml", [_attachment_entity()])])
        self._patch_connect(monkeypatch, conn)

        stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True)

        assert stats["missing_file"] == 1
        assert stats["read_errors"] == 0
        assert stats["updated"] == 0

    async def test_fallback_row_updated_and_counted_separately(self, tmp_path, monkeypatch):
        # A message the typed parse rejects (real anonymized pattern from the
        # 2026-07-14 diagnosis) must be recovered by the fallback, written with
        # the same schema, and labeled parsed_fallback — not a silent success.
        archive_root = self._setup_archive(
            tmp_path, "gmail/2014/09/m1.eml", _CRLF_ENCODED_WORD_EML)
        conn = _FakeConn([_row("env1", "gmail/2014/09/m1.eml", [_attachment_entity()])])
        self._patch_connect(monkeypatch, conn)

        stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True)

        assert stats["updated"] == 1
        assert stats["parsed_fallback"] == 1
        assert stats["parse_errors"] == 0
        _, rows = conn.executemany_calls[0]
        patch = json.loads(rows[0][1])
        assert patch[0]["type"] == "headers"
        assert patch[0]["from"]["address"] == "mailing@example.pl"

    async def test_parse_error_when_fallback_also_fails(self, tmp_path, monkeypatch):
        archive_root = self._setup_archive(
            tmp_path, "gmail/2014/09/m1.eml", _CRLF_ENCODED_WORD_EML)
        conn = _FakeConn([_row("env1", "gmail/2014/09/m1.eml", [_attachment_entity()])])
        self._patch_connect(monkeypatch, conn)

        def _boom(raw):
            raise RuntimeError("fallback failed too")
        monkeypatch.setattr(
            "gmail_header_backfill.backfill.parse_headers_fallback", _boom)

        stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True)

        assert stats["parse_errors"] == 1
        assert stats["parsed_fallback"] == 0
        assert stats["updated"] == 0

    async def test_stats_balance_across_all_outcomes(self, tmp_path, monkeypatch):
        # scanned = updated + already_has_headers + parse_errors + read_errors
        #           + missing_file   (parsed_fallback is a subset of updated)
        ok_raw = _eml({"From": "a@b.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        archive_root = self._setup_archive(tmp_path, "gmail/ok.eml", ok_raw)
        (archive_root / "gmail/fb.eml").write_bytes(_CRLF_ENCODED_WORD_EML)

        already = [{"type": "headers", "from": None, "to": [], "cc": [],
                    "delivered_to": [], "subject": None, "date_raw": None}]
        conn = _FakeConn([
            _row("env-ok", "gmail/ok.eml", [_attachment_entity()]),
            _row("env-already", "gmail/ok.eml", already),
            _row("env-fallback", "gmail/fb.eml", [_attachment_entity()]),
            _row("env-missing", "gmail/nope.eml", [_attachment_entity()]),
        ])
        self._patch_connect(monkeypatch, conn)

        stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True)

        assert stats["scanned"] == 4
        assert stats["scanned"] == (
            stats["updated"] + stats["already_has_headers"]
            + stats["parse_errors"] + stats["read_errors"] + stats["missing_file"]
        )
        assert stats["updated"] == 2
        assert stats["parsed_fallback"] == 1
        assert stats["already_has_headers"] == 1
        assert stats["missing_file"] == 1

    async def test_limit_and_offset_are_passed_to_query(self, tmp_path, monkeypatch):
        captured = {}

        class _FakeConnCapturing(_FakeConn):
            async def fetch(self, query, *params):
                captured["params"] = params
                return self._rows

        conn = _FakeConnCapturing([])
        self._patch_connect(monkeypatch, conn)
        archive_root = tmp_path / "archive"
        archive_root.mkdir()

        await run(dsn="postgresql://fake", archive_root=archive_root, limit=50, offset=200, apply=False)

        assert captured["params"] == (50, 200)

    async def test_batch_flush_multiple_rows_in_one_executemany(self, tmp_path, monkeypatch):
        raw = _eml({"From": "a@b.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"})
        archive_root = self._setup_archive(tmp_path, "gmail/2025/06/m1.eml", raw)
        rows = [_row(f"env{i}", "gmail/2025/06/m1.eml", [_attachment_entity()]) for i in range(3)]
        conn = _FakeConn(rows)
        self._patch_connect(monkeypatch, conn)

        stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True)

        assert stats["updated"] == 3
        assert len(conn.executemany_calls) == 1  # under UPDATE_BATCH_SIZE, one flush at end
        _, flushed_rows = conn.executemany_calls[0]
        assert len(flushed_rows) == 3


class TestFetchBatchQueryShape:
    async def test_selects_only_gmail_source_ordered_by_id(self, monkeypatch):
        captured = {}

        class _FakeConnCapturing(_FakeConn):
            async def fetch(self, query, *params):
                captured["query"] = query
                captured["params"] = params
                return []

        conn = _FakeConnCapturing([])
        await fetch_batch(conn, limit=10, offset=5)

        assert "source = 'gmail'" in captured["query"]
        assert "ORDER BY id" in captured["query"]
        assert captured["params"] == (10, 5)