316 lines
11 KiB
Python
316 lines
11 KiB
Python
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()
|
|
module._ALERT_INFLIGHT.clear()
|
|
monkeypatch.setattr(module, "NEMOTRON_OLLAMA_FIRST", False)
|
|
yield
|
|
module._ALERT_CACHE.clear()
|
|
module._ALERT_INFLIGHT.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 {"durable": True}
|
|
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_forced_review_failure_is_counted_as_skipped(monkeypatch):
|
|
import services.nemoton_dispatcher_service as module
|
|
|
|
monkeypatch.setattr(module, "NIM_API_KEY", "")
|
|
dispatcher = module.NemotronDispatcher()
|
|
monkeypatch.setattr(
|
|
dispatcher,
|
|
"_exec_flag_for_human_review",
|
|
lambda *_args, **_kwargs: {"durable": False},
|
|
)
|
|
threat = FakeThreat("SKU-FORCED", "斷崖品")
|
|
threat.sales_7d_delta_pct = -100.0
|
|
|
|
result = dispatcher.dispatch([threat], hermes_stats={"duration_sec": 1})
|
|
|
|
assert result["dispatched"] == 0
|
|
assert result["skipped"] == 1
|
|
assert len(result["errors"]) == 1
|
|
assert "durable_side_effect_not_confirmed" in result["errors"][0]
|
|
|
|
|
|
def test_hermes_failure_is_counted_as_skipped(monkeypatch):
|
|
import services.nemoton_dispatcher_service as module
|
|
|
|
monkeypatch.setattr(module, "NIM_API_KEY", "")
|
|
dispatcher = module.NemotronDispatcher()
|
|
monkeypatch.setattr(
|
|
dispatcher,
|
|
"_exec_trigger_price_alert",
|
|
lambda *_args, **_kwargs: {"durable": False},
|
|
)
|
|
|
|
result = dispatcher.dispatch(
|
|
[FakeThreat("SKU-HERMES-FAIL", "告警品")],
|
|
hermes_stats={"duration_sec": 1},
|
|
)
|
|
|
|
assert result["dispatched"] == 0
|
|
assert result["skipped"] == 1
|
|
assert len(result["errors"]) == 1
|
|
assert "durable_side_effect_not_confirmed" in result["errors"][0]
|
|
|
|
|
|
def test_forced_review_commit_failure_keeps_durable_result_quarantined(monkeypatch):
|
|
import services.nemoton_dispatcher_service as module
|
|
|
|
monkeypatch.setattr(module, "NIM_API_KEY", "")
|
|
dispatcher = module.NemotronDispatcher()
|
|
monkeypatch.setattr(
|
|
dispatcher,
|
|
"_exec_flag_for_human_review",
|
|
lambda *_args, **_kwargs: {"durable": True},
|
|
)
|
|
monkeypatch.setattr(module, "_mark_alert_dispatched", lambda *_args: False)
|
|
released = []
|
|
monkeypatch.setattr(
|
|
module,
|
|
"_release_alert_reservation",
|
|
lambda *args: (released.append(args) or True),
|
|
)
|
|
threat = FakeThreat("SKU-FORCED-COMMIT", "斷崖品")
|
|
threat.sales_7d_delta_pct = -100.0
|
|
|
|
result = dispatcher.dispatch([threat], hermes_stats={"duration_sec": 1})
|
|
|
|
assert result["dispatched"] == 1
|
|
assert result["skipped"] == 0
|
|
assert result["errors"] == [
|
|
"dedupe_commit_unverified:flag_for_human_review(SKU-FORCED-COMMIT)"
|
|
]
|
|
assert released == []
|
|
|
|
|
|
def test_hermes_commit_failure_keeps_durable_result_quarantined(monkeypatch):
|
|
import services.nemoton_dispatcher_service as module
|
|
|
|
monkeypatch.setattr(module, "NIM_API_KEY", "")
|
|
dispatcher = module.NemotronDispatcher()
|
|
_patch_execution_methods(monkeypatch, dispatcher)
|
|
monkeypatch.setattr(module, "_mark_alert_dispatched", lambda *_args: False)
|
|
released = []
|
|
monkeypatch.setattr(
|
|
module,
|
|
"_release_alert_reservation",
|
|
lambda *args: (released.append(args) or True),
|
|
)
|
|
|
|
result = dispatcher.dispatch(
|
|
[FakeThreat("SKU-HERMES-COMMIT", "告警品")],
|
|
hermes_stats={"duration_sec": 1},
|
|
)
|
|
|
|
assert result["dispatched"] == 1
|
|
assert result["skipped"] == 0
|
|
assert result["errors"] == [
|
|
"dedupe_commit_unverified:hermes_rule_fallback(SKU-HERMES-COMMIT)"
|
|
]
|
|
assert released == []
|
|
|
|
|
|
def test_mixed_batch_cleanup_never_releases_forced_durable_quarantine(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()
|
|
_patch_execution_methods(monkeypatch, dispatcher)
|
|
forced = FakeThreat("SKU-FORCED-MIXED", "斷崖品")
|
|
forced.sales_7d_delta_pct = -100.0
|
|
candidate = FakeThreat("SKU-NIM-MIXED", "模型候選")
|
|
monkeypatch.setattr(
|
|
dispatcher,
|
|
"_call_nim",
|
|
lambda _threats, **_kwargs: (
|
|
[{
|
|
"tool": "trigger_price_alert",
|
|
"args": {
|
|
"sku": candidate.sku,
|
|
"name": candidate.name,
|
|
"gap_pct": candidate.gap_pct,
|
|
"sales_delta": candidate.sales_7d_delta_pct,
|
|
"action": candidate.recommended_action,
|
|
"confidence": candidate.confidence,
|
|
},
|
|
}],
|
|
{"provider": "nim", "total_tokens": 1},
|
|
),
|
|
)
|
|
real_mark = module._mark_alert_dispatched
|
|
monkeypatch.setattr(
|
|
module,
|
|
"_mark_alert_dispatched",
|
|
lambda sku, token: (
|
|
False if str(sku) == forced.sku else real_mark(sku, token)
|
|
),
|
|
)
|
|
|
|
result = dispatcher.dispatch(
|
|
[forced, candidate],
|
|
hermes_stats={"duration_sec": 1},
|
|
)
|
|
|
|
assert result["dispatched"] == 2
|
|
assert result["skipped"] == 0
|
|
assert result["errors"] == [
|
|
"dedupe_commit_unverified:flag_for_human_review(SKU-FORCED-MIXED)"
|
|
]
|
|
assert module._is_duplicate_alert(forced.sku) is True
|
|
assert module._is_duplicate_alert(candidate.sku) is True
|
|
|
|
|
|
def test_mixed_batch_counts_failed_model_candidate_as_skipped(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()
|
|
forced = FakeThreat("SKU-FORCED-COUNT", "斷崖品")
|
|
forced.sales_7d_delta_pct = -100.0
|
|
candidate = FakeThreat("SKU-NIM-COUNT", "模型候選")
|
|
monkeypatch.setattr(
|
|
dispatcher,
|
|
"_exec_flag_for_human_review",
|
|
lambda *_args, **_kwargs: {"durable": True},
|
|
)
|
|
monkeypatch.setattr(
|
|
dispatcher,
|
|
"_exec_trigger_price_alert",
|
|
lambda *_args, **_kwargs: {"durable": False},
|
|
)
|
|
monkeypatch.setattr(
|
|
dispatcher,
|
|
"_call_nim",
|
|
lambda _threats, **_kwargs: (
|
|
[{
|
|
"tool": "trigger_price_alert",
|
|
"args": {
|
|
"sku": candidate.sku,
|
|
"name": candidate.name,
|
|
"gap_pct": candidate.gap_pct,
|
|
"sales_delta": candidate.sales_7d_delta_pct,
|
|
"action": candidate.recommended_action,
|
|
"confidence": candidate.confidence,
|
|
},
|
|
}],
|
|
{"provider": "nim", "total_tokens": 1},
|
|
),
|
|
)
|
|
|
|
result = dispatcher.dispatch(
|
|
[forced, candidate],
|
|
hermes_stats={"duration_sec": 1},
|
|
)
|
|
|
|
assert result["dispatched"] == 1
|
|
assert result["skipped"] == 1
|
|
assert len(result["errors"]) == 1
|
|
assert "durable_side_effect_not_confirmed" in result["errors"][0]
|
|
assert result["dispatched"] + result["skipped"] == 2
|
|
|
|
|
|
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, **_kwargs: (_ 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, **_kwargs: ([], {"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"
|