from __future__ import annotations import base64 import hashlib import json import tempfile import threading import time import unittest from datetime import date, datetime from pathlib import Path from typing import Any, Dict, Mapping, Optional from zoneinfo import ZoneInfo from arr_web.app import PortalApplication, RuntimeHealth, SessionLedger from arr_web.auth import LoginCredentials from arr_web.company_jobs import ( COMPANY_NAMES, PersistentCompanyReportCoordinator, period_complete_at, period_release_at, period_to_as_of, ) from arr_web.contracts import PortalError from arr_web.downloads import ArtifactDescriptor, ControlledProjectArtifactReader from company_reports.contracts import ENGLISH_MONTH_NAMES, RESULT_SCHEMA_VERSION BANGKOK = ZoneInfo("Asia/Bangkok") TEST_CREDENTIALS = LoginCredentials( username="arr-test-operator", password="arr-test-password", ) def decoded(response: Any) -> Dict[str, Any]: return json.loads(response.body.decode("utf-8")) def login_headers(app: PortalApplication) -> Dict[str, str]: response = app.handle( "POST", "/api/login", {"Content-Type": "application/json"}, json.dumps( { "username": TEST_CREDENTIALS.username, "password": TEST_CREDENTIALS.password, } ).encode("utf-8"), ) if response.status != 200: raise AssertionError(f"test login failed with HTTP {response.status}") payload = decoded(response) cookie = response.headers["Set-Cookie"].split(";", 1)[0] return { "Cookie": cookie, "X-ARR-CSRF": payload["data"]["csrf_token"], } class FixtureExecutor: def __init__(self, project_root: Path, *, failed_company: str = "") -> None: self.project_root = project_root self.output_root = project_root / "outputs" / "company_reports" self.failed_company = failed_company self.calls: list[tuple[int, int, object]] = [] def run(self, year: int, month: int, as_of_date: object) -> Mapping[str, Any]: self.calls.append((year, month, as_of_date)) final_day = 29 if year == 2028 and month == 2 else 31 if month in {4, 6, 9, 11}: final_day = 30 elif month == 2 and year != 2028: final_day = 28 companies = [] for index, company in enumerate(COMPANY_NAMES, 1): counts = {"01-10": index, "11-20": index + 1, f"21-{final_day:02d}": 0} if company == self.failed_company: companies.append( { "company": company, "status": "failed", "row_count": 0, "period_row_counts": counts, "warnings": [], "errors": [ { "code": "COMPANY_REPORT_GROUP_CODE_NOT_FOUND", "stage": "source", "period": "01-10", "record_ids": [100 + index], } ], } ) continue filename = f"{company}-{ENGLISH_MONTH_NAMES[month]}-{year}.xlsx" path = self.output_root / f"{year:04d}" / f"{month:02d}" / "test" / filename path.parent.mkdir(parents=True, exist_ok=True) payload = f"safe workbook fixture {company}".encode("utf-8") path.write_bytes(payload) companies.append( { "company": company, "status": "success", "row_count": sum(counts.values()), "period_row_counts": counts, "version_no": index, "warnings": ( [ { "code": "COMPANY_REPORT_MULTI_PRICE_REVIEW", "stage": "pricing", "period": "01-10", "record_ids": [index], } ] if company == "QBD" else [] ), "errors": [], "artifact": { "filename": filename, "storage_key": path.relative_to(self.project_root).as_posix(), "sha256": hashlib.sha256(payload).hexdigest(), "semantic_sha256": "a" * 64, }, } ) successful = len(COMPANY_NAMES) - (1 if self.failed_company else 0) status = "success" if successful == len(COMPANY_NAMES) else "partial_failure" return { "schema_version": RESULT_SCHEMA_VERSION, "status": status, "report_year": year, "report_month": month, "as_of_date": str(as_of_date), "requested_companies": list(COMPANY_NAMES), "companies": companies, } class BlockingFailureExecutor: def __init__(self) -> None: self.started = threading.Event() self.release = threading.Event() def run(self, year: int, month: int, as_of_date: object) -> Mapping[str, Any]: self.started.set() self.release.wait(timeout=3) counts = {"01-10": 0, "11-20": 0, "21-31": 0} return { "schema_version": RESULT_SCHEMA_VERSION, "status": "failed", "report_year": year, "report_month": month, "as_of_date": str(as_of_date), "requested_companies": list(COMPANY_NAMES), "companies": [ { "company": company, "status": "failed", "row_count": 0, "period_row_counts": counts, "warnings": [], "errors": [ { "code": "COMPANY_REPORT_SOURCE_VERSION_MISSING", "stage": "source", "period": "2026-07", "record_ids": [], } ], } for company in COMPANY_NAMES ], } def wait_terminal( coordinator: PersistentCompanyReportCoordinator, job_id: str, timeout: float = 3.0, ) -> Dict[str, Any]: deadline = time.monotonic() + timeout while time.monotonic() < deadline: value = coordinator.get_job(job_id) if value["state"] in {"succeeded", "partial_failure", "failed"}: return value time.sleep(0.01) raise AssertionError("company report job did not finish") class CompanyReportCoordinatorTests(unittest.TestCase): def test_period_mapping_and_bangkok_completion_instants(self) -> None: self.assertEqual(str(period_to_as_of("2028-02", "21-month-end")), "2028-02-29") self.assertEqual( period_complete_at("2026-07", "01-10").isoformat(), "2026-07-11T00:00:00+07:00", ) self.assertEqual( period_complete_at("2026-12", "21-month-end").isoformat(), "2027-01-01T00:00:00+07:00", ) self.assertEqual( period_release_at("2026-07", "01-10"), period_complete_at("2026-07", "01-10"), ) def test_success_is_private_and_download_is_reverified(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) executor = FixtureExecutor(root) coordinator = PersistentCompanyReportCoordinator( root, root / "outputs" / "company_reports", executor, now=lambda: datetime(2026, 8, 1, tzinfo=BANGKOK), ) try: source = { "source_batch_id": 7, "source_type": "excel", "filename": "Booking 报表.xlsx", "source_rows": 12, "worksheet_count": 2, "distinct_group_codes": 8, "room_quantity": 14, "activated_at": "2026-07-31T08:00:00+07:00", "disposition": "current", } created = coordinator.create("2026-07", "11-20", source=source) self.assertEqual(created["source"], source) job = wait_terminal(coordinator, created["job_id"]) self.assertEqual(job["state"], "succeeded") self.assertEqual(job["source"], source) self.assertEqual(job["requested_companies"], list(COMPANY_NAMES)) self.assertEqual(len(job["company_results"]), 5) self.assertEqual(job["company_results"][1]["warnings"][0]["record_ids"], [2]) public_text = json.dumps(job, ensure_ascii=False) self.assertNotIn("storage_key", public_text) self.assertNotIn("sha256", public_text) descriptor = coordinator.resolve_download(created["job_id"], "QBD") self.assertEqual(descriptor.file_kind, "company_ten_day_xlsx") body = ControlledProjectArtifactReader(root).read(descriptor) self.assertIn(b"safe workbook fixture", body) finally: coordinator.close() def test_partial_failure_keeps_successful_company_downloads(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) coordinator = PersistentCompanyReportCoordinator( root, root / "outputs" / "company_reports", FixtureExecutor(root, failed_company="HanaTour"), now=lambda: datetime(2026, 8, 1, tzinfo=BANGKOK), ) try: created = coordinator.create("2026-07", "01-10") job = wait_terminal(coordinator, created["job_id"]) self.assertEqual(job["state"], "partial_failure") self.assertEqual(set(job["downloads"]), set(COMPANY_NAMES) - {"HanaTour"}) with self.assertRaises(PortalError) as caught: coordinator.resolve_download(created["job_id"], "HanaTour") self.assertEqual(caught.exception.status, 404) finally: coordinator.close() def test_incomplete_period_can_run_and_future_month_is_rejected(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) executor = FixtureExecutor(root) coordinator = PersistentCompanyReportCoordinator( root, root / "outputs" / "company_reports", executor, now=lambda: datetime(2026, 7, 10, 23, 59, tzinfo=BANGKOK), ) try: created = coordinator.create("2026-07", "01-10") job = wait_terminal(coordinator, created["job_id"]) self.assertEqual(job["state"], "succeeded") self.assertEqual(executor.calls, [(2026, 7, date(2026, 7, 10))]) finally: coordinator.close() with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) coordinator = PersistentCompanyReportCoordinator( root, root / "outputs" / "company_reports", FixtureExecutor(root), now=lambda: datetime(2026, 7, 10, 23, 59, tzinfo=BANGKOK), ) try: with self.assertRaises(PortalError) as caught: coordinator.create("2026-08", "01-10") self.assertEqual(caught.exception.code, "COMPANY_REPORT_MONTH_IN_FUTURE") finally: coordinator.close() with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) executor = BlockingFailureExecutor() coordinator = PersistentCompanyReportCoordinator( root, root / "outputs" / "company_reports", executor, now=lambda: datetime(2026, 8, 1, tzinfo=BANGKOK), ) try: first = coordinator.create("2026-07", "01-10") self.assertTrue(executor.started.wait(timeout=1)) with self.assertRaises(PortalError) as caught: coordinator.create("2026-07", "01-10") self.assertEqual(caught.exception.code, "COMPANY_REPORT_ALREADY_RUNNING") executor.release.set() failed_job = wait_terminal(coordinator, first["job_id"]) self.assertEqual(failed_job["state"], "failed") self.assertIsNone(failed_job["message"]) finally: executor.release.set() coordinator.close() class FakeCompanyCoordinator: def __init__(self) -> None: self.created: list[tuple[str, str, Optional[Mapping[str, Any]]]] = [] self.listed: list[tuple[object, int, int]] = [] self.descriptor = ArtifactDescriptor( "company_ten_day_xlsx", "QBD-July-2026.xlsx", "outputs/company_reports/QBD-July-2026.xlsx", "a" * 64, 4, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) def create( self, report_month: str, period: str, *, source: Optional[Mapping[str, Any]] = None, ) -> Dict[str, Any]: self.created.append((report_month, period, source)) return {"job_id": "a" * 32, "state": "queued", "source": source} def list_jobs( self, report_month: object = None, limit: int = 50, offset: int = 0, ) -> tuple[list[Dict[str, Any]], int]: self.listed.append((report_month, limit, offset)) return [ { "job_id": f"{index:032x}", "report_month": report_month, "limit": limit, "offset": offset, } for index in range(offset, min(offset + limit, 72)) ], 72 def list_month_counts(self) -> list[Dict[str, Any]]: return [ {"month_key": "2026-07", "company_count": 4}, {"month_key": "2026-06", "company_count": 1}, ] def get_job(self, job_id: str) -> Dict[str, Any]: return {"job_id": job_id, "state": "succeeded"} def resolve_download(self, job_id: str, company: str) -> ArtifactDescriptor: return self.descriptor class FakeReader: def read(self, descriptor: ArtifactDescriptor) -> bytes: return b"xlsx" class FakeBookingSources: def __init__(self) -> None: self.uploads: list[tuple[str, bytes]] = [] self.draft_page: Dict[str, Any] | None = None self.review_updates: list[tuple[Any, int, Any, Any]] = [] self.review_deletes: list[tuple[Any, int]] = [] self.review_bulk_deletes: list[tuple[Any, tuple[int, ...]]] = [] self.review_discards: list[Any] = [] self.review_activations: list[Any] = [] self.source: Dict[str, Any] | None = { "source_batch_id": 7, "source_type": "excel", "filename": "Booking 报表.xlsx", "source_rows": 12, "worksheet_count": 2, "distinct_group_codes": 8, "room_quantity": 14, "activated_at": "2026-07-31T08:00:00+07:00", "disposition": "current", } def current(self) -> Dict[str, Any] | None: return self.source def draft(self, limit: int = 50, offset: int = 0) -> Dict[str, Any] | None: _ = (limit, offset) return self.draft_page def submit(self, filename: str, payload: bytes) -> Dict[str, Any]: self.uploads.append((filename, payload)) self.draft_page = { "summary": { "draft_id": "bookingdraft-" + "a" * 32, "filename": filename, "pending_items": 1, }, "items": [], "pagination": {"total": 0, "limit": 50, "offset": 0}, } return self.draft_page def update_item( self, draft_id: Any, item_id: int, room_type: Any, quantity: Any, ) -> Dict[str, Any]: self.review_updates.append((draft_id, item_id, room_type, quantity)) return {"item_id": item_id, "room_type": room_type, "quantity": quantity} def delete_item(self, draft_id: Any, item_id: int) -> Dict[str, Any]: self.review_deletes.append((draft_id, item_id)) return {"item_id": item_id, "deleted": True} def delete_items(self, draft_id: Any, item_ids: object) -> Dict[str, Any]: assert isinstance(item_ids, list) values = tuple(item_ids) self.review_bulk_deletes.append((draft_id, values)) return { "draft_id": draft_id, "item_ids": list(values), "deleted_count": len(values), "deleted": True, } def discard(self, draft_id: Any) -> Dict[str, Any]: self.review_discards.append(draft_id) self.draft_page = None return {"draft_id": draft_id, "discarded": True} def activate(self, draft_id: Any) -> Dict[str, Any]: self.review_activations.append(draft_id) self.draft_page = None assert self.source is not None return self.source class CompanyReportRouteTests(unittest.TestCase): def setUp(self) -> None: self.coordinator = FakeCompanyCoordinator() self.booking_sources = FakeBookingSources() self.app = PortalApplication( company_reports=self.coordinator, booking_sources=self.booking_sources, artifact_reader=FakeReader(), health=RuntimeHealth(False, False, False, True, True, True), sessions=SessionLedger(), credentials=TEST_CREDENTIALS, ) self.auth_headers = login_headers(self.app) def session_headers(self) -> Dict[str, str]: return dict(self.auth_headers) def test_create_is_csrf_protected_and_strict(self) -> None: body = b'{"report_month":"2026-07","period":"01-10"}' self.assertEqual( self.app.handle( "POST", "/api/company-reports/jobs", {"Cookie": self.auth_headers["Cookie"]}, body, ).status, 403, ) response = self.app.handle( "POST", "/api/company-reports/jobs", {**self.session_headers(), "Content-Type": "application/json"}, body, ) self.assertEqual(response.status, 202) self.assertEqual( self.coordinator.created, [("2026-07", "01-10", self.booking_sources.source)], ) self.assertEqual(decoded(response)["data"]["source"]["source_batch_id"], 7) invalid = self.app.handle( "POST", "/api/company-reports/jobs", {**self.session_headers(), "Content-Type": "application/json"}, b'{"report_month":"2026-07","period":"01-10","companies":[]}', ) self.assertEqual(invalid.status, 400) self.assertEqual(decoded(invalid)["error"]["code"], "COMPANY_REPORT_REQUEST_INVALID") def test_source_read_and_upload_are_authenticated_and_csrf_protected(self) -> None: current = self.app.handle( "GET", "/api/company-reports/source", self.auth_headers, ) self.assertEqual(decoded(current)["data"]["filename"], "Booking 报表.xlsx") encoded = base64.urlsafe_b64encode("七月预订.xlsx".encode("utf-8")).decode("ascii").rstrip("=") without_csrf = self.app.handle( "POST", "/api/company-reports/source", { "Cookie": self.auth_headers["Cookie"], "X-ARR-Filename-B64": encoded, }, b"xlsx", ) self.assertEqual(without_csrf.status, 403) uploaded = self.app.handle( "POST", "/api/company-reports/source", {**self.session_headers(), "X-ARR-Filename-B64": encoded}, b"xlsx", ) self.assertEqual(uploaded.status, 200) self.assertEqual(self.booking_sources.uploads, [("七月预订.xlsx", b"xlsx")]) self.assertEqual(decoded(uploaded)["data"]["summary"]["pending_items"], 1) def test_review_routes_are_authenticated_csrf_protected_and_strict(self) -> None: draft_id = "bookingdraft-" + "a" * 32 self.booking_sources.draft_page = { "summary": {"draft_id": draft_id, "pending_items": 1}, "items": [], "pagination": {"total": 0, "limit": 50, "offset": 0}, } page = self.app.handle( "GET", "/api/company-reports/source/draft?limit=50&offset=0", self.auth_headers, ) self.assertEqual(decoded(page)["data"]["summary"]["draft_id"], draft_id) update_body = json.dumps( {"draft_id": draft_id, "room_type": "EXTRA BED", "quantity": 2} ).encode("utf-8") without_csrf = self.app.handle( "PATCH", "/api/company-reports/source/draft/items/9", {"Cookie": self.auth_headers["Cookie"]}, update_body, ) self.assertEqual(without_csrf.status, 403) updated = self.app.handle( "PATCH", "/api/company-reports/source/draft/items/9", {**self.session_headers(), "Content-Type": "application/json"}, update_body, ) self.assertEqual(updated.status, 200) self.assertEqual( self.booking_sources.review_updates, [(draft_id, 9, "EXTRA BED", 2)], ) deleted = self.app.handle( "DELETE", "/api/company-reports/source/draft/items/9", {**self.session_headers(), "Content-Type": "application/json"}, json.dumps({"draft_id": draft_id}).encode("utf-8"), ) self.assertEqual(deleted.status, 200) self.assertEqual(self.booking_sources.review_deletes, [(draft_id, 9)]) bulk_body = json.dumps({"draft_id": draft_id, "item_ids": [9, 10]}).encode("utf-8") without_bulk_csrf = self.app.handle( "DELETE", "/api/company-reports/source/draft/items", {"Cookie": self.auth_headers["Cookie"]}, bulk_body, ) self.assertEqual(without_bulk_csrf.status, 403) bulk_deleted = self.app.handle( "DELETE", "/api/company-reports/source/draft/items", {**self.session_headers(), "Content-Type": "application/json"}, bulk_body, ) self.assertEqual(bulk_deleted.status, 200) self.assertEqual(decoded(bulk_deleted)["data"]["deleted_count"], 2) self.assertEqual( self.booking_sources.review_bulk_deletes, [(draft_id, (9, 10))], ) invalid_bulk = self.app.handle( "DELETE", "/api/company-reports/source/draft/items", {**self.session_headers(), "Content-Type": "application/json"}, json.dumps({"draft_id": draft_id, "item_ids": [9], "extra": True}).encode( "utf-8" ), ) self.assertEqual(invalid_bulk.status, 400) self.assertEqual( decoded(invalid_bulk)["error"]["code"], "BOOKING_EXCEL_REVIEW_REQUEST_INVALID", ) activated = self.app.handle( "POST", "/api/company-reports/source/draft/activate", {**self.session_headers(), "Content-Type": "application/json"}, json.dumps({"draft_id": draft_id}).encode("utf-8"), ) self.assertEqual(activated.status, 200) self.assertEqual(self.booking_sources.review_activations, [draft_id]) def test_open_review_draft_blocks_company_generation(self) -> None: self.booking_sources.draft_page = { "summary": {"draft_id": "bookingdraft-" + "a" * 32}, "items": [], "pagination": {"total": 0, "limit": 1, "offset": 0}, } response = self.app.handle( "POST", "/api/company-reports/jobs", {**self.session_headers(), "Content-Type": "application/json"}, b'{"report_month":"2026-07","period":"01-10"}', ) self.assertEqual(response.status, 409) self.assertEqual(decoded(response)["error"]["code"], "BOOKING_EXCEL_REVIEW_OPEN") def test_generation_requires_an_accepted_source(self) -> None: self.booking_sources.source = None response = self.app.handle( "POST", "/api/company-reports/jobs", {**self.session_headers(), "Content-Type": "application/json"}, b'{"report_month":"2026-07","period":"01-10"}', ) self.assertEqual(response.status, 409) self.assertEqual( decoded(response)["error"]["code"], "BOOKING_EXCEL_SOURCE_REQUIRED", ) def test_list_detail_and_download_routes(self) -> None: listing = self.app.handle( "GET", "/api/company-reports/jobs?month=2026-07&limit=50&offset=50", self.auth_headers, ) listing_payload = decoded(listing) self.assertEqual(listing_payload["data"][0]["report_month"], "2026-07") self.assertEqual(self.coordinator.listed, [("2026-07", 50, 50)]) self.assertEqual(listing_payload["pagination"]["total"], 72) self.assertTrue(listing_payload["pagination"]["has_previous"]) self.assertFalse(listing_payload["pagination"]["has_next"]) detail = self.app.handle( "GET", f"/api/company-reports/jobs/{'a' * 32}", self.auth_headers, ) self.assertEqual(decoded(detail)["data"]["state"], "succeeded") download = self.app.handle( "GET", f"/api/company-reports/jobs/{'a' * 32}/downloads/QBD", self.auth_headers, ) self.assertEqual(download.status, 200) self.assertEqual(download.body, b"xlsx") self.assertNotIn("outputs/company_reports", str(download.headers)) if __name__ == "__main__": unittest.main()