211 lines
7.2 KiB
Python
211 lines
7.2 KiB
Python
from datetime import datetime
|
||
import json
|
||
|
||
|
||
def test_build_defined_ppt_jobs_uses_latest_date():
|
||
from services.ppt_auto_generation_service import build_defined_ppt_jobs
|
||
|
||
jobs = build_defined_ppt_jobs(latest_date="2026-05-11")
|
||
by_type = {job.report_type: job for job in jobs}
|
||
|
||
assert list(by_type) == [
|
||
"daily", "weekly", "monthly", "quarterly", "half_yearly", "annual", "ttm",
|
||
"strategy", "competitor", "competitor_v4", "promo", "promo_compare",
|
||
"forecast_pre_event", "vendor", "category", "customer", "new_product",
|
||
"market_intel", "price_elasticity",
|
||
]
|
||
assert by_type["daily"].sub_arg == "2026/05/11"
|
||
assert by_type["monthly"].sub_arg == "2026/05"
|
||
assert by_type["quarterly"].sub_arg == "2026/Q2"
|
||
assert by_type["half_yearly"].sub_arg == "2026/H1"
|
||
assert by_type["annual"].sub_arg == "2026"
|
||
assert by_type["strategy"].sub_arg == "2026/05/01-2026/05/11"
|
||
assert by_type["market_intel"].sub_arg == "2026/05/11 起一週"
|
||
assert by_type["competitor"].sub_arg == "monthly"
|
||
assert by_type["promo"].sub_arg == "2026/05/05-2026/05/11"
|
||
assert by_type["promo"].expected_params["label"] == "2026/05/05~2026/05/11"
|
||
assert by_type["strategy"].expected_params == {
|
||
"report_type": "strategy",
|
||
"start": "2026/05/01",
|
||
"end": "2026/05/11",
|
||
"label": "2026/05 月策略（截至 05/11）",
|
||
}
|
||
|
||
|
||
def test_auto_generation_respects_disabled_flag(monkeypatch):
|
||
monkeypatch.setenv("PPT_AUTO_GENERATION_ENABLED", "false")
|
||
|
||
from services.ppt_auto_generation_service import generate_defined_ppt_reports
|
||
|
||
result = generate_defined_ppt_reports(report_types=["daily"])
|
||
|
||
assert result["ok"] is False
|
||
assert result["status"] == "disabled"
|
||
|
||
|
||
def test_dry_run_does_not_generate(monkeypatch):
|
||
monkeypatch.setenv("PPT_AUTO_GENERATION_ENABLED", "true")
|
||
|
||
from services import ppt_auto_generation_service as svc
|
||
|
||
monkeypatch.setattr(svc, "_latest_sales_date", lambda: "2026-05-11")
|
||
|
||
result = svc.generate_defined_ppt_reports(
|
||
report_types=["daily", "monthly"],
|
||
dry_run=True,
|
||
)
|
||
|
||
assert result["ok"] is True
|
||
assert result["status"] == "planned"
|
||
assert [job["report_type"] for job in result["jobs"]] == ["daily", "monthly"]
|
||
|
||
|
||
def test_coverage_marks_ready_from_database(monkeypatch):
|
||
from services import ppt_auto_generation_service as svc
|
||
|
||
class _Rows:
|
||
def fetchall(self):
|
||
return [
|
||
(
|
||
"daily",
|
||
json.dumps({"report_type": "daily", "date": "2026/05/11"}),
|
||
"/tmp/ocbot_daily_ok.pptx",
|
||
datetime(2026, 5, 11, 20, 30),
|
||
),
|
||
(
|
||
"monthly",
|
||
json.dumps({"report_type": "monthly", "month": "2026/05"}),
|
||
"/tmp/ocbot_monthly_ok.pptx",
|
||
datetime(2026, 5, 11, 20, 30),
|
||
),
|
||
]
|
||
|
||
class _Session:
|
||
def execute(self, *_args, **_kwargs):
|
||
return _Rows()
|
||
|
||
def close(self):
|
||
return None
|
||
|
||
monkeypatch.setattr(svc, "get_session", lambda: _Session())
|
||
monkeypatch.setattr(svc, "_latest_sales_date", lambda: "2026-05-11")
|
||
monkeypatch.setenv("PPT_AUTO_GENERATION_ENABLED", "true")
|
||
|
||
result = svc.get_defined_report_coverage(
|
||
month_start=datetime(2026, 5, 1),
|
||
month_end=datetime(2026, 6, 1),
|
||
reports_dir="/tmp/does-not-exist-for-test",
|
||
report_types=["daily", "monthly", "weekly"],
|
||
)
|
||
|
||
by_key = {item["key"]: item for item in result["items"]}
|
||
assert by_key["daily"]["ready"] is True
|
||
assert by_key["daily"]["status"] == "ready"
|
||
assert by_key["daily"]["status_label"] == "已產出"
|
||
assert by_key["monthly"]["ready"] is True
|
||
assert by_key["weekly"]["ready"] is False
|
||
assert by_key["weekly"]["status"] == "missing"
|
||
assert by_key["weekly"]["status_label"] == "待排程補齊"
|
||
assert result["missing_count"] == 1
|
||
|
||
|
||
def test_due_schedule_kinds_include_periodic_boundaries():
|
||
from services.ppt_auto_generation_service import get_due_schedule_kinds
|
||
|
||
assert get_due_schedule_kinds(datetime(2026, 5, 18)) == ["daily", "weekly"]
|
||
assert get_due_schedule_kinds(datetime(2026, 7, 1)) == [
|
||
"daily",
|
||
"monthly",
|
||
"quarterly",
|
||
"half_yearly",
|
||
]
|
||
assert get_due_schedule_kinds(datetime(2026, 1, 1)) == [
|
||
"daily",
|
||
"monthly",
|
||
"quarterly",
|
||
"half_yearly",
|
||
"annual",
|
||
]
|
||
|
||
|
||
def test_schedule_cadence_status_exposes_all_periodic_contracts():
|
||
from services.ppt_auto_generation_service import get_schedule_cadence_status
|
||
|
||
cadences = get_schedule_cadence_status([
|
||
{"key": "daily", "ready": True},
|
||
{"key": "weekly", "ready": False},
|
||
{"key": "market_intel", "ready": True},
|
||
])
|
||
by_key = {cadence["key"]: cadence for cadence in cadences}
|
||
|
||
assert [cadence["schedule_text"] for cadence in cadences] == [
|
||
"每日 20:30",
|
||
"每週一 20:40",
|
||
"每月 1 日 20:50",
|
||
"每季首日 21:00",
|
||
"每半年首日 21:10",
|
||
"每年 1/1 21:20",
|
||
]
|
||
assert by_key["weekly"]["report_types"] == ["weekly", "market_intel"]
|
||
assert by_key["weekly"]["ready_count"] == 1
|
||
assert by_key["weekly"]["missing_report_types"] == ["weekly"]
|
||
assert by_key["weekly"]["missing_report_labels"] == ["週報"]
|
||
assert by_key["weekly"]["status_label"] == "已完成 1/2"
|
||
assert by_key["weekly"]["coverage_text"] == "1/2"
|
||
assert "TTM 滾動 12 月" in by_key["monthly"]["report_labels"]
|
||
|
||
|
||
def test_scheduled_generation_uses_profile_without_generating(monkeypatch):
|
||
from services import ppt_auto_generation_service as svc
|
||
|
||
calls = []
|
||
|
||
def fake_generate_defined_ppt_reports(**kwargs):
|
||
calls.append(kwargs)
|
||
return {"ok": True, "ready": 2, "errors": 0, "jobs": [{"report_type": "weekly"}]}
|
||
|
||
monkeypatch.setattr(svc, "generate_defined_ppt_reports", fake_generate_defined_ppt_reports)
|
||
|
||
result = svc.generate_scheduled_ppt_reports(schedule_kind="weekly", force=True)
|
||
|
||
assert result["ok"] is True
|
||
assert result["schedule_kinds"] == ["weekly"]
|
||
assert calls == [{
|
||
"report_types": ("weekly", "market_intel"),
|
||
"schedule_kind": "weekly",
|
||
"force": True,
|
||
}]
|
||
|
||
|
||
def test_ppt_preview_reports_missing_converter(monkeypatch, tmp_path):
|
||
from services import ppt_preview_service as svc
|
||
|
||
pptx = tmp_path / "demo.pptx"
|
||
pptx.write_bytes(b"fake pptx")
|
||
monkeypatch.setattr(svc.shutil, "which", lambda _name: None)
|
||
|
||
result = svc.build_ppt_preview(pptx, cache_dir=tmp_path / "cache")
|
||
|
||
assert result.ok is False
|
||
assert "LibreOffice" in (result.error or "")
|
||
|
||
|
||
def test_ppt_preview_cache_info_is_read_only(tmp_path):
|
||
from pathlib import Path
|
||
from services import ppt_preview_service as svc
|
||
|
||
pptx = tmp_path / "demo.pptx"
|
||
pptx.write_bytes(b"fake pptx")
|
||
cache_dir = tmp_path / "cache"
|
||
|
||
miss = svc.get_ppt_preview_cache_info(pptx, cache_dir=cache_dir)
|
||
assert miss.pdf_path
|
||
assert miss.cache_exists is False
|
||
|
||
Path(miss.pdf_path).parent.mkdir(parents=True, exist_ok=True)
|
||
Path(miss.pdf_path).write_bytes(b"%PDF-1.4\n")
|
||
|
||
hit = svc.get_ppt_preview_cache_info(pptx, cache_dir=cache_dir)
|
||
assert hit.cache_exists is True
|
||
assert hit.cache_size_kb is not None
|