francwa
e07c9ec77b
Several weeks of work accumulated without being committed. Grouped here for clarity; see CHANGELOG.md [Unreleased] for the user-facing summary. Highlights ---------- P1 #2 — ISO 639-2/B canonical migration - New Language VO + LanguageRegistry (alfred/domain/shared/knowledge/). - iso_languages.yaml as single source of truth for language codes. - SubtitleKnowledgeBase now delegates lookup to LanguageRegistry; subtitles.yaml only declares subtitle-specific tokens (vostfr, vf, vff, …). - SubtitlePreferences default → ["fre", "eng"]; subtitle filenames written as {iso639_2b}.srt (legacy fr.srt still read via alias). - Scanner: dropped _LANG_KEYWORDS / _SDH_TOKENS / _FORCED_TOKENS / SUBTITLE_EXTENSIONS hardcoded dicts. - Fixed: 'hi' token no longer marks SDH (conflicted with Hindi alias). - Added settings.min_movie_size_bytes (was a module constant). P1 #3 — Release parser unification + data-driven tokenizer - parse_release() is now the single source of truth for release-name parsing. - alfred/knowledge/release/separators.yaml declares the token separators used by the tokenizer (., space, [, ], (, ), _). New conventions can be added without code changes. - Tokenizer now splits on any configured separator instead of name.split('.'). Releases like 'The Father (2020) [1080p] [WEBRip] [5.1] [YTS.MX]' parse via the direct path without sanitization fallback. - Site-tag extraction always runs first; well-formedness only rejects truly forbidden chars. - _parse_season_episode() extended with NxNN / NxNNxNN alt forms. - Removed dead helpers: _sanitize, _normalize. Domain cleanup - Deleted fossil services with zero production callers: alfred/domain/movies/services.py alfred/domain/tv_shows/services.py alfred/domain/subtitles/services.py (replaced by subtitles/services/ package) alfred/domain/subtitles/repositories.py - Split monolithic subtitle services into a package (identifier, matcher, placer, pattern_detector, utils) + dedicated knowledge/ package. - MediaInfo split into dedicated package (alfred/domain/shared/media/: audio, video, subtitle, info, matching). Persistence cleanup - Removed dead JSON repositories (movie/subtitle/tvshow_repository.py). Tests - Major expansion of the test suite organized to mirror the source tree. - Removed obsolete *_edge_cases test files superseded by structured tests. - Suite: 990 passed, 8 skipped. Misc - .gitignore: exclude env_backup/ and *.bak. - Adjustments across agent/llm, app.py, application/filesystem, and infrastructure/filesystem to align with the new domain layout.
260 lines
8.2 KiB
Python
260 lines
8.2 KiB
Python
"""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"
|