182 lines
7.6 KiB
Python
182 lines
7.6 KiB
Python
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)
|