303 lines
19 KiB
Python
303 lines
19 KiB
Python
"""Source-evidence integrity tests; fixtures do not assert ARR equivalence."""
|
|
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_day as audit
|
|
from integrations.ohip import collect_arr_day as day
|
|
from integrations.ohip import collect_arr_source as capture
|
|
from integrations.ohip import rate_info
|
|
from integrations.ohip import source_facts as facts
|
|
from integrations.ohip import validate_source_facts as validator
|
|
from integrations.ohip.source_facts_contract import canonical
|
|
from tests.test_ohip_arr_collection import DAY, HOTEL, SECRET
|
|
from tests.test_ohip_day_capture import DayService
|
|
|
|
|
|
class SourceFactsTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.directory = self.root / "capture"
|
|
self.service = DayService()
|
|
self.options = day.Options(DAY, DAY, DAY, HOTEL, page_size=2)
|
|
|
|
def prepare(self, transport=None):
|
|
sink = capture.Archive(self.directory)
|
|
transport = transport or self.service
|
|
result = day.collect(self.options, sink,
|
|
capture.Reader(sink, HOTEL, transport, key=SECRET, sleep=lambda _: None),
|
|
rate_info.RateInfoReader(sink, HOTEL, transport, key=SECRET, sleep=lambda _: None))
|
|
self.assertTrue(result["candidate_capture_complete"], result.get("error"))
|
|
self.pin = result["manifest_sha256"]
|
|
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 observations(self, field, variant=0, record=0, group="fields"):
|
|
return self.document["records"][record][group][field]["variants"][variant]["observations"]
|
|
|
|
def test_handwritten_fixture_preserves_paths_types_and_alternatives(self):
|
|
row = self.service.base.rows[0]
|
|
row["reservationGuest"] = {"fullName": "PRIVATE_DISPLAY_NAME"}
|
|
row["roomStay"].update(departureDate=DAY, guestCounts={"adults": 2, "children": 0},
|
|
currentRoomInfo={"roomId": "0007"})
|
|
segment = row["roomStay"]["roomRates"][0]
|
|
segment.update(start=DAY, end=DAY, numberOfUnits=1, ratePlanCode="UNKNOWN_CODE", roomType="CODE_ONLY")
|
|
row["reservationProfiles"] = {"reservationProfile": [
|
|
{"reservationProfileType": "Company", "profile": {"company": {"companyName": "SAME_NAME"}}},
|
|
{"reservationProfileType": "TravelAgent", "profile": {"company": {"companyName": "SAME_NAME"}}}]}
|
|
row["comments"].append({"comment": {"type": "OTHER", "internal": False, "text": {"value": "SECOND_NOTE"}}})
|
|
row["traces"] = [{"traceText": "RESOLVED_TRACE", "resolveInfo": {"resolvedOn": DAY}},
|
|
{"traceText": "UNKNOWN_TRACE", "resolveInfo": {}}]
|
|
self.prepare()
|
|
first = self.document["records"][0]
|
|
self.assertEqual(first["sources"]["detail"]["file"], "request-000003.response.bin")
|
|
self.assertEqual(first["sources"]["rate"]["request"]["body"],
|
|
{"id": "id0", "type": "Reservation", "summaryInfo": False, "detailDate": DAY})
|
|
self.assertEqual(self.observations("ADULTS"), [{"pointer": "/data/reservations/reservation/0/roomStay/guestCounts/adults",
|
|
"state": "present", "kind": "integer", "value": "2"}])
|
|
self.assertEqual(self.observations("DISP_ROOM_NO")[0]["value"], "0007")
|
|
self.assertEqual([v["value"] for v in self.observations("COMPANY_NAME")], ["SAME_NAME", "SAME_NAME"])
|
|
self.assertEqual([v["value"] for v in self.observations("PROFILE_ROLE", group="context")], ["Company", "TravelAgent"])
|
|
self.assertEqual([v["value"] for v in self.observations("CONFIRMATION_NO")], ["id0", "confirmation-id0"])
|
|
self.assertEqual([v["value"] for v in self.observations("ID_TYPE", group="context")], ["Reservation", "Confirmation"])
|
|
self.assertEqual([v["value"] for v in self.observations("RES_COMMENT")], ["PRIVATE_NOTE", "SECOND_NOTE"])
|
|
self.assertEqual([v["value"] for v in self.observations("TRACE_TEXT")], ["RESOLVED_TRACE", "UNKNOWN_TRACE"])
|
|
self.assertEqual(self.observations("FULL_NAME")[0]["value"], "PRIVATE_DISPLAY_NAME")
|
|
self.assertEqual(self.observations("RATE_CODE")[0]["value"], "UNKNOWN_CODE")
|
|
self.assertEqual(first["fields"]["BLOCK_CODE"], {"mapping_state": "unresolved", "variants": []})
|
|
self.assertTrue(validator.verify_source_facts(self.archive, self.raw)["facts_verified"])
|
|
|
|
def test_missing_null_empty_blank_malformed_and_non_scalar_remain_distinct(self):
|
|
rows = self.service.base.rows
|
|
rows[0]["reservationPackages"] = []
|
|
rows[1]["reservationPackages"] = None
|
|
rows[2]["reservationPackages"] = [None, {}, 3, {"packageCode": ""}, {"packageCode": " "},
|
|
{"packageCode": []}, {"packageCode": {}}, {"packageCode": False}]
|
|
rows[0]["roomStay"]["guestCounts"] = {"adults": None, "children": ""}
|
|
self.prepare()
|
|
self.assertEqual(self.observations("PRODUCTS")[0]["state"], "empty_collection")
|
|
self.assertEqual(self.observations("PRODUCTS", record=1)[0]["state"], "null")
|
|
observations = self.observations("PRODUCTS", record=2)
|
|
self.assertEqual([v["state"] for v in observations],
|
|
["null", "missing", "invalid_container", "explicit_blank", "explicit_blank", "non_scalar", "non_scalar", "present"])
|
|
self.assertEqual(observations[-1]["value"], False)
|
|
self.assertEqual(observations[1]["pointer"], "/data/reservations/reservation/0/reservationPackages/1/packageCode")
|
|
self.assertEqual(self.observations("ADULTS")[0]["state"], "null")
|
|
self.assertEqual(self.observations("CHILDREN")[0]["state"], "explicit_blank")
|
|
self.assertEqual(self.observations("FULL_NAME")[0],
|
|
{"pointer": "/data/reservations/reservationInfo/0/reservationGuest", "state": "missing"})
|
|
validator.verify_source_facts(self.archive, self.raw)
|
|
|
|
def test_money_over_float_and_decimal_context_precision_is_exact(self):
|
|
amount = b"900719925474099312345678901234567890.0123456789"
|
|
def transport(method, path, raw):
|
|
status, headers, response = self.service(method, path, raw)
|
|
if path.endswith("rate-info/searches"):
|
|
response = response.replace(b'"totalRateAmount": 0', b'"totalRateAmount": ' + amount)
|
|
return status, headers, response
|
|
self.prepare(transport)
|
|
self.assertEqual(self.observations("EFFECTIVE_RATE_AMOUNT")[0]["value"], amount.decode())
|
|
self.assertEqual(self.observations("EFFECTIVE_RATE_AMOUNT")[0]["kind"], "decimal")
|
|
self.assertEqual(self.observations("EFFECTIVE_RATE_AMOUNT", variant=1)[0]["value"], amount.decode())
|
|
validator.verify_source_facts(self.archive, self.raw)
|
|
|
|
def test_all_statuses_unassigned_unknown_codes_and_price_issues_retained(self):
|
|
for row, status in zip(self.service.base.rows, ["NoShow", "Cancelled", "InHouse"]):
|
|
row["reservationStatus"] = status
|
|
self.service.rate_edit = lambda data, body: {"detail": {"rateSuppressed": True}} if body["id"] == "id1" else data
|
|
self.prepare()
|
|
self.assertEqual(len(self.document["records"]), 3)
|
|
self.assertEqual(self.observations("EFFECTIVE_RATE_AMOUNT", variant=1, record=1)[0]["state"], "missing")
|
|
self.assertEqual(self.observations("RATE_ISSUE", record=1, group="context")[0]["value"], "rate_suppressed")
|
|
result = validator.verify_source_facts(self.archive, self.raw)
|
|
for flag in ("finance_ready", "source_mapping_verified", "report_equivalence_verified"):
|
|
self.assertIs(result[flag], False)
|
|
self.assertEqual(self.document["record_order"], "capture_order_not_report_order")
|
|
|
|
def test_successful_retry_bound_not_failed_attempt(self):
|
|
attempts = {"search": 0, "rate": 0}
|
|
def transport(method, path, raw):
|
|
key = "rate" if path.endswith("rate-info/searches") else "search"
|
|
attempts[key] += 1
|
|
if attempts[key] == 1:
|
|
if key == "search":
|
|
raise TimeoutError()
|
|
return 429, {}, b'{}'
|
|
return self.service(method, path, raw)
|
|
self.prepare(transport)
|
|
source = self.document["records"][0]["sources"]
|
|
self.assertEqual(source["search"]["request"]["file"], "request-000002.json")
|
|
self.assertEqual(source["rate"]["request"]["file"], "rate-000002.json")
|
|
self.assertEqual(source["rate"]["request"]["attempt"], 2)
|
|
validator.verify_source_facts(self.archive, self.raw)
|
|
|
|
def test_independent_verifier_does_not_use_extractor_or_network(self):
|
|
self.prepare()
|
|
with patch.object(facts, "build_source_facts", side_effect=AssertionError("extractor called")), \
|
|
patch.object(facts, "_walk", side_effect=AssertionError("extractor traversal called")), \
|
|
patch.object(facts, "_bindings", side_effect=AssertionError("extractor join called")), \
|
|
patch.object(capture, "HTTPTransport", side_effect=AssertionError("network called")), \
|
|
patch.object(capture, "load_key", side_effect=AssertionError("credentials read")):
|
|
self.assertTrue(validator.verify_source_facts(self.archive, self.raw)["facts_verified"])
|
|
|
|
def test_records_omitted_added_reordered_or_duplicated_rejected(self):
|
|
self.prepare()
|
|
alterations = [lambda d: d["records"].pop(), lambda d: d["records"].append(d["records"][0]),
|
|
lambda d: d["records"].reverse(), lambda d: d["records"].__setitem__(1, d["records"][0])]
|
|
for alter in alterations:
|
|
document = copy.deepcopy(self.document)
|
|
alter(document)
|
|
with self.subTest(alter=alter), self.assertRaises(capture.CollectionError):
|
|
validator.verify_source_facts(self.archive, canonical(document))
|
|
|
|
def test_field_provenance_type_scope_and_claim_mutations_rejected(self):
|
|
self.prepare()
|
|
alterations = [
|
|
lambda d: d["records"][0]["fields"].pop("BLOCK_CODE"),
|
|
lambda d: d["records"][0]["fields"]["RES_COMMENT"].update(mapping_state="accepted"),
|
|
lambda d: d["records"][0]["fields"]["RES_COMMENT"]["variants"][0]["observations"].clear(),
|
|
lambda d: d["records"][0]["fields"]["RES_COMMENT"]["variants"][0]["observations"][0].update(pointer="/wrong"),
|
|
lambda d: d["records"][0]["sources"].__setitem__("detail", d["records"][1]["sources"]["detail"]),
|
|
lambda d: d["records"][0]["sources"]["rate"]["request"]["body"].update(detailDate="2026-09-14"),
|
|
lambda d: d["records"][0]["sources"]["rate"]["request"].update(attempt=True),
|
|
lambda d: d["records"][0]["context"].pop("COMMENT_TYPE"),
|
|
lambda d: d["records"][0].update(capture_sequence=True),
|
|
lambda d: d["records"][0].update(capture_sequence=1.0),
|
|
lambda d: d["records"][0]["fields"]["EFFECTIVE_RATE_AMOUNT"]["variants"][0]["observations"][0].update(value=0),
|
|
lambda d: d.update(finance_ready=True), lambda d: d.update(source_mapping_verified=True),
|
|
lambda d: d.update(report_equivalence_verified=True), lambda d: d.update(accepted=True),
|
|
lambda d: d.update(hotel_id="OTHER"), lambda d: d.update(rate_date="2026-09-14"),
|
|
lambda d: d.update(contract_sha256="0" * 64), lambda d: d.update(business_blockers=[]),
|
|
lambda d: d.update(record_order="report_order"), lambda d: d.update(capture_manifest_sha256="0" * 64),
|
|
]
|
|
for index, alter in enumerate(alterations):
|
|
document = copy.deepcopy(self.document)
|
|
alter(document)
|
|
with self.subTest(index=index), self.assertRaises(capture.CollectionError):
|
|
validator.verify_source_facts(self.archive, canonical(document))
|
|
|
|
def test_equal_values_swapped_array_positions_rejected(self):
|
|
self.service.base.rows[0]["comments"] *= 2
|
|
self.prepare()
|
|
observations = self.observations("RES_COMMENT")
|
|
self.assertEqual(observations[0]["value"], observations[1]["value"])
|
|
observations.reverse()
|
|
with self.assertRaisesRegex(capture.CollectionError, "facts_record_mismatch"):
|
|
validator.verify_source_facts(self.archive, canonical(self.document))
|
|
|
|
def test_same_money_cannot_be_rebound_to_another_reservation(self):
|
|
self.service.rate_edit = lambda data, body: {"detail": {"totalRateAmount": 0,
|
|
"rateSuppressed": False, "revenue": {"currencyCode": "USD"}}}
|
|
self.prepare()
|
|
first, second = self.document["records"][:2]
|
|
first["sources"]["rate"], second["sources"]["rate"] = second["sources"]["rate"], first["sources"]["rate"]
|
|
self.assertEqual([v["observations"][0]["value"] for v in first["fields"]["EFFECTIVE_RATE_AMOUNT"]["variants"]],
|
|
[v["observations"][0]["value"] for v in second["fields"]["EFFECTIVE_RATE_AMOUNT"]["variants"]])
|
|
with self.assertRaisesRegex(capture.CollectionError, "facts_record_mismatch"):
|
|
validator.verify_source_facts(self.archive, canonical(self.document))
|
|
|
|
def test_nested_array_branch_order_and_primary_name_context(self):
|
|
row = self.service.base.rows[0]
|
|
row["reservationGuests"] = [{"primary": True, "profileInfo": {"profile": {"customer": {
|
|
"personName": [{"nameType": "Primary", "surname": "LAST", "givenName": "FIRST",
|
|
"middleName": "MIDDLE", "nameTitle": "TITLE"}]}}}}]
|
|
row["roomStay"]["roomRates"][0]["stayProfiles"] = [{"reservationProfileType": "Company",
|
|
"profile": {"company": {"companyName": "A"}}}, None]
|
|
row["roomStay"]["roomRates"].append({"stayProfiles": []})
|
|
self.prepare()
|
|
observations = self.observations("COMPANY_NAME", variant=1)
|
|
self.assertEqual([v["state"] for v in observations], ["present", "null", "empty_collection"])
|
|
self.assertEqual([v["pointer"] for v in observations], [
|
|
"/data/reservations/reservation/0/roomStay/roomRates/0/stayProfiles/0/profile/company/companyName",
|
|
"/data/reservations/reservation/0/roomStay/roomRates/0/stayProfiles/1",
|
|
"/data/reservations/reservation/0/roomStay/roomRates/1/stayProfiles"])
|
|
self.assertEqual(self.observations("NAME_TITLE", group="context")[0]["value"], "TITLE")
|
|
self.assertIs(self.observations("GUEST_PRIMARY", group="context")[0]["value"], True)
|
|
self.assertEqual(self.observations("FULL_NAME")[0]["state"], "missing")
|
|
validator.verify_source_facts(self.archive, self.raw)
|
|
|
|
def test_resealed_archive_wrong_day_fails_protocol_even_with_new_pin(self):
|
|
self.prepare()
|
|
path = self.directory / "rate-000001.json"
|
|
request = json.loads(path.read_bytes())
|
|
request["body"]["detailDate"] = "2026-09-14"
|
|
raw = capture.json_bytes(request)
|
|
path.write_bytes(raw)
|
|
manifest_path = self.directory / "result.json"
|
|
manifest = json.loads(manifest_path.read_bytes())
|
|
for item in manifest["files"]:
|
|
if item["name"] == path.name:
|
|
item.update(bytes=len(raw), sha256=hashlib.sha256(raw).hexdigest())
|
|
manifest_path.write_bytes(capture.json_bytes(manifest))
|
|
new_pin = hashlib.sha256(manifest_path.read_bytes()).hexdigest()
|
|
archive = audit.VerifiedArchive(self.directory, new_pin)
|
|
self.document["capture_manifest_sha256"] = new_pin
|
|
with self.assertRaisesRegex(capture.CollectionError, "capture_protocol_replay_failed"):
|
|
validator.verify_source_facts(archive, canonical(self.document))
|
|
|
|
def test_malformed_duplicate_keys_and_budget_rejected(self):
|
|
self.prepare()
|
|
for raw in (b'[]', b'{', b'{"records":[],"records":[]}', b'{"records":NaN}'):
|
|
with self.subTest(raw=raw), self.assertRaises(capture.CollectionError):
|
|
validator.verify_source_facts(self.archive, raw)
|
|
with patch.object(validator, "MAX_FACTS_BYTES", 1), self.assertRaisesRegex(capture.CollectionError, "byte_budget"):
|
|
validator.verify_source_facts(self.archive, self.raw)
|
|
|
|
def test_archive_changed_after_build_is_rejected_and_not_exported(self):
|
|
self.prepare()
|
|
path = self.directory / "rate-000001.response.bin"
|
|
path.write_bytes(path.read_bytes().replace(b'"totalRateAmount": 0', b'"totalRateAmount": 9'))
|
|
with self.assertRaises(capture.CollectionError):
|
|
validator.verify_source_facts(self.archive, self.raw)
|
|
output = self.root / "facts"
|
|
with self.assertRaises(capture.CollectionError):
|
|
facts.export_source_facts(self.directory, self.pin, output)
|
|
self.assertFalse(output.exists())
|
|
|
|
def test_private_export_immutable_capture_repeat_determinism_and_cli_privacy(self):
|
|
self.prepare()
|
|
before = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in self.directory.iterdir()}
|
|
output = self.root / "facts"
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = facts.main(["--capture-dir", str(self.directory), "--capture-sha256", self.pin,
|
|
"--output-dir", str(output)])
|
|
self.assertEqual(code, 0)
|
|
report = 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)
|
|
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()})
|
|
for private in (SECRET, "PRIVATE_NOTE", "confirmation-id0", '"reservation_id"', '"pointer"', str(self.root)):
|
|
self.assertNotIn(private, stdout.getvalue())
|
|
with self.assertRaises(FileExistsError):
|
|
facts.export_source_facts(self.directory, self.pin, output)
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(validator.main(["--capture-dir", str(self.directory), "--capture-sha256", self.pin,
|
|
"--facts", str(output / "source-facts.json"), "--facts-sha256", report["evidence_sha256"]]), 0)
|
|
self.assertEqual(validator.main(["--capture-dir", str(self.directory), "--capture-sha256", self.pin,
|
|
"--facts", str(output / "source-facts.json"), "--facts-sha256", "0" * 64]), 1)
|
|
|
|
def test_repository_export_refused_before_source_read(self):
|
|
repository = Path(facts.__file__).resolve().parents[2]
|
|
with patch.object(audit, "VerifiedArchive", side_effect=AssertionError("should not read")), \
|
|
self.assertRaisesRegex(capture.CollectionError, "output_must_be_outside_repository"):
|
|
facts.export_source_facts(self.directory, "0" * 64, repository / "private-facts-test")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|