63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.models.event import EventType
|
|
from app.repositories.event_store import event_store
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def _truncate_store() -> None:
|
|
"""Mark all pending events as delivered and remove old events.
|
|
|
|
Tests share the global SQLite-backed event store, so we clean it up
|
|
before each test to avoid cross-test contamination.
|
|
"""
|
|
for event in event_store.fetch_pending(limit=10000):
|
|
event_store.mark_delivered(event.event_id)
|
|
event_store.cleanup(0)
|
|
|
|
|
|
def test_send_text_creates_send_result_event():
|
|
_truncate_store()
|
|
client.post("/v1/wechat-personal/login/start", json={"account_id": "send-bot", "tenant_code": "rescue"})
|
|
client.post("/v1/wechat-personal/login/send-bot/confirm", json={"tenant_code": "rescue"})
|
|
|
|
r = client.post("/v1/wechat-personal/send/text", json={
|
|
"account_id": "send-bot",
|
|
"conversation_id": "room-001",
|
|
"text": "hello",
|
|
"record_id": "rec-123",
|
|
"tenant_code": "rescue",
|
|
})
|
|
assert r.status_code == 200
|
|
assert r.json()["accepted"] is True
|
|
|
|
pending = event_store.fetch_pending(limit=100)
|
|
assert any(e.event_type == EventType.SEND_RESULT and e.data.get("record_id") == "rec-123" for e in pending)
|
|
|
|
|
|
def test_send_text_blocked_by_risk_control():
|
|
_truncate_store()
|
|
client.post("/v1/wechat-personal/login/start", json={"account_id": "risk-bot", "tenant_code": "rescue"})
|
|
client.post("/v1/wechat-personal/login/risk-bot/confirm", json={"tenant_code": "rescue"})
|
|
client.post("/v1/heartbeat", json={"account_id": "risk-bot", "tenant_code": "rescue", "risk_level": 90})
|
|
|
|
r = client.post("/v1/wechat-personal/send/text", json={
|
|
"account_id": "risk-bot",
|
|
"conversation_id": "room-001",
|
|
"text": "hello",
|
|
"record_id": "rec-risk",
|
|
"tenant_code": "rescue",
|
|
})
|
|
assert r.status_code == 200
|
|
assert r.json()["accepted"] is False
|
|
|
|
pending = event_store.fetch_pending(limit=100)
|
|
event = next(
|
|
(e for e in pending if e.event_type == EventType.SEND_RESULT and e.data.get("record_id") == "rec-risk"),
|
|
None,
|
|
)
|
|
assert event is not None
|
|
assert event.data["channel_result"]["error_code"] == "RISK_CONTROL_BLOCKED"
|