`tests/test_cycle3_naupa_and_blocking.py`

106 lines
"""Cycle 3 tests — NAUPA II fixed-width adapter parity + candidate-generation blocking.

Run:  python -m tests.test_cycle3_naupa_and_blocking

Proves:
  1. The NAUPA II adapter ingests via the SAME ingest_authorized_feed pipeline (registry
     dispatch), yields the canonical shape, parses cents->dollars, and masks.
  2. Header/trailer records are skipped; property records accepted.
  3. Same-file idempotency holds for the NAUPA path too.
  4. blocking_keys() puts spelling variants of the same owner in a shared bucket (so they
     get compared) — the candidate-generation step that makes matching sub-O(n²).
"""
from __future__ import annotations

import sys
import tempfile
from decimal import Decimal
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
from services.ingestion.naupa2 import build_naupa2_line
from services.matching.entity_match import MatchInput, blocking_keys, entity_match


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


def _naupa2_bytes() -> bytes:
    lines = [
        build_naupa2_line({"record_type": "HD", "holder_name": "SYNTHETIC STATE FILE"}),  # header, skipped
        build_naupa2_line({
            "record_type": "PR", "source_property_id": "SP-N001",
            "holder_name": "First Synthetic Bank", "owner_last": "ONEIL",
            "owner_first": "CATHERINE", "address": "100 TEST ST", "city": "SPRINGFIELD",
            "state": "SM", "zip": "000010000", "property_type": "CHECK",
            "amount_cents": "4250",  # $42.50
        }),
        build_naupa2_line({
            "record_type": "PR", "source_property_id": "SP-N002",
            "holder_name": "Made-Up Securities", "owner_last": "",
            "owner_first": "ACME WIDGETS INC", "address": "300 NOWHERE BLVD",
            "city": "RIVERTOWN", "state": "SM", "zip": "000020000",
            "property_type": "SECURITIES", "amount_cents": "530000",  # $5,300.00
        }),
        build_naupa2_line({"record_type": "TR", "holder_name": "TRAILER 2 RECORDS"}),  # trailer, skipped
    ]
    return ("\n".join(lines) + "\n").encode("utf-8")


def main() -> int:
    tmp = Path(tempfile.mkdtemp(prefix="upp-cycle3-"))
    store = FileObjectStore(tmp / "obj")
    (tmp / "obj").mkdir(parents=True, exist_ok=True)
    store.write_bytes("incoming/naupa2_feed.dat", _naupa2_bytes())
    repo = SqliteRepository(str(tmp / "p.db"))

    print("1) NAUPA II ingest via the shared pipeline (registry dispatch)")
    feed = FeedDefinition("SAMPLE", "incoming/naupa2_feed.dat", format_name="naupa2_v1")
    r = ingest_authorized_feed(feed, store, repo)
    assert r["status"] == "completed", r
    assert r["accepted"] == 2, f"expected 2 property records, got {r['accepted']}"
    assert r["rejected"] == 0, r
    ok(f"accepted={r['accepted']} (2 PR records; HD/TR skipped)")

    print("2) cents parsed to dollars; masking applied")
    row = repo.conn.execute(
        "SELECT amount FROM property WHERE source_property_id='SP-N001'").fetchone()
    assert Decimal(str(row["amount"])) == Decimal("42.50"), row["amount"]
    hits = repo.masked_search("Oneil")
    assert hits and "ONEIL" not in (hits[0]["owner_name_masked"] or "").upper(), hits
    ok(f"amount 4250c -> $42.50; masked hit {hits[0]['owner_name_masked']}")

    print("3) idempotency on the NAUPA path")
    n_before = repo.count_properties("SAMPLE")
    r2 = ingest_authorized_feed(feed, store, repo)
    assert r2["status"] == "duplicate", r2
    assert repo.count_properties("SAMPLE") == n_before
    ok(f"re-ingest -> duplicate; count stable at {n_before}")

    print("4) blocking keys put same-owner variants in a shared bucket")
    a = MatchInput("Catherine O'Neil", address="100 Test St", city="Springfield",
                   postal_code="00001", region="SM")
    b = MatchInput("Kathryn ONeill", address="100 Test Street", city="Springfield",
                   postal_code="00001", region="SM")
    c = MatchInput("Jonathan Doe", address="9 Elsewhere Rd", city="Rivertown",
                   postal_code="99999", region="XX")
    ka, kb, kc = blocking_keys(a), blocking_keys(b), blocking_keys(c)
    phon_a = {k for k in ka if k.startswith("ph:")}
    phon_b = {k for k in kb if k.startswith("ph:")}
    assert phon_a & phon_b, f"variants should share a phonetic block key: {phon_a} vs {phon_b}"
    assert not (phon_a & {k for k in kc if k.startswith('ph:')}), "unrelated names shouldn't share phonetic key"
    # and the blocked pair actually scores as a match candidate
    assert entity_match(a, b).disposition in ("manual_review", "likely_duplicate_candidate")
    ok(f"variants share {sorted(phon_a & phon_b)}; distinct owner excluded from that bucket")

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


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