external_demand/tests/test_festival_detection.py

"""节日检测流程集成测试（含代码兜底）。"""

from __future__ import annotations

import unittest
from datetime import date
from unittest.mock import patch

from app.festival_demand.demand_generation import generate_demands_from_matches
from app.festival_demand.detection import detect_active_festivals
from app.festival_demand.exceptions import FestivalDemandError
from app.festival_demand.fixed_dates import merge_festival_detections
from app.festival_demand.types import FestivalDemandConfig


def _test_config() -> FestivalDemandConfig:
    return FestivalDemandConfig(
        cron_hours="7",
        cron_minute=0,
        llm_model="test-model",
        llm_max_attempts=1,
        llm_retry_sleep_seconds=0.0,
        llm_max_tokens=100,
        llm_temperature=0.0,
        demand_pool_source_table="dwd_multi_demand_pool_di",
        demand_pool_strategy="去年同期阳历",
        output_strategy="去年同期阳历-节点事件",
        match_batch_size=200,
    )


class FestivalDetectionIntegrationTest(unittest.TestCase):
    def test_code_fallback_when_llm_returns_empty(self) -> None:
        llm_result = {
            "date": "2026-08-03",
            "festivals": [],
            "festival_names": [],
        }
        with patch(
            "app.festival_demand.detection._detect_active_festivals_via_llm",
            return_value=llm_result,
        ):
            result = detect_active_festivals(date(2026, 8, 3), _test_config())

        self.assertIn("立秋", result["festival_names"])
        liqiu = next(item for item in result["festivals"] if item["name"] == "立秋")
        self.assertEqual(liqiu["detection_source"], "code")
        self.assertEqual(liqiu["event_type"], "节日/节气")
        self.assertEqual(liqiu["phase"], "prewarm")

    def test_code_fallback_when_llm_fails(self) -> None:
        with patch(
            "app.festival_demand.detection._detect_active_festivals_via_llm",
            side_effect=FestivalDemandError("llm unavailable"),
        ):
            result = detect_active_festivals(date(2026, 8, 3), _test_config())

        self.assertIn("立秋", result["festival_names"])

    def test_code_overrides_llm_for_same_event(self) -> None:
        llm_festivals = [
            {
                "name": "立秋",
                "kind": "non_statutory",
                "event_date": "2026-08-08",
                "festival_start": "2026-08-08",
                "festival_end": "2026-08-08",
                "phase": "prewarm",
                "days_to_festival": 5,
                "event_type": "节日/节气",
                "detection_source": "llm",
            }
        ]
        code_festivals = [
            {
                "name": "立秋",
                "kind": "non_statutory",
                "event_date": "2026-08-07",
                "festival_start": "2026-08-07",
                "festival_end": "2026-08-07",
                "phase": "prewarm",
                "days_to_festival": 4,
                "event_type": "节日/节气",
                "detection_source": "code",
            }
        ]
        merged = merge_festival_detections(llm_festivals, code_festivals)
        liqiu = next(item for item in merged if item["name"] == "立秋")
        self.assertEqual(liqiu["detection_source"], "code")
        self.assertEqual(liqiu["event_date"], "2026-08-07")

    def test_downstream_demand_generation_accepts_code_festival(self) -> None:
        matches = [
            {
                "demand_name": "贴秋膘",
                "festival_name": "立秋",
                "event_type": "节日/节气",
            }
        ]
        generated = generate_demands_from_matches(matches, query_date=date(2026, 8, 3))
        self.assertIn("立秋", generated["generated_demand_names"])
        self.assertIn("2026 立秋", generated["generated_demand_names"])
        self.assertIn("2026 立秋 贴秋膘", generated["generated_demand_names"])


if __name__ == "__main__":
    unittest.main()