email-connect/tests/test_sender_ledger.py

"""Synthetic ledger observations; never connects to an SMTP service."""

import importlib.util
import json
import sqlite3
import subprocess
import sys
from pathlib import Path

import pytest
from test_ledger_backup import synthetic_store

from email_connect.transactional import ProviderError, SQLiteDeliveryStore

SCRIPT = Path(__file__).resolve().parents[1] / "scripts/inspect-sender-ledger.py"
spec = importlib.util.spec_from_file_location("inspect_sender_ledger", SCRIPT)
observer = importlib.util.module_from_spec(spec)
spec.loader.exec_module(observer)


def test_matches_archive_fingerprint_and_preserves_every_row(tmp_path, monkeypatch):
    monkeypatch.syspath_prepend(str(SCRIPT.parent))
    from ledger_archive import fingerprint

    path = tmp_path / "ledger.sqlite3"
    synthetic_store(path)
    before = fingerprint(path)
    result = observer.inspect(path)
    assert result["ledger_fingerprint"] == before == fingerprint(path)
    assert not result["mail_sent"] and not result["private_rows_printed"]
    assert not result["writers_fenced"] and not result["rollout_authorized"]
    assert result["consistent_read_transaction"]
    assert "example.test" not in json.dumps(result)


def test_accepted_invitation_in_flight_row_is_not_unknown(tmp_path):
    path = tmp_path / "ledger.sqlite3"
    store = SQLiteDeliveryStore(str(path))
    assert store.reserve_send("accepted-event", "fingerprint")
    store.record("accepted-event", "synthetic-provider-reference")
    assert store.db.execute("SELECT state FROM send_attempts").fetchone()[0] == "in_flight"
    result = observer.inspect(path)
    assert result["outcomes"]["accepted_invitations"] == 1
    assert result["outcomes"]["unknown_invitations"] == 0
    assert not result["needs_outcome_reconciliation"]
    assert store.reserve_send("accepted-event", "fingerprint") is False
    store.db.close()


def test_unknown_invitation_survives_inspection_and_restart(tmp_path):
    path = tmp_path / "ledger.sqlite3"
    store = SQLiteDeliveryStore(str(path))
    store.reserve_send("uncertain-event", "fingerprint")
    store.db.close()
    result = observer.inspect(path)
    assert result["outcomes"]["unknown_invitations"] == 1
    assert result["needs_outcome_reconciliation"]
    store = SQLiteDeliveryStore(str(path))
    with pytest.raises(ProviderError, match="delivery_outcome_unknown"):
        store.reserve_send("uncertain-event", "fingerprint")
    assert observer.inspect(path)["ledger_fingerprint"] == result["ledger_fingerprint"]
    store.db.close()


@pytest.mark.parametrize("state", ["in_flight", "accepted", "failed"])
def test_verification_outcomes_and_unlinked_history_are_distinct(tmp_path, state):
    path = tmp_path / "ledger.sqlite3"
    store = SQLiteDeliveryStore(str(path))
    payload = {"tenant": "fixture", "client_id": "client", "registration_id": "one",
               "normalized_email": "one@example.test", "preferred_username": "one"}
    request, _ = store.reserve_verification(payload, "synthetic-hash", "2030-01-01T00:00:00Z")
    if state == "accepted":
        store.verification_accepted(request, "synthetic-reference")
    elif state == "failed":
        store.verification_failed(request, ProviderError("definitive", retryable=False))
    payload["registration_id"] = "historical"
    store.create_verification(payload, "other-hash", "2020-01-01T00:00:00Z")
    result = observer.inspect(path)
    label = {"in_flight": "unknown", "accepted": "accepted", "failed": "failed"}[state]
    assert result["outcomes"][label + "_verifications"] == 1
    assert result["outcomes"]["unlinked_verifications"] == 1
    assert result["needs_outcome_reconciliation"] == (state == "in_flight")
    store.db.close()


def test_concurrent_wal_commit_cannot_split_the_observation(tmp_path, monkeypatch):
    path = tmp_path / "ledger.sqlite3"
    store = SQLiteDeliveryStore(str(path))
    store.db.execute("PRAGMA journal_mode=WAL")
    store.reserve_send("accepted", "fp")
    store.record("accepted", "reference")
    real_connect = sqlite3.connect
    changed = []

    class ObservedConnection(sqlite3.Connection):
        def execute(self, statement, *args, **kwargs):
            if statement == "SELECT COUNT(*) FROM deliveries" and not changed:
                changed.append(True)
                store.reserve_send("later-uncertain", "fp2")
            return super().execute(statement, *args, **kwargs)

    monkeypatch.setattr(observer.sqlite3, "connect", lambda *a, **kw:
                        real_connect(*a, **kw, factory=ObservedConnection))
    result = observer.inspect(path)
    assert changed
    assert result["ledger_fingerprint"]["tables"]["send_attempts"]["rows"] == 1
    assert result["outcomes"]["unknown_invitations"] == 0
    assert store.db.execute("SELECT COUNT(*) FROM send_attempts").fetchone()[0] == 2
    store.db.close()


@pytest.mark.parametrize("invalid", ["missing", "symlink", "schema", "size"])
def test_invalid_inputs_are_refused_without_writing(tmp_path, monkeypatch, invalid):
    path = tmp_path / "ledger.sqlite3"
    if invalid != "missing":
        store = SQLiteDeliveryStore(str(path))
        store.db.close()
    if invalid == "symlink":
        target = tmp_path / "target.sqlite3"
        path.rename(target)
        path.symlink_to(target)
    elif invalid == "schema":
        with sqlite3.connect(path) as db:
            db.execute("CREATE TABLE unexpected (value TEXT)")
    elif invalid == "size":
        monkeypatch.setattr(observer, "MAX_BYTES", 1)
    before = path.read_bytes() if path.exists() else None
    with pytest.raises((observer.Refused, OSError)):
        observer.inspect(path)
    assert (path.read_bytes() if path.exists() else None) == before


def test_cli_errors_do_not_echo_private_paths(tmp_path):
    result = subprocess.run([sys.executable, str(SCRIPT), "--database",
                             str(tmp_path / "private-recipient@example.test")],
                            capture_output=True, text=True, timeout=10, check=False)
    assert result.returncode != 0 and not result.stdout
    assert result.stderr.strip() == "ledger_observation_refused"