補強 AI 自動化閉環與安全降級

tests/test_code_review_pipeline_security.py

@@ -0,0 +1,49 @@
+def test_verify_internal_token_requires_env_by_default(monkeypatch):
+    import services.code_review_pipeline_service as module
+
+    monkeypatch.setattr(module, "INTERNAL_TOKEN", "")
+    monkeypatch.setattr(module, "ALLOW_INSECURE_WEBHOOK", False)
+
+    assert module.verify_internal_token("") is False
+    assert module.verify_internal_token("anything") is False
+
+
+def test_verify_internal_token_allows_explicit_dev_override(monkeypatch):
+    import services.code_review_pipeline_service as module
+
+    monkeypatch.setattr(module, "INTERNAL_TOKEN", "")
+    monkeypatch.setattr(module, "ALLOW_INSECURE_WEBHOOK", True)
+
+    assert module.verify_internal_token("") is True
+
+
+def test_code_review_guard_blocks_high_risk_auto_fix(monkeypatch):
+    import services.code_review_pipeline_service as module
+
+    monkeypatch.setattr(module, "AUTO_FIX_ENABLED", True)
+    pipeline = module.CodeReviewPipeline("abcdef123456", ["services/example.py"])
+    pipeline.state["severity_summary"] = {"critical": 0, "high": 1, "medium": 0, "low": 0}
+
+    guarded = pipeline._guard_ea_decision(
+        {"priority": "high", "auto_fix": True, "reasoning": "建議修復", "fix_files": ["services/example.py"]},
+        [{"severity": "HIGH", "file": "services/example.py"}],
+    )
+
+    assert guarded["auto_fix"] is False
+    assert guarded["human_review_needed"] is True
+
+
+def test_code_review_guard_requires_auto_fix_feature_flag(monkeypatch):
+    import services.code_review_pipeline_service as module
+
+    monkeypatch.setattr(module, "AUTO_FIX_ENABLED", False)
+    pipeline = module.CodeReviewPipeline("abcdef123456", ["services/example.py"])
+    pipeline.state["severity_summary"] = {"critical": 0, "high": 0, "medium": 1, "low": 0}
+
+    guarded = pipeline._guard_ea_decision(
+        {"priority": "medium", "auto_fix": True, "reasoning": "建議修復", "fix_files": ["services/example.py"]},
+        [{"severity": "MEDIUM", "file": "services/example.py"}],
+    )
+
+    assert guarded["auto_fix"] is False
+    assert guarded["human_review_needed"] is True

tests/test_event_router.py

@@ -0,0 +1,87 @@
+import asyncio
+import json
+
+
+def test_dispatch_degrades_and_queues_when_ai_or_telegram_fails(tmp_path, monkeypatch):
+    import services.event_router as event_router
+
+    queue_path = tmp_path / "event_router_failed.jsonl"
+    monkeypatch.setattr(event_router, "_QUEUE_PATH", str(queue_path))
+    monkeypatch.setattr(event_router, "_is_event_silenced", lambda event: False)
+
+    async def broken_l1(event, session_id):
+        raise RuntimeError("hermes unavailable")
+
+    monkeypatch.setattr(event_router, "_handle_l1", broken_l1)
+    monkeypatch.setattr(
+        event_router,
+        "send_telegram_with_result",
+        lambda *args, **kwargs: {
+            "ok": False,
+            "sent": 0,
+            "failed": 1,
+            "chat_ids": [123],
+            "errors": ["123:HTTP 500"],
+        },
+    )
+
+    result = asyncio.run(event_router.dispatch({
+        "source": "Scheduler.Test",
+        "event_type": "crawler_timeout",
+        "severity": "warning",
+        "title": "測試任務異常",
+        "summary": "timeout",
+        "trace": "Traceback...",
+    }))
+
+    assert result["tier"] == "L1"
+    assert result["delivered"] is False
+    assert result["queued"] is True
+    assert result["payload"]["status"] == "degraded"
+    queued = json.loads(queue_path.read_text(encoding="utf-8").strip())
+    assert queued["reason"] == "telegram_delivery_failed"
+    assert queued["event_key"] == "Scheduler.Test:crawler_timeout"
+
+
+def test_dispatch_executes_only_auto_safe_actions(monkeypatch):
+    import services.agent_actions as agent_actions
+    import services.event_router as event_router
+
+    calls = []
+    monkeypatch.setattr(event_router, "_is_event_silenced", lambda event: False)
+    monkeypatch.setitem(
+        agent_actions.SAFE_ACTIONS,
+        "retry_task",
+        lambda **params: calls.append(params) or {"status": "scheduled"},
+    )
+
+    async def planned_l2(event, session_id):
+        return {
+            "dispatch_to": "safe_action",
+            "auto_execute": True,
+            "action_plan": [
+                {"action": "retry_task", "params": {"task_name": "run_momo_task"}},
+                {"action": "restart_container", "params": {"container": "momo-db"}},
+            ],
+        }
+
+    monkeypatch.setattr(event_router, "_handle_l2", planned_l2)
+    monkeypatch.setattr(
+        event_router,
+        "send_telegram_with_result",
+        lambda *args, **kwargs: {"ok": True, "sent": 1, "failed": 0, "chat_ids": [123], "errors": []},
+    )
+
+    result = asyncio.run(event_router.dispatch({
+        "source": "Scheduler.MOMO",
+        "event_type": "scheduler_task_failure",
+        "severity": "alert",
+        "title": "MOMO 任務異常",
+        "summary": "boom",
+        "payload": {"task_name": "run_momo_task"},
+    }))
+
+    assert result["tier"] == "L2"
+    assert calls == [{"task_name": "run_momo_task"}]
+    assert result["payload"]["executed_actions"][0]["status"] == "ok"
+    assert result["payload"]["executed_actions"][1]["status"] == "rejected"

tests/test_nemotron_fallback.py

@@ -0,0 +1,79 @@
+from dataclasses import dataclass
+
+
+@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_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"