117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
import sys
|
|
import types
|
|
|
|
|
|
def test_price_decision_envelope_has_hilt_guardrails():
|
|
import services.nemoton_dispatcher_service as module
|
|
|
|
envelope = module._build_price_decision_envelope(
|
|
decision_type="price_alert",
|
|
sku="10413050",
|
|
name="測試精華液",
|
|
gap_pct=16.7,
|
|
sales_delta=-31.2,
|
|
confidence=0.91,
|
|
analysis="建議進入人工價格覆核",
|
|
momo_price=1200,
|
|
comp_price=999,
|
|
revenue_loss_7d=65000,
|
|
recommended_price=999,
|
|
match_type="exact",
|
|
price_basis="total_price",
|
|
alert_tier="price_alert_exact",
|
|
match_score=0.93,
|
|
competitor_product_id="PCHOME-1",
|
|
competitor_product_name="PChome 測試精華液",
|
|
)
|
|
|
|
assert envelope["decision_type"] == "price_alert"
|
|
assert envelope["severity"] == "P1"
|
|
assert envelope["confidence"] == 0.91
|
|
assert envelope["subject"]["sku"] == "10413050"
|
|
assert envelope["subject"]["momo_price"] == 1200
|
|
assert envelope["subject"]["competitor_price"] == 999
|
|
assert envelope["subject"]["competitor_product_id"] == "PCHOME-1"
|
|
assert envelope["guardrails"]["can_auto_execute"] is False
|
|
assert envelope["guardrails"]["data_quality"] == "complete"
|
|
assert envelope["guardrails"]["match_type"] == "exact"
|
|
assert envelope["guardrails"]["price_basis"] == "total_price"
|
|
assert envelope["guardrails"]["alert_tier"] == "price_alert_exact"
|
|
assert envelope["recommended_action"]["requires_hitl"] is True
|
|
assert envelope["expected_impact"]["revenue_loss_7d"] == 65000
|
|
assert envelope["expected_impact"]["gap_amount"] == 201
|
|
assert envelope["expected_impact"]["momo_price"] == 1200
|
|
assert envelope["expected_impact"]["competitor_price"] == 999
|
|
assert envelope["expected_impact"]["candidate_gap_pct"] == 16.7
|
|
assert envelope["expected_impact"]["sales_7d_delta_pct"] == -31.2
|
|
assert any(e["metric"] == "match_score" and e["value"] == 0.93 for e in envelope["evidence"])
|
|
|
|
|
|
def test_send_telegram_attaches_decision_envelope_to_event_router(monkeypatch):
|
|
import services.nemoton_dispatcher_service as module
|
|
|
|
captured = []
|
|
fake_event_router = types.ModuleType("services.event_router")
|
|
fake_event_router.dispatch_sync = lambda event: captured.append(event)
|
|
monkeypatch.setitem(sys.modules, "services.event_router", fake_event_router)
|
|
|
|
envelope = {
|
|
"decision_id": "nemotron:price_alert:SKU-1:abcdef12",
|
|
"decision_type": "price_alert",
|
|
"severity": "P2",
|
|
"guardrails": {"can_auto_execute": False},
|
|
}
|
|
module.NemotronDispatcher()._send_telegram("hello price alert", decision_envelope=envelope)
|
|
|
|
assert len(captured) == 1
|
|
event = captured[0]
|
|
assert event["event_type"] == "nemoton_dispatch_alert"
|
|
assert event["id"] == "nemotron:price_alert:SKU-1:abcdef12"
|
|
assert event["decision_envelope"] == envelope
|
|
assert event["payload"]["decision_envelope"] == envelope
|
|
assert event["payload"]["raw_message"] == "hello price alert"
|
|
|
|
|
|
def test_exec_trigger_price_alert_persists_decision_envelope(monkeypatch):
|
|
import services.nemoton_dispatcher_service as module
|
|
|
|
dispatcher = module.NemotronDispatcher()
|
|
sent = []
|
|
stored = []
|
|
monkeypatch.setattr(
|
|
dispatcher,
|
|
"_send_telegram",
|
|
lambda message, decision_envelope=None: sent.append((message, decision_envelope)),
|
|
)
|
|
monkeypatch.setattr(
|
|
dispatcher,
|
|
"_sink_insight_to_km",
|
|
lambda *args, **kwargs: stored.append((args, kwargs)),
|
|
)
|
|
|
|
dispatcher._exec_trigger_price_alert(
|
|
sku="SKU-1",
|
|
name="測試商品",
|
|
gap_pct=12.5,
|
|
sales_delta=-10.0,
|
|
action="請人工確認是否跟價",
|
|
confidence=0.88,
|
|
momo_price=1000,
|
|
comp_price=875,
|
|
revenue_loss_7d=32000,
|
|
recommended_price=875,
|
|
match_type="exact",
|
|
price_basis="total_price",
|
|
alert_tier="price_alert_exact",
|
|
match_score=0.9,
|
|
competitor_product_id="PC-1",
|
|
)
|
|
|
|
assert sent
|
|
_, envelope = sent[0]
|
|
assert envelope["decision_type"] == "price_alert"
|
|
assert envelope["guardrails"]["can_auto_execute"] is False
|
|
assert envelope["subject"]["competitor_product_id"] == "PC-1"
|
|
assert stored
|
|
assert stored[0][1]["metadata"]["decision_envelope"] == envelope
|