from __future__ import annotations import hashlib import json import sys import unittest import zipfile from decimal import Decimal from pathlib import Path from openpyxl import load_workbook PROJECT_ROOT = Path(__file__).resolve().parents[1] SKILL_ROOT = PROJECT_ROOT / "opera-daily-channel-report" SCRIPTS = SKILL_ROOT / "scripts" PRICE_REFERENCE = SKILL_ROOT / "references" / "价格对照.xlsx" ARCHIVE = PROJECT_ROOT / "opera-daily-channel-report.zip" sys.path.insert(0, str(SCRIPTS)) import process_reports as core # noqa: E402 def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() class PackageContractTests(unittest.TestCase): def test_required_skill_layout(self): required = { "SKILL.md", "scripts/process_reports.py", "scripts/validate_outputs.py", "references/business-rules.md", "references/field-contracts.md", "references/error-contract.md", "references/codex-result.schema.json", "references/structured-output.md", "references/structured-result.schema.json", "references/价格对照.xlsx", "assets/daily-template.xlsx", "assets/channel-report-template.xlsx", } actual = { str(path.relative_to(SKILL_ROOT)) for path in SKILL_ROOT.rglob("*") if path.is_file() and path.name != ".DS_Store" and "__pycache__" not in path.parts } self.assertEqual(actual, required) def test_frozen_constants_and_company_identities(self): self.assertEqual(core.VERSION, "1.0") self.assertEqual(core.DAILY_RESULT_VERSION, "2.0") self.assertEqual(core.PROCESSOR_VERSION, "2.2.0") self.assertEqual(core.STRUCTURED_RESULT_SCHEMA_VERSION, "1.0") self.assertEqual(core.DAILY_STRUCTURED_RESULT_SCHEMA_VERSION, "2.0") self.assertEqual(core.PROCESSING_MODES, {"daily", "daily-monthly"}) self.assertEqual( core.STANDARD_SHEETS, ["LIANTAI-GROUP", "LIANTAI-FIT", "QBD", "DY-AI-Easy-KB", "FENGRUN"], ) self.assertEqual(len(core.DAILY_HEADERS), 19) self.assertEqual(core.DAILY_HEADERS[-2:], ["REAL PRICE", "TOTAL PRICE"]) self.assertEqual(len(core.CHANNEL_HEADERS), 19) self.assertEqual(core.CHANNEL_HEADERS[-2:], ["REAL PRICE", "TOTAL PRICE"]) self.assertEqual(len(core.KB_CHANNEL_HEADERS), 20) self.assertEqual(core.KB_CHANNEL_HEADERS[-1], "KB（100/晚/间）") self.assertEqual(len(core.RATE_WHITELIST), 20) self.assertEqual(core.ZERO_TOTAL_COMPANIES, {"RAINBOW/AI", "GUANGZHOU GO EASY"}) self.assertEqual(core.ZERO_TOTAL_RATE_CODES, {"LBMS", "LBSM"}) self.assertEqual(core.price_company_key("T- Rainbow Holiday Se"), "RAINBOW/AI") self.assertEqual(core.price_company_key("T- FENGRUN TRAVEL TRA"), "FENGRUN") self.assertEqual(core.price_company_key("T- HANATOUR TD CO., L"), "HANA TOUR") self.assertEqual(core.price_company_key("T- HONGTAI TRAVEL (TH"), "HONGTAI") self.assertEqual(core.price_company_key("T- Guangzhou Go Easy"), "GUANGZHOU GO EASY") def test_price_reference_is_unique_and_contains_alias_targets(self): price_map = core.load_price_map(PRICE_REFERENCE.resolve()) self.assertEqual(len(price_map), 73) cases = { ("HANA TOUR", "WHKR2100B", Decimal("2100")): Decimal("2100"), ("HANA TOUR", "WHKR2100B", Decimal("3300")): Decimal("3300"), ("FENGRUN", "GLSPCB", Decimal("1800")): Decimal("1800"), ("HONGTAI", "LBKB", Decimal("900")): Decimal("1800"), ("RAINBOW/AI", "LBSM", Decimal("900")): Decimal("0"), } for key, expected in cases.items(): with self.subTest(key=key): self.assertEqual(price_map[key], expected) def test_templates_have_exact_headers_and_no_data_or_formulas(self): daily_path = SKILL_ROOT / "assets" / "daily-template.xlsx" daily = load_workbook(daily_path, data_only=False) try: for sheet in daily.worksheets: column_count = len(core.DAILY_HEADERS) headers = [ sheet.cell(1, column).value for column in range(1, column_count + 1) ] self.assertEqual(headers, core.DAILY_HEADERS) self.assertIsNone(sheet.cell(1, column_count + 1).value) for row in sheet.iter_rows( min_row=2, max_row=max(sheet.max_row, 2), min_col=1, max_col=column_count, ): for cell in row: self.assertIn(cell.value, (None, "")) self.assertNotEqual(cell.data_type, "f") finally: daily.close() monthly_path = SKILL_ROOT / "assets" / "channel-report-template.xlsx" monthly = load_workbook(monthly_path, data_only=False) try: self.assertEqual(monthly.sheetnames, core.STANDARD_SHEETS) for sheet in monthly.worksheets: expected_headers = core.channel_headers_for_sheet(sheet.title) column_count = len(expected_headers) headers = [ sheet.cell(1, column).value for column in range(1, column_count + 1) ] self.assertEqual(headers, expected_headers) self.assertIsNone(sheet.cell(1, column_count + 1).value) for row in sheet.iter_rows( min_row=2, max_row=max(sheet.max_row, 2), min_col=1, max_col=column_count, ): for cell in row: self.assertIn(cell.value, (None, "")) self.assertNotEqual(cell.data_type, "f") finally: monthly.close() def test_result_schema_strictly_supports_legacy_and_daily_only_versions(self): schema_path = SKILL_ROOT / "references" / "codex-result.schema.json" schema = json.loads(schema_path.read_text(encoding="utf-8")) self.assertFalse(schema["additionalProperties"]) self.assertEqual( schema["properties"]["version"], {"type": "string", "enum": ["1.0", "2.0"]}, ) self.assertEqual(schema["properties"]["processing_mode"], {"type": "string", "const": "daily"}) self.assertEqual(schema["properties"]["status"]["enum"], ["success", "failed"]) self.assertEqual( schema["required"], [ "version", "status", "business_date", "month_key", "message", "metrics", "outputs", "errors", ], ) def test_zip_payload_matches_skill_source(self): source_files = { str(path.relative_to(PROJECT_ROOT)): path for path in SKILL_ROOT.rglob("*") if path.is_file() and path.name != ".DS_Store" and "__pycache__" not in path.parts } with zipfile.ZipFile(ARCHIVE) as archive: archived_files = { name: archive.read(name) for name in archive.namelist() if not name.endswith("/") } self.assertEqual(set(archived_files), set(source_files)) for name, path in source_files.items(): with self.subTest(path=name): self.assertEqual(sha256_bytes(archived_files[name]), sha256_bytes(path.read_bytes())) if __name__ == "__main__": unittest.main(verbosity=2)