from __future__ import annotations import copy import hashlib import json from pathlib import Path import subprocess import sys import tempfile import unittest from unittest.mock import patch from integrations.ohip import audit_arr_capture as audit from integrations.ohip import collect_arr_source as collector from tests.test_ohip_arr_collection import DAY, HOTEL, FakeService, row SCRIPT = Path(__file__).resolve().parents[1] / "integrations/ohip/audit_arr_capture.py" class ArchiveAuditTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.directory = Path(self.temp.name) / "capture" archive = collector.Archive(self.directory) self.service = FakeService(count=3) reader = collector.Reader(archive, HOTEL, self.service, sleep=lambda _: None) self.result = collector.collect(collector.Options(DAY, HOTEL, page_size=2), archive, reader) self.pin = self.result["manifest_sha256"] def inspect(self, expected_error=None): result = audit.audit_capture(self.directory, self.pin) if expected_error: self.assertEqual(result["status"], "invalid_capture") self.assertEqual(result["error"], expected_error) self.assertNotIn("analysis", result) self.assertFalse(result["finance_ready"]) return result def repin(self, result): # Adversarial tests deliberately mint a new test pin to reach deeper validation. raw = collector.json_bytes(result) (self.directory / "result.json").write_bytes(raw) self.pin = hashlib.sha256(raw).hexdigest() def edit_file(self, name, edit): path = self.directory / name doc = json.loads(path.read_bytes()) edit(doc) raw = collector.json_bytes(doc) path.write_bytes(raw) manifest = json.loads((self.directory / "result.json").read_bytes()) entry = next(item for item in manifest["files"] if item["name"] == name) entry.update(bytes=len(raw), sha256=hashlib.sha256(raw).hexdigest()) self.repin(manifest) def test_complete_capture_audits_without_network_or_writes(self): before = {p.name: p.read_bytes() for p in self.directory.iterdir()} with patch.object(collector, "HTTPTransport", side_effect=AssertionError("network forbidden")): result = self.inspect() self.assertEqual(result["status"], "verified_capture_audited") self.assertEqual(result["analysis"]["source_records"], 3) self.assertEqual(result["verified_files"], len(before)) self.assertEqual(result["network_calls"], 0) self.assertEqual({p.name: p.read_bytes() for p in self.directory.iterdir()}, before) text = json.dumps(result) for sensitive in ("PRIVATE_NOTE", "id0", "confirmation-id0"): self.assertNotIn(sensitive, text) def test_manifest_pin_not_computed_from_candidate_implicitly(self): self.pin = "0" * 64 self.inspect("manifest_hash_mismatch") def test_changed_raw_source_fails_hash_check(self): path = self.directory / "request-000001.response.bin" raw = path.read_bytes().replace(b"PRIVATE_NOTE", b"CHANGED_NOTE") path.write_bytes(raw) self.inspect("archive_hash_mismatch") def test_missing_or_unlisted_file_fails(self): extra = self.directory / "extra.txt" extra.write_text("private") self.inspect("archive_file_set_mismatch") extra.unlink() (self.directory / "request-000001.response.bin").unlink() self.inspect("archive_file_set_mismatch") def test_inventory_traversal_cannot_read_outside_archive(self): doc = json.loads((self.directory / "result.json").read_bytes()) doc["files"][0]["name"] = "../private.json" self.repin(doc) self.inspect("unsafe_inventory_name") def test_duplicate_inventory_rejected(self): doc = json.loads((self.directory / "result.json").read_bytes()) doc["files"].append(copy.deepcopy(doc["files"][0])) self.repin(doc) self.inspect("duplicate_inventory_name") def test_symlink_file_refused(self): path = self.directory / "request-000001.response.bin" outside = self.directory.parent / "outside.bin" path.rename(outside) path.symlink_to(outside) self.inspect("archive_or_shape_invalid") def test_world_readable_file_or_directory_refused(self): path = self.directory / "result.json" path.chmod(0o644) self.inspect("unsafe_archive_file") path.chmod(0o600) self.directory.chmod(0o755) self.inspect("unsafe_archive_directory") def test_failed_capture_and_business_ready_claim_rejected(self): doc = json.loads((self.directory / "result.json").read_bytes()) doc["status"] = "failed" self.repin(doc) self.inspect("incomplete_capture") doc["status"] = "complete_candidate_capture" doc["finance_ready"] = True self.repin(doc) self.inspect("unsupported_capture_claim") def test_rehashed_wrong_detail_cannot_pass_protocol_replay(self): def change(doc): doc["data"]["reservations"]["reservation"][0]["reservationIdList"][0]["id"] = "wrong-private-id" self.edit_file("request-000003.response.bin", change) self.inspect("capture_protocol_replay_failed") def test_rehashed_wrong_request_cannot_pass(self): self.edit_file("request-000001.json", lambda doc: doc["body"].update(arrivalEndDate="2026-09-14")) self.inspect("capture_protocol_replay_failed") def test_rehashed_wrong_attempt_cannot_pass(self): self.edit_file("request-000001.json", lambda doc: doc.update(attempt=2)) self.inspect("capture_protocol_replay_failed") def test_summary_counts_not_trusted(self): doc = json.loads((self.directory / "result.json").read_bytes()) doc["verified_details"] = 2 self.repin(doc) self.inspect("capture_summary_mismatch") def test_capture_context_not_trusted(self): self.edit_file("capture.json", lambda doc: doc.update(hotel_id="wrong")) self.inspect("capture_context_mismatch") def test_capture_grants_not_expanded_during_replay(self): self.edit_file("capture.json", lambda doc: doc["operations"].append("postReservation")) self.inspect("capture_contract_mismatch") def test_successful_retried_capture_replays_without_sleep(self): directory = self.directory.parent / "retried" archive = collector.Archive(directory) service = FakeService(count=1) count = 0 def transport(*args): nonlocal count count += 1 if count == 1: raise collector.urllib.error.URLError("private-error") if count == 2: return 429, {"Retry-After": "0"}, b"limited" return service(*args) reader = collector.Reader(archive, HOTEL, transport, sleep=lambda _: None) result = collector.collect(collector.Options(DAY, HOTEL), archive, reader) inspected = audit.audit_capture(directory, result["manifest_sha256"]) self.assertEqual(inspected["status"], "verified_capture_audited") self.assertEqual(inspected["network_calls"], 0) def test_cli_produces_safe_summary_and_nonzero_invalid_input(self): result = subprocess.run([sys.executable, str(SCRIPT), "--capture-dir", str(self.directory), "--manifest-sha256", self.pin], capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(result.stdout)["analysis"]["source_records"], 3) self.assertNotIn("PRIVATE_NOTE", result.stdout + result.stderr) result = subprocess.run([sys.executable, str(SCRIPT), "--capture-dir", str(self.directory), "--manifest-sha256", "0" * 64], capture_output=True, text=True) self.assertEqual(result.returncode, 1) self.assertEqual(json.loads(result.stdout)["error"], "manifest_hash_mismatch") self.assertEqual(result.stderr, "") class CandidateFieldAuditTests(unittest.TestCase): def pair(self): search = row() detail = copy.deepcopy(search) detail["roomStay"].update(departureDate=DAY, guestCounts={"adults": 1, "children": 0}, currentRoomInfo={"roomId": "PRIVATE_ROOM"}, roomRates=[{ "start": DAY, "end": DAY, "numberOfUnits": 1, "roomId": "PRIVATE_ROOM", "ratePlanCode": "PRIVATE_RATE", "roomType": "PRIVATE_TYPE", "rates": {"rate": [{"effectiveRate": {"amountBeforeTax": 0}}]}}]) search["reservationGuest"] = {"fullName": "PRIVATE_FULL_NAME"} return search, detail def analyze(self, pairs): result = audit.analyze_fields([pair[0] for pair in pairs], [pair[1] for pair in pairs]) self.assertNotIn("PRIVATE_", json.dumps(result)) self.assertEqual(len(result["fields"]), 16) self.assertTrue(all(field["report_mapping_verified"] is False for field in result["fields"].values())) return result def test_zero_values_zero_night_and_single_day_segment_present(self): result = self.analyze([self.pair()]) self.assertEqual(result["fields"]["CHILDREN"]["records_with_candidate"], 1) self.assertEqual(result["fields"]["EFFECTIVE_RATE_AMOUNT"]["records_with_candidate"], 1) self.assertEqual(result["scenarios"]["zero_night_records"], 1) self.assertEqual(result["scenarios"]["arrival_interval_one_candidate"], 1) self.assertEqual(result["scenarios"]["single_day_rate_segments"], 1) self.assertEqual(result["scenarios"]["rate_segments_total"], 1) def test_overlapping_segments_detected_without_choosing_first(self): search, detail = self.pair() another = copy.deepcopy(detail["roomStay"]["roomRates"][0]) another.update(ratePlanCode="PRIVATE_SECOND_RATE", numberOfUnits=2) detail["roomStay"]["roomRates"].append(another) result = self.analyze([(search, detail)]) self.assertEqual(result["scenarios"]["arrival_interval_multiple_candidates"], 1) self.assertEqual(result["fields"]["RATE_CODE"]["records_with_distinct_candidates"], 1) self.assertEqual(result["scenarios"]["multi_room_records"], 1) def test_missing_rate_is_visible_without_whitelist_exclusion(self): search, detail = self.pair() del detail["roomStay"]["roomRates"][0]["ratePlanCode"] result = self.analyze([(search, detail)]) self.assertEqual(result["source_records"], 1) self.assertEqual(result["fields"]["RATE_CODE"]["records_without_candidate"], 1) self.assertTrue(result["fields"]["RATE_CODE"]["required_for_all_source_rows"]) def test_base_amount_not_used_as_effective_price(self): search, detail = self.pair() detail["roomStay"]["roomRates"][0]["rates"]["rate"] = [{"base": {"amountBeforeTax": 50}, "total": {"amountBeforeTax": 70}}] result = self.analyze([(search, detail)]) self.assertEqual(result["fields"]["EFFECTIVE_RATE_AMOUNT"]["records_without_candidate"], 1) def test_company_roles_not_given_priority(self): search, detail = self.pair() detail["reservationProfiles"] = {"reservationProfile": [ {"reservationProfileType": role, "profile": {"company": {"companyName": "PRIVATE_" + role}}} for role in ["Company", "TravelAgent"]]} result = self.analyze([(search, detail)]) self.assertEqual(result["fields"]["COMPANY_NAME"]["records_with_distinct_candidates"], 1) self.assertEqual(result["profile_roles"], {"Company": 1, "TravelAgent": 1}) def test_unknown_role_does_not_leak_value(self): search, detail = self.pair() detail["reservationProfiles"] = {"reservationProfile": [{"reservationProfileType": "PRIVATE_ROLE"}]} self.assertEqual(self.analyze([(search, detail)])["profile_roles"], {"Other": 1}) def test_multiple_gen_notes_and_traces_keep_ambiguity(self): search, detail = self.pair() detail["comments"] = [{"comment": {"type": "GEN", "notificationLocation": "RESERVATION", "internal": internal, "text": {"value": "PRIVATE_NOTE_" + str(internal)}}} for internal in (False, True)] detail["traces"] = [{"traceText": "PRIVATE_TRACE_A"}, {"traceText": "PRIVATE_TRACE_B"}] result = self.analyze([(search, detail)]) self.assertEqual(result["scenarios"]["records_with_multiple_nonempty_gen_notes"], 1) self.assertEqual(result["scenarios"]["records_with_internal_gen_note"], 1) self.assertEqual(result["scenarios"]["records_with_internal_and_external_gen_notes"], 1) self.assertEqual(result["scenarios"]["records_with_multiple_nonempty_traces"], 1) def test_duplicates_are_reported_not_removed_and_leading_zero_preserved(self): pairs = [self.pair() for _ in range(3)] for pair, room in zip(pairs, ("0101", "101", " 0101 ")): pair[1]["roomStay"]["currentRoomInfo"]["roomId"] = room result = self.analyze(pairs) self.assertEqual(result["source_records"], 3) self.assertEqual(result["scenarios"]["duplicate_nonblank_room_arrival_groups"], 1) self.assertEqual(result["scenarios"]["records_in_duplicate_nonblank_room_arrival_groups"], 2) def test_current_and_arrival_segment_room_conflict_visible(self): search, detail = self.pair() detail["roomStay"]["currentRoomInfo"]["roomId"] = "PRIVATE_DIFFERENT_ROOM" result = self.analyze([(search, detail)]) self.assertEqual(result["fields"]["DISP_ROOM_NO"]["records_with_distinct_candidates"], 1) self.assertEqual(result["scenarios"]["current_vs_arrival_segment_room_disagreement"], 1) def test_missing_display_name_not_silently_built_from_components(self): search, detail = self.pair() del search["reservationGuest"] detail["reservationGuests"] = [{"primary": True, "profileInfo": {"profile": {"customer": { "personName": [{"nameType": "Primary", "givenName": "PRIVATE_GIVEN", "surname": "PRIVATE_SURNAME"}]}}}}] result = self.analyze([(search, detail)]) self.assertEqual(result["fields"]["FULL_NAME"]["records_without_candidate"], 1) self.assertEqual(result["scenarios"]["records_without_unique_primary_name"], 0) self.assertEqual(result["scenarios"]["records_with_name_components_but_no_display_name"], 1) self.assertEqual(result["scenarios"]["records_with_primary_surname"], 1) self.assertEqual(result["scenarios"]["records_with_primary_givenName"], 1) if __name__ == "__main__": unittest.main()