100 lines
3.6 KiB
Python
100 lines
3.6 KiB
Python
import asyncio
|
||
|
||
|
||
def test_elephant_orchestrator_extracts_fenced_json_object():
|
||
from services.elephant_alpha_orchestrator import ElephantAlphaOrchestrator
|
||
|
||
payload = ElephantAlphaOrchestrator._extract_json_object(
|
||
"""```json
|
||
{"priority":"high","confidence":0.91}
|
||
```"""
|
||
)
|
||
|
||
assert payload == {"priority": "high", "confidence": 0.91}
|
||
|
||
|
||
def test_elephant_orchestrator_extracts_json_object_from_prefaced_text():
|
||
from services.elephant_alpha_orchestrator import ElephantAlphaOrchestrator
|
||
|
||
payload = ElephantAlphaOrchestrator._extract_json_object(
|
||
'先做分析：\n{"priority":"medium","agents_required":["hermes"]}\n以上。'
|
||
)
|
||
|
||
assert payload["priority"] == "medium"
|
||
assert payload["agents_required"] == ["hermes"]
|
||
|
||
|
||
def test_elephant_orchestrator_fallback_uses_concrete_evidence_without_plan():
|
||
from services.elephant_alpha_orchestrator import ElephantAlphaOrchestrator
|
||
|
||
orchestrator = ElephantAlphaOrchestrator()
|
||
decision = orchestrator._fallback_decision(
|
||
{
|
||
"trigger_type": "market_opportunity",
|
||
"conditions": {
|
||
"_prefetched_hermes_threats": [
|
||
"[SKU-1] 測試商品｜MOMO $1,200 vs PChome $990｜每件價差 NT$ 210"
|
||
]
|
||
},
|
||
},
|
||
reason="測試解析失敗",
|
||
)
|
||
|
||
assert decision.priority == "high"
|
||
assert decision.confidence == 0.74
|
||
assert decision.execution_plan == []
|
||
assert "1 筆 DB/Hermes 價格比對實證" in decision.reasoning
|
||
assert "自動調價" in decision.reasoning
|
||
|
||
|
||
def test_elephant_orchestrator_fallback_without_evidence_has_no_openclaw_action():
|
||
from services.elephant_alpha_orchestrator import ElephantAlphaOrchestrator
|
||
|
||
orchestrator = ElephantAlphaOrchestrator()
|
||
decision = orchestrator._fallback_decision(
|
||
{"trigger_type": "market_opportunity", "conditions": {}},
|
||
reason="測試解析失敗",
|
||
)
|
||
|
||
assert decision.confidence == 0.6
|
||
assert decision.execution_plan == []
|
||
assert decision.agents_required == ["elephant_alpha"]
|
||
assert "沒有可稽核的 DB/Hermes 實證" in decision.reasoning
|
||
|
||
|
||
def test_elephant_orchestrator_uses_evidence_fallback_on_non_json_response(monkeypatch):
|
||
from services.elephant_service import ElephantResponse
|
||
from services.elephant_alpha_orchestrator import ElephantAlphaOrchestrator
|
||
|
||
class FakeElephant:
|
||
def generate(self, **_kwargs):
|
||
return ElephantResponse(
|
||
success=True,
|
||
content="\n我無法輸出 JSON，但建議先處理。",
|
||
model="fake-model",
|
||
)
|
||
|
||
async def _noop_log(*_args, **_kwargs):
|
||
raise AssertionError("parse fallback should not log an LLM decision")
|
||
|
||
orchestrator = ElephantAlphaOrchestrator()
|
||
orchestrator.elephant = FakeElephant()
|
||
monkeypatch.setattr(orchestrator, "_get_recent_performance_metrics", lambda: {})
|
||
monkeypatch.setattr(orchestrator, "_get_agent_status", lambda: {})
|
||
monkeypatch.setattr(orchestrator, "_get_system_load", lambda: "normal")
|
||
monkeypatch.setattr(orchestrator, "_get_pending_actions_count", lambda: 0)
|
||
monkeypatch.setattr(orchestrator, "_log_decision", _noop_log)
|
||
|
||
decision = asyncio.run(orchestrator.analyze_and_coordinate({
|
||
"trigger_type": "price_drop_alert",
|
||
"conditions": {
|
||
"_prefetched_hermes_threats": [
|
||
"[SKU-9] 實證商品｜MOMO $1,000 vs PChome $800｜每件價差 NT$ 200"
|
||
]
|
||
},
|
||
}))
|
||
|
||
assert decision.priority == "high"
|
||
assert decision.execution_plan == []
|
||
assert "1 筆 DB/Hermes 價格比對實證" in decision.reasoning
|