211 lines
6.9 KiB
Python
211 lines
6.9 KiB
Python
"""Tests for FastAPI endpoints."""
|
|
import pytest
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
class TestHealthEndpoint:
|
|
"""Tests for /health endpoint."""
|
|
|
|
def test_health_check(self, memory):
|
|
"""Should return healthy status."""
|
|
from app import app
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "healthy"
|
|
|
|
|
|
class TestModelsEndpoint:
|
|
"""Tests for /v1/models endpoint."""
|
|
|
|
def test_list_models(self, memory):
|
|
"""Should return model list."""
|
|
from app import app
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/v1/models")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["object"] == "list"
|
|
assert len(data["data"]) > 0
|
|
assert data["data"][0]["id"] == "agent-media"
|
|
|
|
|
|
class TestMemoryEndpoints:
|
|
"""Tests for memory debug endpoints."""
|
|
|
|
def test_get_memory_state(self, memory):
|
|
"""Should return full memory state."""
|
|
from app import app
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/memory/state")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "ltm" in data
|
|
assert "stm" in data
|
|
assert "episodic" in data
|
|
|
|
def test_get_search_results_empty(self, memory):
|
|
"""Should return empty when no search results."""
|
|
from app import app
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/memory/episodic/search-results")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "empty"
|
|
|
|
def test_get_search_results_with_data(self, memory_with_search_results):
|
|
"""Should return search results when available."""
|
|
from app import app
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/memory/episodic/search-results")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "ok"
|
|
assert data["query"] == "Inception 1080p"
|
|
assert data["result_count"] == 3
|
|
|
|
def test_clear_session(self, memory_with_search_results):
|
|
"""Should clear session memories."""
|
|
from app import app
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/memory/clear-session")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "ok"
|
|
|
|
# Verify cleared
|
|
state = client.get("/memory/state").json()
|
|
assert state["episodic"]["last_search_results"] is None
|
|
|
|
|
|
class TestChatCompletionsEndpoint:
|
|
"""Tests for /v1/chat/completions endpoint."""
|
|
|
|
def test_chat_completion_success(self, memory):
|
|
"""Should return chat completion."""
|
|
from app import app
|
|
# Patch the agent's step method directly
|
|
with patch("app.agent.step", return_value="Hello! How can I help?"):
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/v1/chat/completions", json={
|
|
"model": "agent-media",
|
|
"messages": [{"role": "user", "content": "Hello"}],
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["object"] == "chat.completion"
|
|
assert "Hello" in data["choices"][0]["message"]["content"]
|
|
|
|
def test_chat_completion_no_user_message(self, memory):
|
|
"""Should return error if no user message."""
|
|
from app import app
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/v1/chat/completions", json={
|
|
"model": "agent-media",
|
|
"messages": [{"role": "system", "content": "You are helpful"}],
|
|
})
|
|
|
|
assert response.status_code == 422
|
|
detail = response.json()["detail"]
|
|
# Pydantic returns a list of errors or a string
|
|
if isinstance(detail, list):
|
|
detail_str = str(detail).lower()
|
|
else:
|
|
detail_str = detail.lower()
|
|
assert "user message" in detail_str
|
|
|
|
def test_chat_completion_empty_messages(self, memory):
|
|
"""Should return error for empty messages."""
|
|
from app import app
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/v1/chat/completions", json={
|
|
"model": "agent-media",
|
|
"messages": [],
|
|
})
|
|
|
|
assert response.status_code == 422
|
|
|
|
def test_chat_completion_invalid_json(self, memory):
|
|
"""Should return error for invalid JSON."""
|
|
from app import app
|
|
client = TestClient(app)
|
|
|
|
response = client.post(
|
|
"/v1/chat/completions",
|
|
content="not json",
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
|
|
def test_chat_completion_streaming(self, memory):
|
|
"""Should support streaming mode."""
|
|
from app import app
|
|
with patch("app.agent.step", return_value="Streaming response"):
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/v1/chat/completions", json={
|
|
"model": "agent-media",
|
|
"messages": [{"role": "user", "content": "Hello"}],
|
|
"stream": True,
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
assert "text/event-stream" in response.headers["content-type"]
|
|
|
|
def test_chat_completion_extracts_last_user_message(self, memory):
|
|
"""Should use last user message."""
|
|
from app import app
|
|
with patch("app.agent.step", return_value="Response") as mock_step:
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/v1/chat/completions", json={
|
|
"model": "agent-media",
|
|
"messages": [
|
|
{"role": "user", "content": "First message"},
|
|
{"role": "assistant", "content": "Response"},
|
|
{"role": "user", "content": "Second message"},
|
|
],
|
|
})
|
|
|
|
assert response.status_code == 200
|
|
# Verify the agent received the last user message
|
|
mock_step.assert_called_once_with("Second message")
|
|
|
|
def test_chat_completion_response_format(self, memory):
|
|
"""Should return OpenAI-compatible format."""
|
|
from app import app
|
|
with patch("app.agent.step", return_value="Test response"):
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/v1/chat/completions", json={
|
|
"model": "agent-media",
|
|
"messages": [{"role": "user", "content": "Test"}],
|
|
})
|
|
|
|
data = response.json()
|
|
assert "id" in data
|
|
assert data["id"].startswith("chatcmpl-")
|
|
assert "created" in data
|
|
assert "model" in data
|
|
assert "choices" in data
|
|
assert "usage" in data
|
|
assert data["choices"][0]["finish_reason"] == "stop"
|
|
assert data["choices"][0]["message"]["role"] == "assistant"
|