403 lines
14 KiB
Python
403 lines
14 KiB
Python
"""Tests for API tools - Refactored to use real components with minimal mocking."""
|
|
|
|
from unittest.mock import Mock, patch
|
|
|
|
from alfred.agent.tools import api as api_tools
|
|
from alfred.infrastructure.persistence import get_memory
|
|
|
|
|
|
def create_mock_response(status_code, json_data=None, text=None):
|
|
"""Helper to create properly mocked HTTP response."""
|
|
response = Mock()
|
|
response.status_code = status_code
|
|
response.raise_for_status = Mock()
|
|
if json_data is not None:
|
|
response.json = Mock(return_value=json_data)
|
|
if text is not None:
|
|
response.text = text
|
|
return response
|
|
|
|
|
|
class TestFindMediaImdbId:
|
|
"""Tests for find_media_imdb_id tool."""
|
|
|
|
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
|
|
def test_success(self, mock_get, memory):
|
|
"""Should return movie info on success."""
|
|
|
|
# Mock HTTP responses
|
|
def mock_get_side_effect(url, **kwargs):
|
|
if "search" in url:
|
|
return create_mock_response(
|
|
200,
|
|
json_data={
|
|
"results": [
|
|
{
|
|
"id": 27205,
|
|
"title": "Inception",
|
|
"release_date": "2010-07-16",
|
|
"overview": "A thief...",
|
|
"media_type": "movie",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
elif "external_ids" in url:
|
|
return create_mock_response(200, json_data={"imdb_id": "tt1375666"})
|
|
|
|
mock_get.side_effect = mock_get_side_effect
|
|
|
|
result = api_tools.find_media_imdb_id("Inception")
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["imdb_id"] == "tt1375666"
|
|
assert result["title"] == "Inception"
|
|
|
|
# Verify HTTP calls
|
|
assert mock_get.call_count == 2
|
|
|
|
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
|
|
def test_stores_in_stm(self, mock_get, memory):
|
|
"""Should store result in STM on success."""
|
|
|
|
def mock_get_side_effect(url, **kwargs):
|
|
if "search" in url:
|
|
return create_mock_response(
|
|
200,
|
|
json_data={
|
|
"results": [
|
|
{
|
|
"id": 27205,
|
|
"title": "Inception",
|
|
"release_date": "2010-07-16",
|
|
"media_type": "movie",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
elif "external_ids" in url:
|
|
return create_mock_response(200, json_data={"imdb_id": "tt1375666"})
|
|
|
|
mock_get.side_effect = mock_get_side_effect
|
|
|
|
api_tools.find_media_imdb_id("Inception")
|
|
|
|
mem = get_memory()
|
|
entity = mem.stm.get_entity("last_media_search")
|
|
assert entity is not None
|
|
assert entity["title"] == "Inception"
|
|
assert mem.stm.current_topic == "searching_media"
|
|
|
|
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
|
|
def test_not_found(self, mock_get, memory):
|
|
"""Should return error when not found."""
|
|
mock_get.return_value = create_mock_response(200, json_data={"results": []})
|
|
|
|
result = api_tools.find_media_imdb_id("NonexistentMovie12345")
|
|
|
|
assert result["status"] == "error"
|
|
assert result["error"] == "not_found"
|
|
|
|
@patch("alfred.infrastructure.api.tmdb.client.requests.get")
|
|
def test_does_not_store_on_error(self, mock_get, memory):
|
|
"""Should not store in STM on error."""
|
|
mock_get.return_value = create_mock_response(200, json_data={"results": []})
|
|
|
|
api_tools.find_media_imdb_id("Test")
|
|
|
|
mem = get_memory()
|
|
assert mem.stm.get_entity("last_media_search") is None
|
|
|
|
|
|
class TestFindTorrent:
|
|
"""Tests for find_torrent tool."""
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_success(self, mock_post, memory):
|
|
"""Should return torrents on success."""
|
|
mock_post.return_value = create_mock_response(
|
|
200,
|
|
json_data={
|
|
"hits": [
|
|
{
|
|
"title": "Torrent 1",
|
|
"seeders": 100,
|
|
"leechers": 10,
|
|
"magnetUrl": "magnet:?xt=...",
|
|
"size": "2.5 GB",
|
|
},
|
|
{
|
|
"title": "Torrent 2",
|
|
"seeders": 50,
|
|
"leechers": 5,
|
|
"magnetUrl": "magnet:?xt=...",
|
|
"size": "1.8 GB",
|
|
},
|
|
]
|
|
},
|
|
)
|
|
|
|
result = api_tools.find_torrent("Inception 1080p")
|
|
|
|
assert result["status"] == "ok"
|
|
assert len(result["torrents"]) == 2
|
|
|
|
# Verify HTTP payload
|
|
payload = mock_post.call_args[1]["json"]
|
|
assert payload["query"] == "Inception 1080p"
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_stores_in_episodic(self, mock_post, memory):
|
|
"""Should store results in episodic memory."""
|
|
mock_post.return_value = create_mock_response(
|
|
200,
|
|
json_data={
|
|
"hits": [
|
|
{
|
|
"title": "Torrent 1",
|
|
"seeders": 100,
|
|
"leechers": 10,
|
|
"magnetUrl": "magnet:?xt=...",
|
|
"size": "2.5 GB",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
api_tools.find_torrent("Inception")
|
|
|
|
mem = get_memory()
|
|
assert mem.episodic.last_search_results is not None
|
|
assert mem.episodic.last_search_results["query"] == "Inception"
|
|
assert mem.stm.current_topic == "selecting_torrent"
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_results_have_indexes(self, mock_post, memory):
|
|
"""Should add indexes to results."""
|
|
mock_post.return_value = create_mock_response(
|
|
200,
|
|
json_data={
|
|
"hits": [
|
|
{
|
|
"title": "Torrent 1",
|
|
"seeders": 100,
|
|
"leechers": 10,
|
|
"magnetUrl": "magnet:?xt=1",
|
|
"size": "1GB",
|
|
},
|
|
{
|
|
"title": "Torrent 2",
|
|
"seeders": 50,
|
|
"leechers": 5,
|
|
"magnetUrl": "magnet:?xt=2",
|
|
"size": "2GB",
|
|
},
|
|
{
|
|
"title": "Torrent 3",
|
|
"seeders": 25,
|
|
"leechers": 2,
|
|
"magnetUrl": "magnet:?xt=3",
|
|
"size": "3GB",
|
|
},
|
|
]
|
|
},
|
|
)
|
|
|
|
api_tools.find_torrent("Test")
|
|
|
|
mem = get_memory()
|
|
results = mem.episodic.last_search_results["results"]
|
|
assert results[0]["index"] == 1
|
|
assert results[1]["index"] == 2
|
|
assert results[2]["index"] == 3
|
|
|
|
@patch("alfred.infrastructure.api.knaben.client.requests.post")
|
|
def test_not_found(self, mock_post, memory):
|
|
"""Should return error when no torrents found."""
|
|
mock_post.return_value = create_mock_response(200, json_data={"hits": []})
|
|
|
|
result = api_tools.find_torrent("NonexistentMovie12345")
|
|
|
|
assert result["status"] == "error"
|
|
|
|
|
|
class TestGetTorrentByIndex:
|
|
"""Tests for get_torrent_by_index tool."""
|
|
|
|
def test_success(self, memory_with_search_results):
|
|
"""Should return torrent at index."""
|
|
result = api_tools.get_torrent_by_index(2)
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["torrent"]["name"] == "Inception.2010.1080p.WEB-DL.x265"
|
|
|
|
def test_first_index(self, memory_with_search_results):
|
|
"""Should return first torrent."""
|
|
result = api_tools.get_torrent_by_index(1)
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["torrent"]["name"] == "Inception.2010.1080p.BluRay.x264"
|
|
|
|
def test_last_index(self, memory_with_search_results):
|
|
"""Should return last torrent."""
|
|
result = api_tools.get_torrent_by_index(3)
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["torrent"]["name"] == "Inception.2010.720p.BluRay"
|
|
|
|
def test_index_out_of_range(self, memory_with_search_results):
|
|
"""Should return error for invalid index."""
|
|
result = api_tools.get_torrent_by_index(10)
|
|
|
|
assert result["status"] == "error"
|
|
assert result["error"] == "not_found"
|
|
|
|
def test_index_zero(self, memory_with_search_results):
|
|
"""Should return error for index 0."""
|
|
result = api_tools.get_torrent_by_index(0)
|
|
|
|
assert result["status"] == "error"
|
|
assert result["error"] == "not_found"
|
|
|
|
def test_negative_index(self, memory_with_search_results):
|
|
"""Should return error for negative index."""
|
|
result = api_tools.get_torrent_by_index(-1)
|
|
|
|
assert result["status"] == "error"
|
|
assert result["error"] == "not_found"
|
|
|
|
def test_no_search_results(self, memory):
|
|
"""Should return error if no search results."""
|
|
result = api_tools.get_torrent_by_index(1)
|
|
|
|
assert result["status"] == "error"
|
|
assert result["error"] == "not_found"
|
|
assert "Search for torrents first" in result["message"]
|
|
|
|
|
|
class TestAddTorrentToQbittorrent:
|
|
"""Tests for add_torrent_to_qbittorrent tool.
|
|
|
|
Note: These tests mock the qBittorrent client because:
|
|
1. The client requires authentication/session management
|
|
2. We want to test the tool's logic (memory updates, workflow management)
|
|
3. The client itself is tested separately in infrastructure tests
|
|
|
|
This is acceptable mocking because we're testing the TOOL logic, not the client.
|
|
"""
|
|
|
|
@patch("alfred.agent.tools.api.qbittorrent_client")
|
|
def test_success(self, mock_client, memory):
|
|
"""Should add torrent successfully and update memory."""
|
|
mock_client.add_torrent.return_value = True
|
|
|
|
result = api_tools.add_torrent_to_qbittorrent("magnet:?xt=urn:btih:abc123")
|
|
|
|
# Test tool logic
|
|
assert result["status"] == "ok"
|
|
# Verify client was called correctly
|
|
mock_client.add_torrent.assert_called_once_with("magnet:?xt=urn:btih:abc123")
|
|
|
|
@patch("alfred.agent.tools.api.qbittorrent_client")
|
|
def test_adds_to_active_downloads(self, mock_client, memory_with_search_results):
|
|
"""Should add to active downloads on success."""
|
|
mock_client.add_torrent.return_value = True
|
|
|
|
api_tools.add_torrent_to_qbittorrent("magnet:?xt=urn:btih:abc123")
|
|
|
|
# Test memory update logic
|
|
mem = get_memory()
|
|
assert len(mem.episodic.active_downloads) == 1
|
|
assert (
|
|
mem.episodic.active_downloads[0]["name"]
|
|
== "Inception.2010.1080p.BluRay.x264"
|
|
)
|
|
|
|
@patch("alfred.agent.tools.api.qbittorrent_client")
|
|
def test_sets_topic_and_ends_workflow(self, mock_client, memory):
|
|
"""Should set topic and end workflow."""
|
|
mock_client.add_torrent.return_value = True
|
|
memory.stm.start_workflow("download", {"title": "Test"})
|
|
|
|
api_tools.add_torrent_to_qbittorrent("magnet:?xt=...")
|
|
|
|
# Test workflow management logic
|
|
mem = get_memory()
|
|
assert mem.stm.current_topic == "downloading"
|
|
assert mem.stm.current_workflow is None
|
|
|
|
@patch("alfred.agent.tools.api.qbittorrent_client")
|
|
def test_error_handling(self, mock_client, memory):
|
|
"""Should handle client errors correctly."""
|
|
from alfred.infrastructure.api.qbittorrent.exceptions import QBittorrentAPIError
|
|
|
|
mock_client.add_torrent.side_effect = QBittorrentAPIError("Connection failed")
|
|
|
|
result = api_tools.add_torrent_to_qbittorrent("magnet:?xt=...")
|
|
|
|
# Test error handling logic
|
|
assert result["status"] == "error"
|
|
|
|
|
|
class TestAddTorrentByIndex:
|
|
"""Tests for add_torrent_by_index tool.
|
|
|
|
These tests verify the tool's logic:
|
|
- Getting torrent from memory by index
|
|
- Extracting magnet link
|
|
- Calling add_torrent_to_qbittorrent
|
|
- Error handling for edge cases
|
|
"""
|
|
|
|
@patch("alfred.agent.tools.api.qbittorrent_client")
|
|
def test_success(self, mock_client, memory_with_search_results):
|
|
"""Should get torrent by index and add it."""
|
|
mock_client.add_torrent.return_value = True
|
|
|
|
result = api_tools.add_torrent_by_index(1)
|
|
|
|
# Test tool logic
|
|
assert result["status"] == "ok"
|
|
assert result["torrent_name"] == "Inception.2010.1080p.BluRay.x264"
|
|
# Verify correct magnet was extracted and used
|
|
mock_client.add_torrent.assert_called_once_with("magnet:?xt=urn:btih:abc123")
|
|
|
|
@patch("alfred.agent.tools.api.qbittorrent_client")
|
|
def test_uses_correct_magnet(self, mock_client, memory_with_search_results):
|
|
"""Should extract correct magnet from index."""
|
|
mock_client.add_torrent.return_value = True
|
|
|
|
api_tools.add_torrent_by_index(2)
|
|
|
|
# Test magnet extraction logic
|
|
mock_client.add_torrent.assert_called_once_with("magnet:?xt=urn:btih:def456")
|
|
|
|
def test_invalid_index(self, memory_with_search_results):
|
|
"""Should return error for invalid index."""
|
|
result = api_tools.add_torrent_by_index(99)
|
|
|
|
# Test error handling logic (no mock needed)
|
|
assert result["status"] == "error"
|
|
assert result["error"] == "not_found"
|
|
|
|
def test_no_search_results(self, memory):
|
|
"""Should return error if no search results."""
|
|
result = api_tools.add_torrent_by_index(1)
|
|
|
|
# Test error handling logic (no mock needed)
|
|
assert result["status"] == "error"
|
|
assert result["error"] == "not_found"
|
|
|
|
def test_no_magnet_link(self, memory):
|
|
"""Should return error if torrent has no magnet."""
|
|
memory.episodic.store_search_results(
|
|
"test",
|
|
[{"name": "Torrent without magnet", "seeders": 100}],
|
|
)
|
|
|
|
result = api_tools.add_torrent_by_index(1)
|
|
|
|
# Test error handling logic (no mock needed)
|
|
assert result["status"] == "error"
|
|
assert result["error"] == "no_magnet"
|