178 lines
7.2 KiB
Python
178 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from dataclasses import replace
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from company_reports.publishing import (
|
|
ArtifactToolBuilder,
|
|
AtomicReportPublisher,
|
|
BuiltWorkbook,
|
|
sha256_file,
|
|
)
|
|
from company_reports.repository import ReservedReport
|
|
from company_reports.service import CompanyReportService, RunRequest
|
|
from tests.test_company_reports_service import FakeRepository, synthetic_snapshot
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
BUILDER_SCRIPT = PROJECT_ROOT / "company_reports" / "xlsx" / "build_workbook.mjs"
|
|
|
|
|
|
class CompanyReportIntegrationTests(unittest.TestCase):
|
|
def test_missing_and_unmatched_group_codes_export_blank_booking_room(self):
|
|
class CapturingBuilder:
|
|
def __init__(self, delegate):
|
|
self.delegate = delegate
|
|
self.reports = []
|
|
|
|
def build(self, report, work_dir):
|
|
self.reports.append(report)
|
|
return self.delegate.build(report, work_dir)
|
|
|
|
with tempfile.TemporaryDirectory(prefix="company-report-integration-") as temp_dir:
|
|
project_root = Path(temp_dir) / "project"
|
|
output_root = project_root / "outputs" / "company_reports"
|
|
project_root.mkdir(parents=True)
|
|
repository = FakeRepository(
|
|
replace(
|
|
synthetic_snapshot(include_invalid_hana=True),
|
|
room_items=(),
|
|
group_parse_versions={},
|
|
)
|
|
)
|
|
builder = CapturingBuilder(
|
|
ArtifactToolBuilder(BUILDER_SCRIPT, timeout_seconds=90)
|
|
)
|
|
publisher = AtomicReportPublisher(project_root, output_root)
|
|
service = CompanyReportService(
|
|
repository,
|
|
builder,
|
|
publisher,
|
|
output_root / ".staging",
|
|
)
|
|
|
|
result = service.run(
|
|
RunRequest(2026, 7, date(2026, 7, 10), ("QBD", "HanaTour"))
|
|
)
|
|
|
|
self.assertEqual(result.status, "success")
|
|
self.assertEqual([item.row_count for item in result.companies], [1, 1])
|
|
self.assertTrue(all(item.errors == () for item in result.companies))
|
|
self.assertEqual(len(builder.reports), 2)
|
|
qbd_report, hana_report = builder.reports
|
|
self.assertEqual(qbd_report.periods[0].rows[0].res_comment, "SYN-GROUP-A")
|
|
self.assertEqual(qbd_report.periods[0].rows[0].booking_room, "")
|
|
self.assertEqual(hana_report.periods[0].rows[0].res_comment, "")
|
|
self.assertEqual(hana_report.periods[0].rows[0].booking_room, "")
|
|
self.assertTrue(
|
|
(output_root / "2026" / "07" / "QBD-July-2026.xlsx").is_file()
|
|
)
|
|
self.assertTrue(
|
|
(output_root / "2026" / "07" / "HanaTour-July-2026.xlsx").is_file()
|
|
)
|
|
|
|
def test_same_snapshot_rerun_reuses_publication_when_binary_changes(self):
|
|
class StableVersionRepository(FakeRepository):
|
|
def reserve_report(self, report):
|
|
reservation = ReservedReport(9001, 1, report.company)
|
|
self.reserved.append(reservation)
|
|
return reservation
|
|
|
|
class RebuildingBuilder:
|
|
def __init__(self, delegate):
|
|
self.delegate = delegate
|
|
self.binary_sha256s = []
|
|
|
|
def build(self, report, work_dir):
|
|
built = self.delegate.build(report, work_dir)
|
|
with built.path.open("ab") as handle:
|
|
handle.write(
|
|
f"\nsynthetic-rebuild-marker-{len(self.binary_sha256s)}".encode(
|
|
"utf-8"
|
|
)
|
|
)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.chmod(built.path, 0o600)
|
|
rebuilt = BuiltWorkbook(
|
|
path=built.path,
|
|
sha256=sha256_file(built.path),
|
|
byte_size=built.path.stat().st_size,
|
|
summary=built.summary,
|
|
)
|
|
self.binary_sha256s.append(rebuilt.sha256)
|
|
return rebuilt
|
|
|
|
with tempfile.TemporaryDirectory(prefix="company-report-integration-") as temp_dir:
|
|
project_root = Path(temp_dir) / "project"
|
|
output_root = project_root / "outputs" / "company_reports"
|
|
project_root.mkdir(parents=True)
|
|
repository = StableVersionRepository(synthetic_snapshot())
|
|
builder = RebuildingBuilder(
|
|
ArtifactToolBuilder(BUILDER_SCRIPT, timeout_seconds=90)
|
|
)
|
|
publisher = AtomicReportPublisher(project_root, output_root)
|
|
service = CompanyReportService(
|
|
repository,
|
|
builder,
|
|
publisher,
|
|
output_root / ".staging",
|
|
)
|
|
request = RunRequest(2026, 7, date(2026, 7, 10), ("QBD",))
|
|
|
|
first = service.run(request)
|
|
second = service.run(request)
|
|
|
|
self.assertEqual(first.status, "success")
|
|
self.assertEqual(second.status, "success")
|
|
self.assertEqual(first.companies[0].row_count, 1)
|
|
self.assertEqual(second.companies[0].row_count, 1)
|
|
first_semantic = first.companies[0].artifact["semantic_sha256"]
|
|
second_semantic = second.companies[0].artifact["semantic_sha256"]
|
|
self.assertEqual(len(first_semantic), 64)
|
|
self.assertEqual(first_semantic, second_semantic)
|
|
self.assertNotEqual(builder.binary_sha256s[0], builder.binary_sha256s[1])
|
|
self.assertEqual(
|
|
first.companies[0].artifact["sha256"],
|
|
second.companies[0].artifact["sha256"],
|
|
)
|
|
self.assertEqual(
|
|
first.companies[0].period_row_counts,
|
|
second.companies[0].period_row_counts,
|
|
)
|
|
self.assertEqual(first.companies[0].version_no, 1)
|
|
self.assertEqual(second.companies[0].version_no, 1)
|
|
|
|
month_root = output_root / "2026" / "07"
|
|
filename = "QBD-July-2026.xlsx"
|
|
current = month_root / filename
|
|
first_archive = month_root / "archive" / "v0001" / filename
|
|
self.assertTrue(first_archive.is_file())
|
|
self.assertFalse((month_root / "archive" / "v0002").exists())
|
|
first_archive_bytes = first_archive.read_bytes()
|
|
self.assertEqual(current.read_bytes(), first_archive_bytes)
|
|
self.assertEqual(current.stat().st_mode & 0o777, 0o600)
|
|
self.assertEqual(len(repository.activated), 2)
|
|
|
|
first_result = json.loads(
|
|
(month_root / "archive" / "v0001" / "QBD-v0001.result.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
first_result["artifact"]["semantic_sha256"],
|
|
second.companies[0].artifact["semantic_sha256"],
|
|
)
|
|
serialized = json.dumps(second.to_dict(), ensure_ascii=False)
|
|
self.assertNotIn("SYN-GROUP-A", serialized)
|
|
self.assertNotIn("SYN-BLOCK-A", serialized)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|