292 lines
8.2 KiB
Python
292 lines
8.2 KiB
Python
"""Pytest configuration and shared fixtures."""
|
|
|
|
# TODO: Moved directory, should not be necessary anymore but need to check !!
|
|
# Ajouter le dossier parent (brain) au PYTHONPATH
|
|
# sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
import pytest
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, Mock
|
|
|
|
from alfred.infrastructure.persistence import Memory, set_memory
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
"""Create a temporary directory for tests."""
|
|
dirpath = tempfile.mkdtemp()
|
|
yield Path(dirpath)
|
|
shutil.rmtree(dirpath)
|
|
|
|
|
|
@pytest.fixture
|
|
def memory(temp_dir):
|
|
"""Create a fresh Memory instance for testing."""
|
|
mem = Memory(storage_dir=str(temp_dir))
|
|
set_memory(mem)
|
|
yield mem
|
|
|
|
|
|
@pytest.fixture
|
|
def memory_with_config(memory):
|
|
"""Memory with pre-configured folders."""
|
|
memory.ltm.set_config("download_folder", "/tmp/downloads")
|
|
memory.ltm.set_config("movie_folder", "/tmp/movies")
|
|
memory.ltm.set_config("tvshow_folder", "/tmp/tvshows")
|
|
memory.ltm.set_config("torrent_folder", "/tmp/torrents")
|
|
return memory
|
|
|
|
|
|
@pytest.fixture
|
|
def memory_with_search_results(memory):
|
|
"""Memory with pre-populated search results."""
|
|
memory.episodic.store_search_results(
|
|
query="Inception 1080p",
|
|
results=[
|
|
{
|
|
"name": "Inception.2010.1080p.BluRay.x264",
|
|
"size": "2.5 GB",
|
|
"seeders": 150,
|
|
"leechers": 10,
|
|
"magnet": "magnet:?xt=urn:btih:abc123",
|
|
"tracker": "ThePirateBay",
|
|
},
|
|
{
|
|
"name": "Inception.2010.1080p.WEB-DL.x265",
|
|
"size": "1.8 GB",
|
|
"seeders": 80,
|
|
"leechers": 5,
|
|
"magnet": "magnet:?xt=urn:btih:def456",
|
|
"tracker": "1337x",
|
|
},
|
|
{
|
|
"name": "Inception.2010.720p.BluRay",
|
|
"size": "1.2 GB",
|
|
"seeders": 45,
|
|
"leechers": 2,
|
|
"magnet": "magnet:?xt=urn:btih:ghi789",
|
|
"tracker": "RARBG",
|
|
},
|
|
],
|
|
search_type="torrent",
|
|
)
|
|
return memory
|
|
|
|
|
|
@pytest.fixture
|
|
def memory_with_history(memory):
|
|
"""Memory with conversation history."""
|
|
memory.stm.add_message("user", "Hello")
|
|
memory.stm.add_message("assistant", "Hi! How can I help you?")
|
|
memory.stm.add_message("user", "Find me Inception")
|
|
memory.stm.add_message("assistant", "I found Inception (2010)...")
|
|
return memory
|
|
|
|
|
|
@pytest.fixture
|
|
def memory_with_library(memory):
|
|
"""Memory with movies in library."""
|
|
memory.ltm.library["movies"] = [
|
|
{
|
|
"imdb_id": "tt1375666",
|
|
"title": "Inception",
|
|
"release_year": 2010,
|
|
"quality": "1080p",
|
|
"file_path": "/movies/Inception.2010.1080p.mkv",
|
|
"added_at": "2024-01-15T10:30:00",
|
|
},
|
|
{
|
|
"imdb_id": "tt0816692",
|
|
"title": "Interstellar",
|
|
"release_year": 2014,
|
|
"quality": "4K",
|
|
"file_path": "/movies/Interstellar.2014.4K.mkv",
|
|
"added_at": "2024-01-16T14:20:00",
|
|
},
|
|
]
|
|
memory.ltm.library["tv_shows"] = [
|
|
{
|
|
"imdb_id": "tt0944947",
|
|
"title": "Game of Thrones",
|
|
"seasons_count": 8,
|
|
"status": "ended",
|
|
"added_at": "2024-01-10T09:00:00",
|
|
},
|
|
]
|
|
return memory
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_llm():
|
|
"""Create a mock LLM client that returns OpenAI-compatible format."""
|
|
llm = Mock()
|
|
|
|
# Return OpenAI-style message dict without tool calls
|
|
def complete_func(messages, tools=None):
|
|
return {"role": "assistant", "content": "I found what you're looking for!"}
|
|
|
|
llm.complete = Mock(side_effect=complete_func)
|
|
return llm
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_llm_with_tool_call():
|
|
"""Create a mock LLM that returns a tool call then a response."""
|
|
llm = Mock()
|
|
|
|
# First call returns a tool call, second returns final response
|
|
def complete_side_effect(messages, tools=None):
|
|
if not hasattr(complete_side_effect, "call_count"):
|
|
complete_side_effect.call_count = 0
|
|
complete_side_effect.call_count += 1
|
|
|
|
if complete_side_effect.call_count == 1:
|
|
# First call: return tool call
|
|
return {
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [
|
|
{
|
|
"id": "call_123",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "find_torrent",
|
|
"arguments": '{"media_title": "Inception"}',
|
|
},
|
|
}
|
|
],
|
|
}
|
|
else:
|
|
# Second call: return final response
|
|
return {"role": "assistant", "content": "I found 3 torrents for Inception!"}
|
|
|
|
llm.complete = Mock(side_effect=complete_side_effect)
|
|
return llm
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_tmdb_client():
|
|
"""Create a mock TMDB client."""
|
|
client = Mock()
|
|
client.search_movie = Mock(
|
|
return_value=Mock(
|
|
results=[
|
|
Mock(
|
|
id=27205,
|
|
title="Inception",
|
|
release_date="2010-07-16",
|
|
overview="A thief who steals corporate secrets...",
|
|
)
|
|
]
|
|
)
|
|
)
|
|
client.get_external_ids = Mock(return_value={"imdb_id": "tt1375666"})
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_knaben_client():
|
|
"""Create a mock Knaben client."""
|
|
client = Mock()
|
|
client.search = Mock(
|
|
return_value=[
|
|
Mock(
|
|
title="Inception.2010.1080p.BluRay",
|
|
size="2.5 GB",
|
|
seeders=150,
|
|
leechers=10,
|
|
magnet="magnet:?xt=urn:btih:abc123",
|
|
info_hash="abc123",
|
|
tracker="TPB",
|
|
upload_date="2024-01-01",
|
|
category="Movies",
|
|
),
|
|
]
|
|
)
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_qbittorrent_client():
|
|
"""Create a mock qBittorrent client."""
|
|
client = Mock()
|
|
client.add_torrent = Mock(return_value=True)
|
|
client.get_torrents = Mock(return_value=[])
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def real_folder(temp_dir):
|
|
"""Create a real folder structure for filesystem tests."""
|
|
downloads = temp_dir / "downloads"
|
|
movies = temp_dir / "movies"
|
|
tvshows = temp_dir / "tvshows"
|
|
|
|
downloads.mkdir()
|
|
movies.mkdir()
|
|
tvshows.mkdir()
|
|
|
|
# Create some test files
|
|
(downloads / "test_movie.mkv").touch()
|
|
(downloads / "test_series").mkdir()
|
|
(downloads / "test_series" / "episode1.mkv").touch()
|
|
|
|
return {
|
|
"root": temp_dir,
|
|
"downloads": downloads,
|
|
"movies": movies,
|
|
"tvshows": tvshows,
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_deepseek():
|
|
"""
|
|
Mock DeepSeekClient for individual tests that need it.
|
|
This prevents real API calls in tests that use this fixture.
|
|
|
|
Usage:
|
|
def test_something(mock_deepseek):
|
|
# Your test code here
|
|
"""
|
|
from unittest.mock import Mock
|
|
|
|
# Save the original module if it exists
|
|
original_module = sys.modules.get("agent.llm.deepseek")
|
|
|
|
# Create a mock module for deepseek
|
|
mock_deepseek_module = MagicMock()
|
|
|
|
class MockDeepSeekClient:
|
|
def __init__(self, *args, **kwargs):
|
|
self.complete = Mock(return_value="Mocked LLM response")
|
|
|
|
mock_deepseek_module.DeepSeekClient = MockDeepSeekClient
|
|
|
|
# Inject the mock
|
|
sys.modules["agent.llm.deepseek"] = mock_deepseek_module
|
|
|
|
yield mock_deepseek_module
|
|
|
|
# Restore the original module
|
|
if original_module is not None:
|
|
sys.modules["agent.llm.deepseek"] = original_module
|
|
elif "agent.llm.deepseek" in sys.modules:
|
|
del sys.modules["agent.llm.deepseek"]
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_agent_step():
|
|
"""
|
|
Fixture to easily mock the agent's step method in API tests.
|
|
Returns a context manager that patches app.agent.step.
|
|
"""
|
|
from unittest.mock import patch
|
|
|
|
def _mock_step(return_value="Mocked agent response"):
|
|
return patch("app.agent.step", return_value=return_value)
|
|
|
|
return _mock_step
|