312 lines
16 KiB
Python
312 lines
16 KiB
Python
"""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()
|