"""Loopback acquisition, independent XML validation, refusal and persistent SQL.""" from datetime import date import hashlib import json import os from pathlib import Path import tempfile import unittest from unittest.mock import patch import xml.etree.ElementTree as ET 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.arr_download_executor import CapturedARRExecutor from arr_web.local_api_fixture import HOTEL, LocalAPIAdapter, NativeBaselineValidator, build_fixture from arr_web.local_api_server import LocalTransport, serve_fixture from arr_web.local_api_simulation import create_simulation, open_simulation from arr_web.local_replay import open_instance from arr_web.local_xml_replay import NativeXMLSnapshot, document from integrations.ohip import audit_arr_day as audit, collect_arr_day as day, collect_arr_source as source, rate_info from tests.test_arr_opera_daily_ingest import reservation, xml_document from tests.test_arr_local_xml_replay import PORT, REQUEST, login from tests import test_arr_local_xml_replay as native_tests from tests.test_ohip_processing_handoff import CountingProcessor DAY = date(2026, 7, 27) def xml(*, rate="900", note=None): rows = [] for index in range(1, 4): row = reservation(index, confirmation=f"SYNTHETIC-{4-index}", rate_amount=rate, res_comment=note) row = row.replace("", f"LOCAL{index}" "THB") row = row.replace("", "IGNORED TRACE" "") rows.append(row) return xml_document(*rows).encode() class LocalAPITests(unittest.TestCase): @classmethod def setUpClass(cls): cls.policy = load_processor_policy(Path(__file__).resolve().parents[1]) def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.raw = xml() self.input = self.root / "input.xml" self.input.write_bytes(self.raw) self.snapshot = NativeXMLSnapshot.create(self.root / "fixture", self.input, hashlib.sha256(self.raw).hexdigest(), DAY.isoformat()) self.fixture = build_fixture(self.snapshot) self.adapter = LocalAPIAdapter(self.fixture["source_sha256"], 3) self.validator = NativeBaselineValidator(self.snapshot) 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) @staticmethod def readers(archive, api): return (source.Reader(archive, HOTEL, api.transport, key=api.key, sleep=lambda _: None), rate_info.RateInfoReader(archive, HOTEL, api.transport, key=api.key, sleep=lambda _: None)) def capture(self, api, suffix="capture"): archive = source.Archive(self.root / suffix) options = day.Options(DAY.isoformat(), DAY.isoformat(), DAY.isoformat(), HOTEL, page_size=2) result = day.collect(options, archive, *self.readers(archive, api)) return result, archive def executor(self, api): return CapturedARRExecutor(root=self.root / "executor", hotel_id=HOTEL, adapter_contract="local-test/v1", adapter=self.adapter, mapping_validator=self.validator, reader_factory=lambda archive, _: self.readers(archive, api), policy=self.policy, object_store=self.store, repository=self.repository, ingestion=self.ingestion, processor=self.processor, page_size=2) @staticmethod def execute(executor): return executor.execute(request_id=REQUEST, from_date=DAY, to_date=DAY, report_stage=lambda _: None) def test_real_http_paging_decimal_notes_source_order_and_baseline(self): self.raw = xml(rate="900.0123456789", note="SYNTHETIC GEN NOTE preserved") self.input.write_bytes(self.raw) self.snapshot = NativeXMLSnapshot.create(self.root / "decimal-fixture", self.input, hashlib.sha256(self.raw).hexdigest(), DAY.isoformat()) self.fixture = build_fixture(self.snapshot) self.adapter = LocalAPIAdapter(self.fixture["source_sha256"], 3) self.validator = NativeBaselineValidator(self.snapshot) with serve_fixture(self.fixture) as api: result, archive = self.capture(api) self.assertTrue(result["candidate_capture_complete"]) self.assertEqual(dict(api.counts), {source.SEARCH: 4, source.DETAIL: 3, rate_info.POST: 3}) verified = audit.VerifiedArchive(archive.path, result["manifest_sha256"]) rows, _, _ = audit.replay(verified) self.assertEqual(source.reservation_id(rows[0]), "local000003") output = self.adapter.adapt(verified) self.validator.validate(verified, output) parsed = ET.fromstring(output) self.assertEqual(parsed.findtext(".//RESV_NAME_ID"), "local000001") self.assertEqual(parsed.findtext(".//EFFECTIVE_RATE_AMOUNT"), "900.0123456789") self.assertIn("\r", parsed.findtext(".//RES_COMMENT")) self.assertFalse(parsed.findall(".//TRACE_TEXT")) self.assertEqual(self.snapshot.payload(), self.raw) with self.assertRaises(source.CollectionError): self.validator.validate(verified, output.replace(b"900.0123456789", b"1", 1)) def test_one_503_recovers_and_permanent_503_stops_after_three_without_finance(self): with serve_fixture(self.fixture) as api: api.test_reply = lambda operation, count, _: (503, b"{}") if operation == source.DETAIL and count == 1 else None result, _ = self.capture(api) self.assertTrue(result["candidate_capture_complete"]) self.assertEqual(api.counts[source.DETAIL], 4) with serve_fixture(self.fixture) as api: api.test_reply = lambda operation, *_: (503, b"{}") if operation == source.DETAIL else None outcome = self.execute(self.executor(api)) self.assertEqual(outcome.status, "failed") self.assertEqual(api.counts[source.DETAIL], 3) self.assertEqual(self.processor.calls, 0) self.assertFalse(self.repository._versions) def test_identity_and_pagination_and_final_search_drift_are_refused(self): for scenario in ("identity", "pagination", "drift"): with self.subTest(scenario=scenario), serve_fixture(self.fixture) as api: def change(operation, count, envelope): if scenario == "identity" and operation == source.DETAIL: envelope["data"]["reservations"]["reservation"][0]["reservationIdList"][0]["id"] = "foreign" elif scenario == "pagination" and operation == source.SEARCH: envelope["data"]["reservations"]["offset"] = 0 elif scenario == "drift" and operation == source.SEARCH and count == 3: envelope["data"]["reservations"]["reservationInfo"][0]["lastModifyDateTime"] = "changed" else: return None return 200, json.dumps(envelope).encode() api.test_reply = change result, _ = self.capture(api, scenario) self.assertFalse(result["candidate_capture_complete"]) def test_local_marker_missing_and_price_missing_cannot_reach_processor(self): for scenario in ("marker", "price"): with self.subTest(scenario=scenario), serve_fixture(self.fixture) as api: def change(operation, _, envelope): if scenario == "marker" and operation == source.DETAIL: envelope.pop("__localSimulation") elif scenario == "price" and operation == rate_info.POST: envelope["data"]["detail"].pop("totalRateAmount") else: return None return 200, json.dumps(envelope).encode() api.test_reply = change result, archive = self.capture(api, scenario) self.assertTrue(result["candidate_capture_complete"]) with self.assertRaises(source.CollectionError): self.adapter.adapt(audit.VerifiedArchive(archive.path, result["manifest_sha256"])) self.assertEqual(self.processor.calls, 0) def test_fixture_field_corruption_is_rejected_by_independent_native_baseline(self): self.fixture["records"][0]["detail"]["__localARRReport"]["full_name"] = "SYNTHETIC WRONG NAME" with serve_fixture(self.fixture) as api, self.assertRaises(source.CollectionError): self.execute(self.executor(api)) self.assertEqual(self.processor.calls, 0) self.assertFalse(self.repository._versions) def test_unknown_commit_ack_reuses_archive_and_one_version(self): with serve_fixture(self.fixture) as api: executor = self.executor(api) original = self.ingestion.ingest def lose_ack(*args, **kwargs): original(*args, **kwargs) raise ConnectionError("synthetic_unknown_commit") with patch.object(self.ingestion, "ingest", side_effect=lose_ack), self.assertRaises(ConnectionError): self.execute(executor) counts = dict(api.counts) self.assertEqual(self.execute(executor).status, "succeeded") self.assertEqual(dict(api.counts), counts) self.assertEqual(len(self.repository._versions), 1) self.assertEqual(self.processor.calls, 1) def test_local_auth_origin_pin_wrong_date_and_proxy_boundary(self): with patch.dict(os.environ, {"http_proxy": "http://invalid.example:1", "https_proxy": "http://invalid.example:1"}): with serve_fixture(self.fixture) as api: body = source.json_bytes({"arrivalStartDate": "2026-07-28", "arrivalEndDate": "2026-07-28", "offset": 0, "limit": 2, "orderBy": ["ConfirmationNo"], "sortOrder": ["Asc"]}) self.assertEqual(api.transport("POST", "/api/v1/reservations/searches", body)[0], 400) wrong = LocalTransport(int(api.transport.authority.rsplit(":", 1)[1]), "wrong", self.fixture["source_sha256"]) self.assertEqual(wrong("POST", "/api/v1/reservations/searches", body)[0], 403) api.transport.source_sha256 = "0" * 64 with self.assertRaises(source.CollectionError): api.transport("POST", "/api/v1/reservations/searches", body) with self.assertRaises(source.CollectionError): api.transport("POST", "https://invalid.example/", body) self.assertFalse(api.counts) def test_native_entrypoint_cannot_open_simulation_and_tampering_precedes_sql(self): root = create_simulation(self.root, self.input, hashlib.sha256(self.raw).hexdigest(), DAY.isoformat()) with self.assertRaisesRegex(ValueError, "mode_mismatch"): with open_instance(root, PORT): self.fail("simulation opened as native replay") (root / "simulation-data.json").write_text("{}") with self.assertRaises(source.CollectionError): with open_simulation(root, PORT): self.fail("modified fixture accepted") self.assertFalse((root / "postgres").exists()) @unittest.skipUnless(os.environ.get("ARR_TEST_LOCAL_POSTGRES") == "1", "owned SQL opt-in required") class LocalAPISQLTests(unittest.TestCase): def test_button_to_sql_monthly_restart_and_isolated_cookie(self): with tempfile.TemporaryDirectory() as temporary: parent = Path(temporary) source_path = parent / "synthetic.xml" source_path.write_bytes(xml()) root = create_simulation(parent, source_path, hashlib.sha256(xml()).hexdigest(), DAY.isoformat()) with open_simulation(root, PORT) as runtime: headers = login(runtime.app, document(root / "login.json")) self.assertTrue(headers["Cookie"].startswith("arr_api_simulation_session=")) wrong_cookie = {**headers, "Cookie": headers["Cookie"].replace("arr_api_simulation_session", "arr_xml_replay_session")} self.assertEqual(runtime.app.handle("GET", "/api/session", wrong_cookie).status, 401) config = json.loads(runtime.app.handle("GET", "/api/arr-downloads", headers).body)["data"] self.assertEqual(config["source_kind"], "local_api_simulation") self.assertFalse(config["oracle_connected"]) invalid = runtime.app.handle("POST", "/api/arr-downloads", headers, json.dumps({"request_id": "b" * 32, "report_date": "2026-07-28"}).encode()) self.assertEqual(invalid.status, 409) response = runtime.app.handle("POST", "/api/arr-downloads", headers, json.dumps({"request_id": REQUEST, "report_date": DAY.isoformat()}).encode()) self.assertEqual(response.status, 202) result = native_tests.ReplaySQLTests.wait(runtime.queue) self.assertEqual(result["status"], "succeeded") daily = runtime.app.handle("GET", "/api/download/daily?job_id=" + result["job_id"], headers) self.assertEqual(daily.status, 200) self.assertIn("LOCAL-API-SIMULATION-", daily.headers["Content-Disposition"]) self.assertEqual(runtime.worker.process_next().status, "published") with runtime.database.connect() as connection: self.assertEqual(connection.execute("SELECT count(*) FROM finance.daily_versions").fetchone()[0], 1) self.assertEqual(connection.execute("SELECT count(*) FROM reporting.monthly_runs").fetchone()[0], 1) self.assertEqual(sum(runtime.local_api.counts.values()), 8) with open_simulation(root, PORT) as runtime: runtime.queue.create(DAY.isoformat(), REQUEST) self.assertEqual(runtime.queue.get(REQUEST)["status"], "succeeded") self.assertFalse(runtime.local_api.counts) self.assertEqual(runtime.worker.process_next().status, "idle")