"""Offline status evidence checks against the real synthetic ARR collector.""" import copy import hashlib import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from integrations.ohip import arr_data as data from integrations.ohip import collect_arr_source as base from integrations.ohip import reservation_status as status from tests.test_ohip_arr_data import DAY, HOTEL, REQUEST, SimulatedOHIP def digest(raw): return hashlib.sha256(raw).hexdigest() class SavedStatusEvidenceTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.service = SimulatedOHIP(2) self.capture_number = 0 self.collect() self.snapshot = {file.name: file.read_bytes() for file in self.capture.iterdir()} def collect(self): self.capture_number += 1 source = data.ARRDataSource(self.root / f"source-{self.capture_number}", HOTEL, transport_factory=lambda: self.service, sleep=lambda _: None, page_size=1) summary = source.fetch(DAY, REQUEST) self.assertTrue(summary["collection_complete"], summary) self.capture = Path(summary["data_path"]).parent self.original = (self.capture / "arr-data.json").read_bytes() self.pin = summary["manifest_sha256"] def restore(self): for file in self.capture.iterdir(): file.unlink() for name, raw in self.snapshot.items(): file = self.capture / name file.write_bytes(raw) file.chmod(0o600) self.original = self.snapshot["arr-data.json"] self.pin = digest(self.snapshot["result.json"]) self.capture.chmod(0o700) def document(self, name): return json.loads((self.capture / name).read_bytes()) def write(self, name, value): (self.capture / name).write_bytes(base.json_bytes(value)) def repin(self, *, original_changed=False): manifest = self.document("result.json") for entry in manifest["files"]: raw = (self.capture / entry["name"]).read_bytes() entry.update(bytes=len(raw), sha256=digest(raw)) manifest["data_sha256"] = digest((self.capture / "arr-data.json").read_bytes()) self.write("result.json", manifest) self.pin = digest((self.capture / "result.json").read_bytes()) if original_changed: self.original = (self.capture / "arr-data.json").read_bytes() def requests(self, operation): return sorted(file.name for file in self.capture.glob("data-request-*.json") if not file.name.endswith(".meta.json") and self.document(file.name)["operation_id"] == operation) def response(self, request): return request.removesuffix(".json") + ".response.bin" def evidence(self): return status.saved_status_evidence(self.capture, self.pin, self.original) def assert_rejected(self, code=None): with self.assertRaises(base.CollectionError) as caught: self.evidence() if code: self.assertEqual(str(caught.exception), code) def test_complete_paginated_capture_is_read_only_and_offline(self): calls = len(self.service.calls) with patch.object(base, "HTTPTransport", side_effect=AssertionError("network forbidden")), \ patch.object(base, "load_key", side_effect=AssertionError("credentials forbidden")): receipt = self.evidence() self.assertEqual(receipt["version"], status.VERSION) self.assertEqual(receipt["policy_id"], status.POLICY_ID) self.assertEqual(receipt["original_sha256"], digest(self.original)) self.assertEqual(receipt["source_manifest_sha256"], self.pin) self.assertEqual((receipt["report_date"], receipt["hotel_id"]), (DAY, HOTEL)) self.assertEqual([(r["source_sequence"], r["reservation_id"], r["reservation_status"]) for r in receipt["records"]], [(1, "res0", "InHouse"), (2, "res1", "InHouse")]) for row in receipt["records"]: self.assertEqual(len(row["sources"]), 9) self.assertEqual(len({item["file"] for item in row["sources"]}), 9) for item in row["sources"]: self.assertEqual(item["sha256"], digest((self.capture / item["file"]).read_bytes())) self.assertEqual(len(self.service.calls), calls) self.assertEqual({file.name: file.read_bytes() for file in self.capture.iterdir()}, self.snapshot) def test_legacy_original_without_status_is_supported(self): payload = self.document("arr-data.json") for row in payload["records"]: row.pop("reservation_status", None) self.write("arr-data.json", payload) self.repin(original_changed=True) self.assertEqual([r["reservation_status"] for r in self.evidence()["records"]], ["InHouse", "InHouse"]) self.assertNotIn("reservation_status", self.document("arr-data.json")["records"][0]) def test_unknown_nonempty_status_is_preserved_without_interpretation(self): for row, value in zip(self.service.rows, ("CA", "FutureOracleStatus")): row["reservationStatus"] = value self.collect() self.assertEqual([r["reservation_status"] for r in self.evidence()["records"]], ["CA", "FutureOracleStatus"]) def test_retry_failure_is_never_selected_as_evidence(self): self.service.calls.clear() self.service.status = lambda operation, count: 503 if operation == base.SEARCH and count == 1 else 200 self.collect() failed = self.response(self.requests(base.SEARCH)[0]) receipt = self.evidence() self.assertTrue(all(failed not in {item["file"] for item in row["sources"]} for row in receipt["records"])) # A source row cannot claim that failed response as its status evidence. payload = self.document("arr-data.json") payload["records"][0]["sources"].append(failed) self.write("arr-data.json", payload) self.repin(original_changed=True) self.assert_rejected("status_evidence_source_binding_failed") def test_pins_detect_manifest_raw_data_and_source_format_changes(self): for mutation, code in ( (lambda: setattr(self, "pin", "0" * 64), "status_evidence_manifest_changed"), (lambda: setattr(self, "original", self.original + b"\n"), "status_evidence_original_changed"), (lambda: (self.capture / self.response(self.requests(base.DETAIL)[0])).write_bytes(b"{}"), "status_evidence_archive_changed"), ): with self.subTest(code=code): self.restore() mutation() self.assert_rejected(code) def test_inventory_rejects_unsafe_duplicate_unlisted_and_orphan_files(self): for name in ("../escape", "/tmp/escape", "data-request-000001.json/child"): with self.subTest(name=name): self.restore() manifest = self.document("result.json") manifest["files"][0]["name"] = name self.write("result.json", manifest) self.pin = digest((self.capture / "result.json").read_bytes()) self.assert_rejected("status_evidence_inventory_name_invalid") self.restore() manifest = self.document("result.json") manifest["files"].append(copy.deepcopy(manifest["files"][0])) self.write("result.json", manifest) self.pin = digest((self.capture / "result.json").read_bytes()) self.assert_rejected("status_evidence_inventory_name_invalid") self.restore() extra = self.capture / "data-request-999999.response.bin" extra.write_bytes(b"{}") extra.chmod(0o600) self.assert_rejected("status_evidence_inventory_changed") manifest = self.document("result.json") manifest["files"].append({"name": extra.name, "bytes": 2, "sha256": digest(b"{}")}) self.write("result.json", manifest) self.pin = digest((self.capture / "result.json").read_bytes()) self.assert_rejected("status_evidence_request_inventory_invalid") def test_incomplete_manifest_or_invalid_options_rejected(self): for target, key, value in (("result.json", "collection_complete", False), ("result.json", "error", "http_failure"), ("result.json", "status", "failed"), ("result.json", "finance_ready", True), ("capture.json", "options", {"unexpected": 1})): with self.subTest(target=target, key=key): self.restore() document = self.document(target) document[key] = value self.write(target, document) self.repin() self.assert_rejected() def test_request_metadata_and_envelope_fail_closed(self): request_name = self.requests(base.DETAIL)[0] response_name = self.response(request_name) meta_name = request_name.removesuffix(".json") + ".meta.json" for target, key, value in ((meta_name, "http_status", 500), (meta_name, "error", "transport_failure"), (meta_name, "oversized", True), (response_name, "operation_id", base.SEARCH), (response_name, "hotel_id", "OTHER_HOTEL"), (response_name, "oracle_request_id", " "), (response_name, "upstream_status", 500), (response_name, "warnings", [{"message": "partial result"}]), (request_name, "method", "POST"), (request_name, "path", "/api/v1/reservations/res0"), (request_name, "path", None)): with self.subTest(target=target, key=key, value=value): self.restore() document = self.document(target) document[key] = value self.write(target, document) self.repin() self.assert_rejected() def test_typed_identity_hotel_date_and_detail_cardinality_must_match(self): response_name = self.response(self.requests(base.DETAIL)[0]) for key, value in (("reservationIdList", [{"type": "Reservation", "id": "res99"}]), ("reservationIdList", [{"type": "Confirmation", "id": "res0"}]), ("reservationIdList", [{"type": "Reservation", "id": "res0"}] * 2), ("hotelId", "OTHER_HOTEL"), ("roomStay", {"arrivalDate": "2026-09-16"})): with self.subTest(key=key, value=value): self.restore() envelope = self.document(response_name) envelope["data"]["reservations"]["reservation"][0][key] = value self.write(response_name, envelope) self.repin() self.assert_rejected() for count in (0, 2): with self.subTest(count=count): self.restore() envelope = self.document(response_name) envelope["data"]["reservations"]["reservation"] *= count self.write(response_name, envelope) self.repin() self.assert_rejected("status_evidence_detail_ambiguous") def test_conflicting_missing_empty_and_nonstring_status_rejected(self): response_name = self.response(self.requests(base.DETAIL)[0]) for value in (None, "", " ", 0, False, {}, [], "Cancelled"): with self.subTest(value=value): self.restore() envelope = self.document(response_name) envelope["data"]["reservations"]["reservation"][0]["reservationStatus"] = value self.write(response_name, envelope) self.repin() self.assert_rejected("status_evidence_status_conflict") self.restore() envelope = self.document(response_name) envelope["data"]["reservations"]["reservation"][0].pop("reservationStatus") self.write(response_name, envelope) self.repin() self.assert_rejected("status_evidence_status_conflict") def test_modified_timestamp_and_final_search_drift_rejected(self): detail_name = self.response(self.requests(base.DETAIL)[0]) envelope = self.document(detail_name) envelope["data"]["reservations"]["reservation"][0]["lastModifyDateTime"] = "later" self.write(detail_name, envelope) self.repin() self.assert_rejected("status_evidence_state_conflict") self.restore() final_name = self.response(self.requests(base.SEARCH)[-1]) envelope = self.document(final_name) envelope["data"]["reservations"]["reservationInfo"][0]["reservationStatus"] = "Cancelled" self.write(final_name, envelope) self.repin() self.assert_rejected("status_evidence_search_changed") def test_original_order_existing_status_and_source_refs_are_bound(self): for mutate in ( lambda rows: rows.reverse(), lambda rows: rows[0].update(source_sequence=2), lambda rows: rows[0].update(reservation_id="res99"), lambda rows: rows[0].update(reservation_status="Cancelled"), lambda rows: rows[0].update(sources=rows[1]["sources"]), lambda rows: rows[0]["sources"].append(self.response(self.requests(base.DETAIL)[1])), lambda rows: rows[0]["sources"].append(rows[0]["sources"][0]), ): with self.subTest(mutation=mutate): self.restore() payload = self.document("arr-data.json") mutate(payload["records"]) self.write("arr-data.json", payload) self.repin(original_changed=True) self.assert_rejected() def test_search_body_cannot_change_date_status_scope_or_pagination(self): request_name = self.requests(base.SEARCH)[0] for key, value in (("arrivalStartDate", "2026-09-16"), ("reservationStatus", ["InHouse"]), ("offset", 1), ("limit", 100)): with self.subTest(key=key): self.restore() request = self.document(request_name) request["body"][key] = value self.write(request_name, request) self.repin() self.assert_rejected() def test_archive_private_permissions_and_no_symlink_files(self): self.capture.chmod(0o755) self.assert_rejected("status_evidence_directory_unsafe") self.capture.chmod(0o700) target = self.capture / "arr-data.json" target.chmod(0o644) self.assert_rejected("unsafe_archive_file") target.chmod(0o600) outside = self.root / "outside.json" outside.write_bytes(target.read_bytes()) outside.chmod(0o600) target.unlink() target.symlink_to(outside) with self.assertRaises(OSError): self.evidence() if __name__ == "__main__": unittest.main()