317 lines
18 KiB
Python
317 lines
18 KiB
Python
"""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()
|