Spaces:
Sleeping
Sleeping
| 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"] | |