`tests/test_cycle8_suppression.py`

85 lines
"""Cycle 7 test — state suppression / reinstatement of a record from public search.

Run:  python -m tests.test_cycle8_suppression

Proves the brief's near-real-time suppression requirement:
  1. a state can remove a record from public search immediately (is_suppressed flip);
  2. the underlying data + version history are RETAINED (nothing deleted);
  3. every suppress/reinstate writes an immutable audit_event;
  4. reinstate returns it to search;
  5. suppressing an unknown record returns False (no silent no-op success).
"""
from __future__ import annotations

import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from services.common.sqlite_repo import FileObjectStore, SqliteRepository
from services.ingestion.ingest import FeedDefinition, ingest_authorized_feed

SAMPLE = Path(__file__).resolve().parents[1] / "data" / "sample" / "sample_state_feed.csv"


def ok(msg: str) -> None:
    print(f"  ✓ {msg}")


def _refs(repo, q):
    return {h["public_reference"] for h in repo.masked_search(q)}


def main() -> int:
    tmp = Path(tempfile.mkdtemp(prefix="upp-cycle8-"))
    store = FileObjectStore(tmp / "obj")
    (tmp / "obj").mkdir(parents=True, exist_ok=True)
    store.write_bytes("incoming/s.csv", SAMPLE.read_bytes())
    repo = SqliteRepository(str(tmp / "p.db"))
    ingest_authorized_feed(FeedDefinition("SAMPLE", "incoming/s.csv"), store, repo)

    pid = repo._property_id("SAMPLE", "SP-0001")
    assert pid, "SP-0001 should exist"

    print("1) record is publicly searchable before suppression")
    assert pid in _refs(repo, "Catherine"), "SP-0001 should be searchable"
    ok("SP-0001 (Catherine O'Neil) appears in masked search")

    print("2) suppress -> gone from search immediately")
    assert repo.suppress_property("SAMPLE", "SP-0001", actor="state:CA", reason="privacy")
    assert pid not in _refs(repo, "Catherine"), "suppressed record must not appear in search"
    ok("after suppress, SP-0001 no longer in search results")

    print("3) data + version history RETAINED (nothing deleted)")
    assert repo._property_id("SAMPLE", "SP-0001") == pid, "property row must remain"
    owner = repo.conn.execute(
        "SELECT owner_name_raw FROM owner WHERE property_id=?", (pid,)).fetchone()
    assert owner and owner["owner_name_raw"], "owner row must remain"
    vers = repo.conn.execute(
        "SELECT COUNT(*) FROM property_version WHERE property_id=?", (pid,)).fetchone()[0]
    assert vers >= 1, "version history must remain"
    ok(f"property + owner + {vers} version(s) all retained")

    print("4) audit trail records the action")
    trail = repo.audit_trail(pid)
    assert any(e["action"] == "suppress" and e["actor"] == "state:CA" for e in trail), trail
    ok(f"audit_event 'suppress' by state:CA recorded ({len(trail)} event(s))")

    print("5) reinstate -> back in search + audit event")
    assert repo.reinstate_property("SAMPLE", "SP-0001", actor="state:CA")
    assert pid in _refs(repo, "Catherine"), "reinstated record should be searchable again"
    assert any(e["action"] == "reinstate" for e in repo.audit_trail(pid))
    ok("reinstate restores search visibility + writes audit event")

    print("6) suppressing an unknown record returns False")
    assert repo.suppress_property("SAMPLE", "NO-SUCH-ID", actor="state:CA") is False
    ok("unknown record -> False (no silent success)")

    print("\nALL CYCLE-7 SUPPRESSION ASSERTIONS PASSED ✅")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())