203 lines
11 KiB
Python
203 lines
11 KiB
Python
"""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()
|