homelab-codex-ws/jobs/mail-imap-sync/tests/test_sync.py

"""Unit tests for the IMAP sync job — no network, no DB, no live mailbox.

Every test drives the real `ImapClient` over a fake `imaplib` connection and the real
`run()` over an in-memory `envelope` / `mail_sync_state` pair, so what is being tested is the
job's actual behaviour: what it fetches, what it archives, where the cursor ends up. The
scenarios mirror the failure modes the recon named — new mail, UIDVALIDITY invalidation,
dedup, resumption after an interrupted run — plus the traps it warned would be silent.
"""
from __future__ import annotations

import json
from datetime import datetime, timezone
from pathlib import Path

import pytest
from conftest import FakeConn, FakeIMAP, eml

from kb_mail.imap import ImapAccount, ImapClient
from mail_imap_sync.prom import read_prev_value, render_prom
from mail_imap_sync.sync import (
    balance_errors,
    build_metrics,
    measure,
    run,
    total_errors,
)

GMAIL_FOLDER = "[Gmail]/Wszystkie"


def gmail_account(**kwargs) -> ImapAccount:
    base = dict(name="gmail", host="imap.example.com", user="u@example.com",
                password="secret", special_use=("\\All",))
    base.update(kwargs)
    return ImapAccount(**base)


def fastmail_account(**kwargs) -> ImapAccount:
    base = dict(name="fastmail", host="imap.example.net", user="u@example.net",
                password="secret", special_use=(), folders=("INBOX",))
    base.update(kwargs)
    return ImapAccount(**base)


def gmail_server(messages, **kwargs) -> FakeIMAP:
    return FakeIMAP({GMAIL_FOLDER: messages},
                    list_attrs={GMAIL_FOLDER: ["\\HasNoChildren", "\\All"]}, **kwargs)


def factory_for(servers: dict):
    """Client factory returning a real ImapClient bound to the named account's fake server."""
    def _factory(account: ImapAccount) -> ImapClient:
        return ImapClient(account, connection_factory=lambda acc, timeout: servers[acc.name])
    return _factory


async def do_run(servers, accounts, conn, archive, *, apply=True, limit=None, monkeypatch=None):
    async def _connect(_dsn):
        return conn
    import asyncpg
    monkeypatch.setattr(asyncpg, "connect", _connect)
    return await run(
        dsn="postgresql://fake",
        accounts=accounts,
        archive_root=archive,
        apply=apply,
        limit=limit,
        client_factory=factory_for(servers),
    )


class TestNewMessages:
    async def test_fetches_archives_and_inserts(self, tmp_path: Path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
        conn = FakeConn()
        result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                              conn, tmp_path, monkeypatch=monkeypatch)

        assert result["totals"]["envelopes_inserted"] == 2
        assert set(conn.envelopes) == {"a@x", "b@x"}
        assert sorted(p.name for p in tmp_path.rglob("*.eml")) == ["a@x.eml", "b@x.eml"]
        assert not result["failed"]

    async def test_archive_path_is_source_year_month(self, tmp_path: Path, monkeypatch):
        server = gmail_server({1: eml("a@x")})
        await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                     FakeConn(), tmp_path, monkeypatch=monkeypatch)
        assert (tmp_path / "gmail" / "2026" / "06" / "a@x.eml").exists()

    async def test_cursor_advances_to_the_highest_processed_uid(self, tmp_path: Path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
        conn = FakeConn()
        await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                     conn, tmp_path, monkeypatch=monkeypatch)
        assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2)

    async def test_envelope_carries_headers_at_insert_time(self, tmp_path: Path, monkeypatch):
        # Recon §2.5 (i): without this, mail_body_ingest.build_prefix silently produces
        # "Temat: (brak tematu) | Od: ?" for every new mail — no error, just worse retrieval.
        server = gmail_server({1: eml("a@x", subject="Faktura 07/2026")})
        conn = FakeConn()
        await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                     conn, tmp_path, monkeypatch=monkeypatch)
        entities = json.loads(conn.envelopes["a@x"]["entities"])
        headers = next(e for e in entities if e["type"] == "headers")
        assert headers["subject"] == "Faktura 07/2026"
        assert headers["from"]["address"] == "a@example.com"

    async def test_attachment_manifest_is_appended_next_to_headers(self, tmp_path: Path, monkeypatch):
        raw = (
            b"Message-ID: <att@x>\r\nFrom: a@example.com\r\nSubject: s\r\n"
            b"Date: Mon, 15 Jun 2026 14:00:00 +0200\r\n"
            b'Content-Type: multipart/mixed; boundary="b"\r\n\r\n--b\r\n'
            b"Content-Type: text/plain\r\n\r\nhello\r\n--b\r\n"
            b"Content-Type: application/pdf\r\n"
            b'Content-Disposition: attachment; filename="doc.pdf"\r\n\r\n%PDF\r\n--b--\r\n'
        )
        conn = FakeConn()
        await do_run({"gmail": gmail_server({1: raw})}, [gmail_account(initial_mode="full")],
                     conn, tmp_path, monkeypatch=monkeypatch)
        types = [e["type"] for e in json.loads(conn.envelopes["att@x"]["entities"])]
        assert types == ["headers", "attachment"]

    async def test_never_marks_mail_as_read(self, tmp_path: Path, monkeypatch):
        # A plain SELECT + FETCH RFC822 would set \Seen on the operator's own inbox.
        server = gmail_server({1: eml("a@x")})
        await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                     FakeConn(), tmp_path, monkeypatch=monkeypatch)
        assert all(cmd[2] is True for cmd in server.commands if cmd[0] == "SELECT")
        assert all(cmd[-1] == "(BODY.PEEK[])" for cmd in server.fetch_commands())


class TestIncremental:
    async def test_second_run_fetches_only_what_arrived_since(self, tmp_path: Path, monkeypatch):
        server = gmail_server({1: eml("a@x")})
        conn = FakeConn()
        account = gmail_account(initial_mode="full")
        await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)

        server.folders[GMAIL_FOLDER][2] = eml("b@x")
        server.commands.clear()
        result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)

        assert [int(c[2]) for c in server.fetch_commands()] == [2]
        assert result["totals"]["envelopes_inserted"] == 1

    async def test_idle_mailbox_reports_zero_new(self, tmp_path: Path, monkeypatch):
        # The `n:*` range trailer means an unfiltered implementation would report one "new"
        # mail on every idle tick, forever.
        server = gmail_server({1: eml("a@x")})
        conn = FakeConn()
        account = gmail_account(initial_mode="full")
        await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
        server.commands.clear()

        result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
        assert result["totals"]["uids_seen"] == 0
        assert result["totals"]["envelopes_inserted"] == 0
        assert server.fetch_commands() == []

    async def test_rerun_of_the_same_uids_is_a_no_op_at_every_layer(self, tmp_path: Path, monkeypatch):
        # Force a re-fetch by rewinding the cursor: archive says FileExistsError, DB says
        # ON CONFLICT DO NOTHING, and nothing is double-counted as new.
        server = gmail_server({1: eml("a@x")})
        conn = FakeConn()
        account = gmail_account(initial_mode="full")
        await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
        conn.state[("gmail", GMAIL_FOLDER)] = (42, 0)

        result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
        totals = result["totals"]
        assert totals["envelopes_inserted"] == 0
        assert totals["envelopes_skipped_dup"] == 1
        assert totals["archive_exists"] == 1
        assert totals["archived"] == 0
        assert len(list(tmp_path.rglob("*.eml"))) == 1


class TestUidvalidityInvalidation:
    async def test_changed_uidvalidity_triggers_a_full_sweep(self, tmp_path: Path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
        conn = FakeConn()
        account = gmail_account(initial_mode="full")
        await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)

        server._uidvalidity = 99
        server.commands.clear()
        result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)

        assert result["totals"]["uidvalidity_resets"] == 1
        assert sorted(int(c[2]) for c in server.fetch_commands()) == [1, 2]

    async def test_resweep_inserts_nothing_new_thanks_to_message_id_dedup(self, tmp_path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
        conn = FakeConn()
        account = gmail_account(initial_mode="full")
        await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)

        server._uidvalidity = 99
        result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
        assert result["totals"]["envelopes_inserted"] == 0
        assert result["totals"]["envelopes_skipped_dup"] == 2

    async def test_new_uidvalidity_is_stored_with_the_recomputed_cursor(self, tmp_path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
        conn = FakeConn()
        account = gmail_account(initial_mode="full")
        await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)

        server._uidvalidity = 99
        await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
        assert conn.state[("gmail", GMAIL_FOLDER)] == (99, 2)

    async def test_renumbered_mailbox_with_lower_uids_is_not_skipped(self, tmp_path, monkeypatch):
        # The stored cursor (90) is meaningless under a new UIDVALIDITY; treating it as a
        # floor would skip every renumbered message below it.
        conn = FakeConn(state={("gmail", GMAIL_FOLDER): (42, 90)})
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")}, uidvalidity=99)
        result = await do_run({"gmail": server}, [gmail_account()], conn, tmp_path,
                              monkeypatch=monkeypatch)
        assert result["totals"]["envelopes_inserted"] == 2
        assert conn.state[("gmail", GMAIL_FOLDER)] == (99, 2)


class TestResumeAfterInterruption:
    async def test_cursor_stops_before_a_failed_message(self, tmp_path: Path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")},
                              fetch_errors={2})
        conn = FakeConn()
        result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                              conn, tmp_path, monkeypatch=monkeypatch)

        assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 1)
        assert result["totals"]["errors"] == 1
        assert result["failed"] is True

    async def test_messages_after_the_failure_are_still_stored(self, tmp_path: Path, monkeypatch):
        # Not skipped, not lost: they land in the archive and DB now, and the re-fetch that
        # the un-advanced cursor forces next tick is absorbed by dedup.
        server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")}, fetch_errors={2})
        conn = FakeConn()
        await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                     conn, tmp_path, monkeypatch=monkeypatch)
        assert set(conn.envelopes) == {"a@x", "c@x"}

    async def test_next_run_retries_from_the_stalled_uid(self, tmp_path: Path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")}, fetch_errors={2})
        conn = FakeConn()
        account = gmail_account(initial_mode="full")
        await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)

        server.fetch_errors.clear()
        server.commands.clear()
        result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)

        assert sorted(int(c[2]) for c in server.fetch_commands()) == [2, 3]
        assert result["totals"]["envelopes_inserted"] == 1   # only b@x was actually missing
        assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 3)

    async def test_a_failed_db_insert_leaves_the_cursor_behind(self, tmp_path: Path, monkeypatch):
        # The archive write already happened; the cursor must not move until the row exists.
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
        conn = FakeConn(insert_errors={"a@x"})
        result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                              conn, tmp_path, monkeypatch=monkeypatch)
        assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 0)
        assert result["totals"]["errors"] == 1

    async def test_limit_caps_a_tick_and_the_rest_follows_next_time(self, tmp_path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")})
        conn = FakeConn()
        account = gmail_account(initial_mode="full")
        await do_run({"gmail": server}, [account], conn, tmp_path, limit=2, monkeypatch=monkeypatch)
        assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2)

        await do_run({"gmail": server}, [account], conn, tmp_path, limit=2, monkeypatch=monkeypatch)
        assert set(conn.envelopes) == {"a@x", "b@x", "c@x"}


class TestFirstTickModes:
    async def test_new_only_records_the_high_water_mark_without_fetching(self, tmp_path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
        conn = FakeConn()
        result = await do_run({"gmail": server}, [gmail_account(initial_mode="new-only")],
                              conn, tmp_path, monkeypatch=monkeypatch)
        assert server.fetch_commands() == []
        assert conn.envelopes == {}
        assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2)
        assert not result["failed"]

    async def test_since_mode_issues_an_imap_date_search(self, tmp_path, monkeypatch):
        from datetime import date
        server = gmail_server({1: eml("a@x")})
        await do_run({"gmail": server},
                     [gmail_account(initial_mode="since", initial_since=date(2026, 6, 15))],
                     FakeConn(), tmp_path, monkeypatch=monkeypatch)
        search = [c for c in server.commands if c[:2] == ("UID", "SEARCH")][0]
        assert search[-2:] == ("SINCE", "15-Jun-2026")

    async def test_initial_mode_no_longer_applies_once_a_cursor_exists(self, tmp_path, monkeypatch):
        conn = FakeConn(state={("gmail", GMAIL_FOLDER): (42, 1)})
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
        await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                     conn, tmp_path, monkeypatch=monkeypatch)
        assert [int(c[2]) for c in server.fetch_commands()] == [2]

    async def test_empty_mailbox_first_tick_stores_a_zero_cursor(self, tmp_path, monkeypatch):
        server = gmail_server({})
        conn = FakeConn()
        await do_run({"gmail": server}, [gmail_account(initial_mode="new-only")],
                     conn, tmp_path, monkeypatch=monkeypatch)
        assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 0)


class TestCrossAccount:
    async def test_shared_message_id_counts_as_a_cross_source_conflict(self, tmp_path, monkeypatch):
        # Recon §2.4: one row, whichever account got there first — but the overlap must be
        # visible, or "new fastmail mails" reads low for no discoverable reason.
        servers = {
            "gmail": gmail_server({1: eml("shared@list")}),
            "fastmail": FakeIMAP({"INBOX": {1: eml("shared@list")}}),
        }
        conn = FakeConn()
        result = await do_run(servers,
                              [gmail_account(initial_mode="full"),
                               fastmail_account(initial_mode="full")],
                              conn, tmp_path, monkeypatch=monkeypatch)

        assert result["accounts"]["gmail"]["envelopes_inserted"] == 1
        assert result["accounts"]["fastmail"]["envelopes_inserted"] == 0
        assert result["accounts"]["fastmail"]["envelopes_conflict_other_source"] == 1
        assert conn.envelopes["shared@list"]["source"] == "gmail"

    async def test_both_accounts_keep_their_own_archive_copy(self, tmp_path, monkeypatch):
        servers = {
            "gmail": gmail_server({1: eml("shared@list")}),
            "fastmail": FakeIMAP({"INBOX": {1: eml("shared@list")}}),
        }
        await do_run(servers,
                     [gmail_account(initial_mode="full"), fastmail_account(initial_mode="full")],
                     FakeConn(), tmp_path, monkeypatch=monkeypatch)
        assert (tmp_path / "gmail" / "2026" / "06" / "shared@list.eml").exists()
        assert (tmp_path / "fastmail" / "2026" / "06" / "shared@list.eml").exists()

    async def test_a_broken_account_does_not_stop_the_other(self, tmp_path, monkeypatch):
        servers = {
            "gmail": gmail_server({1: eml("a@x")}, login_error="AUTHENTICATIONFAILED"),
            "fastmail": FakeIMAP({"INBOX": {1: eml("b@x")}}),
        }
        conn = FakeConn()
        result = await do_run(servers,
                              [gmail_account(initial_mode="full"),
                               fastmail_account(initial_mode="full")],
                              conn, tmp_path, monkeypatch=monkeypatch)

        assert set(conn.envelopes) == {"b@x"}
        assert result["accounts"]["gmail"]["account_errors"] == 1
        assert result["failed"] is True

    async def test_fastmail_syncs_its_three_configured_folders(self, tmp_path, monkeypatch):
        server = FakeIMAP({"INBOX": {1: eml("i@x")}, "Archive": {1: eml("ar@x")},
                           "Sent": {1: eml("s@x")}})
        conn = FakeConn()
        account = fastmail_account(folders=("INBOX", "Archive", "Sent"), initial_mode="full")
        await do_run({"fastmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
        assert set(conn.envelopes) == {"i@x", "ar@x", "s@x"}
        assert len(conn.state) == 3


class TestVanishedAndErrors:
    async def test_expunged_uid_is_not_an_error_and_does_not_block_the_cursor(self, tmp_path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")}, vanished={1})
        conn = FakeConn()
        result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                              conn, tmp_path, monkeypatch=monkeypatch)
        assert result["totals"]["vanished"] == 1
        assert result["totals"]["errors"] == 0
        assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2)
        assert not result["failed"]

    async def test_missing_special_use_folder_fails_the_account_loudly(self, tmp_path, monkeypatch):
        # Rather than "synced 0 mails, all good" every hour.
        server = FakeIMAP({"INBOX": {}})
        result = await do_run({"gmail": server}, [gmail_account()], FakeConn(), tmp_path,
                              monkeypatch=monkeypatch)
        assert result["failed"] is True
        assert result["accounts"]["gmail"]["account_errors"] == 1


class TestBalance:
    async def test_counters_balance_on_a_mixed_run(self, tmp_path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x"), 4: eml("d@x")},
                              fetch_errors={3}, vanished={2})
        result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                              FakeConn(), tmp_path, monkeypatch=monkeypatch)
        assert balance_errors(result["totals"]) == []
        assert result["balance_problems"] == []

    def test_balance_errors_names_each_broken_identity(self):
        broken = {"uids_seen": 5, "processed": 1, "vanished": 0, "errors": 0,
                  "archived": 0, "archive_exists": 0, "envelopes_inserted": 0,
                  "envelopes_skipped_dup": 0, "envelopes_conflict_other_source": 0}
        assert len(balance_errors(broken)) == 3

    def test_total_errors_sums_all_three_levels(self):
        assert total_errors({"errors": 1, "folder_errors": 2, "account_errors": 3}) == 6


class TestDryRun:
    async def test_dry_run_fetches_nothing_and_writes_nothing(self, tmp_path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
        conn = FakeConn()
        result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                              conn, tmp_path, apply=False, monkeypatch=monkeypatch)

        assert server.fetch_commands() == []
        assert conn.envelopes == {}
        assert conn.state == {}
        assert list(tmp_path.rglob("*.eml")) == []
        assert not result["failed"]

    async def test_dry_run_reports_how_much_the_real_run_would_pull(self, tmp_path, monkeypatch):
        server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
        result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                              FakeConn(), tmp_path, apply=False, monkeypatch=monkeypatch)
        folder = result["reports"][0]["folders"][0]
        assert folder["candidates"] == 2
        assert folder["mode"] == "initial-full"

    async def test_dry_run_is_not_flagged_unbalanced(self, tmp_path, monkeypatch):
        # It never fetched, so there are no message-level outcomes to balance.
        server = gmail_server({1: eml("a@x")})
        result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
                              FakeConn(), tmp_path, apply=False, monkeypatch=monkeypatch)
        assert result["balance_problems"] == []


class TestMeasure:
    async def test_reports_message_counts_per_folder(self):
        server = FakeIMAP({"INBOX": {1: eml("a@x"), 2: eml("b@x")}, "Archive": {1: eml("c@x")}})
        rows = await measure([fastmail_account(folders=("INBOX", "Archive"))],
                             client_factory=factory_for({"fastmail": server}))
        assert [(r["folder"], r["messages"]) for r in rows] == [("INBOX", 2), ("Archive", 1)]

    async def test_touches_neither_archive_nor_mail_flags(self):
        server = FakeIMAP({"INBOX": {1: eml("a@x")}})
        await measure([fastmail_account()], client_factory=factory_for({"fastmail": server}))
        assert server.fetch_commands() == []

    async def test_a_failing_account_is_reported_not_raised(self):
        server = FakeIMAP({"INBOX": {}}, login_error="nope")
        rows = await measure([fastmail_account()], client_factory=factory_for({"fastmail": server}))
        assert "error" in rows[0]


class TestMetrics:
    def _result(self, *, failed=False, inserted=3):
        return {
            "totals": {},
            "accounts": {
                "gmail": {"envelopes_inserted": inserted, "envelopes_skipped_dup": 1,
                          "envelopes_conflict_other_source": 0, "uidvalidity_resets": 0,
                          "errors": 0, "folder_errors": 0, "account_errors": 0},
                "fastmail": {"envelopes_inserted": 0, "envelopes_skipped_dup": 0,
                             "envelopes_conflict_other_source": 2, "uidvalidity_resets": 1,
                             "errors": 0, "folder_errors": 0, "account_errors": 0},
            },
            "last_message_ts": {"gmail": 1786000000.0},
            "failed": failed,
        }

    def test_counters_are_labelled_per_account(self, tmp_path: Path):
        text = render_prom(build_metrics(self._result(), 1786000123.0,
                                         tmp_path / "kb-mail-sync.prom"))
        assert 'kb_mail_sync_envelopes_inserted{account="gmail"} 3' in text
        assert 'kb_mail_sync_conflict_other_source{account="fastmail"} 2' in text
        assert 'kb_mail_sync_uidvalidity_resets{account="fastmail"} 1' in text

    def test_success_timestamp_is_set_on_a_clean_run(self, tmp_path: Path):
        text = render_prom(build_metrics(self._result(), 1786000123.0, tmp_path / "x.prom"))
        assert "kb_mail_sync_last_success_timestamp 1786000123" in text
        assert "kb_mail_sync_last_exit_code 0" in text

    def test_success_timestamp_carries_forward_across_a_failure(self, tmp_path: Path):
        path = tmp_path / "kb-mail-sync.prom"
        path.write_text(render_prom(build_metrics(self._result(), 1786000000.0, path)))
        text = render_prom(build_metrics(self._result(failed=True), 1786009999.0, path))
        # The staleness alert measures time since the last GOOD tick — a transient failure
        # must not reset its clock, and must not pretend this run succeeded either.
        assert "kb_mail_sync_last_success_timestamp 1786000000" in text
        assert "kb_mail_sync_last_run_timestamp 1786009999" in text
        assert "kb_mail_sync_last_exit_code 1" in text

    def test_first_ever_run_failing_leaves_the_clock_at_zero(self, tmp_path: Path):
        text = render_prom(build_metrics(self._result(failed=True), 1786000123.0,
                                          tmp_path / "absent.prom"))
        assert "kb_mail_sync_last_success_timestamp 0" in text

    def test_last_message_ts_is_omitted_for_an_account_with_no_rows(self, tmp_path: Path):
        text = render_prom(build_metrics(self._result(), 1786000123.0, tmp_path / "x.prom"))
        assert 'kb_mail_sync_last_message_ts{account="gmail"}' in text
        assert 'kb_mail_sync_last_message_ts{account="fastmail"}' not in text

    def test_read_prev_value_ignores_labelled_series(self, tmp_path: Path):
        path = tmp_path / "x.prom"
        path.write_text('kb_mail_sync_envelopes_inserted{account="gmail"} 3\n'
                        "kb_mail_sync_last_success_timestamp 1786000000\n")
        assert read_prev_value(path, "kb_mail_sync_last_success_timestamp") == 1786000000.0
        assert read_prev_value(path, "kb_mail_sync_envelopes_inserted") is None