PacketStation/tests/server/test_integrity.py

"""The integrity job: a clean run, then a tampered seal, a changed file, a missing file, an orphan file and an
edited audit row (through a superuser connection with the append-only trigger disabled)."""

from __future__ import annotations

import hashlib
import os
import time
import uuid
from collections.abc import Iterator
from datetime import datetime
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo

import psycopg
import pytest
from closehelpers import filed
from fakedir import FakeDirectory
from pshelpers import PG_URL, app_conn, needs_pg
from psycopg.conninfo import conninfo_to_dict, make_conninfo
from shophelpers import Shop, open_shop

from packetstation.domain import assets
from packetstation.settings import Settings
from packetstation.worker import jobs, runs
from packetstation.worker.jobs import integrity
from packetstation.worker.scheduler import Scheduler


def test_full_on_sundays_site_time() -> None:
    tz = ZoneInfo("America/Denver")
    assert integrity.full_day(datetime(2026, 10, 25, 4, tzinfo=tz)) is True  # a Sunday
    assert integrity.full_day(datetime(2026, 10, 21, 4, tzinfo=tz)) is False


@pytest.fixture
def shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]:
    yield from open_shop(pg_settings, fake_dir)


class Events:
    def __init__(self) -> None:
        self.calls: list[tuple[str, str, dict[str, Any]]] = []

    def raise_event(self, conn: Any, event_type: str, **kw: Any) -> int:
        self.calls.append(("raise", event_type, kw))
        return 1

    def resolve_event(self, conn: Any, event_type: str, **kw: Any) -> None:
        self.calls.append(("resolve", event_type, kw))


def superuser(settings: Settings) -> psycopg.Connection[Any]:
    assert PG_URL
    info = conninfo_to_dict(PG_URL)
    info["dbname"] = settings.pg_database
    return psycopg.connect(make_conninfo(**info), autocommit=True)


def run_integrity(settings: Settings) -> dict[str, Any]:
    sch = Scheduler(settings, build="pytest", jobs={"integrity": integrity.SPEC})
    return runs.run_job(sch, "integrity", "schedule")


@needs_pg
def test_integrity(shop: Shop, monkeypatch: pytest.MonkeyPatch) -> None:
    s = shop.settings
    ev = Events()
    monkeypatch.setattr(jobs, "raise_event", ev.raise_event)
    monkeypatch.setattr(jobs, "resolve_event", ev.resolve_event)
    monkeypatch.setattr(integrity, "full_day", lambda now: False)
    pkt = filed(shop)
    pid = pkt["id"]

    row = run_integrity(s)
    assert row["status"] == "ok", row["errors"]
    c = row["counts"]
    assert c["seals"] == 1 and c["total"] == 0 and c["assets"] >= 1 and c["hashed"] == c["assets"]
    assert c["audit_rows"] > 0 and c["full"] is False
    assert ev.calls == [("resolve", "integrity_failure", {"subject": "integrity", "title": "Integrity check failure"})]

    with app_conn(s) as conn:
        assert (
            conn.execute("SELECT count(*) FROM assets WHERE hash_ok AND hash_checked_at IS NOT NULL").fetchone()[0]
            == c["assets"]
        )
        db_id, db_key = conn.execute(
            "SELECT a.id, a.storage_key FROM packet_seals s JOIN assets a ON a.id = s.databook_asset_id"
            " WHERE s.packet_id = %s",
            (pid,),
        ).fetchone()
        # An old asset whose file changed: only the Sunday run rehashes it
        old_id = uuid.uuid4()
        old_key = assets.storage_key(old_id)
        old_path = assets.path_of(s, old_key)
        old_path.parent.mkdir(parents=True, exist_ok=True)
        old_path.write_bytes(b"changed later")
        conn.execute(
            "INSERT INTO assets (id, sha256, size, mime, storage_key, uploaded_at)"
            " VALUES (%s, %s, 5, 'application/pdf', %s, now() - interval '30 days')",
            (old_id, hashlib.sha256(b"first").digest(), old_key),
        )
        # An asset row with no file
        gone_id = uuid.uuid4()
        conn.execute(
            "INSERT INTO assets (id, sha256, size, mime, storage_key) VALUES (%s, %s, 1, 'application/pdf', %s)",
            (gone_id, hashlib.sha256(b"x").digest(), assets.storage_key(gone_id)),
        )
        # Tamper with the filed packet behind the app's back
        conn.execute("UPDATE packets SET title = title || ' (edited)' WHERE id = %s", (pid,))

    # The data book's bytes change on disk
    with assets.path_of(s, str(db_key)).open("ab") as f:
        f.write(b"\n% tampered\n")
    # A file nobody's row names, old enough to count; and a fresh one, skipped as an upload in flight
    orphan = Path(s.assets_dir) / "zz" / "orphan.bin"
    orphan.parent.mkdir(parents=True, exist_ok=True)
    orphan.write_bytes(b"orphan")
    old = time.time() - 2 * 3600
    os.utime(orphan, (old, old))
    (Path(s.assets_dir) / "zz" / "fresh.bin").write_bytes(b"in flight")

    # An audit row edited through a superuser connection with the append-only trigger off
    with superuser(s) as su:
        aid = su.execute("SELECT id FROM audit_events ORDER BY id LIMIT 1 OFFSET 2").fetchone()[0]
        su.execute("ALTER TABLE audit_events DISABLE TRIGGER USER")
        try:
            su.execute("UPDATE audit_events SET detail = '{\"edited\": true}' WHERE id = %s", (aid,))
        finally:
            su.execute("ALTER TABLE audit_events ENABLE TRIGGER USER")

    ev.calls.clear()
    row = run_integrity(s)
    assert row["status"] == "ok", row["errors"]
    f = row["counts"]["findings"]
    assert f == {"seal": 1, "asset_hash": 1, "asset_missing": 1, "blob_orphan": 1, "audit_chain": 1}
    assert orphan.exists()  # reported, never deleted
    ((kind, etype, kw),) = ev.calls
    assert (kind, etype) == ("raise", "integrity_failure") and kw["condition"] is True and kw["subject"] == "integrity"
    assert kw["title"].startswith("Integrity check found 5 problems") and "file changed" in kw["title"]

    with app_conn(s) as conn:
        found = {
            r[0]: (r[1], r[2])
            for r in conn.execute(
                "SELECT kind, subject, detail FROM integrity_findings WHERE run_id = %s", (row["id"],)
            ).fetchall()
        }
        assert found["seal"][0] == f"packet:{pid}"
        assert found["asset_hash"][0] == str(db_id)
        assert found["asset_missing"][0] == str(gone_id)
        assert found["blob_orphan"][0] == "zz/orphan.bin"
        assert found["audit_chain"][0] == str(aid)
        oks = dict(conn.execute("SELECT id, hash_ok FROM assets").fetchall())
        assert oks[db_id] is False and oks[gone_id] is False and oks[old_id] is None  # not rehashed on a weekday

    # Sunday: every file is rehashed, so the old one counts too
    monkeypatch.setattr(integrity, "full_day", lambda now: True)
    row = run_integrity(s)
    assert row["counts"]["full"] is True and row["counts"]["findings"]["asset_hash"] == 2
    with app_conn(s) as conn:
        assert conn.execute("SELECT hash_ok FROM assets WHERE id = %s", (old_id,)).fetchone()[0] is False
        assert conn.execute("SELECT count(*) FROM integrity_findings").fetchone()[0] == 11