252 lines
15 KiB
Python
252 lines
15 KiB
Python
"""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("<G_RESERVATION>", f"<G_RESERVATION><RESORT>LOCAL</RESORT><RESV_NAME_ID>{index}</RESV_NAME_ID>"
|
|
"<CURRENCY_CODE>THB</CURRENCY_CODE>")
|
|
row = row.replace("</G_RESERVATION>", "<LIST_G_DEPT_ID><G_DEPT_ID><TRACE_TEXT>IGNORED TRACE</TRACE_TEXT>"
|
|
"</G_DEPT_ID></LIST_G_DEPT_ID></G_RESERVATION>")
|
|
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")
|