`tests/test_cycle6_naupa3.py`

110 lines
"""Cycle 5 tests — NAUPA III XML adapter: parity, namespace tolerance, XXE guard.

Run:  python -m tests.test_cycle6_naupa3

Proves:
  1. XML ingests via the same pipeline; canonical shape + cents/dollars + masking match.
  2. Namespace-tolerant: an xmlns-defaulted document parses identically.
  3. XXE / entity-expansion guard: a DTD/ENTITY document is REFUSED (XmlSecurityError).
  4. Malformed record (missing PropertyId) is rejected, not fatal.
  5. Same-file idempotency holds for the XML path.
"""
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.naupa3 import XmlSecurityError, build_naupa3_xml, parse_naupa3_feed


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


RECORDS = [
    {"property_id": "SP-X001", "holder_name": "First Synthetic Bank",
     "first_name": "Catherine", "last_name": "O'Neil", "street": "100 Test St",
     "city": "Springfield", "state": "SM", "zip": "00001",
     "property_type": "UNCASHED_CHECK", "amount": "42.50"},
    {"property_id": "SP-X002", "holder_name": "Made-Up Securities",
     "first_name": "", "last_name": "ACME WIDGETS INC", "street": "300 Nowhere Blvd",
     "city": "Rivertown", "state": "SM", "zip": "00002",
     "property_type": "SECURITIES", "amount": "5300.00"},
]


def _repo(tmp: Path, data: bytes, uri: str) -> tuple:
    store = FileObjectStore(tmp / "obj")
    (tmp / "obj").mkdir(parents=True, exist_ok=True)
    store.write_bytes(uri, data)
    repo = SqliteRepository(str(tmp / "p.db"))
    return store, repo


def main() -> int:
    tmp = Path(tempfile.mkdtemp(prefix="upp-cycle6-"))

    print("1) NAUPA III ingest parity (no namespace)")
    store, repo = _repo(tmp / "a", build_naupa3_xml(RECORDS), "incoming/n3.xml")
    feed = FeedDefinition("SAMPLE", "incoming/n3.xml", format_name="naupa3_v1")
    r = ingest_authorized_feed(feed, store, repo)
    assert r["status"] == "completed" and r["accepted"] == 2, r
    amt = repo.conn.execute(
        "SELECT amount FROM property WHERE source_property_id='SP-X001'").fetchone()["amount"]
    assert Decimal(str(amt)) == Decimal("42.50"), amt
    # NB: normalize_text turns "O'Neil" -> "O NEIL" (apostrophe->space), so search the
    # given name. (Apostrophe/hyphen search-recall is a logged backlog item; matching is
    # unaffected because the phonetic key still links O'Neil<->ONeill.)
    hits = repo.masked_search("Catherine")
    assert hits and "CATHERINE" not in (hits[0]["owner_name_masked"] or "").upper()
    ok(f"accepted=2; $42.50 parsed; masked {hits[0]['owner_name_masked']}")

    print("2) namespace tolerance (default xmlns)")
    store2, repo2 = _repo(tmp / "b",
                          build_naupa3_xml(RECORDS, namespace="http://naupa.org/naupa3"),
                          "incoming/n3ns.xml")
    r2 = ingest_authorized_feed(
        FeedDefinition("SAMPLE", "incoming/n3ns.xml", format_name="naupa3_v1"), store2, repo2)
    assert r2["accepted"] == 2, r2
    ok("xmlns-defaulted document parsed identically (accepted=2)")

    print("3) XXE / entity-expansion guard")
    bomb = (b'<?xml version="1.0"?>\n'
            b'<!DOCTYPE lolz [ <!ENTITY lol "lol"> <!ENTITY lol2 "&lol;&lol;"> ]>\n'
            b'<UnclaimedProperty><Property><PropertyId>X</PropertyId>'
            b'<Owner><LastName>&lol2;</LastName></Owner></Property></UnclaimedProperty>')
    raised = False
    try:
        list(parse_naupa3_feed(bomb, "SAMPLE"))
    except XmlSecurityError:
        raised = True
    assert raised, "DTD/ENTITY document must be refused"
    ok("DOCTYPE/ENTITY document refused (XmlSecurityError)")

    print("4) malformed record (missing PropertyId) rejected, not fatal")
    bad = list(RECORDS) + [{"first_name": "No", "last_name": "Id", "amount": "1.00"}]
    store3, repo3 = _repo(tmp / "c", build_naupa3_xml(bad), "incoming/n3bad.xml")
    r3 = ingest_authorized_feed(
        FeedDefinition("SAMPLE", "incoming/n3bad.xml", format_name="naupa3_v1"), store3, repo3)
    assert r3["accepted"] == 2 and r3["rejected"] == 1, r3
    ok(f"accepted=2, rejected=1 (missing PropertyId), status={r3['status']}")

    print("5) idempotency on the XML path")
    n = repo.count_properties("SAMPLE")
    again = ingest_authorized_feed(feed, store, repo)
    assert again["status"] == "duplicate" and repo.count_properties("SAMPLE") == n
    ok(f"re-ingest -> duplicate; count stable at {n}")

    print("\nALL CYCLE-5 NAUPA-III ASSERTIONS PASSED ✅")
    return 0


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