"""Named evidence joins are checked against originals, not display-name equality.""" import contextlib import copy import hashlib import io import json from pathlib import Path import stat import tempfile import unittest from unittest.mock import patch from integrations.ohip import audit_arr_named_day as audit, collect_arr_named_day as named from integrations.ohip import collect_arr_source as source, profile_summary as probe, source_facts as facts from integrations.ohip import validate_source_facts as validator from integrations.ohip.profile_reader import ProfileSummaryReader from integrations.ohip.rate_info import RateInfoReader from integrations.ohip.source_facts_contract import canonical, NAMED_VERSION, NAMED_CONTRACT_SHA256 from tests.test_ohip_named_day import NamedService, DAY, HOTEL, SECRET class NamedSourceFactsTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.directory = self.root / "capture" self.service = NamedService() self.options = named.Options(DAY, DAY, DAY, HOTEL, page_size=2, max_profiles=3) def collect(self, transport=None): archive = source.Archive(self.directory) transport = transport or self.service result = named.collect(self.options, archive, source.Reader(archive, HOTEL, transport, key=SECRET, sleep=lambda _: None), RateInfoReader(archive, HOTEL, transport, key=SECRET, sleep=lambda _: None), ProfileSummaryReader(archive, HOTEL, transport, max_profiles=3, key=SECRET, sleep=lambda _: None)) self.pin = result["manifest_sha256"] return result def prepare(self, transport=None): self.assertTrue(self.collect(transport)["candidate_capture_complete"]) self.archive = audit.VerifiedArchive(self.directory, self.pin) self.raw = facts.build_source_facts(self.archive) self.document = json.loads(self.raw) return self.document def verify(self, document=None): with patch.object(facts, "build_source_facts", side_effect=AssertionError("verifier must not extract")), \ patch.object(source, "HTTPTransport", side_effect=AssertionError("no network")), \ patch.object(source, "load_key", side_effect=AssertionError("no credentials")): return validator.verify_source_facts(self.archive, canonical(document or self.document)) def names(self, record=0): return self.document["records"][record]["fields"]["FULL_NAME"]["variants"] def test_exact_name_provenance_and_new_version_without_report_claim(self): self.prepare() first = self.document["records"][0] self.assertEqual((self.document["version"], self.document["contract_sha256"]), (NAMED_VERSION, NAMED_CONTRACT_SHA256)) self.assertEqual(first["sources"]["profile"]["request"]["body"], {"profileIds": ["p0"], "summaryInfo": True, "limit": 1, "offset": 0}) self.assertEqual(first["sources"]["profile"]["file"], "profile-000001.response.bin") self.assertEqual([v["observations"][0]["value"] for v in self.names()[1:]], [" Surname0, Given · 测试 "] * 2) self.assertEqual(self.names()[1]["observations"][0]["pointer"], "/data/profileSummaries/profileInfo/0/profile/formerName/fullName") self.assertEqual(first["context"]["SUMMARY_NAME_TYPE"]["variants"][0]["observations"][0]["state"], "missing") self.assertEqual(first["context"]["SUMMARY_NAME_PREFIX"]["variants"][0]["observations"][0]["value"], "Dr") result = self.verify() self.assertTrue(result["facts_verified"]) for flag in ("source_mapping_verified", "report_equivalence_verified", "finance_ready"): self.assertIs(result[flag], False) self.assertNotIn("note_and_trace_scope_order", self.document["business_blockers"]) def test_shared_profile_keeps_raw_name_but_rejects_second_reservation_candidate(self): first, second = self.service.base.rows[:2] second["reservationGuests"] = copy.deepcopy(first["reservationGuests"]) second["reservationGuests"][0]["profileInfo"]["profile"]["customer"]["personName"][0]["givenName"] = "Other" self.prepare() records = self.document["records"] self.assertEqual(records[0]["sources"]["profile"], records[1]["sources"]["profile"]) self.assertEqual(self.names(1)[1]["observations"][0]["state"], "present") self.assertEqual(self.names(1)[2]["observations"][0]["state"], "missing") self.assertEqual(records[1]["context"]["NAME_ISSUE"]["variants"][0]["observations"][0]["value"], "profile_name_components_mismatch") self.verify() def test_no_query_is_distinct_even_if_same_profile_is_queried_for_another_reservation(self): first, second = self.service.base.rows[:2] second["reservationGuests"] = copy.deepcopy(first["reservationGuests"]) first["reservationGuests"][0]["profileInfo"]["profile"]["customer"]["personName"] = [] self.prepare() self.assertIsNone(self.document["records"][0]["sources"]["profile"]) self.assertEqual(self.names()[1]["observations"], [{"state": "source_not_acquired"}]) self.assertEqual(self.names()[2]["observations"][0]["state"], "missing") self.assertIsNotNone(self.document["records"][1]["sources"]["profile"]) self.verify() def test_acquired_missing_or_blank_name_has_real_provenance_and_no_validated_value(self): self.service.profiles["p0"].pop("fullName") self.service.profiles["p1"]["fullName"] = " " self.prepare() self.assertEqual(self.names(0)[1]["observations"][0]["state"], "missing") self.assertEqual(self.names(1)[1]["observations"][0]["state"], "explicit_blank") for i in (0, 1): self.assertIsNotNone(self.document["records"][i]["sources"]["profile"]) self.assertEqual(self.names(i)[2]["observations"][0]["state"], "missing") self.verify() def test_recovery_binds_successful_attempt_instead_of_first_failed_response(self): attempts = 0 def transport(method, path, body): nonlocal attempts if path == probe.PATH: attempts += 1 if attempts == 1: return 503, {}, b'{}' if attempts == 2: raise TimeoutError() return self.service(method, path, body) self.prepare(transport) reference = self.document["records"][0]["sources"]["profile"] self.assertEqual(reference["file"], "profile-000003.response.bin") self.assertEqual(reference["request"]["attempt"], 3) self.verify() def test_equal_display_names_cannot_hide_wrong_profile_provenance(self): self.service.profiles["p0"]["fullName"] = "PRIVATE SAME DISPLAY" self.service.profiles["p1"]["fullName"] = "PRIVATE SAME DISPLAY" self.prepare() first, second = self.document["records"][:2] first["sources"]["profile"], second["sources"]["profile"] = second["sources"]["profile"], first["sources"]["profile"] self.assertEqual(self.names(0)[1]["observations"][0]["value"], self.names(1)[1]["observations"][0]["value"]) with self.assertRaisesRegex(source.CollectionError, "facts_record_mismatch"): self.verify() def test_forged_observation_identity_or_omitted_context_is_rejected(self): self.prepare() changes = [lambda d: d["records"][0]["fields"]["FULL_NAME"]["variants"][1]["observations"][0].update(value="wrong"), lambda d: d["records"][0]["context"]["SUMMARY_PROFILE_ID"]["variants"][0]["observations"][0].update(value="p1"), lambda d: d["records"][0]["context"].pop("SUMMARY_NAME_PREFIX"), lambda d: d["records"][0]["sources"]["profile"]["request"]["body"].update(summaryInfo=1), lambda d: d["records"].reverse()] for change in changes: modified = copy.deepcopy(self.document); change(modified) with self.subTest(change=change), self.assertRaisesRegex(source.CollectionError, "facts_record_mismatch"): self.verify(modified) def test_missing_query_cannot_be_fabricated_as_empty_response(self): self.service.base.rows[0]["reservationGuests"] = [] self.prepare() self.document["records"][0]["sources"]["profile"] = {} with self.assertRaisesRegex(source.CollectionError, "facts_record_mismatch"): self.verify() def test_version_downgrade_or_business_claim_is_rejected(self): self.prepare() for changes in ({"version": "arr-source-field-evidence/v1"}, {"source_mapping_verified": True}, {"finance_ready": True}, {"contract_sha256": "0" * 64}): modified = {**self.document, **changes} with self.subTest(changes=changes), self.assertRaisesRegex(source.CollectionError, "facts_header_mismatch"): self.verify(modified) def test_failed_capture_cannot_export_partial_field_evidence(self): def transport(method, path, body): return (503, {}, b'{}') if path == probe.PATH else self.service(method, path, body) self.assertFalse(self.collect(transport)["candidate_capture_complete"]) output = self.root / "facts" with self.assertRaisesRegex(source.CollectionError, "incomplete_capture"): facts.export_source_facts(self.directory, self.pin, output, capture_version="v3") self.assertFalse(output.exists()) def test_explicit_v3_cli_export_and_independent_verify_are_private_and_repeatable(self): self.prepare() before = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in self.directory.iterdir()} output = self.root / "facts" args = ["--capture-dir", str(self.directory), "--capture-sha256", self.pin, "--output-dir", str(output)] stdout = io.StringIO() with contextlib.redirect_stdout(stdout): self.assertEqual(facts.main(args), 1) # Old default must not silently upgrade. self.assertFalse(output.exists()) stdout = io.StringIO() with contextlib.redirect_stdout(stdout): self.assertEqual(facts.main([*args, "--capture-version", "v3"]), 0) result = json.loads(stdout.getvalue()) self.assertEqual((output / "source-facts.json").read_bytes(), self.raw) self.assertEqual(stat.S_IMODE(output.stat().st_mode), 0o700) for path in output.iterdir(): self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) for value in ("Surname0", '"p0"', '"profileIds"', SECRET, str(self.root)): self.assertNotIn(value, stdout.getvalue()) self.assertEqual(facts.build_source_facts(self.archive), self.raw) self.assertEqual(before, {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in self.directory.iterdir()}) with contextlib.redirect_stdout(io.StringIO()): self.assertEqual(validator.main(["--capture-dir", str(self.directory), "--capture-sha256", self.pin, "--capture-version", "v3", "--facts", str(output / "source-facts.json"), "--facts-sha256", result["evidence_sha256"]]), 0) if __name__ == "__main__": unittest.main()