from dataclasses import replace import hashlib import json from pathlib import Path import stat import tempfile import unittest from unittest.mock import patch from integrations.ohip import audit_arr_capture as v1audit 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 source from integrations.ohip import rate_info from tests.test_ohip_arr_collection import DAY, HOTEL, SECRET, FakeService, envelope class DayService: def __init__(self, count=3): self.base = FakeService(count) for row in self.base.rows: row["roomStay"]["roomRates"] = [{"rates": {"rate": [{"base": { "currencyCode": "USD", "amountBeforeTax": 999}}]}}] self.calls = [] self.rate_edit = lambda data, body: data def __call__(self, method, path, raw): self.calls.append((method, path, raw)) if path == rate_info.PATH + "/searches": body = json.loads(raw) data = {"detail": {"totalRateAmount": int(body["id"][2:]), "rateSuppressed": False, "revenue": {"currencyCode": "USD"}}} return 200, {}, source.json_bytes(envelope(rate_info.POST, self.rate_edit(data, body))) return self.base(method, path, raw) class DayCaptureTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.output = self.root / "capture" self.service = DayService() self.options = day.Options(DAY, DAY, DAY, HOTEL, page_size=2) def capture(self, transport=None): archive = source.Archive(self.output) transport = transport or self.service return day.collect(self.options, archive, source.Reader(archive, HOTEL, transport, key=SECRET, sleep=lambda _: None), rate_info.RateInfoReader(archive, HOTEL, transport, key=SECRET, sleep=lambda _: None)) def verify(self, result): verified = audit.VerifiedArchive(self.output, result["manifest_sha256"]) with patch.object(source, "HTTPTransport", side_effect=AssertionError("offline only")), \ patch.object(source, "load_key", side_effect=AssertionError("offline only")): return audit.replay(verified) def repin(self, name, edit): path = self.output / name document = json.loads(path.read_bytes()) edit(document) raw = source.json_bytes(document) path.write_bytes(raw) result_path = self.output / "result.json" if name != "result.json": result = json.loads(result_path.read_bytes()) for item in result["files"]: if item["name"] == name: item.update(bytes=len(raw), sha256=hashlib.sha256(raw).hexdigest()) result_path.write_bytes(source.json_bytes(result)) return {"manifest_sha256": hashlib.sha256(result_path.read_bytes()).hexdigest()} def test_complete_day_dates_identity_zero_privacy_and_offline_replay(self): result = self.capture() self.assertTrue(result["candidate_capture_complete"]) self.assertTrue(result["all_rates_valid"]) self.assertEqual((result["search_records"], result["verified_details"], result["rate_responses"]), (3, 3, 3)) self.assertEqual((result["source_http_attempts"], result["rate_http_attempts"], result["http_attempts"]), (7, 3, 10)) self.assertEqual(result["explicit_zero_rates"], 1) self.assertEqual(result["currencies"], {"USD": 3}) self.assertEqual(result["rate_issues"], {}) for field in ("finance_ready", "atomic_snapshot", "report_equivalence_verified"): self.assertFalse(result[field]) rate_requests = [json.loads(raw) for _, path, raw in self.service.calls if path.endswith("rate-info/searches")] self.assertEqual(rate_requests, [{"id": f"id{i}", "type": "Reservation", "summaryInfo": False, "detailDate": DAY} for i in range(3)]) searches = [json.loads(raw) for _, path, raw in self.service.calls if path == "/api/v1/reservations/searches"] self.assertEqual([r["offset"] for r in searches], [0, 2, 0, 2]) self.assertTrue(all(r["arrivalStartDate"] == r["arrivalEndDate"] == DAY for r in searches)) rows, details, assessments = self.verify(result) self.assertEqual(len(rows), len(details)) self.assertEqual(assessments["records"][0]["effective_rate"], "0") self.assertEqual(stat.S_IMODE(self.output.stat().st_mode), 0o700) for path in self.output.iterdir(): self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) self.assertNotIn(SECRET.encode(), path.read_bytes()) def test_system_dates_required_equal_valid_and_no_clock_default(self): # A delayed retry or backfill is valid; there is no today/yesterday gate. day.Options("2001-01-01", "2001-01-01", "2001-01-01", HOTEL).validate() for field in ("from_date", "to_date", "rate_date"): for value in (None, "", "2026-02-31", "20260915", "2026-09-14"): with self.subTest(field=field, value=value), self.assertRaises(source.CollectionError): replace(self.options, **{field: value}).validate() def test_invalid_dates_rejected_before_any_read_or_archive_write(self): self.options = replace(self.options, rate_date="2026-09-14") with self.assertRaisesRegex(source.CollectionError, "report_dates_must_match"): self.capture() self.assertEqual(self.service.calls, []) self.assertEqual(list(self.output.iterdir()), []) def test_missing_rate_is_diagnostic_full_day_kept_and_replayable(self): self.service.rate_edit = lambda data, body: {} if body["id"] == "id1" else data result = self.capture() self.assertTrue(result["candidate_capture_complete"]) self.assertFalse(result["all_rates_valid"]) self.assertFalse(result["finance_ready"]) self.assertEqual(result["valid_rate_candidates"], 2) self.assertEqual(result["rate_issues"], {"empty_rate_info": 1}) rows, details, assessments = self.verify(result) self.assertEqual(len(assessments["records"]), 3) field = audit.analyze_fields(rows, details, assessments)["fields"]["EFFECTIVE_RATE_AMOUNT"] self.assertEqual(field["records_with_candidate"], 2) self.assertEqual(field["records_without_candidate"], 1) self.assertFalse(field["report_mapping_verified"]) def test_suppressed_summary_and_invalid_amounts_are_never_zero(self): cases = [({"detail": {"rateSuppressed": True}}, "rate_suppressed"), ({"summary": {}}, "unexpected_rate_summary"), ({"detail": {"totalRateAmount": -1}}, "missing_or_invalid_effective_rate"), ({"detail": {"totalRateAmount": "0"}}, "missing_or_invalid_effective_rate")] for index, (data, code) in enumerate(cases): with self.subTest(code=code): self.output = self.root / f"case{index}" self.service.rate_edit = lambda _, body, data=data: data result = self.capture() self.assertTrue(result["candidate_capture_complete"]) self.assertFalse(result["all_rates_valid"]) self.assertEqual(result["explicit_zero_rates"], 0) self.assertEqual(result["rate_issues"], {code: 3}) self.verify(result) def test_currency_mismatch_does_not_convert_or_use_base(self): self.service.base.rows[0]["roomStay"]["roomRates"][0]["rates"]["rate"][0]["base"]["currencyCode"] = "THB" result = self.capture() self.assertEqual(result["rate_issues"], {"rate_currency_mismatch": 1}) self.assertEqual(result["valid_rate_candidates"], 2) self.verify(result) def test_missing_and_mixed_detail_currency_require_review(self): self.service.base.rows[0]["roomStay"].pop("roomRates") self.service.base.rows[1]["roomStay"]["roomRates"][0]["rates"]["rate"].append({"base": {"currencyCode": "THB"}}) result = self.capture() self.assertEqual(result["rate_issues"], {"missing_detail_rate_currency": 1, "mixed_detail_rate_currencies": 1}) self.assertEqual(result["valid_rate_candidates"], 1) self.verify(result) def test_money_decimal_bytes_preserved_through_assessment_and_replay(self): 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": 9007199254740993.01') return status, headers, response result = self.capture(transport) self.assertEqual(self.verify(result)[2]["records"][0]["effective_rate"], "9007199254740993.01") def test_no_show_cancelled_unassigned_and_missing_codes_are_not_filtered(self): for row, status in zip(self.service.base.rows, ["NoShow", "Cancelled", "InHouse"]): row["reservationStatus"] = status result = self.capture() self.assertEqual(result["verified_details"], 3) self.assertEqual(result["rate_responses"], 3) def test_detail_identity_drift_or_note_loss_stops_before_its_rate(self): for index, (change, code) in enumerate([ (lambda d: d["reservationIdList"][0].update(id="other"), "detail_identity_mismatch"), (lambda d: d.update(reservationStatus="Cancelled"), "search_detail_state_mismatch"), (lambda d: d["roomStay"].update(arrivalDate="2026-09-14"), "record_arrival_mismatch"), ]): with self.subTest(code=code): self.service = DayService() self.output = self.root / f"bad-detail-{index}" def edit(document, _, change=change): if document["operation_id"] == source.DETAIL: change(document["data"]["reservations"]["reservation"][0]) return document self.service.base.edit = edit result = self.capture() self.assertEqual(result["error"], code) self.assertEqual(result["rate_responses"], 0) self.assertFalse(result["candidate_capture_complete"]) self.service = DayService() self.output = self.root / "note-loss" self.service.base.rows[0]["reservationIndicators"] = [{"indicatorName": "COMMENT", "count": 2}] self.assertEqual(self.capture()["error"], "note_or_trace_count_mismatch") def test_final_search_drift_refuses_capture_even_after_all_prices(self): def edit(document, index): if index == 6: document["data"]["reservations"]["reservationInfo"][0]["lastModifyDateTime"] = "changed" return document self.service.base.edit = edit result = self.capture() self.assertEqual(result["error"], "source_changed_during_collection") self.assertEqual(result["valid_rate_candidates"], 3) self.assertFalse(result["candidate_capture_complete"]) self.assertFalse(result["all_rates_valid"]) with self.assertRaisesRegex(source.CollectionError, "incomplete_capture"): self.verify(result) def test_empty_day_requires_review_without_rates(self): self.service = DayService(0) result = self.capture() self.assertEqual(result["error"], "empty_source_requires_review") self.assertEqual(result["rate_responses"], 0) def test_rate_warning_wrong_hotel_or_auth_failure_stops_whole_day(self): for index, kind in enumerate(("warning", "hotel", "auth")): self.output = self.root / kind self.service = DayService() def transport(method, path, raw, kind=kind): status, headers, response = self.service(method, path, raw) if path.endswith("rate-info/searches"): document = json.loads(response) if kind == "warning": document["data"]["warnings"] = [{"code": "SYNTHETIC"}] elif kind == "hotel": document["hotel_id"] = "OTHER" else: status = 403 response = source.json_bytes(document) return status, headers, response result = self.capture(transport) self.assertFalse(result["candidate_capture_complete"]) self.assertEqual(result["rate_http_attempts"], 1) self.assertEqual(result["rate_issues"], {}) def test_transient_rate_and_source_retries_replay_without_network(self): counts = {"search": 0, "rate": 0} def transport(method, path, raw): key = "rate" if path.endswith("rate-info/searches") else "search" counts[key] += 1 if key == "rate" and counts[key] == 1: return 429, {"Retry-After": "0"}, b"{}" if key == "search" and counts[key] == 1: raise TimeoutError() return self.service(method, path, raw) result = self.capture(transport) self.assertEqual(result["http_attempts"], 12) self.assertEqual(result["rate_http_attempts"], 4) self.verify(result) def test_modified_raw_and_manifest_require_original_pin(self): result = self.capture() path = self.output / "rate-000001.response.bin" path.write_bytes(path.read_bytes() + b" ") with self.assertRaisesRegex(source.CollectionError, "archive_(file_too_large|hash_mismatch)"): self.verify(result) self.repin("result.json", lambda d: d.update(valid_rate_candidates=99)) with self.assertRaisesRegex(source.CollectionError, "manifest_hash_mismatch"): self.verify(result) def test_resealed_wrong_rate_date_or_identity_fails_protocol(self): for field, value in (("detailDate", "2026-09-14"), ("id", "other")): self.output = self.root / field self.capture() result = self.repin("rate-000001.json", lambda d: d["body"].update({field: value})) with self.assertRaisesRegex(source.CollectionError, "capture_protocol_replay_failed"): self.verify(result) def test_resealed_summary_and_assessment_do_not_override_raw(self): for file, change, code in [ ("result.json", lambda d: d.update(valid_rate_candidates=999), "capture_summary_mismatch"), ("rate-assessments.json", lambda d: d["records"][0].update(effective_rate="100"), "rate_assessment_mismatch"), ]: self.output = self.root / file self.capture() result = self.repin(file, change) with self.assertRaisesRegex(source.CollectionError, code): self.verify(result) def test_resealed_protocol_dates_operations_and_safety_claims_rejected(self): cases = [("result.json", lambda d: d.update(rate_date="2026-09-14"), "capture_date_mismatch"), ("capture.json", lambda d: d.update(operations=[source.SEARCH, source.DETAIL]), "capture_contract_mismatch"), ("result.json", lambda d: d.update(finance_ready=True), "unsupported_capture_claim"), ("capture.json", lambda d: d["options"].update(to_date="2026-09-14"), "report_dates_must_match")] for index, (file, change, code) in enumerate(cases): self.output = self.root / f"contract-{index}" self.capture() result = self.repin(file, change) with self.assertRaisesRegex(source.CollectionError, code): self.verify(result) def test_unsafe_file_permissions_and_extra_files_refused(self): result = self.capture() target = self.output / "rate-000001.response.bin" target.chmod(0o644) with self.assertRaisesRegex(source.CollectionError, "unsafe_archive_file"): self.verify(result) target.chmod(0o600) (self.output / "extra.json").write_text("{}") with self.assertRaisesRegex(source.CollectionError, "archive_file_set_mismatch"): self.verify(result) def test_v1_verifier_never_silently_accepts_v2(self): result = self.capture() with self.assertRaisesRegex(source.CollectionError, "incomplete_capture"): v1audit.VerifiedArchive(self.output, result["manifest_sha256"]) def test_field_audit_requires_rate_identity_alignment(self): rows, details, assessments = self.verify(self.capture()) assessments["records"].reverse() with self.assertRaisesRegex(source.CollectionError, "rate_assessment_identity_mismatch"): audit.analyze_fields(rows, details, assessments) if __name__ == "__main__": unittest.main()