"""Real orchestration with synthetic transport and XML; not source acceptance.""" from dataclasses import replace from datetime import date, datetime import hashlib import json from pathlib import Path import tempfile import time import unittest from unittest.mock import patch from arr_ingestion.repository import InMemoryIngestionRepository from arr_ingestion.service import IngestionService from arr_ingestion.validation import DeliveryValidator from arr_processing.policy import load_processor_policy from arr_storage.filesystem import FilesystemObjectBackend from arr_storage.store import ManagedObjectStore from arr_web.app import PortalApplication from arr_web.arr_downloads import PersistentARRDownloads from arr_web import arr_download_executor as execution from integrations.ohip import collect_arr_source as source from integrations.ohip.rate_info import RateInfoReader from tests.test_arr_web import TEST_CREDENTIALS, login from tests.test_arr_web_download_handoff import FixtureExecutor, REQUEST_ID, DAY as REPORT_DATE from tests.test_ohip_day_capture import DayService, DAY, HOTEL from tests.test_ohip_processing_handoff import CountingProcessor class FixtureAdapter: """Fixed synthetic XML, deliberately not an API-to-report mapping rule.""" def __init__(self): self.calls = 0 self.payload = FixtureExecutor.xml() def adapt(self, archive): self.calls += 1 return self.payload class FixtureValidator: """Call/refusal probe; actual mapping validator remains unimplemented.""" def __init__(self): self.calls = 0 self.refuse = False def validate(self, archive, payload): self.calls += 1 if self.refuse: raise ValueError("synthetic mapping rejection") class CaptureExecutorTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.policy = load_processor_policy(Path(__file__).resolve().parents[1]) def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.repository = InMemoryIngestionRepository() self.store = ManagedObjectStore(FilesystemObjectBackend(self.root / "objects", create=True)) self.ingestion = IngestionService(DeliveryValidator(self.store, self.policy), self.repository) self.processor = CountingProcessor(self.policy) self.adapter, self.validator = FixtureAdapter(), FixtureValidator() self.transport = DayService() self.factory_calls = 0 self.stages = [] self.executor = self.make_executor() def factory(self, archive, hotel): self.factory_calls += 1 return (source.Reader(archive, hotel, self.transport, sleep=lambda _: None), RateInfoReader(archive, hotel, self.transport, sleep=lambda _: None)) def make_executor(self, **changes): args = dict(root=self.root / "executor", hotel_id=HOTEL, adapter_contract="synthetic-only/v1", adapter=self.adapter, mapping_validator=self.validator, reader_factory=self.factory, policy=self.policy, object_store=self.store, repository=self.repository, ingestion=self.ingestion, processor=self.processor) args.update(changes) return execution.CapturedARRExecutor(**args) def run_executor(self, executor=None, **changes): args = dict(request_id=REQUEST_ID, from_date=REPORT_DATE, to_date=REPORT_DATE, report_stage=self.stages.append) args.update(changes) return (executor or self.executor).execute(**args) def checkpoint(self): return self.executor.root / "requests" / REQUEST_ID / "prepared.json" def hashes(self, folder): return {str(p.relative_to(folder)): hashlib.sha256(p.read_bytes()).hexdigest() for p in folder.rglob("*") if p.is_file()} def lose_commit_ack(self): original = self.ingestion.ingest def lost(raw): original(raw) raise ConnectionError("synthetic lost commit response") with patch.object(self.ingestion, "ingest", side_effect=lost), self.assertRaises(ConnectionError): self.run_executor() def test_web_task_runs_capture_and_handoff_with_explicit_equal_dates(self): queue = PersistentARRDownloads(self.root / "queue", self.executor) try: app = PortalApplication(arr_downloads=queue, credentials=TEST_CREDENTIALS) _, headers = login(app) response = app.handle("POST", "/api/arr-downloads", headers, json.dumps({"request_id": REQUEST_ID, "report_date": DAY}).encode()) self.assertEqual(response.status, 202) deadline = time.monotonic() + 15 while queue.get(REQUEST_ID)["status"] in {"queued", "downloading", "processing"}: self.assertLess(time.monotonic(), deadline) time.sleep(.01) self.assertEqual(queue.get(REQUEST_ID)["status"], "succeeded") finally: queue.close() searches = [json.loads(raw) for _, path, raw in self.transport.calls if path.endswith("reservations/searches")] rates = [json.loads(raw) for _, path, raw in self.transport.calls if path.endswith("rate-info/searches")] self.assertTrue(searches and rates) self.assertTrue(all(p["arrivalStartDate"] == p["arrivalEndDate"] == DAY for p in searches)) self.assertTrue(all(p["detailDate"] == DAY for p in rates)) prepared = json.loads(self.checkpoint().read_bytes()) self.assertEqual(prepared["binding"]["batch_id"], "web-" + REQUEST_ID) self.assertEqual(prepared["binding"]["arrival_date"], DAY) self.assertEqual((self.factory_calls, self.adapter.calls, self.validator.calls, self.processor.calls), (1, 1, 1, 1)) self.assertEqual(len(self.repository._versions), 1) for p in self.executor.root.rglob("*"): self.assertEqual(p.stat().st_mode & 0o777, 0o700 if p.is_dir() else 0o600) def test_lost_commit_ack_restarts_without_capture_adapter_or_reprocessing(self): self.lose_commit_ack() self.assertTrue(self.checkpoint().exists()) captured = self.hashes(self.executor.root / "captures") frozen = self.hashes(next((self.executor.root / "handoffs").iterdir()) / "frozen") restarted = self.make_executor(reader_factory=lambda *_: self.fail("must not recapture")) self.stages.clear() with patch.object(self.adapter, "adapt", side_effect=AssertionError("must not readapt")), \ patch.object(self.validator, "validate", side_effect=AssertionError("must not remap")): result = self.run_executor(restarted) self.assertEqual(result.status, "succeeded") self.assertEqual(self.stages, ["processing"]) self.assertEqual(self.hashes(self.executor.root / "captures"), captured) self.assertEqual(self.hashes(next((self.executor.root / "handoffs").iterdir()) / "frozen"), frozen) self.assertEqual((len(self.repository._jobs), len(self.repository._callbacks), len(self.repository._versions)), (1, 1, 1)) self.assertEqual(self.processor.calls, 1) def test_mapping_refusal_prevents_preparation_and_retry_reuses_capture(self): self.validator.refuse = True with self.assertRaisesRegex(ValueError, "mapping rejection"): self.run_executor() self.assertFalse(self.checkpoint().exists()) self.assertFalse(self.repository._jobs) self.assertEqual(self.processor.calls, 0) with patch.object(self.validator, "validate", return_value=False), self.assertRaisesRegex( source.CollectionError, "invalid_mapping_validation_result" ): self.run_executor() self.validator.refuse = False self.assertEqual(self.run_executor().status, "succeeded") self.assertEqual(self.factory_calls, 1) def test_capture_failure_preserves_attempt_and_retries_same_batch(self): transport = self.transport self.transport = lambda method, path, raw: (403, {}, b"{}") if path.endswith("rate-info/searches") else transport(method, path, raw) failed = self.run_executor() self.assertEqual((failed.status, failed.retryable), ("failed", True)) self.assertEqual(self.adapter.calls, 0) self.assertFalse(self.repository._jobs) first = self.executor.root / "captures" / ("web-" + REQUEST_ID) / "attempt-0001" before = self.hashes(first) self.transport = transport self.assertEqual(self.run_executor().status, "succeeded") self.assertEqual(self.hashes(first), before) self.assertTrue((first.parent / "attempt-0002").is_dir()) self.assertEqual(len(self.repository._versions), 1) def test_crash_before_prepared_checkpoint_reuses_frozen_package(self): publish = execution.atomic_json def crash(path, document, **kwargs): if path.name == "prepared.json": raise OSError("synthetic checkpoint failure") return publish(path, document, **kwargs) with patch.object(execution, "atomic_json", side_effect=crash), self.assertRaises(OSError): self.run_executor() self.assertFalse(self.checkpoint().exists()) self.assertFalse(self.repository._jobs) self.assertEqual(self.processor.calls, 1) self.assertEqual(self.run_executor().status, "succeeded") self.assertEqual((self.factory_calls, self.processor.calls), (1, 1)) def test_tampered_package_pin_is_rejected_before_delivery_writes(self): self.lose_commit_ack() checkpoint = json.loads(self.checkpoint().read_bytes()) checkpoint["manifest_sha256"] = "0" * 64 self.checkpoint().write_text(json.dumps(checkpoint)) with patch.object(self.store, "upload_committed", side_effect=AssertionError("unexpected write")) as upload, \ self.assertRaisesRegex(source.CollectionError, "prepared_pin_mismatch"): self.run_executor() upload.assert_not_called() self.assertEqual(self.factory_calls, 1) self.assertEqual(len(self.repository._versions), 1) def test_prepared_capture_binding_mismatch_is_rejected_before_delivery_writes(self): # Stop at the first delivery boundary; the prepared checkpoint exists, # but no object or Finance write has happened yet. with patch.object(execution.handoff, "deliver", side_effect=ConnectionError("before delivery")), \ self.assertRaises(ConnectionError): self.run_executor() self.assertTrue(self.checkpoint().exists()) self.assertFalse(self.repository._jobs) checkpoint = json.loads(self.checkpoint().read_bytes()) checkpoint["binding"]["manifest_sha256"] = "b" * 64 self.checkpoint().write_text(json.dumps(checkpoint)) with patch.object(self.store, "upload_committed", side_effect=AssertionError("unexpected write")) as upload, \ self.assertRaisesRegex(source.CollectionError, "handoff_expected_binding_mismatch"): self.run_executor() upload.assert_not_called() self.assertFalse(self.repository._jobs) self.assertEqual((self.factory_calls, self.processor.calls), (1, 1)) def test_changed_context_refused_without_new_capture_or_commit(self): self.run_executor() for changes in ({"hotel_id": "OTHER"}, {"adapter_contract": "synthetic-only/v2"}, {"policy": replace(self.policy, rule_set_sha256="c" * 64)}): with self.subTest(changes=changes), self.assertRaisesRegex(source.CollectionError, "executor_request_conflict"): self.run_executor(self.make_executor(**changes)) with self.assertRaisesRegex(source.CollectionError, "executor_request_conflict"): self.run_executor(from_date=date(2026, 9, 16), to_date=date(2026, 9, 16)) self.assertEqual((self.factory_calls, self.processor.calls, len(self.repository._versions)), (1, 1, 1)) def test_missing_adapter_or_validator_does_not_create_runtime_state(self): for changes in ({"adapter": None}, {"mapping_validator": None}): with self.subTest(changes=changes), self.assertRaisesRegex(ValueError, "adapter and mapping validator"): self.make_executor(**changes) self.assertFalse(self.executor.root.exists()) self.assertEqual(self.factory_calls, 0) def test_invalid_collection_limits_refused_before_state_or_factory(self): for field, maximum in (("page_size", 100), ("max_pages", 100), ("max_records", 10000)): for value in (None, True, False, 0, -1, 1.0, "2", maximum + 1): with self.subTest(field=field, value=value), self.assertRaisesRegex( source.CollectionError, "invalid_collection_limit" ): self.make_executor(**{field: value}) self.assertFalse(self.executor.root.exists()) self.assertEqual(self.factory_calls, 0) def test_collection_bounds_are_enforced_and_frozen_on_failed_retry(self): for index, (limits, error) in enumerate([ ({"page_size": 2, "max_pages": 1, "max_records": 3}, "page_limit_exceeded"), ({"page_size": 2, "max_pages": 2, "max_records": 3}, "record_limit_exceeded"), ]): with self.subTest(error=error): # Four rows exceed the record budget in the second case. The # first keeps three so it reaches the page budget instead. self.transport = type(self.transport)(count=3 + index) executor = self.make_executor(root=self.root / f"bounded-{index}", **limits) failed = self.run_executor(executor) self.assertEqual((failed.status, failed.retryable), ("failed", True)) self.assertEqual(len(self.transport.calls), 1) self.assertTrue(self.transport.calls[0][1].endswith("reservations/searches")) self.assertEqual(json.loads(self.transport.calls[0][2])["limit"], 2) job = executor.root / "captures" / ("web-" + REQUEST_ID) state = json.loads((job / "state.json").read_bytes()) self.assertEqual(state["attempts"][0]["error"], error) before = self.hashes(executor.root) for field in limits: changed = dict(limits) changed[field] += 1 with self.assertRaisesRegex(source.CollectionError, "executor_request_conflict"): self.run_executor(self.make_executor(root=executor.root, **changed)) self.assertEqual(before, self.hashes(executor.root)) self.assertEqual((self.factory_calls, self.adapter.calls, self.validator.calls, self.processor.calls), (2, 0, 0, 0)) self.assertFalse(self.repository._jobs) def test_explicit_collection_limits_reach_capture_and_mapping(self): self.executor = self.make_executor(page_size=2, max_pages=2, max_records=3) seen = [] original = self.adapter.adapt def inspect(archive): seen.append((archive.options.page_size, archive.options.max_pages, archive.options.max_records)) return original(archive) with patch.object(self.adapter, "adapt", side_effect=inspect): self.assertEqual(self.run_executor().status, "succeeded") self.assertEqual(seen, [(2, 2, 3)]) searches = [json.loads(raw) for _, path, raw in self.transport.calls if path.endswith("reservations/searches")] self.assertEqual([body["offset"] for body in searches], [0, 2, 0, 2]) self.assertTrue(all(body["limit"] == 2 for body in searches)) def test_invalid_or_range_dates_rejected_before_state_and_capture(self): for changes in ({"from_date": None}, {"to_date": date(2026, 9, 16)}, {"from_date": DAY}, {"from_date": datetime(2026, 9, 15)}): with self.subTest(changes=changes), self.assertRaises(ValueError): self.run_executor(**changes) self.assertFalse(self.executor.root.exists()) self.assertEqual(self.factory_calls, 0) def test_adapted_wrong_day_rejected_before_any_business_write(self): self.adapter.payload = self.adapter.payload.replace(b"20260915", b"20260916").replace(b"15-09-26", b"16-09-26") with self.assertRaisesRegex(source.CollectionError, "handoff_date_mismatch"): self.run_executor() self.assertFalse(self.repository._jobs) self.assertFalse(self.checkpoint().exists()) if __name__ == "__main__": unittest.main()