File size: 5,124 Bytes
102dd4f fcf2812 102dd4f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | from types import SimpleNamespace
from fastapi.testclient import TestClient
from app.main import create_app
from app.models.domain import ToolResult
from tests.conftest import signed_body, whatsapp_payload
def test_whatsapp_verification_success(test_app):
with TestClient(test_app) as client:
response = client.get(
"/webhooks/whatsapp",
params={
"hub.mode": "subscribe",
"hub.verify_token": "verify-token",
"hub.challenge": "challenge-code",
},
)
assert response.status_code == 200
assert response.text == "challenge-code"
def test_whatsapp_verification_rejects_wrong_token(test_app):
with TestClient(test_app) as client:
response = client.get(
"/webhooks/whatsapp",
params={
"hub.mode": "subscribe",
"hub.verify_token": "wrong",
"hub.challenge": "challenge-code",
},
)
assert response.status_code == 401
def test_post_webhook_validates_signature_and_accepts_text(test_app):
payload = whatsapp_payload(message_id="wamid.100", text="I need Aden to Mukalla")
body, signature = signed_body(payload, "app-secret")
with TestClient(test_app) as client:
response = client.post(
"/webhooks/whatsapp",
content=body,
headers={
"content-type": "application/json",
"x-hub-signature-256": signature,
},
)
conversation = test_app.state.test_container.conversation
assert response.status_code == 200
assert response.json() == {"status": "accepted", "messages": 1}
assert len(conversation.calls) == 1
assert conversation.calls[0].text == "I need Aden to Mukalla"
def test_post_webhook_debug_returns_reply_and_status(test_app):
payload = whatsapp_payload(message_id="wamid.200", text="Debug this message")
body, signature = signed_body(payload, "app-secret")
with TestClient(test_app) as client:
response = client.post(
"/webhooks/whatsapp/debug",
content=body,
headers={
"content-type": "application/json",
"x-hub-signature-256": signature,
},
)
conversation = test_app.state.test_container.conversation
assert response.status_code == 200
assert response.json() == {
"status": "accepted",
"messages": 1,
"replies": ["ok"],
}
assert len(conversation.calls) == 1
def test_post_webhook_rejects_bad_signature(test_app):
payload = whatsapp_payload()
body, _ = signed_body(payload, "app-secret")
with TestClient(test_app) as client:
response = client.post(
"/webhooks/whatsapp",
content=body,
headers={
"content-type": "application/json",
"x-hub-signature-256": "sha256=bad",
},
)
assert response.status_code == 401
def test_admin_routes_require_key(test_app):
with TestClient(test_app) as client:
rejected = client.post("/admin/seed-info")
accepted = client.post("/admin/sync-trips", headers={"X-Admin-Api-Key": "admin-secret"})
assert rejected.status_code == 401
assert accepted.status_code == 200
assert accepted.json() == {"indexed_trips": 3}
def test_admin_driver_debug_returns_llm_and_tool_results(settings):
class DummyPrimaryProvider:
async def chat(self, messages, tools, tool_choice, temperature):
return SimpleNamespace(
content="Driver debug reply",
tool_calls=[
SimpleNamespace(
id="call-1",
name="add_trip_by_driver",
arguments='{"departure":"A","destination":"B","departure_date":"2026-06-20","departure_time":"morning","available_seats":2,"total_seats":4,"price":30}',
)
],
)
class DummyConversation:
def _tool_registry(self, customer, *, remoteJid, user_mode="driver", current_message=None):
class DummyRegistry:
async def execute(self, name, arguments):
return ToolResult(ok=True, data={"name": name, "arguments": arguments})
return DummyRegistry()
container = SimpleNamespace(
settings=settings,
conversation=DummyConversation(),
ai=SimpleNamespace(primary=DummyPrimaryProvider()),
)
app = create_app(settings=settings, container=container)
with TestClient(app) as client:
response = client.post(
"/admin/driver-debug",
json={"message": "debug driver", "client_number": "967700000002"},
headers={"X-Admin-Api-Key": "admin-secret"},
)
assert response.status_code == 200
payload = response.json()
assert payload["llm_response"] == "Driver debug reply"
assert payload["tool_calls"][0]["name"] == "add_trip_by_driver"
assert payload["tool_results"][0]["result"]["ok"] is True
|