Spaces:
Sleeping
Sleeping
File size: 4,024 Bytes
5bb9b0e | 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 | import json
from app.services import model_client
def _tool_call(call_id: str, name: str, arguments: dict) -> dict:
return {
"id": call_id,
"function": {"name": name, "arguments": json.dumps(arguments)},
}
def test_dispatch_chat_requires_login(client):
resp = client.post("/v1/dispatch/chat", json={"messages": [], "context": {}})
assert resp.status_code == 401
def test_plain_reply_with_no_tool_call_passes_through(client, auth_headers, monkeypatch):
headers = auth_headers("+15556660001", "Driver A")
async def fake(messages, system_prompt, tools):
return "Yeah, that's a solid rate for that lane.", [], "groq"
monkeypatch.setattr(model_client, "get_reply_with_tools", fake)
resp = client.post(
"/v1/dispatch/chat",
headers=headers,
json={"messages": [{"role": "user", "content": "is this a good rate?"}], "context": {}},
)
assert resp.status_code == 200
assert resp.json()["reply"] == "Yeah, that's a solid rate for that lane."
def test_save_load_tool_call_actually_persists(client, auth_headers, monkeypatch):
headers = auth_headers("+15556660002", "Driver B")
calls = {"n": 0}
async def fake(messages, system_prompt, tools):
calls["n"] += 1
if calls["n"] == 1:
return "", [_tool_call("call_1", "save_load", {"load_id": "LOAD-99"})], "groq"
return "Saved LOAD-99 for you.", [], "groq"
monkeypatch.setattr(model_client, "get_reply_with_tools", fake)
resp = client.post(
"/v1/dispatch/chat",
headers=headers,
json={"messages": [{"role": "user", "content": "save this load"}], "context": {}},
)
assert resp.status_code == 200
assert resp.json()["reply"] == "Saved LOAD-99 for you."
saved = client.get("/v1/saved-loads", headers=headers)
assert saved.json() == ["LOAD-99"]
def test_send_email_tool_call_actually_creates_thread(client, auth_headers, monkeypatch):
headers = auth_headers("+15556660003", "Driver C")
calls = {"n": 0}
email_args = {
"load_reference": "LOAD-7",
"broker_name": "Acme Logistics",
"broker_email": "broker@example.com",
"subject": "Availability for LOAD-7",
"body": "Still available, can pick up tomorrow at 8am.",
}
async def fake(messages, system_prompt, tools):
calls["n"] += 1
if calls["n"] == 1:
return "", [_tool_call("call_1", "send_email", email_args)], "groq"
return "Email's sent.", [], "groq"
monkeypatch.setattr(model_client, "get_reply_with_tools", fake)
resp = client.post(
"/v1/dispatch/chat",
headers=headers,
json={"messages": [{"role": "user", "content": "email the broker, send it now"}], "context": {}},
)
assert resp.status_code == 200
assert resp.json()["reply"] == "Email's sent."
threads = client.get("/v1/inbox/threads", headers=headers).json()
assert len(threads) == 1
assert threads[0]["broker_email"] == "broker@example.com"
def test_unknown_tool_name_reports_error_without_crashing(client, auth_headers, monkeypatch):
headers = auth_headers("+15556660004", "Driver D")
calls = {"n": 0}
async def fake(messages, system_prompt, tools):
calls["n"] += 1
if calls["n"] == 1:
return "", [_tool_call("call_1", "delete_everything", {})], "groq"
# The model should see the {"ok": False, "error": ...} tool result
# and explain the failure in plain language rather than the whole
# request blowing up.
last = json.loads(messages[-1]["content"])
return f"Can't do that: {last['error']}", [], "groq"
monkeypatch.setattr(model_client, "get_reply_with_tools", fake)
resp = client.post(
"/v1/dispatch/chat",
headers=headers,
json={"messages": [{"role": "user", "content": "delete everything"}], "context": {}},
)
assert resp.status_code == 200
assert "Unknown tool" in resp.json()["reply"]
|