367 lines
20 KiB
Python
367 lines
20 KiB
Python
"""Opt-in, owned disposable PostgreSQL acceptance; never reads a production DSN.
|
|
|
|
Run ARR_TEST_LOCAL_POSTGRES=1 python -m unittest tests.test_arr_download_postgres_integration.
|
|
Hotel transport/source mapping are synthetic, storage is local, SQL is real.
|
|
"""
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from contextlib import contextmanager
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
from arr_ingestion.contracts import IngestionError
|
|
from arr_ingestion.postgres import DatabaseConfig, PostgresIngestionRepository
|
|
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_download_executor import CapturedARRExecutor
|
|
from arr_web.arr_downloads import PersistentARRDownloads
|
|
from arr_web.programmatic import ProgrammaticUploadCoordinator
|
|
from arr_web.repository import PostgresPortalRepository
|
|
from integrations.ohip import collect_arr_source as source
|
|
from integrations.ohip.rate_info import RateInfoReader
|
|
from integrations.ohip.profile_reader import ProfileSummaryReader
|
|
from monthly_reports.publishing import AtomicReportPublisher, OpenpyxlWorkbookBuilder
|
|
from monthly_reports.repository import DatabaseConfig as MonthlyConfig, PostgresReportRepository
|
|
from monthly_reports.service import MonthlyReportService
|
|
from monthly_reports.worker import MonthlyOutboxWorker, PostgresOutboxRepository, WorkerError
|
|
from tests.local_postgres import TemporaryPostgres
|
|
from tests.test_arr_web import TEST_CREDENTIALS, login
|
|
from tests.test_arr_web_capture_executor import FixtureAdapter, FixtureValidator
|
|
from tests.test_arr_web_download_handoff import FixtureExecutor, REQUEST_ID, DAY
|
|
from tests.test_ohip_day_capture import DayService, HOTEL
|
|
from tests.test_ohip_named_day import NamedService
|
|
from tests.test_ohip_processing_handoff import CountingProcessor
|
|
|
|
|
|
class LoseCommitConnection:
|
|
"""Close an actual PG connection after its COMMIT, then lose the client ack."""
|
|
def __init__(self, connection, state):
|
|
self.connection, self.state = connection, state
|
|
self.wrote_finance = False
|
|
|
|
@contextmanager
|
|
def transaction(self):
|
|
with self.connection.transaction():
|
|
yield
|
|
if self.wrote_finance and self.state["armed"]:
|
|
self.state["armed"] = False
|
|
self.connection.close()
|
|
import psycopg
|
|
raise psycopg.OperationalError("synthetic connection loss after real COMMIT")
|
|
|
|
@contextmanager
|
|
def cursor(self):
|
|
with self.connection.cursor() as cursor:
|
|
owner = self
|
|
class Cursor:
|
|
def execute(self, query, params=None):
|
|
if "INSERT INTO finance.daily_versions" in query:
|
|
owner.wrote_finance = True
|
|
return cursor.execute(query, params)
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(cursor, name)
|
|
yield Cursor()
|
|
|
|
def close(self):
|
|
self.connection.close()
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("ARR_TEST_LOCAL_POSTGRES") == "1", "explicit disposable PostgreSQL opt-in required")
|
|
class PostgresDownloadIntegrationTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.database = TemporaryPostgres().__enter__()
|
|
cls.addClassCleanup(cls.database.__exit__, None, None, None)
|
|
cls.policy = load_processor_policy(Path(__file__).resolve().parents[1])
|
|
|
|
def setUp(self):
|
|
self.database.reset_database()
|
|
temporary = tempfile.TemporaryDirectory(prefix="arr-pg-artifacts-")
|
|
self.addCleanup(temporary.cleanup)
|
|
self.files = Path(temporary.name)
|
|
self.store = ManagedObjectStore(FilesystemObjectBackend(self.files / "objects", create=True))
|
|
self.repository = PostgresIngestionRepository(DatabaseConfig("owned-fixture"), connect=self.database.connect)
|
|
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.reader_calls = 0
|
|
self.executor = self.make_executor()
|
|
|
|
def reader_factory(self, archive, hotel):
|
|
self.reader_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):
|
|
arguments = dict(root=self.files / "executor", hotel_id=HOTEL,
|
|
adapter_contract="synthetic-pg-fixture/v1", adapter=self.adapter, mapping_validator=self.validator,
|
|
reader_factory=self.reader_factory, policy=self.policy, object_store=self.store,
|
|
repository=self.repository, ingestion=self.ingestion, processor=self.processor)
|
|
arguments.update(changes)
|
|
return CapturedARRExecutor(**arguments)
|
|
|
|
def execute(self):
|
|
return self.executor.execute(request_id=REQUEST_ID, from_date=DAY, to_date=DAY, report_stage=lambda _: None)
|
|
|
|
def sql(self, query, parameters=None):
|
|
with self.database.connect() as connection:
|
|
cursor = connection.execute(query, parameters)
|
|
return cursor.fetchall() if cursor.description else []
|
|
|
|
def count(self, table):
|
|
# Only fixed test-owned schema names supplied below; no user SQL input.
|
|
return self.sql(f"SELECT count(*) FROM {table}")[0][0]
|
|
|
|
def daily_event_count(self):
|
|
return self.sql("SELECT count(*) FROM ingestion.outbox_events WHERE event_type='arr.daily_version_committed'")[0][0]
|
|
|
|
def prepared_directory(self):
|
|
return next((self.files / "executor" / "handoffs").iterdir())
|
|
|
|
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))
|
|
_, headers = login(app)
|
|
return queue, app, headers
|
|
|
|
def wait_task(self, queue):
|
|
deadline = time.monotonic() + 20
|
|
while True:
|
|
result = queue.get(REQUEST_ID)
|
|
if result["status"] not in {"queued", "downloading", "processing"}:
|
|
return result
|
|
self.assertLess(time.monotonic(), deadline, "PostgreSQL task did not settle")
|
|
time.sleep(.01)
|
|
|
|
def monthly_worker(self):
|
|
config = MonthlyConfig("owned-fixture")
|
|
repository = PostgresReportRepository(config, connect=self.database.connect)
|
|
output = self.files / "outputs" / "monthly"
|
|
service = MonthlyReportService(repository, OpenpyxlWorkbookBuilder(),
|
|
AtomicReportPublisher(self.files, output, object_store=self.store), output / ".staging")
|
|
return MonthlyOutboxWorker(PostgresOutboxRepository(config, connect=self.database.connect), repository, service)
|
|
|
|
def assert_one_commit(self):
|
|
self.assertEqual(self.count("finance.daily_versions"), 1)
|
|
self.assertEqual(self.count("finance.current_daily_versions"), 1)
|
|
self.assertEqual(self.daily_event_count(), 1)
|
|
|
|
def assert_one_rejection(self, failure_code):
|
|
versions = self.sql("SELECT id, version_status, version_no, business_date, failure_code FROM finance.daily_versions")
|
|
self.assertEqual(len(versions), 1)
|
|
self.assertGreater(versions[0][0], 0)
|
|
self.assertEqual(versions[0][1:], ("rejected", None, None, failure_code))
|
|
self.assertEqual(self.count("finance.current_daily_versions"), 0)
|
|
self.assertEqual(self.count("ingestion.processing_deliveries"), 1)
|
|
self.assertEqual(self.daily_event_count(), 0)
|
|
self.assertEqual(self.sql("SELECT event_type FROM ingestion.outbox_events"), [("arr.processing_failed",)])
|
|
self.assertEqual(self.monthly_worker().process_next().status, "idle")
|
|
self.assertEqual(self.count("reporting.monthly_runs"), 0)
|
|
|
|
def test_no_eligible_rows_records_terminal_failure_in_web(self):
|
|
self.adapter.payload = FixtureExecutor.xml(rate_code="NOT-ALLOWED")
|
|
queue, app, headers = self.start_web()
|
|
response = app.handle("POST", "/api/arr-downloads", headers,
|
|
json.dumps({"request_id": REQUEST_ID, "report_date": DAY.isoformat()}).encode())
|
|
self.assertEqual(response.status, 202)
|
|
task = self.wait_task(queue)
|
|
self.assert_one_rejection("XML_NO_ELIGIBLE_ROWS")
|
|
self.assertEqual(task["status"], "failed")
|
|
self.assertFalse(task["can_retry"])
|
|
self.assertIsNotNone(task["job_id"])
|
|
response = app.handle("POST", "/api/arr-downloads/" + REQUEST_ID + "/retry", headers, b"{}")
|
|
self.assertEqual(response.status, 409)
|
|
response = app.handle("POST", "/api/arr-downloads", headers,
|
|
json.dumps({"request_id": REQUEST_ID, "report_date": DAY.isoformat()}).encode())
|
|
self.assertEqual(json.loads(response.body)["data"]["job_id"], task["job_id"])
|
|
self.assertEqual((self.reader_calls, self.adapter.calls, self.validator.calls, self.processor.calls), (1, 1, 1, 1))
|
|
|
|
def test_rejected_commit_lost_ack_recovers_same_failure_without_new_capture(self):
|
|
self.adapter.payload = FixtureExecutor.xml(rate_code="")
|
|
state = {"armed": True}
|
|
self.repository._connect = lambda dsn: LoseCommitConnection(self.database.connect(dsn), state)
|
|
queue, _, _ = self.start_web()
|
|
queue.create(DAY.isoformat(), REQUEST_ID)
|
|
self.assertEqual(self.wait_task(queue)["status"], "interrupted")
|
|
self.assertFalse(state["armed"])
|
|
self.assert_one_rejection("XML_RATE_CODE_MISSING")
|
|
queue.close(wait=True)
|
|
self.executor = self.make_executor(reader_factory=lambda *_: self.fail("must not recapture"))
|
|
resumed, app, headers = self.start_web()
|
|
response = app.handle("POST", "/api/arr-downloads/" + REQUEST_ID + "/retry", headers, b"{}")
|
|
self.assertEqual(response.status, 202)
|
|
task = self.wait_task(resumed)
|
|
self.assertEqual(task["status"], "failed")
|
|
self.assertFalse(task["can_retry"])
|
|
self.assertIsNotNone(task["job_id"])
|
|
self.assert_one_rejection("XML_RATE_CODE_MISSING")
|
|
self.assertEqual((self.reader_calls, self.adapter.calls, self.validator.calls, self.processor.calls), (1, 1, 1, 1))
|
|
|
|
def test_web_commit_two_monthly_workers_and_verified_workbook(self):
|
|
queue, app, headers = self.start_web()
|
|
response = app.handle("POST", "/api/arr-downloads", headers,
|
|
json.dumps({"request_id": REQUEST_ID, "report_date": DAY.isoformat()}).encode())
|
|
self.assertEqual(response.status, 202)
|
|
self.assertEqual(self.wait_task(queue)["status"], "succeeded")
|
|
self.assert_one_commit()
|
|
workers = [self.monthly_worker(), self.monthly_worker()]
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results = list(pool.map(lambda worker: worker.process_next(), workers))
|
|
self.assertEqual(sorted(result.status for result in results), ["idle", "published"], results)
|
|
self.assertEqual(self.count("reporting.monthly_runs"), 1)
|
|
report = self.sql("SELECT report_status, as_of_date, version_no FROM reporting.monthly_runs")[0]
|
|
self.assertEqual(report, ("active", DAY, 1))
|
|
self.assertEqual(self.monthly_worker().process_next().status, "idle")
|
|
key, digest, size = self.sql("""SELECT artifact.object_key, artifact.sha256, artifact.byte_size
|
|
FROM reporting.monthly_runs AS report JOIN ingestion.artifacts AS artifact
|
|
ON artifact.id=report.workbook_artifact_id""")[0]
|
|
workbook_file = self.files / "verified-monthly.xlsx"
|
|
self.store.materialize(key, workbook_file, size)
|
|
self.assertEqual(hashlib.sha256(workbook_file.read_bytes()).hexdigest(), digest)
|
|
workbook = load_workbook(workbook_file, read_only=True, data_only=False)
|
|
try:
|
|
formulas = [cell.value for sheet in workbook for row in sheet.iter_rows() for cell in row
|
|
if cell.data_type == "f"]
|
|
self.assertTrue(formulas)
|
|
self.assertTrue(all(str(value).startswith("=R") and "*C" in value and "*G" in value for value in formulas))
|
|
finally:
|
|
workbook.close()
|
|
|
|
def test_connection_lost_after_commit_recovers_one_database_result(self):
|
|
state = {"armed": True}
|
|
self.repository._connect = lambda dsn: LoseCommitConnection(self.database.connect(dsn), state)
|
|
queue, app, headers = self.start_web()
|
|
queue.create(DAY.isoformat(), REQUEST_ID)
|
|
self.assertEqual(self.wait_task(queue)["status"], "interrupted")
|
|
self.assertFalse(state["armed"])
|
|
self.assert_one_commit()
|
|
queue.close()
|
|
self.executor = self.make_executor()
|
|
resumed, _, _ = self.start_web()
|
|
self.assertEqual(resumed.get(REQUEST_ID)["status"], "interrupted")
|
|
resumed.retry(REQUEST_ID)
|
|
self.assertEqual(self.wait_task(resumed)["status"], "succeeded")
|
|
self.assert_one_commit()
|
|
self.assertEqual(self.count("ingestion.processing_deliveries"), 1)
|
|
self.assertEqual((self.reader_calls, self.adapter.calls, self.processor.calls), (1, 1, 1))
|
|
|
|
def test_concurrent_exact_deliveries_share_version_and_outbox(self):
|
|
with patch.object(self.ingestion, "ingest", side_effect=ConnectionError("pause before ingest")), self.assertRaises(ConnectionError):
|
|
self.execute()
|
|
self.assertEqual(self.count("finance.daily_versions"), 0)
|
|
self.assertEqual(self.count("ingestion.processing_runs"), 1)
|
|
envelope = (self.prepared_directory() / "frozen" / "delivery.json").read_bytes()
|
|
barrier = threading.Barrier(2)
|
|
def submit(_):
|
|
barrier.wait(timeout=5)
|
|
return self.ingestion.ingest(envelope)
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
outcomes = list(pool.map(submit, range(2)))
|
|
self.assertEqual([item.status for item in outcomes], ["committed", "committed"])
|
|
self.assertEqual(outcomes[0].daily_version_id, outcomes[1].daily_version_id)
|
|
self.assert_one_commit()
|
|
self.assertEqual(self.count("ingestion.processing_deliveries"), 1)
|
|
self.assertEqual(self.execute().status, "succeeded")
|
|
self.assertEqual(self.processor.calls, 1)
|
|
|
|
def test_named_bounded_capture_recovers_after_real_commit_without_new_profile_reads(self):
|
|
self.transport = NamedService()
|
|
def named_factory(archive, hotel):
|
|
self.reader_calls += 1
|
|
return (source.Reader(archive, hotel, self.transport, sleep=lambda _: None),
|
|
RateInfoReader(archive, hotel, self.transport, sleep=lambda _: None),
|
|
ProfileSummaryReader(archive, hotel, self.transport, max_profiles=3, sleep=lambda _: None))
|
|
arguments = dict(capture_version="v3", max_profiles=3, page_size=2, max_pages=2, max_records=3)
|
|
self.executor = self.make_executor(reader_factory=named_factory, **arguments)
|
|
state = {"armed": True}
|
|
self.repository._connect = lambda dsn: LoseCommitConnection(self.database.connect(dsn), state)
|
|
with self.assertRaises(IngestionError):
|
|
self.execute()
|
|
self.assertFalse(state["armed"])
|
|
self.assert_one_commit()
|
|
self.assertEqual(len(self.transport.profile_calls), 3)
|
|
capture = self.files / "executor" / "captures"
|
|
before = {str(p): hashlib.sha256(p.read_bytes()).hexdigest() for p in capture.rglob("*") if p.is_file()}
|
|
self.executor = self.make_executor(reader_factory=lambda *_: self.fail("must not read again"), **arguments)
|
|
self.assertEqual(self.execute().status, "succeeded")
|
|
self.assert_one_commit()
|
|
self.assertEqual(self.count("ingestion.processing_deliveries"), 1)
|
|
self.assertEqual((self.reader_calls, self.adapter.calls, self.validator.calls, self.processor.calls), (1, 1, 1, 1))
|
|
self.assertEqual(before, {str(p): hashlib.sha256(p.read_bytes()).hexdigest() for p in capture.rglob("*") if p.is_file()})
|
|
self.assertEqual(len(self.transport.profile_calls), 3)
|
|
|
|
def test_database_error_rolls_back_finance_delivery_and_event_together(self):
|
|
self.sql("""CREATE FUNCTION ingestion.reject_fixture_outbox() RETURNS trigger LANGUAGE plpgsql AS $$
|
|
BEGIN RAISE EXCEPTION 'synthetic database failure' USING ERRCODE='XX000'; END; $$;
|
|
CREATE TRIGGER fixture_outbox_failure BEFORE INSERT ON ingestion.outbox_events
|
|
FOR EACH ROW WHEN (NEW.event_type='arr.daily_version_committed')
|
|
EXECUTE FUNCTION ingestion.reject_fixture_outbox();""")
|
|
with self.assertRaises(IngestionError):
|
|
self.execute()
|
|
for table in ("finance.daily_versions", "finance.current_daily_versions", "finance.daily_records", "ingestion.processing_deliveries"):
|
|
self.assertEqual(self.count(table), 0, table)
|
|
self.assertEqual(self.daily_event_count(), 0)
|
|
self.sql("DROP TRIGGER fixture_outbox_failure ON ingestion.outbox_events")
|
|
self.assertEqual(self.execute().status, "succeeded")
|
|
self.assert_one_commit()
|
|
self.assertEqual((self.reader_calls, self.processor.calls), (1, 1))
|
|
|
|
def test_price_review_commits_once_and_web_observes_same_job(self):
|
|
self.adapter.payload = FixtureExecutor.xml(rate_amount="1800")
|
|
queue, app, headers = self.start_web()
|
|
queue.create(DAY.isoformat(), REQUEST_ID)
|
|
task = self.wait_task(queue)
|
|
self.assertEqual(task["status"], "needs_review")
|
|
self.assertEqual(self.count("ingestion.daily_review_cases"), 1)
|
|
self.assertEqual(self.count("finance.daily_versions"), 0)
|
|
self.assertEqual(self.daily_event_count(), 0)
|
|
review = ProgrammaticUploadCoordinator(object_store=self.store, ingestion_repository=self.repository,
|
|
ingestion_service=self.ingestion, processor=self.processor, processor_version=self.policy.processor_version,
|
|
rule_set_sha256=self.policy.rule_set_sha256)
|
|
case = review.get_price_review(task["job_id"], 50, 0)
|
|
updated = review.update_price_review_item(task["job_id"], case["items"][0]["item_id"],
|
|
case["case_id"], case["revision"], "0", "synthetic-pg-reviewer")
|
|
result = review.finalize_price_review(task["job_id"], updated["case_id"], updated["revision"], "synthetic-pg-reviewer")
|
|
self.assertEqual(result["status"], "succeeded")
|
|
response = app.handle("GET", "/api/arr-downloads/" + REQUEST_ID, headers)
|
|
self.assertEqual(json.loads(response.body)["data"]["status"], "succeeded")
|
|
repeated = review.finalize_price_review(task["job_id"], updated["case_id"], updated["revision"], "synthetic-pg-reviewer")
|
|
self.assertEqual(result["daily_version_id"], repeated["daily_version_id"])
|
|
self.assert_one_commit()
|
|
self.assertEqual(self.reader_calls, 1)
|
|
|
|
def test_monthly_publication_replay_after_missing_event_ack_has_one_version(self):
|
|
self.execute()
|
|
worker = self.monthly_worker()
|
|
with patch.object(worker._outbox, "mark_published", side_effect=WorkerError("SYNTHETIC_ACK_MISSING", retryable=True)):
|
|
first = worker.process_next()
|
|
self.assertEqual(first.status, "pending")
|
|
self.assertEqual(self.count("reporting.monthly_runs"), 1)
|
|
first_report = self.sql("SELECT id, workbook_artifact_id, result_artifact_id FROM reporting.monthly_runs")[0]
|
|
self.sql("UPDATE ingestion.outbox_events SET available_at=now() WHERE event_type='arr.daily_version_committed'")
|
|
second = self.monthly_worker().process_next()
|
|
self.assertEqual(second.status, "published")
|
|
self.assertEqual(self.sql("SELECT id, workbook_artifact_id, result_artifact_id FROM reporting.monthly_runs"), [first_report])
|
|
self.assertEqual(self.sql("SELECT publish_status FROM ingestion.outbox_events WHERE event_type='arr.daily_version_committed'"), [("published",)])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|