311 lines
12 KiB
Python
311 lines
12 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, patch
|
|
from jellyfin_mcp.client import JellyfinClient
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_env(monkeypatch):
|
|
monkeypatch.setenv("JELLYFIN_URL", "http://test-jellyfin.local")
|
|
monkeypatch.setenv("JELLYFIN_API_KEY", "test-api-key")
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_client():
|
|
return JellyfinClient(base_url="http://test-jellyfin.local", api_key="test-api-key")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_items_tool(mock_client):
|
|
from jellyfin_mcp.server import search_items
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
# Test with no items found
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_items", new_callable=AsyncMock
|
|
) as mock_search:
|
|
mock_search.return_value = []
|
|
result = await search_items(query="nothing")
|
|
assert "No items found" in result
|
|
|
|
# Test with items found
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_items", new_callable=AsyncMock
|
|
) as mock_search:
|
|
mock_search.return_value = [{"Name": "Movie A", "Type": "Movie", "Id": "1"}]
|
|
result = await search_items(query="movie")
|
|
assert "- Movie A (Movie) [ID: 1]" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_active_sessions_tool(mock_client):
|
|
from jellyfin_mcp.server import list_active_sessions
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
# Test with no sessions found
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.list_active_sessions",
|
|
new_callable=AsyncMock,
|
|
) as mock_sessions:
|
|
mock_sessions.return_value = []
|
|
result = await list_active_sessions()
|
|
assert "No active sessions found" in result
|
|
|
|
# Test with active sessions
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.list_active_sessions",
|
|
new_callable=AsyncMock,
|
|
) as mock_sessions:
|
|
mock_sessions.return_value = [{"UserName": "Alice", "DeviceName": "iPad"}]
|
|
result = await list_active_sessions()
|
|
assert "- User: Alice, Device: iPad" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_items_tool_error(mock_client):
|
|
from jellyfin_mcp.server import search_items
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_items",
|
|
side_effect=Exception("API Error"),
|
|
):
|
|
result = await search_items(query="error")
|
|
assert "Error searching items: API Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_active_sessions_tool_error(mock_client):
|
|
from jellyfin_mcp.server import list_active_sessions
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.list_active_sessions",
|
|
side_effect=Exception("API Error"),
|
|
):
|
|
result = await list_active_sessions()
|
|
assert "Error listing sessions: API Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_by_genre_tool(mock_client):
|
|
from jellyfin_mcp.server import search_by_genre
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_by_genre", new_callable=AsyncMock
|
|
) as mock_search:
|
|
mock_search.return_value = []
|
|
result = await search_by_genre(genre="Horror")
|
|
assert "No items found for genre: 'Horror'" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_by_genre", new_callable=AsyncMock
|
|
) as mock_search:
|
|
mock_search.return_value = [{"Name": "The Shining", "Type": "Movie", "Id": "abc1"}]
|
|
result = await search_by_genre(genre="Horror")
|
|
assert "The Shining (Movie) [ID: abc1]" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_by_genre",
|
|
side_effect=Exception("API Error"),
|
|
):
|
|
result = await search_by_genre(genre="Horror")
|
|
assert "Error searching by genre: API Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_by_director_tool(mock_client):
|
|
from jellyfin_mcp.server import search_by_director
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_by_director", new_callable=AsyncMock
|
|
) as mock_search:
|
|
mock_search.return_value = []
|
|
result = await search_by_director(director="Kubrick")
|
|
assert "No items found for director: 'Kubrick'" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_by_director", new_callable=AsyncMock
|
|
) as mock_search:
|
|
mock_search.return_value = [{"Name": "Full Metal Jacket", "Type": "Movie", "Id": "abc2"}]
|
|
result = await search_by_director(director="Kubrick")
|
|
assert "Full Metal Jacket (Movie) [ID: abc2]" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_by_director",
|
|
side_effect=Exception("API Error"),
|
|
):
|
|
result = await search_by_director(director="Kubrick")
|
|
assert "Error searching by director: API Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_by_cast_tool(mock_client):
|
|
from jellyfin_mcp.server import search_by_cast
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_by_cast", new_callable=AsyncMock
|
|
) as mock_search:
|
|
mock_search.return_value = []
|
|
result = await search_by_cast(person="Nicholson")
|
|
assert "No items found for cast member: 'Nicholson'" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_by_cast", new_callable=AsyncMock
|
|
) as mock_search:
|
|
mock_search.return_value = [{"Name": "The Departed", "Type": "Movie", "Id": "abc3"}]
|
|
result = await search_by_cast(person="Nicholson")
|
|
assert "The Departed (Movie) [ID: abc3]" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.search_by_cast",
|
|
side_effect=Exception("API Error"),
|
|
):
|
|
result = await search_by_cast(person="Nicholson")
|
|
assert "Error searching by cast: API Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_users_tool(mock_client):
|
|
from jellyfin_mcp.server import get_users
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_users", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.return_value = []
|
|
result = await get_users()
|
|
assert "No users found." in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_users", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.return_value = [
|
|
{"Name": "Alice", "Id": "u1", "Policy": {"IsAdministrator": False}},
|
|
{"Name": "Bob", "Id": "u2", "Policy": {"IsAdministrator": True}},
|
|
]
|
|
result = await get_users()
|
|
assert "- Alice [ID: u1]" in result
|
|
assert "(admin)" not in result.split("Alice")[1].split("\n")[0]
|
|
assert "- Bob (admin) [ID: u2]" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_users",
|
|
side_effect=Exception("API Error"),
|
|
):
|
|
result = await get_users()
|
|
assert "Error retrieving users: API Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_item_data_tool(mock_client):
|
|
from jellyfin_mcp.server import get_user_item_data
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_user_item_data", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.return_value = {
|
|
"Played": False,
|
|
"PlayCount": 0,
|
|
"PlaybackPositionTicks": 0,
|
|
"LastPlayedDate": None,
|
|
"IsFavorite": False,
|
|
}
|
|
result = await get_user_item_data(user_id="u1", item_id="i1")
|
|
assert "Played: No" in result
|
|
assert "Play count: 0" in result
|
|
assert "Last played: Never" in result
|
|
assert "Playback position: 0s" in result
|
|
assert "Favorite: No" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_user_item_data", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.return_value = {
|
|
"Played": True,
|
|
"PlayCount": 3,
|
|
"PlaybackPositionTicks": 50_000_000_000,
|
|
"LastPlayedDate": "2024-01-15T20:00:00Z",
|
|
"IsFavorite": True,
|
|
}
|
|
result = await get_user_item_data(user_id="u1", item_id="i1")
|
|
assert "Played: Yes" in result
|
|
assert "Play count: 3" in result
|
|
assert "Last played: 2024-01-15T20:00:00Z" in result
|
|
assert "Playback position: 5000s" in result
|
|
assert "Favorite: Yes" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_user_item_data",
|
|
side_effect=Exception("API Error"),
|
|
):
|
|
result = await get_user_item_data(user_id="u1", item_id="i1")
|
|
assert "Error retrieving user item data: API Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_series_seasons_tool(mock_client):
|
|
from jellyfin_mcp.server import get_series_seasons
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_series_seasons", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.return_value = []
|
|
result = await get_series_seasons(series_id="ser1")
|
|
assert "No seasons found" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_series_seasons", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.return_value = [
|
|
{"Name": "Season 1", "Id": "s1", "ChildCount": 10},
|
|
{"Name": "Season 2", "Id": "s2", "ChildCount": 8},
|
|
]
|
|
result = await get_series_seasons(series_id="ser1")
|
|
assert "Seasons: 2 | Total episodes: 18" in result
|
|
assert "Season 1 - 10 episodes [ID: s1]" in result
|
|
assert "Season 2 - 8 episodes [ID: s2]" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_series_seasons",
|
|
side_effect=Exception("API Error"),
|
|
):
|
|
result = await get_series_seasons(series_id="ser1")
|
|
assert "Error retrieving seasons: API Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_season_episodes_tool(mock_client):
|
|
from jellyfin_mcp.server import get_season_episodes
|
|
|
|
with patch("jellyfin_mcp.server.client", mock_client):
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_season_episodes", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.return_value = []
|
|
result = await get_season_episodes(series_id="ser1", season_number=1)
|
|
assert "No episodes found" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_season_episodes", new_callable=AsyncMock
|
|
) as mock_get:
|
|
mock_get.return_value = [
|
|
{"Name": "Pilot", "Id": "e1", "IndexNumber": 1, "ParentIndexNumber": 1},
|
|
{"Name": "The Second One", "Id": "e2", "IndexNumber": 2, "ParentIndexNumber": 1},
|
|
]
|
|
result = await get_season_episodes(series_id="ser1", season_number=1)
|
|
assert "Season 1 - 2 episodes" in result
|
|
assert "S01E01 - Pilot [ID: e1]" in result
|
|
assert "S01E02 - The Second One [ID: e2]" in result
|
|
|
|
with patch(
|
|
"jellyfin_mcp.client.JellyfinClient.get_season_episodes",
|
|
side_effect=Exception("API Error"),
|
|
):
|
|
result = await get_season_episodes(series_id="ser1", season_number=1)
|
|
assert "Error retrieving episodes: API Error" in result
|