from __future__ import annotations import json import tempfile import unittest from datetime import date from pathlib import Path from typing import Optional from openpyxl import load_workbook from company_reports.contracts import ( BatchSnapshot, CompanyReport, DailyVersionPin, ErrorCode, PeriodReport, ReportRow, ) from company_reports.core import build_company_report from company_reports.publishing import ( OpenpyxlWorkbookBuilder, AtomicReportPublisher, BuiltWorkbook, PublicationError, sha256_file, ) from company_reports.repository import FileMetadata, RepositoryError, ReservedReport class FakeRepository: def __init__(self, fail_activation: bool = False): self.fail_activation = fail_activation self.activated: Optional[tuple] = None self.failed: Optional[tuple] = None def activate_report( self, reservation: ReservedReport, artifact: FileMetadata, result_json: FileMetadata, ) -> None: if self.fail_activation: raise RepositoryError(ErrorCode.PUBLISH_FAILED, "synthetic activation failed") self.activated = (reservation, artifact, result_json) def mark_failed( self, reservation: ReservedReport, code: str, safe_message: str, ) -> None: self.failed = (reservation, code, safe_message) def empty_report(): return build_company_report( "HanaTour", 2026, 7, date(2026, 7, 10), BatchSnapshot((), (), (), {}), ) def workbook_report() -> CompanyReport: row = ReportRow( arrival=date(2026, 7, 8), departure=date(2026, 7, 10), nights=2, block_code="=BLOCK-A", res_comment="SYN-GROUP-A", booking_room="【DBL】1", total_booking_price="+900", normalized_group_code="SYN-GROUP-A", record_ids=(1,), duplicate_group=True, multi_price_review=True, ) return CompanyReport( company="QBD", report_year=2026, report_month=7, as_of_date=date(2026, 7, 10), filename="QBD-July-2026.xlsx", periods=( PeriodReport(1, 10, True, "QBD 01-10 Jul 2026", (row,)), PeriodReport(11, 20, False, "QBD 11-20 Jul 2026", ()), PeriodReport(21, 31, False, "QBD 21-31 Jul 2026", ()), ), warnings=(), errors=(), daily_versions=(DailyVersionPin(date(2026, 7, 8), 100),), booking_versions={"SYN-GROUP-A": 700}, ) def synthetic_built_workbook( work_dir: Path, filename: str, content: bytes, semantic_sha256: str, ) -> BuiltWorkbook: path = work_dir / filename path.write_bytes(content) return BuiltWorkbook( path=path, sha256=sha256_file(path), byte_size=path.stat().st_size, summary={"semantic_sha256": semantic_sha256}, ) class CompanyReportPublishingTests(unittest.TestCase): def test_builder_creates_xlsx_with_openpyxl_only(self): with tempfile.TemporaryDirectory(prefix="company-report-builder-test-") as temp_dir: root = Path(temp_dir) report = workbook_report() builder = OpenpyxlWorkbookBuilder() built = builder.build(report, root) self.assertTrue(built.path.is_file()) self.assertEqual(built.path.stat().st_mode & 0o777, 0o600) self.assertEqual(built.summary["status"], "success") self.assertEqual(built.summary["row_counts"], [1, 0, 0]) self.assertEqual(built.summary["formula_count"], 0) self.assertEqual(len(built.summary["semantic_sha256"]), 64) workbook = load_workbook(built.path) self.assertEqual(workbook.sheetnames, [period.sheet_name for period in report.periods]) sheet = workbook[report.periods[0].sheet_name] self.assertEqual( [cell.value for cell in sheet[1]], [ "ARRIVAL", "DEPARTURE", "NIGHTS", "BLOCK_CODE", "RES_COMMENT", "Booking Room", "Total Booking Price", ], ) self.assertEqual(sheet["D2"].value, "'=BLOCK-A") self.assertEqual(sheet["G2"].value, "'+900") def test_publish_archives_prior_current_and_activates_new_version(self): with tempfile.TemporaryDirectory(prefix="company-report-publish-test-") as temp_dir: project_root = Path(temp_dir) / "project" output_root = project_root / "outputs" / "company_reports" work_dir = project_root / "staging" project_root.mkdir(parents=True) work_dir.mkdir() report = empty_report() month_root = output_root / "2026" / "07" month_root.mkdir(parents=True) current = month_root / report.filename current.write_bytes(b"synthetic-old-workbook") built_path = work_dir / report.filename built_path.write_bytes(b"synthetic-new-workbook") built = BuiltWorkbook( path=built_path, sha256=sha256_file(built_path), byte_size=built_path.stat().st_size, summary={}, ) reservation = ReservedReport(44, 2, report.company) repository = FakeRepository() publisher = AtomicReportPublisher(project_root, output_root) outcome = publisher.publish( report, reservation, built, repository, work_dir ) self.assertEqual(current.read_bytes(), b"synthetic-new-workbook") self.assertEqual(outcome.archive_path.read_bytes(), b"synthetic-new-workbook") legacy = list((month_root / "archive" / "legacy").rglob(report.filename)) self.assertEqual(len(legacy), 1) self.assertEqual(legacy[0].read_bytes(), b"synthetic-old-workbook") self.assertEqual(current.stat().st_mode & 0o777, 0o600) result = json.loads(outcome.result_path.read_text(encoding="utf-8")) self.assertEqual(result["status"], "success") self.assertEqual(result["row_count"], 0) self.assertNotIn("group_code", outcome.result_path.read_text(encoding="utf-8").lower()) self.assertIsNotNone(repository.activated) artifact = repository.activated[1] self.assertEqual( artifact.storage_key, "outputs/company_reports/2026/07/archive/v0002/HanaTour-July-2026.xlsx", ) self.assertIsNone(repository.failed) def test_activation_failure_restores_prior_current_and_removes_new_version_files(self): with tempfile.TemporaryDirectory(prefix="company-report-publish-test-") as temp_dir: project_root = Path(temp_dir) / "project" output_root = project_root / "outputs" / "company_reports" work_dir = project_root / "staging" project_root.mkdir(parents=True) work_dir.mkdir() report = empty_report() month_root = output_root / "2026" / "07" month_root.mkdir(parents=True) current = month_root / report.filename current.write_bytes(b"synthetic-old-workbook") built_path = work_dir / report.filename built_path.write_bytes(b"synthetic-new-workbook") built = BuiltWorkbook( path=built_path, sha256=sha256_file(built_path), byte_size=built_path.stat().st_size, summary={}, ) reservation = ReservedReport(45, 3, report.company) repository = FakeRepository(fail_activation=True) publisher = AtomicReportPublisher(project_root, output_root) with self.assertRaises(PublicationError) as caught: publisher.publish(report, reservation, built, repository, work_dir) self.assertEqual(caught.exception.code, ErrorCode.PUBLISH_FAILED) self.assertEqual(current.read_bytes(), b"synthetic-old-workbook") version_root = month_root / "archive" / "v0003" self.assertFalse((version_root / report.filename).exists()) self.assertFalse((version_root / "HanaTour-v0003.result.json").exists()) self.assertEqual(repository.failed[0], reservation) self.assertEqual(repository.failed[1], ErrorCode.PUBLISH_FAILED) def test_same_semantic_retry_reuses_first_publication_when_binary_changes(self): with tempfile.TemporaryDirectory(prefix="company-report-publish-test-") as temp_dir: project_root = Path(temp_dir) / "project" output_root = project_root / "outputs" / "company_reports" work_dir = project_root / "staging" project_root.mkdir(parents=True) work_dir.mkdir() report = empty_report() semantic_sha256 = "a" * 64 reservation = ReservedReport(46, 4, report.company) repository = FakeRepository() publisher = AtomicReportPublisher(project_root, output_root) first_built = synthetic_built_workbook( work_dir, report.filename, b"synthetic-first-workbook", semantic_sha256, ) first = publisher.publish( report, reservation, first_built, repository, work_dir ) archive_before = first.archive_path.read_bytes() result_before = first.result_path.read_bytes() current_before = first.current_path.read_bytes() second_built = synthetic_built_workbook( work_dir, report.filename, b"synthetic-rebuilt-workbook-with-different-bytes", semantic_sha256, ) self.assertNotEqual(first_built.sha256, second_built.sha256) second = publisher.publish( report, reservation, second_built, repository, work_dir ) self.assertEqual(second.archive_path.read_bytes(), archive_before) self.assertEqual(second.result_path.read_bytes(), result_before) self.assertEqual(second.current_path.read_bytes(), current_before) self.assertEqual(second.artifact.sha256, first.artifact.sha256) self.assertEqual(second.result_json.sha256, first.result_json.sha256) self.assertEqual(repository.activated[1].sha256, first.artifact.sha256) self.assertIsNone(repository.failed) def test_semantic_mismatch_does_not_replace_existing_publication(self): with tempfile.TemporaryDirectory(prefix="company-report-publish-test-") as temp_dir: project_root = Path(temp_dir) / "project" output_root = project_root / "outputs" / "company_reports" work_dir = project_root / "staging" project_root.mkdir(parents=True) work_dir.mkdir() report = empty_report() reservation = ReservedReport(47, 5, report.company) repository = FakeRepository() publisher = AtomicReportPublisher(project_root, output_root) first_built = synthetic_built_workbook( work_dir, report.filename, b"synthetic-first-workbook", "a" * 64 ) first = publisher.publish( report, reservation, first_built, repository, work_dir ) archive_before = first.archive_path.read_bytes() result_before = first.result_path.read_bytes() current_before = first.current_path.read_bytes() second_built = synthetic_built_workbook( work_dir, report.filename, b"synthetic-conflicting-workbook", "b" * 64 ) with self.assertRaises(PublicationError) as caught: publisher.publish(report, reservation, second_built, repository, work_dir) self.assertEqual(caught.exception.code, ErrorCode.PUBLISH_FAILED) self.assertEqual(first.archive_path.read_bytes(), archive_before) self.assertEqual(first.result_path.read_bytes(), result_before) self.assertEqual(first.current_path.read_bytes(), current_before) self.assertEqual(repository.failed[0], reservation) self.assertEqual(repository.failed[1], ErrorCode.PUBLISH_FAILED) def test_partial_existing_publication_fails_closed(self): with tempfile.TemporaryDirectory(prefix="company-report-publish-test-") as temp_dir: project_root = Path(temp_dir) / "project" output_root = project_root / "outputs" / "company_reports" work_dir = project_root / "staging" project_root.mkdir(parents=True) work_dir.mkdir() report = empty_report() reservation = ReservedReport(48, 6, report.company) repository = FakeRepository() publisher = AtomicReportPublisher(project_root, output_root) built = synthetic_built_workbook( work_dir, report.filename, b"synthetic-first-workbook", "c" * 64 ) first = publisher.publish(report, reservation, built, repository, work_dir) first.result_path.unlink() with self.assertRaises(PublicationError) as caught: publisher.publish(report, reservation, built, repository, work_dir) self.assertEqual(caught.exception.code, ErrorCode.PUBLISH_FAILED) self.assertTrue(first.archive_path.is_file()) self.assertFalse(first.result_path.exists()) self.assertEqual(repository.failed[0], reservation) def test_corrupt_existing_archive_fails_closed(self): with tempfile.TemporaryDirectory(prefix="company-report-publish-test-") as temp_dir: project_root = Path(temp_dir) / "project" output_root = project_root / "outputs" / "company_reports" work_dir = project_root / "staging" project_root.mkdir(parents=True) work_dir.mkdir() report = empty_report() reservation = ReservedReport(49, 7, report.company) repository = FakeRepository() publisher = AtomicReportPublisher(project_root, output_root) built = synthetic_built_workbook( work_dir, report.filename, b"synthetic-first-workbook", "d" * 64 ) first = publisher.publish(report, reservation, built, repository, work_dir) first.archive_path.write_bytes(b"corrupted-archive") with self.assertRaises(PublicationError) as caught: publisher.publish(report, reservation, built, repository, work_dir) self.assertEqual(caught.exception.code, ErrorCode.PUBLISH_FAILED) self.assertEqual(first.current_path.read_bytes(), b"synthetic-first-workbook") self.assertEqual(repository.failed[0], reservation) if __name__ == "__main__": unittest.main()