"""Tests for the FastAPI endpoints exposed by ``alfred.app``. Covers the OpenAI-compatible surface that LibreChat consumes: - ``GET /health`` — version + status. - ``GET /v1/models`` — single ``agent-media`` entry. - ``POST /v1/chat/completions`` — both blocking and streaming modes, request validation (empty messages, missing user role, invalid JSON), and the OpenAI-compatible response envelope (``choices[0].message``). - ``GET /memory/state`` and ``GET /memory/episodic/search-results`` — debug introspection endpoints. - ``POST /memory/clear-session`` — STM/episodic reset. Tests patch ``alfred.app.agent.step`` rather than running the real LLM. The app module degrades gracefully when no LLM provider is configured at import time (placeholder LLM that 503s on use), which is what lets these tests collect under pytest without ``DEEPSEEK_API_KEY``. """ from unittest.mock import patch from fastapi.testclient import TestClient class TestHealthEndpoint: """Tests for /health endpoint.""" def test_health_check(self, memory): """Should return healthy status.""" from alfred.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 alfred.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 alfred.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 alfred.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 alfred.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 alfred.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 alfred.app import app # Patch the agent's step method directly with patch("alfred.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 alfred.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 alfred.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 alfred.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 alfred.app import app with patch("alfred.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 alfred.app import app with patch("alfred.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 alfred.app import app with patch("alfred.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"