509 lines
18 KiB
Python
509 lines
18 KiB
Python
"""Tests for Copilot chat, action, history, and voice endpoints with mocked OpenRouter."""
|
|
|
|
import base64
|
|
import json
|
|
import uuid
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
from app.models.copilot import CopilotChat
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def sample_vehicle_data():
|
|
"""Valid vehicle data for creation."""
|
|
return {
|
|
"make": "Mercedes-Benz",
|
|
"model": "Actros",
|
|
"fin": "WDB9066351L123456",
|
|
"year": 2020,
|
|
"power_kw": 300,
|
|
"fuel_type": "Diesel",
|
|
"condition": "used",
|
|
"availability": "available",
|
|
"price": 45000.00,
|
|
"vehicle_type": "lkw",
|
|
"lkw_type": "sattelzugmaschine",
|
|
"mileage_km": 120000,
|
|
}
|
|
|
|
|
|
def _mock_chat_json(response_text: str, actions: list | None = None) -> str:
|
|
"""Build a JSON response string as the AI would return."""
|
|
return json.dumps(
|
|
{
|
|
"response": response_text,
|
|
"actions": actions or [],
|
|
}
|
|
)
|
|
|
|
|
|
def _patch_openrouter_chat(content: str):
|
|
"""Patch _call_openrouter_chat to return fixed content.
|
|
|
|
Usage:
|
|
with _patch_openrouter_chat(content):
|
|
response = await client.post(...)
|
|
"""
|
|
return patch(
|
|
"app.services.copilot_service._call_openrouter_chat",
|
|
new_callable=AsyncMock,
|
|
return_value=content,
|
|
)
|
|
|
|
|
|
class TestCopilotChat:
|
|
"""POST /api/v1/copilot/chat tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_returns_200_with_response_and_actions(self, admin_client):
|
|
"""POST /copilot/chat with valid message returns 200 + response + actions."""
|
|
mock_content = _mock_chat_json(
|
|
"Ich suche nach allen LKWs im Bestand.",
|
|
[{"type": "search_vehicles", "params": {"type": "lkw"}}],
|
|
)
|
|
with _patch_openrouter_chat(mock_content):
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": "Zeige alle LKWs"},
|
|
)
|
|
|
|
assert response.status_code == 200, response.text
|
|
data = response.json()
|
|
assert "response" in data
|
|
assert "actions" in data
|
|
assert "session_id" in data
|
|
assert "message_id" in data
|
|
assert isinstance(data["response"], str)
|
|
assert len(data["response"]) > 0
|
|
assert len(data["actions"]) == 1
|
|
assert data["actions"][0]["type"] == "search_vehicles"
|
|
assert data["actions"][0]["params"]["type"] == "lkw"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_empty_message_returns_422(self, admin_client):
|
|
"""POST /copilot/chat with empty message returns 422."""
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": ""},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_no_actions_returns_empty_list(self, admin_client):
|
|
"""POST /copilot/chat with a general question returns empty actions list."""
|
|
mock_content = _mock_chat_json("Hallo! Wie kann ich helfen?", [])
|
|
with _patch_openrouter_chat(mock_content):
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": "Hallo"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["actions"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_persists_messages_to_db(self, admin_client, db_session):
|
|
"""Chat messages are persisted to the database."""
|
|
mock_content = _mock_chat_json("Test response", [])
|
|
with _patch_openrouter_chat(mock_content):
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": "Test message"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
session_id = response.json()["session_id"]
|
|
|
|
# Verify messages in DB via a fresh query
|
|
from sqlalchemy import select
|
|
|
|
stmt = select(CopilotChat).where(
|
|
CopilotChat.session_id == uuid.UUID(session_id)
|
|
)
|
|
result = await db_session.execute(stmt)
|
|
messages = list(result.scalars().all())
|
|
assert len(messages) == 2 # user + assistant
|
|
roles = [m.role.value for m in messages]
|
|
assert "user" in roles
|
|
assert "assistant" in roles
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_contact_search_action(self, admin_client):
|
|
"""Copilot can propose a contact search action."""
|
|
mock_content = _mock_chat_json(
|
|
"Ich suche nach Kontakten mit dem Namen Müller.",
|
|
[{"type": "search_contacts", "params": {"search": "Müller"}}],
|
|
)
|
|
with _patch_openrouter_chat(mock_content):
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": "Suche Kontakt Müller"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["actions"]) == 1
|
|
assert data["actions"][0]["type"] == "search_contacts"
|
|
assert data["actions"][0]["params"]["search"] == "Müller"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_requires_auth(self, client):
|
|
"""POST /copilot/chat without auth returns 401."""
|
|
response = await client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": "test"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_continues_existing_session(self, admin_client):
|
|
"""POST /copilot/chat with session_id continues the same session."""
|
|
mock_content1 = _mock_chat_json("Erste Antwort", [])
|
|
with _patch_openrouter_chat(mock_content1):
|
|
resp1 = await admin_client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": "Erste Nachricht"},
|
|
)
|
|
assert resp1.status_code == 200
|
|
session_id = resp1.json()["session_id"]
|
|
|
|
mock_content2 = _mock_chat_json("Zweite Antwort", [])
|
|
with _patch_openrouter_chat(mock_content2):
|
|
resp2 = await admin_client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": "Zweite Nachricht", "session_id": session_id},
|
|
)
|
|
assert resp2.status_code == 200
|
|
assert resp2.json()["session_id"] == session_id
|
|
|
|
|
|
class TestCopilotAction:
|
|
"""POST /api/v1/copilot/action tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_search_vehicles_returns_200(
|
|
self, admin_client, sample_vehicle_data
|
|
):
|
|
"""POST /copilot/action with search_vehicles returns 200 + results."""
|
|
# First create a vehicle
|
|
await admin_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
|
|
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/action",
|
|
json={"action": "search_vehicles", "params": {"type": "lkw"}},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
data = response.json()
|
|
assert data["action"] == "search_vehicles"
|
|
assert data["success"] is True
|
|
assert "result" in data
|
|
assert "items" in data["result"]
|
|
assert data["result"]["total"] >= 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_search_contacts_returns_200(self, admin_client):
|
|
"""POST /copilot/action with search_contacts returns 200 + results."""
|
|
# First create a contact
|
|
await admin_client.post(
|
|
"/api/v1/contacts/",
|
|
json={
|
|
"company_name": "Test GmbH",
|
|
"role": "kaeufer",
|
|
"address_country": "DE",
|
|
},
|
|
)
|
|
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/action",
|
|
json={"action": "search_contacts", "params": {"search": "Test"}},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
data = response.json()
|
|
assert data["action"] == "search_contacts"
|
|
assert data["success"] is True
|
|
assert data["result"]["total"] >= 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_invalid_action_returns_400(self, admin_client):
|
|
"""POST /copilot/action with invalid action returns 400."""
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/action",
|
|
json={"action": "invalid_action", "params": {}},
|
|
)
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert "error" in data["detail"]
|
|
assert data["detail"]["error"]["code"] == "INVALID_ACTION"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_get_sale_overview_returns_200(self, admin_client):
|
|
"""POST /copilot/action with get_sale_overview returns 200."""
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/action",
|
|
json={"action": "get_sale_overview", "params": {}},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
data = response.json()
|
|
assert data["action"] == "get_sale_overview"
|
|
assert data["success"] is True
|
|
assert "items" in data["result"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_requires_auth(self, client):
|
|
"""POST /copilot/action without auth returns 401."""
|
|
response = await client.post(
|
|
"/api/v1/copilot/action",
|
|
json={"action": "search_vehicles", "params": {}},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestCopilotHistory:
|
|
"""GET /api/v1/copilot/history tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_history_returns_200_with_pagination(self, admin_client):
|
|
"""GET /copilot/history returns 200 with paginated history."""
|
|
# First create a chat message
|
|
mock_content = _mock_chat_json("Test", [])
|
|
with _patch_openrouter_chat(mock_content):
|
|
await admin_client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": "Test message"},
|
|
)
|
|
|
|
response = await admin_client.get("/api/v1/copilot/history?page=1&page_size=20")
|
|
assert response.status_code == 200, response.text
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
assert "page" in data
|
|
assert "page_size" in data
|
|
assert data["total"] >= 2 # user + assistant message
|
|
assert len(data["items"]) >= 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_history_empty_returns_200(self, admin_client):
|
|
"""GET /copilot/history with no messages returns 200 + empty list."""
|
|
response = await admin_client.get("/api/v1/copilot/history")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["items"] == []
|
|
assert data["total"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_history_requires_auth(self, client):
|
|
"""GET /copilot/history without auth returns 401."""
|
|
response = await client.get("/api/v1/copilot/history")
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_history_filtered_by_session(self, admin_client):
|
|
"""GET /copilot/history?session_id=... returns only that session's messages."""
|
|
mock_content = _mock_chat_json("Test", [])
|
|
with _patch_openrouter_chat(mock_content):
|
|
resp1 = await admin_client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": "Session 1 message"},
|
|
)
|
|
await admin_client.post(
|
|
"/api/v1/copilot/chat",
|
|
json={"message": "Session 2 message"},
|
|
)
|
|
|
|
session1_id = resp1.json()["session_id"]
|
|
|
|
response = await admin_client.get(
|
|
f"/api/v1/copilot/history?session_id={session1_id}"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 2 # only session 1's user + assistant
|
|
for item in data["items"]:
|
|
assert item["session_id"] == session1_id
|
|
|
|
|
|
class TestCopilotVoice:
|
|
"""POST /api/v1/copilot/voice tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_returns_200_with_transcription_and_response(
|
|
self, admin_client
|
|
):
|
|
"""POST /copilot/voice returns 200 with transcription + chat response."""
|
|
mock_content = _mock_chat_json("Ich helfe dir bei der Suche.", [])
|
|
audio_b64 = base64.b64encode(b"fake-audio-data").decode("utf-8")
|
|
|
|
with (
|
|
patch(
|
|
"app.services.copilot_service.transcribe_audio",
|
|
new_callable=AsyncMock,
|
|
return_value="Zeige alle LKWs",
|
|
),
|
|
_patch_openrouter_chat(mock_content),
|
|
):
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/voice",
|
|
json={"audio": audio_b64, "mime_type": "audio/webm"},
|
|
)
|
|
|
|
assert response.status_code == 200, response.text
|
|
data = response.json()
|
|
assert "transcription" in data
|
|
assert "response" in data
|
|
assert "actions" in data
|
|
assert "session_id" in data
|
|
assert "message_id" in data
|
|
assert isinstance(data["transcription"], str)
|
|
assert len(data["transcription"]) > 0
|
|
assert data["transcription"] == "Zeige alle LKWs"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_empty_audio_returns_422(self, admin_client):
|
|
"""POST /copilot/voice with empty audio returns 422."""
|
|
response = await admin_client.post(
|
|
"/api/v1/copilot/voice",
|
|
json={"audio": ""},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_requires_auth(self, client):
|
|
"""POST /copilot/voice without auth returns 401."""
|
|
audio_b64 = base64.b64encode(b"fake").decode("utf-8")
|
|
response = await client.post(
|
|
"/api/v1/copilot/voice",
|
|
json={"audio": audio_b64},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestCopilotServiceUnit:
|
|
"""Unit tests for copilot_service internal functions."""
|
|
|
|
def test_parse_ai_response_valid_json(self):
|
|
"""_parse_ai_response correctly parses valid JSON."""
|
|
from app.services.copilot_service import _parse_ai_response
|
|
|
|
raw = json.dumps(
|
|
{
|
|
"response": "Ich suche LKWs.",
|
|
"actions": [{"type": "search_vehicles", "params": {"type": "lkw"}}],
|
|
}
|
|
)
|
|
result = _parse_ai_response(raw)
|
|
assert result["response"] == "Ich suche LKWs."
|
|
assert len(result["actions"]) == 1
|
|
assert result["actions"][0]["type"] == "search_vehicles"
|
|
|
|
def test_parse_ai_response_with_code_fence(self):
|
|
"""_parse_ai_response handles markdown code fences."""
|
|
from app.services.copilot_service import _parse_ai_response
|
|
|
|
raw = (
|
|
"```json\n"
|
|
+ json.dumps(
|
|
{
|
|
"response": "Test",
|
|
"actions": [],
|
|
}
|
|
)
|
|
+ "\n```"
|
|
)
|
|
result = _parse_ai_response(raw)
|
|
assert result["response"] == "Test"
|
|
assert result["actions"] == []
|
|
|
|
def test_parse_ai_response_fallback_to_raw_text(self):
|
|
"""_parse_ai_response falls back to raw text when no JSON found."""
|
|
from app.services.copilot_service import _parse_ai_response
|
|
|
|
raw = "Das ist keine JSON-Antwort."
|
|
result = _parse_ai_response(raw)
|
|
assert result["response"] == raw
|
|
assert result["actions"] == []
|
|
|
|
def test_parse_ai_response_invalid_actions_filtered(self):
|
|
"""_parse_ai_response filters out invalid action structures."""
|
|
from app.services.copilot_service import _parse_ai_response
|
|
|
|
raw = json.dumps(
|
|
{
|
|
"response": "Test",
|
|
"actions": [
|
|
{"type": "search_vehicles", "params": {}},
|
|
{"invalid": "no type"},
|
|
"not a dict",
|
|
],
|
|
}
|
|
)
|
|
result = _parse_ai_response(raw)
|
|
assert len(result["actions"]) == 1
|
|
assert result["actions"][0]["type"] == "search_vehicles"
|
|
|
|
def test_system_prompt_contains_erp_context(self):
|
|
"""System prompt includes ERP context and available actions."""
|
|
from app.utils.copilot_prompt import build_system_prompt
|
|
|
|
prompt = build_system_prompt()
|
|
assert "ERP" in prompt
|
|
assert "Fahrzeugtypen" in prompt
|
|
assert "Kontakttypen" in prompt
|
|
assert "Verkaufsworkflow" in prompt
|
|
assert "search_vehicles" in prompt
|
|
assert "search_contacts" in prompt
|
|
assert "get_sale_overview" in prompt
|
|
assert "create_vehicle" in prompt
|
|
assert "create_contact" in prompt
|
|
assert "actions" in prompt
|
|
assert "response" in prompt
|
|
|
|
def test_action_registry_has_all_actions(self):
|
|
"""Action registry contains all 5 defined actions."""
|
|
from app.utils.copilot_actions import ACTION_REGISTRY
|
|
|
|
assert len(ACTION_REGISTRY) == 5
|
|
assert "search_vehicles" in ACTION_REGISTRY
|
|
assert "search_contacts" in ACTION_REGISTRY
|
|
assert "get_sale_overview" in ACTION_REGISTRY
|
|
assert "create_vehicle" in ACTION_REGISTRY
|
|
assert "create_contact" in ACTION_REGISTRY
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_action_unknown_raises_value_error(self, db_session):
|
|
"""execute_action raises ValueError for unknown action type."""
|
|
from app.utils.copilot_actions import execute_action
|
|
|
|
with pytest.raises(ValueError, match="Unknown action type"):
|
|
await execute_action(db_session, "nonexistent_action", {})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_session_creates_new(self, db_session, admin_user):
|
|
"""_get_or_create_session creates a new session when no session_id given."""
|
|
from app.services.copilot_service import _get_or_create_session
|
|
|
|
session = await _get_or_create_session(
|
|
db_session, admin_user.id, first_message="Test message"
|
|
)
|
|
assert session.title == "Test message"
|
|
assert session.user_id == admin_user.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_or_create_session_returns_existing(self, db_session, admin_user):
|
|
"""_get_or_create_session returns existing session when valid session_id given."""
|
|
from app.services.copilot_service import _get_or_create_session
|
|
|
|
session1 = await _get_or_create_session(
|
|
db_session, admin_user.id, first_message="First"
|
|
)
|
|
await db_session.flush()
|
|
|
|
session2 = await _get_or_create_session(
|
|
db_session, admin_user.id, session_id=str(session1.id)
|
|
)
|
|
assert session2.id == session1.id
|