from dataclasses import dataclass import pytest @pytest.fixture(autouse=True) def _force_legacy_nim_first(monkeypatch): """Fallback tests cover the legacy NIM path, not live qwen3/Ollama routing.""" import services.nemoton_dispatcher_service as module module._ALERT_CACHE.clear() monkeypatch.setattr(module, "NEMOTRON_OLLAMA_FIRST", False) yield module._ALERT_CACHE.clear() @dataclass class FakeThreat: sku: str name: str momo_price: float = 100.0 pchome_price: float = 80.0 gap_pct: float = 25.0 sales_7d_delta_pct: float = -20.0 risk: str = "HIGH" recommended_action: str = "請評估價格策略" confidence: float = 0.8 def _patch_execution_methods(monkeypatch, dispatcher): calls = [] def record(kind): def _inner(*args, **kwargs): calls.append({"kind": kind, "args": args, "kwargs": kwargs}) return _inner monkeypatch.setattr(dispatcher, "_exec_trigger_price_alert", record("price_alert")) monkeypatch.setattr(dispatcher, "_exec_add_to_recommendation", record("recommendation")) monkeypatch.setattr(dispatcher, "_exec_flag_for_human_review", record("human_review")) return calls def test_dispatch_falls_back_to_hermes_rules_without_nim_api_key(monkeypatch): import services.nemoton_dispatcher_service as module monkeypatch.setattr(module, "NIM_API_KEY", "") dispatcher = module.NemotronDispatcher() calls = _patch_execution_methods(monkeypatch, dispatcher) result = dispatcher.dispatch([FakeThreat("SKU-1", "測試品")], hermes_stats={"duration_sec": 1}) assert result["dispatched"] == 1 assert result["skipped"] == 0 assert result["nim_stats"]["degraded"] is True assert calls[0]["kind"] == "price_alert" def test_dispatch_routes_non_exact_match_to_human_review(monkeypatch): import services.nemoton_dispatcher_service as module monkeypatch.setattr(module, "NIM_API_KEY", "") dispatcher = module.NemotronDispatcher() calls = _patch_execution_methods(monkeypatch, dispatcher) threat = FakeThreat("SKU-UNIT", "測試品 2入組") threat.match_type = "same_product_different_pack" threat.price_basis = "unit_price" threat.alert_tier = "unit_price_review" threat.match_score = 0.74 result = dispatcher.dispatch([threat], hermes_stats={"duration_sec": 1}) assert result["dispatched"] == 1 assert calls[0]["kind"] == "human_review" assert "unit_price_review" in calls[0]["kwargs"]["concern"] def test_dispatch_falls_back_to_hermes_rules_on_nim_timeout(monkeypatch): import requests import services.nemoton_dispatcher_service as module monkeypatch.setattr(module, "NIM_API_KEY", "test-key") monkeypatch.setattr(module, "_check_nim_quota", lambda: True) dispatcher = module.NemotronDispatcher() calls = _patch_execution_methods(monkeypatch, dispatcher) monkeypatch.setattr(dispatcher, "_call_nim", lambda threats: (_ for _ in ()).throw(requests.Timeout("timeout"))) result = dispatcher.dispatch([FakeThreat("SKU-2", "測試品")], hermes_stats={"duration_sec": 1}) assert result["dispatched"] == 1 assert result["skipped"] == 0 assert result["nim_stats"]["degraded"] is True assert result["errors"] == ["timeout"] assert calls[0]["kind"] == "price_alert" def test_dispatch_falls_back_to_hermes_rules_on_zero_tool_calls(monkeypatch): import services.nemoton_dispatcher_service as module monkeypatch.setattr(module, "NIM_API_KEY", "test-key") monkeypatch.setattr(module, "_check_nim_quota", lambda: True) dispatcher = module.NemotronDispatcher() calls = _patch_execution_methods(monkeypatch, dispatcher) monkeypatch.setattr(dispatcher, "_call_nim", lambda threats: ([], {"total_tokens": 10})) result = dispatcher.dispatch([FakeThreat("SKU-3", "測試品")], hermes_stats={"duration_sec": 1}) assert result["dispatched"] == 1 assert result["skipped"] == 0 assert result["nim_stats"]["degraded"] is True assert calls[0]["kind"] == "price_alert"