`tests/test_ingest_and_match.py`

113 lines
"""End-to-end smoke test — proves the prototype runs at $0 on synthetic data (stdlib only).

Run:  python -m tests.test_ingest_and_match

Asserts:
  1. Ingestion loads the synthetic feed and rejects the empty-owner row.
  2. Ingestion is IDEMPOTENT — re-running the same file is a no-op ('duplicate'), and
     record count does not grow.
  3. Masking works — no raw owner name leaks into the search projection.
  4. Entity matching links spelling variants (person + business) while keeping distinct
     people distinct.
"""
from __future__ import annotations

import sys
from pathlib import Path

# make repo root importable when run as a script
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.matching.entity_match import MatchInput, entity_match

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


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


def main() -> int:
    import tempfile

    tmp = Path(tempfile.mkdtemp(prefix="upp-proto-"))
    store = FileObjectStore(tmp / "objstore")
    # stage the synthetic feed into the object store under a relative uri
    (tmp / "objstore").mkdir(parents=True, exist_ok=True)
    store.write_bytes("incoming/sample.csv", SAMPLE.read_bytes())
    repo = SqliteRepository(str(tmp / "proto.db"))

    feed = FeedDefinition(jurisdiction="SAMPLE", source_uri="incoming/sample.csv")

    print("1) Ingestion + rejection")
    r1 = ingest_authorized_feed(feed, store, repo)
    assert r1["status"] in ("completed", "completed_with_errors"), r1
    # 8 rows, 1 has empty owner_name -> rejected
    assert r1["accepted"] == 7, f"expected 7 accepted, got {r1['accepted']}"
    assert r1["rejected"] == 1, f"expected 1 rejected, got {r1['rejected']}"
    _passed(f"accepted={r1['accepted']} rejected={r1['rejected']} (empty-owner row rejected)")

    print("2) Idempotency")
    count_after_first = repo.count_properties("SAMPLE")
    r2 = ingest_authorized_feed(feed, store, repo)
    assert r2["status"] == "duplicate", r2
    assert repo.count_properties("SAMPLE") == count_after_first, "re-ingest changed count!"
    _passed(f"re-ingest -> 'duplicate', property count stable at {count_after_first}")

    print("3) Masking (no raw name leaks into search projection)")
    hits = repo.masked_search("Catherine")
    assert hits, "expected at least one masked hit for 'Catherine'"
    leaked = [h for h in hits if "CATHERINE" in (h["owner_name_masked"] or "").upper()]
    assert not leaked, f"raw name leaked into masked projection: {leaked}"
    _passed(f"masked hit sample: {hits[0]['owner_name_masked']} | {hits[0]['amount_band']}")

    print("4) Entity matching")
    person = entity_match(
        MatchInput("Catherine O'Neil", city="Springfield", region="SAMPLE",
                   postal_code="00001", address="100 Test St"),
        MatchInput("Kathryn ONeill", city="Springfield", region="SAMPLE",
                   postal_code="00001", address="100 Test Street"),
    )
    assert person.disposition in ("manual_review", "likely_duplicate_candidate"), person
    _passed(f"person variants -> {person.disposition} (p={person.probability})")

    business = entity_match(
        MatchInput("Acme Widgets Inc", city="Rivertown", region="SAMPLE",
                   postal_code="00002", address="300 Nowhere Blvd", is_business=True),
        MatchInput("Acme Widgets Incorporated", city="Rivertown", region="SAMPLE",
                   postal_code="00002", address="300 Nowhere Blvd", is_business=True),
    )
    assert business.disposition in ("manual_review", "likely_duplicate_candidate"), business
    _passed(f"business variants -> {business.disposition} (p={business.probability})")

    distinct = entity_match(
        MatchInput("Catherine O'Neil", city="Springfield", region="SAMPLE", postal_code="00001"),
        MatchInput("Jonathan Doe", city="Springfield", region="SAMPLE", postal_code="00001"),
    )
    assert distinct.disposition == "distinct", distinct
    _passed(f"different people -> {distinct.disposition} (p={distinct.probability})")

    # person vs business must never link
    cross = entity_match(
        MatchInput("Acme Widgets Inc", is_business=True),
        MatchInput("Acme Widgets Inc", is_business=False),
    )
    assert cross.disposition == "distinct", cross
    _passed("person/business type conflict -> distinct (never linked)")

    # Robust contract (survives weight re-tuning): a same-owner variant must ALWAYS
    # out-score two clearly-different people at the same address.
    assert person.probability > distinct.probability, (person, distinct)
    assert business.probability > distinct.probability, (business, distinct)
    _passed(f"ordering: variant p={person.probability} > distinct p={distinct.probability}")

    print("\nALL SMOKE-TEST ASSERTIONS PASSED ✅")
    return 0


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