"""Offline acceptance of direct OHIP data. All hotel values are synthetic.""" import copy from datetime import date from decimal import Decimal import hashlib import json import os from pathlib import Path import tempfile import time import unittest from unittest.mock import patch from openpyxl import load_workbook from arr_ingestion.contracts import DeliveryEnvelope, IngestionError from arr_ingestion.postgres import DatabaseConfig, PostgresIngestionRepository from arr_ingestion.repository import InMemoryIngestionRepository from arr_ingestion.service import IngestionService from arr_ingestion.validation import DeliveryValidator from arr_processing.local import LocalDailyProcessor 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_data_executor import DirectARRExecutor from arr_web.arr_downloads import PersistentARRDownloads from arr_web.programmatic import ProgrammaticUploadCoordinator from arr_web.repository import PostgresPortalRepository from tests.local_postgres import TemporaryPostgres from tests.test_arr_web import TEST_CREDENTIALS, login from tests import test_arr_download_postgres_integration as pg_helpers from tests.test_ohip_arr_data import SimulatedOHIP, DAY, HOTEL, REQUEST, core from integrations.ohip.arr_data import ARRDataSource PROJECT = Path(__file__).resolve().parents[1] class DirectDataTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.policy = load_processor_policy(PROJECT) def setUp(self): temporary = tempfile.TemporaryDirectory(prefix="arr-direct-test-") self.addCleanup(temporary.cleanup) self.files = Path(temporary.name) self.transport = SimulatedOHIP(6) self.source = ARRDataSource(self.files / "source", HOTEL, transport_factory=lambda: self.transport) self.store = ManagedObjectStore(FilesystemObjectBackend(self.files / "objects", create=True)) self.repository = InMemoryIngestionRepository() self.configure() def configure(self): self.processor = LocalDailyProcessor(self.policy) self.ingestion = IngestionService(DeliveryValidator(self.store, self.policy), self.repository) self.executor = DirectARRExecutor(root=self.files / "executor", source=self.source, policy=self.policy, object_store=self.store, repository=self.repository, ingestion=self.ingestion, processor=self.processor) self.coordinator = ProgrammaticUploadCoordinator(self.store, self.repository, self.ingestion, self.processor, self.policy.processor_version, self.policy.rule_set_sha256) def execute(self, request=REQUEST): return self.executor.execute(request_id=request, from_date=date.fromisoformat(DAY), to_date=date.fromisoformat(DAY), report_stage=lambda _: None) def collected(self): result = self.source.fetch(DAY, REQUEST) return Path(result["data_path"]) def test_direct_finance_total_and_frozen_replay_without_xml_or_new_calls(self): with patch.object(core.ET, "Element", side_effect=AssertionError("no synthetic XML")): outcome = self.execute() self.assertEqual(outcome.status, "succeeded") version = self.repository.current_version_id(date.fromisoformat(DAY)) rows = self.repository.version_records(version) self.assertEqual(len(rows), 6) self.assertEqual(sum(Decimal(str(row["total_price"])) for row in rows), Decimal("18200")) self.assertFalse(list(self.files.rglob("*.xml"))) raw = next((self.files / "executor").rglob("delivery.json")).read_bytes() envelope = DeliveryEnvelope.from_dict(json.loads(raw)) self.assertEqual(envelope.source.role, "source_data") self.assertEqual(envelope.result_schema_version, "5.0") calls = len(self.transport.calls) self.assertEqual(self.execute(), outcome) self.assertEqual(len(self.transport.calls), calls) self.assertEqual(len(self.repository._versions), 1) # A JSON source cannot be relabelled as the legacy XML contract. invalid = json.loads(raw); invalid["result_schema_version"] = "4.0" with self.assertRaises(IngestionError): DeliveryEnvelope.from_dict(invalid) def make_missing_price(self): old = self.transport.edit def edit(operation, document, request): document = old(operation, document, request) if operation == "searchRateInfo": # All six price observations use an exact, deliberately unknown key. def change(node): if isinstance(node, dict): for key, value in node.items(): if key == "totalRateAmount": node[key] = 8765 else: change(value) elif isinstance(node, list): for value in node: change(value) change(document) return document self.transport.edit = edit def finish_review(self, outcome): review = self.coordinator.get_price_review(outcome.job_id, 50, 0) for item in review["items"]: review = self.coordinator.update_price_review_item(outcome.job_id, item["item_id"], review["case_id"], review["revision"], "100", "local-test@example.test") return self.coordinator.finalize_price_review(outcome.job_id, review["case_id"], review["revision"], "local-test@example.test") def test_missing_price_replays_original_json_with_manual_prices(self): self.make_missing_price() outcome = self.execute() self.assertEqual(outcome.status, "needs_review") self.assertIsNone(self.repository.current_version_id(date.fromisoformat(DAY))) source = next((self.files / "executor").rglob("source.json")) before = source.read_bytes() calls = len(self.transport.calls) result = self.finish_review(outcome) self.assertEqual(result["status"], "succeeded") self.assertEqual(source.read_bytes(), before) self.assertEqual(len(self.transport.calls), calls) self.assertFalse(list(self.files.rglob("*.xml"))) def test_field_gap_is_failure_not_price_review_and_excluded_rows_keep_same_rule(self): path = self.collected(); data = json.loads(path.read_text()) data["records"][0]["fields"]["DISP_ROOM_NO"] = {"state": "ambiguous", "value": None} path.write_text(json.dumps(data)) output = self.processor.run(path, self.files / "failed") self.assertEqual(output.status, "failed") self.assertNotIn("daily_report", output.artifacts) data["records"][0]["fields"]["RATE_CODE"] = {"state": "available", "value": "NOT-WHITELISTED"} path.write_text(json.dumps(data)) output = self.processor.run(path, self.files / "excluded") self.assertEqual(output.status, "success") result = json.loads(output.artifacts["structured_result_json"].read_text()) self.assertEqual(result["removed_by_rate_code"], 1) self.assertEqual(result["output_rows"], 5) def test_duplicate_room_and_date_keeps_source_order(self): path = self.collected(); data = json.loads(path.read_text()) data["records"][1]["fields"]["DISP_ROOM_NO"] = copy.deepcopy(data["records"][0]["fields"]["DISP_ROOM_NO"]) path.write_text(json.dumps(data)) output = self.processor.run(path, self.files / "out") self.assertEqual(output.status, "success") result = json.loads(output.artifacts["structured_result_json"].read_text()) self.assertEqual(result["removed_as_duplicates"], 1) self.assertEqual(result["records"][1]["duplicate_of_source_sequence"], 1) def test_source_parser_rejects_incomplete_mixed_types_and_duplicates(self): path = self.collected(); data = json.loads(path.read_text()) variants = [] one = copy.deepcopy(data); one["collection_complete"] = False; variants.append(one) one = copy.deepcopy(data); one["records"][0]["fields"]["ADULTS"]["value"] = True; variants.append(one) one = copy.deepcopy(data); one["records"][0]["source_sequence"] = True; variants.append(one) one = copy.deepcopy(data); one["records"][1]["reservation_id"] = one["records"][0]["reservation_id"]; variants.append(one) one = copy.deepcopy(data); one["records"][0]["fields"]["PRODUCTS"] = {"state":"empty", "value":["unexpected"]}; variants.append(one) for variant in variants: path.write_text(json.dumps(variant)) with self.assertRaises(core.ProcessingFailure): core.read_data_source(path) path.write_text('{"version":"x","version":"y"}') with self.assertRaises(core.ProcessingFailure): core.read_data_source(path) def test_notes_packages_and_empty_values_have_explicit_display_rules(self): path = self.collected(); data = json.loads(path.read_text()) fields = data["records"][0]["fields"] fields["RES_COMMENT"] = {"state": "available", "value": ["😀", " GROUP-A ", "second note"]} fields["PRODUCTS"] = {"state": "available", "value": [ {"package": {"packageCode": "PKG1", "schedule": ["preserved"]}}, {"package": {"packageCode": "PKG1"}}, {"package": {"packageCode": "PKG2"}}]} path.write_text(json.dumps(data)); before = path.read_bytes(); cleanup = {} with patch.object(core, "read_xml", side_effect=AssertionError("must not read XML")): _, records = core.read_source(path, input_cleanup=cleanup) self.assertEqual(records[0]["RES_COMMENT"], "GROUP-A") self.assertEqual(records[0]["PRODUCTS"], "PKG1, PKG1, PKG2") self.assertEqual(records[1]["PRODUCTS"], "") self.assertEqual(records[0]["TRACE_TEXT"], "") self.assertEqual(cleanup["ignored_emoji_count"], 1) self.assertEqual(path.read_bytes(), before) def test_runtime_starts_without_fetching_and_pins_configuration(self): from types import SimpleNamespace from arr_web.arr_data_runtime import DirectARRSource, compose_direct_arr_downloads runtime = SimpleNamespace(policy=self.policy, object_store=self.store, repository=self.repository, ingestion=self.ingestion, processor=self.processor) source = DirectARRSource(self.files / "runtime", HOTEL, transport_factory=lambda: self.fail("construction must not fetch")) queue = compose_direct_arr_downloads(source, runtime); queue.close(wait=True) changed = DirectARRSource(self.files / "runtime", HOTEL + "2", transport_factory=lambda: self.transport) with self.assertRaises(Exception): compose_direct_arr_downloads(changed, runtime) def test_independent_replay_rejects_a_changed_report(self): outcome = self.execute(); self.assertEqual(outcome.status, "succeeded") frozen = next((self.files / "executor").rglob("delivery.json")).parent source = frozen / "artifacts" / "source.json" daily = frozen / "artifacts" / "daily-report.xlsx" workbook = load_workbook(daily); sheet = workbook[workbook.sheetnames[0]] sheet.cell(2, 18, 999999); workbook.save(daily); workbook.close() import subprocess command = [self.policy.python_binary, str(PROJECT / "arr-opera-daily-ingest/scripts/validate_daily.py"), "--data-json", str(source), "--daily", str(daily), "--result-json", str(frozen / "artifacts/result.json"), "--structured-result-json", str(frozen / "artifacts/structured-result.json"), "--price-reference", str(PROJECT / "arr-opera-daily-ingest/references/价格对照.xlsx")] self.assertNotEqual(subprocess.run(command, capture_output=True).returncode, 0) with self.assertRaises(Exception): self.execute() @unittest.skipUnless(os.environ.get("ARR_TEST_LOCAL_POSTGRES") == "1", "explicit disposable PostgreSQL opt-in required") class DirectDataPostgresTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.database = TemporaryPostgres().__enter__() cls.addClassCleanup(cls.database.__exit__, None, None, None) cls.policy = load_processor_policy(PROJECT) def setUp(self): self.database.reset_database() DirectDataTests.setUp(self) self.repository = PostgresIngestionRepository(DatabaseConfig("owned-fixture"), connect=self.database.connect) self.configure() configure = DirectDataTests.configure execute = DirectDataTests.execute make_missing_price = DirectDataTests.make_missing_price finish_review = DirectDataTests.finish_review sql = pg_helpers.PostgresDownloadIntegrationTests.sql count = pg_helpers.PostgresDownloadIntegrationTests.count monthly_worker = pg_helpers.PostgresDownloadIntegrationTests.monthly_worker def start_web(self): queue = PersistentARRDownloads(self.files / "queue", self.executor) self.addCleanup(queue.close) app = PortalApplication(arr_downloads=queue, credentials=TEST_CREDENTIALS, repository=PostgresPortalRepository("owned-fixture", connect=self.database.connect), upload=self.coordinator, price_reviews=self.coordinator) _, headers = login(app) return queue, app, headers def wait_task(self, queue): deadline = time.monotonic() + 40 while True: task = queue.get(REQUEST) if task["status"] not in {"queued", "downloading", "processing"}: return task self.assertLess(time.monotonic(), deadline) time.sleep(.01) def test_page_date_to_daily_and_monthly_and_repeat(self): queue, app, headers = self.start_web() body = json.dumps({"request_id": REQUEST, "report_date": DAY}).encode() self.assertEqual(app.handle("POST", "/api/arr-downloads", headers, body).status, 202) task = self.wait_task(queue) self.assertEqual(task["status"], "succeeded", task) self.assertEqual(self.sql("SELECT sum(total_price) FROM finance.daily_records")[0][0], Decimal("18200")) self.assertEqual(self.count("finance.daily_versions"), 1) self.assertEqual(self.sql("SELECT artifact_kind FROM ingestion.artifacts WHERE artifact_kind='ohip_json'"), [("ohip_json",)]) self.assertEqual(self.monthly_worker().process_next().status, "published") self.assertEqual(self.monthly_worker().process_next().status, "idle") key, size = self.sql("SELECT a.object_key,a.byte_size FROM reporting.monthly_runs r JOIN ingestion.artifacts a ON a.id=r.workbook_artifact_id")[0] output = self.files / "monthly.xlsx"; self.store.materialize(key, output, size) workbook = load_workbook(output, data_only=False) formulas = [c.value for sheet in workbook for row in sheet for c in row if c.data_type == "f"] workbook.close() self.assertEqual(len(formulas), 6) self.assertTrue(all(value.startswith("=R") and "*C" in value and "*G" in value for value in formulas)) calls = len(self.transport.calls) self.assertIn(app.handle("POST", "/api/arr-downloads", headers, body).status, {200,202}) self.assertEqual(self.execute().status, "succeeded") self.assertEqual(len(self.transport.calls), calls) self.assertEqual(self.count("finance.daily_versions"), 1) self.assertFalse(list(self.files.rglob("*.xml"))) def test_price_review_only_creates_reports_after_confirmation(self): self.make_missing_price(); outcome = self.execute() self.assertEqual(outcome.status, "needs_review") self.assertEqual(self.count("finance.daily_versions"), 0) self.assertEqual(self.monthly_worker().process_next().status, "idle") calls = len(self.transport.calls) final = self.finish_review(outcome) self.assertEqual(final["status"], "succeeded") self.assertEqual(len(self.transport.calls), calls) self.assertEqual(self.count("finance.daily_versions"), 1) self.assertEqual(self.sql("SELECT result_schema_version FROM ingestion.processing_runs"), [("5.0",)]) self.assertEqual(self.monthly_worker().process_next().status, "published") self.assertEqual(self.coordinator.get_price_review(outcome.job_id, 50, 0)["case_status"], "completed") def test_commit_ack_loss_retry_has_one_result_and_no_new_fetch(self): state = {"armed": True} self.repository._connect = lambda dsn: pg_helpers.LoseCommitConnection(self.database.connect(dsn), state) queue, _, _ = self.start_web(); queue.create(DAY, REQUEST) self.assertEqual(self.wait_task(queue)["status"], "interrupted") self.assertEqual(self.count("finance.daily_versions"), 1) calls = len(self.transport.calls) queue.retry(REQUEST) self.assertEqual(self.wait_task(queue)["status"], "succeeded") self.assertEqual(len(self.transport.calls), calls) self.assertEqual(self.count("finance.daily_versions"), 1) self.assertEqual(self.count("ingestion.processing_deliveries"), 1) def test_schema_preflight_and_guarded_empty_rollback(self): self.repository.assert_data_source_schema() with self.database.connect() as connection: connection.execute((PROJECT / "database/019_ohip_data_source.down.sql").read_text(), prepare=False) with self.assertRaises(IngestionError) as error: self.repository.assert_data_source_schema() self.assertEqual(error.exception.code, "DATABASE_MIGRATION_MISSING") with self.database.connect() as connection: connection.execute((PROJECT / "database/019_ohip_data_source.sql").read_text(), prepare=False) self.repository.assert_data_source_schema()