diff --git a/.gitignore b/.gitignore index 1bb2a4f..b6eefac 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,4 @@ __pycache__/ *$py.class .venv .env -tests/ -src/ .vscode/ \ No newline at end of file diff --git a/src/jellyfin_mcp/client.py b/src/jellyfin_mcp/client.py new file mode 100644 index 0000000..08fb3a3 --- /dev/null +++ b/src/jellyfin_mcp/client.py @@ -0,0 +1,123 @@ +import httpx +import os +from typing import Optional, List, Dict, Any + + +class JellyfinClient: + """A client to interact with the Jellyfin API.""" + + def __init__(self, base_url: str, api_key: str): + """ + Initializes the JellyfinClient. + + Args: + base_url (str): The base URL of the Jellyfin server. + api_key (str): The API key for authentication. + """ + self.base_url = base_url.rstrip("/") + self.api_key = api_key + self.headers = {"Authorization": f"MediaBrowser Token=\"{self.api_key}\"", "Accept": "application/json"} + + async def _request( + self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None + ) -> Dict[str, Any]: + """ + Sends an authenticated request to the Jellyfin API. + + Args: + method (str): The HTTP method (GET, POST, etc.). + endpoint (str): The API endpoint. + params (Optional[Dict[str, Any]]): Query parameters. + + Returns: + Dict[str, Any]: The JSON response from the API. + + Raises: + httpx.HTTPStatusError: If the request was unsuccessful. + """ + url = f"{self.base_url}/{endpoint.lstrip('/')}" + async with httpx.AsyncClient(headers=self.headers) as client: + response = await client.request(method, url, params=params) + response.raise_for_status() + return response.json() + + async def search_items( + self, query: str, item_types: Optional[List[str]] = None + ) -> List[Dict[str, Any]]: + """ + Searches for items in the Jellyfin library. + + Args: + query (str): The search term. + item_types (Optional[List[str]]): A list of item types to filter by (e.g., ["Movie", "Series"]). + + Returns: + List[Dict[str, Any]]: A list of matching items. + """ + params = { + "searchTerm": query, + "Recursive": "true", + } + + if item_types: + params["IncludeItemTypes"] = ",".join(item_types) + + response = await self._request("GET", "/Items", params=params) + # Jellyfin returns an object containing 'Items' list. + return response.get("Items", []) + + async def list_active_sessions(self) -> List[Dict[str, Any]]: + """ + Retrieves a list of active user sessions. + + Returns: + List[Dict[str, Any]]: A list of active sessions. + """ + response = await self._request("GET", "/Sessions") + # Jellyfin returns an object containing 'Sessions' list. + return response + + async def search_by_genre( + self, genre: str, item_types: Optional[List[str]] = None + ) -> List[Dict[str, Any]]: + params = {"Genres": genre, "Recursive": "true"} + if item_types: + params["IncludeItemTypes"] = ",".join(item_types) + response = await self._request("GET", "/Items", params=params) + return response.get("Items", []) + + async def search_by_director( + self, director: str, item_types: Optional[List[str]] = None + ) -> List[Dict[str, Any]]: + params = {"Person": director, "PersonTypes": "Director", "Recursive": "true"} + if item_types: + params["IncludeItemTypes"] = ",".join(item_types) + response = await self._request("GET", "/Items", params=params) + return response.get("Items", []) + + async def search_by_cast( + self, person: str, item_types: Optional[List[str]] = None + ) -> List[Dict[str, Any]]: + params = {"Person": person, "PersonTypes": "Actor", "Recursive": "true"} + if item_types: + params["IncludeItemTypes"] = ",".join(item_types) + response = await self._request("GET", "/Items", params=params) + return response.get("Items", []) + + async def get_users(self) -> List[Dict[str, Any]]: + return await self._request("GET", "/Users") + + async def get_user_item_data(self, user_id: str, item_id: str) -> Dict[str, Any]: + return await self._request("GET", f"/Users/{user_id}/Items/{item_id}/UserData") + + async def get_series_seasons(self, series_id: str) -> List[Dict[str, Any]]: + response = await self._request( + "GET", f"/Shows/{series_id}/Seasons", params={"Fields": "ChildCount"} + ) + return response.get("Items", []) + + async def get_season_episodes(self, series_id: str, season_number: int) -> List[Dict[str, Any]]: + response = await self._request( + "GET", f"/Shows/{series_id}/Episodes", params={"season": season_number} + ) + return response.get("Items", []) diff --git a/src/jellyfin_mcp/server.py b/src/jellyfin_mcp/server.py new file mode 100644 index 0000000..3517236 --- /dev/null +++ b/src/jellyfin_mcp/server.py @@ -0,0 +1,273 @@ +import os +from typing import List, Dict, Any + +from dotenv import load_dotenv +from jellyfin_mcp.client import JellyfinClient +from mcp.server.fastmcp import FastMCP + +# Load environment variables from .env file +load_dotenv() + +JELLYFIN_URL = os.getenv("JELLYFIN_URL") +JELLYFIN_API_KEY = os.getenv("JELLYFIN_API_KEY") + +if not JELLYFIN_URL or not JELLYFIN_API_KEY: + raise ValueError( + "Missing JELLYFIN_URL or JELLYFIN_API_KEY in environment variables." + ) + +# Initialize the MCP server using FastMCP +mcp = FastMCP("jellyfin") + +# Initialize the Jellyfin client +client = JellyfinClient(JELLYFIN_URL, JELLYFIN_API_KEY) + + +@mcp.tool() +async def search_items(query: str, item_types: str = "") -> str: + """ + Search for items (movies, series, etc.) in the Jellyfin library. + + Args: + query: The search term to look for. + item_types: A comma-separated list of item types to filter by (e.g., 'Movie,Series'). + """ + item_types_list = [t.strip() for t in item_types.split(",") if t.strip()] + + try: + items = await client.search_items( + query, item_types=item_types_list if item_types_list else None + ) + if not items: + return f"No items found for query: '{query}'" + + results = [] + for item in items: + name = item.get("Name", "Unknown") + item_type = item.get("Type", "Unknown") + # Including ID and Type for better context in AI interaction + results.append(f"{name} ({item_type}) [ID: {item.get('Id')}]") + + return "\n".join(results) + except Exception as e: + return f"Error searching items: {str(e)}" + + +@mcp.tool() +async def list_active_sessions() -> str: + """ + Retrieve a list of active user sessions and playback information. + """ + try: + sessions = await client.list_active_sessions() + if not sessions: + return "No active sessions found." + + results = [] + for session in sessions: + user_name = session.get("UserName", "Unknown User") + device = session.get("DeviceName", "Unknown Device") + # Use a simplified list for the AI to understand current activity + results.append(f"- User: {user_name}, Device: {device}") + + return "\n".join(results) + except Exception as e: + return f"Error listing sessions: {str(e)}" + + +@mcp.tool() +async def search_by_genre(genre: str, item_types: str = "") -> str: + """ + Search for items in the Jellyfin library by genre. + + Args: + genre: The genre to filter by (e.g., 'Action', 'Comedy'). + item_types: A comma-separated list of item types to filter by (e.g., 'Movie,Series'). + """ + item_types_list = [t.strip() for t in item_types.split(",") if t.strip()] + + try: + items = await client.search_by_genre( + genre, item_types=item_types_list if item_types_list else None + ) + if not items: + return f"No items found for genre: '{genre}'" + + results = [] + for item in items: + name = item.get("Name", "Unknown") + item_type = item.get("Type", "Unknown") + results.append(f"{name} ({item_type}) [ID: {item.get('Id')}]") + + return "\n".join(results) + except Exception as e: + return f"Error searching by genre: {str(e)}" + + +@mcp.tool() +async def search_by_director(director: str, item_types: str = "") -> str: + """ + Search for items in the Jellyfin library directed by a specific person. + + Args: + director: The name of the director to search for. + item_types: A comma-separated list of item types to filter by (e.g., 'Movie,Series'). + """ + item_types_list = [t.strip() for t in item_types.split(",") if t.strip()] + + try: + items = await client.search_by_director( + director, item_types=item_types_list if item_types_list else None + ) + if not items: + return f"No items found for director: '{director}'" + + results = [] + for item in items: + name = item.get("Name", "Unknown") + item_type = item.get("Type", "Unknown") + results.append(f"{name} ({item_type}) [ID: {item.get('Id')}]") + + return "\n".join(results) + except Exception as e: + return f"Error searching by director: {str(e)}" + + +@mcp.tool() +async def search_by_cast(person: str, item_types: str = "") -> str: + """ + Search for items in the Jellyfin library featuring a specific cast member. + + Args: + person: The name of the actor to search for. + item_types: A comma-separated list of item types to filter by (e.g., 'Movie,Series'). + """ + item_types_list = [t.strip() for t in item_types.split(",") if t.strip()] + + try: + items = await client.search_by_cast( + person, item_types=item_types_list if item_types_list else None + ) + if not items: + return f"No items found for cast member: '{person}'" + + results = [] + for item in items: + name = item.get("Name", "Unknown") + item_type = item.get("Type", "Unknown") + results.append(f"{name} ({item_type}) [ID: {item.get('Id')}]") + + return "\n".join(results) + except Exception as e: + return f"Error searching by cast: {str(e)}" + + +@mcp.tool() +async def get_users() -> str: + """ + Retrieve a list of all Jellyfin users. + """ + try: + users = await client.get_users() + if not users: + return "No users found." + + results = [] + for user in users: + name = user.get("Name", "Unknown") + user_id = user.get("Id") + is_admin = user.get("Policy", {}).get("IsAdministrator", False) + suffix = " (admin)" if is_admin else "" + results.append(f"- {name}{suffix} [ID: {user_id}]") + + return "\n".join(results) + except Exception as e: + return f"Error retrieving users: {str(e)}" + + +@mcp.tool() +async def get_user_item_data(user_id: str, item_id: str) -> str: + """ + Retrieve a user's playback data for a specific item. + + Args: + user_id: The ID of the user. + item_id: The ID of the media item. + """ + try: + data = await client.get_user_item_data(user_id, item_id) + + played = data.get("Played", False) + play_count = data.get("PlayCount", 0) + last_played = data.get("LastPlayedDate") or "Never" + position_ticks = data.get("PlaybackPositionTicks", 0) + position_seconds = position_ticks // 10_000_000 + is_favorite = data.get("IsFavorite", False) + + lines = [ + f"Played: {'Yes' if played else 'No'}", + f"Play count: {play_count}", + f"Last played: {last_played}", + f"Playback position: {position_seconds}s", + f"Favorite: {'Yes' if is_favorite else 'No'}", + ] + return "\n".join(lines) + except Exception as e: + return f"Error retrieving user item data: {str(e)}" + + +@mcp.tool() +async def get_series_seasons(series_id: str) -> str: + """ + List all seasons of a TV series, including episode counts and season IDs. + + Args: + series_id: The ID of the TV series. + """ + try: + seasons = await client.get_series_seasons(series_id) + if not seasons: + return f"No seasons found for series ID: '{series_id}'" + + total_episodes = sum(s.get("ChildCount", 0) for s in seasons) + lines = [f"Seasons: {len(seasons)} | Total episodes: {total_episodes}", ""] + for season in seasons: + name = season.get("Name", "Unknown") + season_id = season.get("Id") + episode_count = season.get("ChildCount", 0) + lines.append(f"{name} - {episode_count} episodes [ID: {season_id}]") + + return "\n".join(lines) + except Exception as e: + return f"Error retrieving seasons: {str(e)}" + + +@mcp.tool() +async def get_season_episodes(series_id: str, season_number: int) -> str: + """ + List all episodes in a season of a TV series, including episode IDs. + + Args: + series_id: The ID of the TV series. + season_number: The season number (e.g., 1 for Season 1). + """ + try: + episodes = await client.get_season_episodes(series_id, season_number) + if not episodes: + return f"No episodes found for season {season_number} of series ID: '{series_id}'" + + lines = [f"Season {season_number} - {len(episodes)} episodes", ""] + for episode in episodes: + name = episode.get("Name", "Unknown") + episode_id = episode.get("Id") + ep_num = episode.get("IndexNumber", 0) + season_num = episode.get("ParentIndexNumber", 0) + lines.append(f"S{season_num:02d}E{ep_num:02d} - {name} [ID: {episode_id}]") + + return "\n".join(lines) + except Exception as e: + return f"Error retrieving episodes: {str(e)}" + + +if __name__ == "__main__": + mcp.run() diff --git a/src/mcp_jellyfin.egg-info/PKG-INFO b/src/mcp_jellyfin.egg-info/PKG-INFO new file mode 100644 index 0000000..8d63e6b --- /dev/null +++ b/src/mcp_jellyfin.egg-info/PKG-INFO @@ -0,0 +1,62 @@ +Metadata-Version: 2.4 +Name: mcp-jellyfin +Version: 0.1.0 +Summary: An MCP server to interface with the Jellyfin API. +Requires-Python: >=3.10 +Description-Content-Type: text/markdown +Requires-Dist: mcp +Requires-Dist: httpx +Requires-Dist: python-dotenv +Provides-Extra: dev +Requires-Dist: pytest; extra == "dev" +Requires-Dist: pytest-asyncio; extra == "dev" + +# mcp-jellyfin + +An MCP server to interface with the Jellyfin API. + +## Features +- Search through Jellyfin libraries (Movies, Series, etc.) +- List active user sessions and playback information + +## Setup + +### Prerequisites +- Python 3.10+ +- `uv` (recommended) or `pip` + +### Installation + +Using [uv](https://github.com/astral-sh/uv): +```bash +uv sync +``` + +Using pip: +```bash +pip install -e . +``` + +### Configuration + +Create a `.env` file based on `.env.example`: +```bash +cp .env.example .env +``` + +Add your Jellyfin credentials to the `.env` file: +- `JELLYFIN_URL`: Your Jellyfin server URL (e.g., `http://192.168.1.10:8096`) +- `JELLYFIN_API_KEY`: Your Jellyfin API key + +## Usage + +To run the MCP server: +```bash +python src/jellyfin_mcp/server.py +``` + +## Testing + +```bash +pytest +``` diff --git a/src/mcp_jellyfin.egg-info/SOURCES.txt b/src/mcp_jellyfin.egg-info/SOURCES.txt new file mode 100644 index 0000000..efa504b --- /dev/null +++ b/src/mcp_jellyfin.egg-info/SOURCES.txt @@ -0,0 +1,12 @@ +README.md +pyproject.toml +src/jellyfin_mcp/__init__.py +src/jellyfin_mcp/client.py +src/jellyfin_mcp/server.py +src/mcp_jellyfin.egg-info/PKG-INFO +src/mcp_jellyfin.egg-info/SOURCES.txt +src/mcp_jellyfin.egg-info/dependency_links.txt +src/mcp_jellyfin.egg-info/requires.txt +src/mcp_jellyfin.egg-info/top_level.txt +tests/test_client.py +tests/test_server.py \ No newline at end of file diff --git a/src/mcp_jellyfin.egg-info/dependency_links.txt b/src/mcp_jellyfin.egg-info/dependency_links.txt new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/mcp_jellyfin.egg-info/dependency_links.txt @@ -0,0 +1 @@ + diff --git a/src/mcp_jellyfin.egg-info/requires.txt b/src/mcp_jellyfin.egg-info/requires.txt new file mode 100644 index 0000000..526a8db --- /dev/null +++ b/src/mcp_jellyfin.egg-info/requires.txt @@ -0,0 +1,7 @@ +mcp +httpx +python-dotenv + +[dev] +pytest +pytest-asyncio diff --git a/src/mcp_jellyfin.egg-info/top_level.txt b/src/mcp_jellyfin.egg-info/top_level.txt new file mode 100644 index 0000000..8490cc2 --- /dev/null +++ b/src/mcp_jellyfin.egg-info/top_level.txt @@ -0,0 +1 @@ +jellyfin_mcp diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..f3df23c --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,96 @@ +import pytest +import httpx +from unittest.mock import AsyncMock, patch +from jellyfin_mcp.client import JellyfinClient + + +@pytest.fixture +def client(): + return JellyfinClient(base_url="http://test-jellyfin.local", api_key="test-api-key") + + +@pytest.mark.asyncio +async def test_client_initialization(client): + assert client.base_url == "http://test-jellyfin.local" + assert client.api_key == "test-api-key" + assert client.headers["Authorization"] == 'MediaBrowser Token="test-api-key"' + + +@pytest.mark.asyncio +async def test_request_success(client): + # Mocking the httpx.AsyncClient context manager and the request method. + # In 'async with httpx.AsyncClient(...) as client:', + # the '__aenter__' method of AsyncClient returns the client instance. + # The 'request' method is what we want to mock. + + with patch("httpx.AsyncClient.request", new_callable=AsyncMock) as mock_request: + # We need to simulate the response object returned by the awaitable. + mock_response = AsyncMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = {"test": "data"} + # When 'await client.request(...)' is called, it returns mock_response + mock_request.return_value = mock_response + + result = await client._request("GET", "/test-endpoint") + + assert result == {"test": "data"} + + +@pytest.mark.asyncio +async def test_request_failure(client): + with patch("httpx.AsyncClient.request", new_callable=AsyncMock) as mock_request: + mock_response = AsyncMock(spec=httpx.Response) + # Setup side effect for raise_for_status + mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( + "Error", request=AsyncMock(), response=AsyncMock(status_code=404) + ) + mock_request.return_value = mock_response + + with pytest.raises(httpx.HTTPStatusError): + await client._request("GET", "/bad-endpoint") + + +@pytest.mark.asyncio +async def test_search_items_success(client): + mock_data = {"Items": [{"Id": "1", "Name": "Test Movie", "Type": "Movie"}]} + + with patch("httpx.AsyncClient.request", new_callable=AsyncMock) as mock_request: + mock_response = AsyncMock(spec=httpx.Response) + mock_response.json.return_value = mock_data + mock_request.return_value = mock_response + + results = await client.search_items("test") + + assert len(results) == 1 + assert results[0]["Name"] == "Test Movie" + + +@pytest.mark.asyncio +async def test_search_items_with_types(client): + mock_data = {"Items": [{"Id": "1", "Name": "Test Movie", "Type": "Movie"}]} + + with patch("httpx.AsyncClient.request", new_callable=AsyncMock) as mock_request: + mock_response = AsyncMock(spec=httpx.Response) + mock_response.json.return_value = mock_data + mock_request.return_value = mock_response + + results = await client.search_items("test", item_types=["Movie"]) + + assert len(results) == 1 + args, kwargs = mock_request.call_args + assert kwargs["params"]["IncludeItemTypes"] == "Movie" + + +@pytest.mark.asyncio +async def test_list_active_sessions_success(client): + mock_data = {"Sessions": [{"UserName": "User1", "DeviceName": "Phone"}]} + + with patch("httpx.AsyncClient.request", new_callable=AsyncMock) as mock_request: + mock_response = AsyncMock(spec=httpx.Response) + mock_response.json.return_value = mock_data + mock_request.return_value = mock_response + + sessions = await client.list_active_sessions() + + assert len(sessions) == 1 + assert sessions[0]["UserName"] == "User1" diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..fc162a8 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,310 @@ +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