email-connect/tests/test_sender_transition.py

"""Protocol fault injection uses synthetic Kubernetes state and real SQLite only."""

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

import pytest

from email_connect.transactional import ProviderError, SQLiteDeliveryStore

SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
sys.path.insert(0, str(SCRIPTS))
import sender_transition as transition

spec = importlib.util.spec_from_file_location("sender_observer_fixture", SCRIPTS / "inspect-sender-ledger.py")
observer = importlib.util.module_from_spec(spec)
spec.loader.exec_module(observer)


def write(path, value):
    with path.open("w") as stream:
        json.dump(value, stream)
        stream.flush()
        os.fsync(stream.fileno())


def initial():
    return {
        "identity": {"deployment": "fixture-deployment", "config": "fixture-config"},
        "protected": {"policy": "fixture-policy", "pvc": "fixture-pvc"},
        "config_data": {"EMAIL_CONNECT_SENDER": "forgejo@coulomb.social", "other": "preserve"},
        "deployment_spec": {
            "replicas": 1, "strategy": {"type": "RollingUpdate"},
            "template": {"spec": {"containers": [{
                "name": "email-connect", "image": transition.IMAGE,
                "envFrom": [{"configMapRef": {"name": "email-connect-config"}},
                            {"secretRef": {"name": "email-connect-runtime", "optional": False}}],
                "env": [{"name": "EMAIL_CONNECT_CALLERS_FILE",
                         "value": "/etc/email-connect-callers/callers.json"}],
                "resources": {"requests": {"cpu": "5m", "memory": "64Mi"}},
            }]}},
        },
    }


class FixtureAdapter:
    """File-backed API effects persist across real child-process termination."""

    def __init__(self, root, lost_response=None):
        self.root = Path(root)
        self.lost_response = lost_response

    def observe(self):
        return json.loads((self.root / "cluster.json").read_text())

    def admission(self, proposal):
        if (self.root / "deny-admission").exists():
            raise transition.Refused("fixture_admission_absent")

    def patch(self, key, expected, desired):
        value = self.observe()
        assert value[key] == expected
        value[key] = desired
        write(self.root / "cluster.json", value)
        if self.lost_response == key:
            os._exit(73)

    def patch_deployment(self, expected, desired):
        self.patch("deployment_spec", expected, desired)

    def patch_config(self, expected, desired):
        self.patch("config_data", expected, desired)

    def fence(self, journal):
        assert self.observe()["deployment_spec"]["replicas"] == 0

    def snapshot(self, journal, *, quiesced):
        if quiesced:
            assert self.observe()["deployment_spec"]["replicas"] == 0
        path = self.root / f"snapshot-{len(journal.entries)}.sqlite3"
        assert not path.exists()
        source = sqlite3.connect((self.root / "ledger.sqlite3").as_uri() + "?mode=ro", uri=True)
        target = sqlite3.connect(path)
        source.backup(target)
        source.close()
        target.close()
        path.chmod(0o600)
        return path, observer.inspect(path)

    def verify(self, expected, journal):
        assert self.observe() == expected
        return {"fixture_only": True}

    def cleanup(self, journal):
        pass


@pytest.fixture
def attempt(tmp_path):
    root = tmp_path / "private"
    root.mkdir(mode=0o700)
    before = initial()
    write(root / "cluster.json", before)
    store = SQLiteDeliveryStore(str(root / "ledger.sqlite3"))
    store.db.close()
    proposal = {"before": before, "after": transition.desired(before)}
    write(root / "proposal.json", proposal)
    return root, proposal


def execute(root, proposal):
    journal = transition.Journal(root / "journal.jsonl", proposal=proposal)
    try:
        result = transition.run(FixtureAdapter(root), journal)
        assert journal.entries[-1]["phase"] == "completed"
        return result
    finally:
        journal.close()


def recover(root):
    journal = transition.Journal(root / "journal.jsonl")
    try:
        result = transition.run(FixtureAdapter(root), journal, recover=True)
        assert journal.entries[-1]["phase"] == "recovered"
        return result
    finally:
        journal.close()


def test_only_intended_fields_change_and_replay_refuses(attempt):
    root, proposal = attempt
    result = execute(root, proposal)
    assert result["all_six_tables_checked"] and not result["ledger_restored"]
    assert FixtureAdapter(root).observe() == proposal["after"]
    with pytest.raises(FileExistsError):
        execute(root, proposal)
    journal = transition.Journal(root / "journal.jsonl")
    try:
        with pytest.raises(transition.Refused, match="forward_replay"):
            transition.run(FixtureAdapter(root), journal)
    finally:
        journal.close()


@pytest.mark.parametrize("phase", [
    "forward_pause_intent", "forward_capture_intent", "forward_config_intent",
    "forward_deployment_intent", "forward_resume_intent", "forward_postflight_intent",
    "forward_cleanup_intent", "completed", "lost_config_response", "lost_scale_response",
])
def test_real_process_exit_can_restore_configuration_without_restoring_ledger(attempt, phase):
    root, proposal = attempt
    child = '''
import json,os,sys
from pathlib import Path
from test_sender_transition import transition,FixtureAdapter
root=Path(sys.argv[1]);phase=sys.argv[2]
proposal=json.loads((root/'proposal.json').read_text())
original=transition.Journal.record
def checkpoint(self, name, **data):
 original(self,name,**data)
 if name==phase:os._exit(73)
transition.Journal.record=checkpoint
journal=transition.Journal(root/'journal.jsonl',proposal=proposal)
lost={'lost_config_response':'config_data','lost_scale_response':'deployment_spec'}.get(phase)
transition.run(FixtureAdapter(root,lost),journal)
'''
    result = subprocess.run([sys.executable, "-c", child, str(root), phase],
                            env={**os.environ, "PYTHONPATH": os.pathsep.join([
                                str(SCRIPTS.parent / "src"), str(Path(__file__).parent)])},
                            capture_output=True, timeout=30, check=False)
    assert result.returncode == 73, result.stderr.decode()
    # Writes which occurred after the original attempt must survive recovery.
    store = SQLiteDeliveryStore(str(root / "ledger.sqlite3"))
    store.reserve_send("later-accepted", "fp")
    store.record("later-accepted", "provider-receipt")
    store.reserve_send("later-unknown", "fp2")
    store.db.close()
    baseline = (root / "ledger.sqlite3").read_bytes()
    old_journal = (root / "journal.jsonl").read_bytes()
    result = recover(root)
    assert FixtureAdapter(root).observe() == proposal["before"]
    assert (root / "ledger.sqlite3").read_bytes() == baseline
    assert (root / "journal.jsonl").read_bytes().startswith(old_journal)
    assert not result["ledger_restored"]
    store = SQLiteDeliveryStore(str(root / "ledger.sqlite3"))
    with pytest.raises(ProviderError, match="delivery_outcome_unknown"):
        store.reserve_send("later-unknown", "fp2")
    assert store.reserve_send("later-accepted", "fp") is False
    store.db.close()


@pytest.mark.parametrize("phase", ["recovery_config_intent", "recovery_resume_intent"])
def test_second_recovery_interruption_is_restartable(attempt, monkeypatch, phase):
    root, proposal = attempt
    execute(root, proposal)
    original = transition.Journal.record

    def stop(self, name, **data):
        original(self, name, **data)
        if name == phase:
            raise RuntimeError("second interruption")

    monkeypatch.setattr(transition.Journal, "record", stop)
    with pytest.raises(RuntimeError):
        recover(root)
    monkeypatch.setattr(transition.Journal, "record", original)
    recover(root)
    assert FixtureAdapter(root).observe() == proposal["before"]


def test_unknown_before_switch_holds_old_receiver_and_recovery_preserves_unknown(attempt):
    root, proposal = attempt
    store = SQLiteDeliveryStore(str(root / "ledger.sqlite3"))
    store.reserve_send("unknown", "fp")
    store.db.close()
    with pytest.raises(transition.Refused, match="unknown_outcome"):
        execute(root, proposal)
    value = FixtureAdapter(root).observe()
    assert value["config_data"] == proposal["before"]["config_data"]
    assert value["deployment_spec"]["replicas"] == 0
    recover(root)
    assert observer.inspect(root / "ledger.sqlite3")["needs_outcome_reconciliation"]


@pytest.mark.parametrize("drift", ["image", "protected", "sender"])
def test_foreign_drift_refuses_recovery_without_overwrite(attempt, drift):
    root, proposal = attempt
    execute(root, proposal)
    value = FixtureAdapter(root).observe()
    if drift == "image":
        value["deployment_spec"]["template"]["spec"]["containers"][0]["image"] = "foreign"
    elif drift == "protected":
        value["protected"]["pvc"] = "replaced-pvc"
    else:
        value["config_data"]["EMAIL_CONNECT_SENDER"] = "someone@example.test"
    write(root / "cluster.json", value)
    with pytest.raises(transition.Refused):
        recover(root)
    assert FixtureAdapter(root).observe() == value


def test_absent_admission_does_not_pause_or_patch(attempt):
    root, proposal = attempt
    (root / "deny-admission").touch()
    with pytest.raises(transition.Refused, match="admission_absent"):
        execute(root, proposal)
    assert FixtureAdapter(root).observe() == proposal["before"]


def test_partial_journal_is_held_and_private_modes_enforced(attempt):
    root, proposal = attempt
    execute(root, proposal)
    path = root / "journal.jsonl"
    assert path.stat().st_mode & 0o777 == 0o600
    with path.open("ab") as stream:
        stream.write(b'{"partial":')
    with pytest.raises(transition.Refused, match="incomplete_journal"):
        transition.Journal(path)
    root.chmod(0o755)
    with pytest.raises(transition.Refused, match="private_journal_parent"):
        transition.Journal(path)


def test_continuity_permits_terminal_consumption_and_new_writes_but_not_rewrites(tmp_path):
    path = tmp_path / "before.sqlite3"
    store = SQLiteDeliveryStore(str(path))
    payload = {"tenant": "fixture", "client_id": "client", "registration_id": "registration",
               "normalized_email": "one@example.test", "preferred_username": "one"}
    request, _ = store.reserve_verification(payload, "handle", "2030-01-01T00:00:00+00:00")
    store.verification_accepted(request, "receipt")
    store.request_budget("caller", rate=60, burst=10, now=100)
    store.db.close()
    path.chmod(0o600)
    after = tmp_path / "after.sqlite3"
    after.write_bytes(path.read_bytes())
    after.chmod(0o600)
    store = SQLiteDeliveryStore(str(after))
    store.consume_verification("handle")
    store.request_budget("caller", rate=60, burst=10, now=102)
    store.reserve_send("new", "fp")
    store.record("new", "new-receipt")
    result = transition.continuity(path, after)
    assert result["tables"]["deliveries"]["new_rows"] == 1
    assert result["tables"]["verifications"]["permitted_updates"] == 1
    with store.db:
        store.db.execute("UPDATE verifications SET email='different@example.test'")
    with pytest.raises(transition.Refused, match="history_changed"):
        transition.continuity(path, after)
    store.db.close()


@pytest.mark.parametrize("change", ["delete", "schema", "unconsume"])
def test_continuity_refuses_lost_rows_schema_drift_or_reversed_consumption(tmp_path, change):
    before = tmp_path / "before.sqlite3"
    store = SQLiteDeliveryStore(str(before))
    store.record("event", "provider")
    payload = {"tenant": "fixture", "client_id": "client", "registration_id": "registration",
               "normalized_email": "one@example.test", "preferred_username": "one"}
    store.create_verification(payload, "handle", "2030-01-01T00:00:00+00:00")
    store.consume_verification("handle")
    store.db.close()
    before.chmod(0o600)
    after = tmp_path / "after.sqlite3"
    after.write_bytes(before.read_bytes())
    after.chmod(0o600)
    with sqlite3.connect(after) as db:
        db.execute({"delete": "DELETE FROM deliveries",
                    "schema": "CREATE TABLE foreign_table (id TEXT)",
                    "unconsume": "UPDATE verifications SET consumed_at=NULL"}[change])
    with pytest.raises(transition.Refused):
        transition.continuity(before, after)


def test_proposal_cannot_widen_the_configuration_change(attempt):
    root, proposal = attempt
    proposal = copy.deepcopy(proposal)
    proposal["after"]["deployment_spec"]["template"]["spec"]["containers"][0]["image"] = "other"
    with pytest.raises(transition.Refused, match="unreviewed_configuration_delta"):
        execute(root, proposal)
    assert FixtureAdapter(root).observe() == proposal["before"]


def test_writer_resumed_during_capture_refuses_configuration_change(attempt):
    root, proposal = attempt

    class Interference(FixtureAdapter):
        def snapshot(self, journal, *, quiesced):
            result = super().snapshot(journal, quiesced=quiesced)
            value = self.observe()
            value["deployment_spec"]["replicas"] = 1
            write(root / "cluster.json", value)
            return result

    journal = transition.Journal(root / "journal.jsonl", proposal=proposal)
    try:
        with pytest.raises(transition.Refused, match="resumed_before_config"):
            transition.run(Interference(root), journal)
    finally:
        journal.close()
    assert FixtureAdapter(root).observe() == proposal["before"]


def test_broken_capture_stops_before_switching_configuration(attempt):
    root, proposal = attempt

    class BrokenCapture(FixtureAdapter):
        def snapshot(self, journal, *, quiesced):
            path, outcomes = super().snapshot(journal, quiesced=quiesced)
            path.write_bytes(b"not a SQLite snapshot")
            return path, outcomes

    journal = transition.Journal(root / "journal.jsonl", proposal=proposal)
    try:
        with pytest.raises(sqlite3.DatabaseError):
            transition.run(BrokenCapture(root), journal)
        assert journal.entries[-1]["phase"] == "requires_inspection"
    finally:
        journal.close()
    expected = copy.deepcopy(proposal["before"])
    expected["deployment_spec"]["replicas"] = 0
    assert FixtureAdapter(root).observe() == expected


def test_nullable_sqlite_primary_keys_cannot_hide_rows_during_comparison(tmp_path):
    path = tmp_path / "ambiguous.sqlite3"
    store = SQLiteDeliveryStore(str(path))
    with store.db:
        store.db.execute("INSERT INTO deliveries VALUES (NULL, 'first')")
        store.db.execute("INSERT INTO deliveries VALUES (NULL, 'second')")
    store.db.close()
    path.chmod(0o600)
    with pytest.raises(transition.Refused, match="ambiguous_snapshot_row_identity"):
        transition.continuity(path, path)