diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7134347..e33183d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,10 +22,19 @@ jobs: with: python-version: ${{ matrix.python-version }} - - name: Install dependencies + - name: Install `uv` run: | python -m pip install --upgrade pip - pip install . + pip install uv - - name: Run unit tests - run: pytest + - name: Create virtual environment + run: uv venv + + - name: Install dependencies with `uv` + run: uv pip install -e . + + - name: Set PYTHONPATH + run: echo "PYTHONPATH=$(pwd)/src" >> $GITHUB_ENV + + - name: Run unit tests with `uv` + run: uv run pytest diff --git a/README.md b/README.md index dab6fe8..bf46f2a 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,30 @@ This is a Python-based MCP server that integrates with the Plex Media Server API to search for movies and manage playlists. It uses the PlexAPI library for seamless interaction with your Plex server. +## Screenshots + +Here are some examples of how the Plex MCP server works: + +### 1. Find Movies in Plex Library by Director +Search for movies in your Plex library by specifying a director's name. For example, searching for "Alfred Hitchcock" returns a list of his movies in your library. + +![Find movies by director](images/plex-mcp-search-movies-hitchcock.png) + +--- + +### 2. Find Missing Movies for a Director +Identify movies by a specific director that are missing from your Plex library. This helps you discover gaps in your collection. + +![Find missing movies](images/plex-mcp-missing-movies.png) + +--- + +### 3. Create a Playlist in Your Plex Library +Create a new playlist in your Plex library using the movies found in a search. This allows you to organize your library efficiently. + +![Create a playlist](images/plex-mcp-create-playlist.png) + + ## Setup ### Prerequisites @@ -62,9 +86,9 @@ Add the following configuration to your Claude app: "command": "uv", "args": [ "--directory", - "FULL_PATH_TO_PROJECT/plex-mcp", + "FULL_PATH_TO_PROJECT", "run", - "plex-mcp.py" + "src/plex_mcp/plex_mcp.py" ], "env": { "PLEX_TOKEN": "YOUR_PLEX_TOKEN", @@ -78,18 +102,17 @@ Add the following configuration to your Claude app: ## Available Commands The Plex MCP server exposes these commands: - -| Command | Description | OpenAPI Reference | -|---------|-------------|-------------------| -| `search_movies` | Search for movies in your library by title | `/library/sections/{sectionKey}/search` | -| `get_movie_details` | Get detailed information about a specific movie | `/library/metadata/{ratingKey}` | -| `get_movie_genres` | Get the genres for a specific movie | `/library/sections/{sectionKey}/genre` | -| `list_playlists` | List all playlists on your Plex server | `/playlists` | -| `get_playlist_items` | Get the items in a specific playlist | `/playlists/{playlistID}/items` | -| `create_playlist` | Create a new playlist with specified movies | `/playlists` | -| `delete_playlist` | Delete a playlist from your Plex server | `/playlists/{playlistID}` | -| `add_to_playlist` | Add a movie to an existing playlist | `/playlists/{playlistID}/items` | -| `recent_movies` | Get recently added movies from your library | `/library/recentlyAdded` | +| Command | Description | OpenAPI Reference | +|----------------------|-----------------------------------------------------------------------------|---------------------------------------| +| `search_movies` | Search for movies in your library by various filters (e.g., title, director, genre) with support for a `limit` parameter to control the number of results. | `/library/sections/{sectionKey}/search` | +| `get_movie_details` | Get detailed information about a specific movie. | `/library/metadata/{ratingKey}` | +| `get_movie_genres` | Get the genres for a specific movie. | `/library/sections/{sectionKey}/genre` | +| `list_playlists` | List all playlists on your Plex server. | `/playlists` | +| `get_playlist_items` | Get the items in a specific playlist. | `/playlists/{playlistID}/items` | +| `create_playlist` | Create a new playlist with specified movies. | `/playlists` | +| `delete_playlist` | Delete a playlist from your Plex server. | `/playlists/{playlistID}` | +| `add_to_playlist` | Add a movie to an existing playlist. | `/playlists/{playlistID}/items` | +| `recent_movies` | Get recently added movies from your library. | `/library/recentlyAdded` | ## Running Tests diff --git a/images/plex-mcp-create-playlist.png b/images/plex-mcp-create-playlist.png new file mode 100644 index 0000000..6f8d73c Binary files /dev/null and b/images/plex-mcp-create-playlist.png differ diff --git a/images/plex-mcp-missing-movies.png b/images/plex-mcp-missing-movies.png new file mode 100644 index 0000000..2a3a7e9 Binary files /dev/null and b/images/plex-mcp-missing-movies.png differ diff --git a/images/plex-mcp-search-movies-hitchcock.png b/images/plex-mcp-search-movies-hitchcock.png new file mode 100644 index 0000000..161c18a Binary files /dev/null and b/images/plex-mcp-search-movies-hitchcock.png differ diff --git a/pytest.ini b/pytest.ini index 547d8d2..f20e12a 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,4 +1,5 @@ [pytest] +pythonpath = src asyncio_default_fixture_loop_scope = function addopts = -m "not integration" markers = diff --git a/src/plex_mcp/__init__.py b/src/plex_mcp/__init__.py new file mode 100644 index 0000000..3b7b2ab --- /dev/null +++ b/src/plex_mcp/__init__.py @@ -0,0 +1,27 @@ +from .plex_mcp import ( + search_movies, + get_movie_details, + list_playlists, + get_playlist_items, + create_playlist, + delete_playlist, + add_to_playlist, + recent_movies, + get_movie_genres, + get_plex_server, + MovieSearchParams, +) + +__all__ = [ + "search_movies", + "get_movie_details", + "list_playlists", + "get_playlist_items", + "create_playlist", + "delete_playlist", + "add_to_playlist", + "recent_movies", + "get_movie_genres", + "get_plex_server", + "MovieSearchParams", +] \ No newline at end of file diff --git a/plex_mcp.py b/src/plex_mcp/plex_mcp.py similarity index 72% rename from plex_mcp.py rename to src/plex_mcp/plex_mcp.py index bafd882..0e919ea 100644 --- a/plex_mcp.py +++ b/src/plex_mcp/plex_mcp.py @@ -9,6 +9,7 @@ # --- Import Statements --- from typing import Any, Dict, List, Optional +from dataclasses import dataclass, asdict import os import asyncio import logging @@ -108,13 +109,76 @@ def get_server(self) -> PlexServer: """ if self._server is None: try: + logger.info("Initializing PlexServer with URL: %s", self.server_url) self._server = PlexServer(self.server_url, self.token) logger.info("Successfully initialized PlexServer.") + + # Validate the connection + self._server.library.sections() # Attempt to fetch library sections + logger.info("Plex server connection validated.") + except Unauthorized as exc: + logger.error("Unauthorized: Invalid Plex token provided.") + raise Exception("Unauthorized: Invalid Plex token provided.") from exc except Exception as exc: logger.exception("Error initializing Plex server: %s", exc) raise Exception(f"Error initializing Plex server: {exc}") return self._server +# --- Data Classes --- + +@dataclass +class MovieSearchParams: + title: Optional[str] = None + year: Optional[int] = None + director: Optional[str] = None + studio: Optional[str] = None + genre: Optional[str] = None + actor: Optional[str] = None + rating: Optional[str] = None + country: Optional[str] = None + language: Optional[str] = None + watched: Optional[bool] = None # True=only watched, False=only unwatched + min_duration: Optional[int] = None # in minutes + max_duration: Optional[int] = None # in minutes + + def to_filters(self) -> Dict[str, Any]: + FIELD_MAP = { + "title": "title", + "year": "year", + "director": "director", + "studio": "studio", + "genre": "genre", + "actor": "actor", + "rating": "rating", + "country": "country", + "language": "language", + "watched": "unwatched", + "min_duration": "minDuration", + "max_duration": "maxDuration", + } + + filters: Dict[str, Any] = {"libtype": "movie"} + + for field_name, plex_arg in FIELD_MAP.items(): + value = getattr(self, field_name) + if value is None: + continue + + if field_name == "watched": + # invert for Plex 'unwatched' flag + filters["unwatched"] = not value + continue + + if field_name in ("min_duration", "max_duration"): + # convert minutes to milliseconds + filters[plex_arg] = value * 60_000 + continue + + filters[plex_arg] = value + + return filters + + # --- Global Singleton and Access Functions --- _plex_client_instance: PlexClient = None @@ -152,37 +216,74 @@ async def get_plex_server() -> PlexServer: # --- Tool Methods --- @mcp.tool() -async def search_movies(query: str) -> str: +async def search_movies( + title: Optional[str] = None, + year: Optional[int] = None, + director: Optional[str] = None, + studio: Optional[str] = None, + genre: Optional[str] = None, + actor: Optional[str] = None, + rating: Optional[str] = None, + country: Optional[str] = None, + language: Optional[str] = None, + watched: Optional[bool] = None, + min_duration: Optional[int] = None, + max_duration: Optional[int] = None, + limit: Optional[int] = 5, +) -> str: """ - Search for movies in the Plex library. + Search for movies in your Plex library using optional filters. Parameters: - query: The search term to look up movies. + title: Optional title or substring to match. + year: Optional release year to filter by. + director: Optional director name to filter by. + studio: Optional studio name to filter by. + genre: Optional genre tag to filter by. + actor: Optional actor name to filter by. + rating: Optional rating (e.g., "PG-13") to filter by. + country: Optional country of origin to filter by. + language: Optional audio or subtitle language to filter by. + watched: Optional boolean; True returns only watched movies, False only unwatched. + min_duration: Optional minimum duration in minutes. + max_duration: Optional maximum duration in minutes. Returns: - A formatted string of search results or an error message. + A formatted string of up to 5 matching movies (with a count of any additional results), + or an error message if the search fails or no movies are found. """ - if query is None: - return "ERROR: No query provided. Please provide a search term." + + # Validate the limit parameter + limit = max(1, limit) if limit else 5 # Default to 5 if limit is 0 or negative + + params = MovieSearchParams( + title, year, director, studio, + genre, actor, rating, country, + language, watched, min_duration, max_duration + ) + filters = params.to_filters() + logger.info("Searching Plex with filters: %r", filters) try: plex = await get_plex_server() + movies = await asyncio.to_thread(plex.library.search, **filters) except Exception as e: - return f"ERROR: Could not connect to Plex server. {str(e)}" + logger.exception("search_movies failed connecting to Plex") + return f"ERROR: Could not search Plex. {e}" + + if not movies: + return f"No movies found matching filters {filters!r}." + + logger.info("Found %d movies matching filters: %r", len(movies), filters) - try: - movies = await asyncio.to_thread(plex.library.search, title=query, libtype="movie") - if not movies: - return f"No movies found matching '{query}'." - formatted_results = [] - for i, movie in enumerate(movies[:5], 1): # Limit to 5 results - formatted_results.append(f"Result #{i}:\nKey: {movie.ratingKey}\n{format_movie(movie)}") - if len(movies) > 5: - formatted_results.append(f"\n... and {len(movies) - 5} more results.") - return "\n---\n".join(formatted_results) - except Exception as e: - logger.exception("Failed to search movies with query '%s'", query) - return f"ERROR: Failed to search movies. {str(e)}" + results: List[str] = [] + for i, m in enumerate(movies[:limit], start=1): + results.append(f"Result #{i}:\nKey: {m.ratingKey}\n{format_movie(m)}") + + if len(movies) > limit: + results.append(f"\n... and {len(movies)-limit} more results.") + + return "\n---\n".join(results) @mcp.tool() async def get_movie_details(movie_key: str) -> str: @@ -202,23 +303,10 @@ async def get_movie_details(movie_key: str) -> str: try: key = int(movie_key) - sections = await asyncio.to_thread(plex.library.sections) - movie = None - for section in sections: - if section.type == 'movie': - try: - items = await asyncio.to_thread(lambda s=section, k=key: s.search(filters={"ratingKey": k})) - if items: - movie = items[0] - break - except Exception: - continue - if not movie: - all_movies = await asyncio.to_thread(lambda: plex.library.search(libtype="movie")) - for m in all_movies: - if m.ratingKey == key: - movie = m - break + + all_movies = await asyncio.to_thread(lambda: plex.library.search(libtype="movie")) + movie = next((m for m in all_movies if m.ratingKey == key), None) + if not movie: return f"No movie found with key {movie_key}." return format_movie(movie) @@ -407,36 +495,26 @@ async def add_to_playlist(playlist_key: str, movie_key: str) -> str: try: p_key = int(playlist_key) m_key = int(movie_key) + + # Find the playlist all_playlists = await asyncio.to_thread(plex.playlists) playlist = next((p for p in all_playlists if p.ratingKey == p_key), None) if not playlist: return f"No playlist found with key {playlist_key}." - sections = await asyncio.to_thread(plex.library.sections) - movie_sections = [section for section in sections if section.type == 'movie'] - movie = None - for section in movie_sections: - try: - items = await asyncio.to_thread(lambda s=section, k=m_key: s.search(filters={"ratingKey": k})) - if items: - movie = items[0] - break - except Exception: - continue - if not movie: - all_movies = await asyncio.to_thread(lambda: plex.library.search(libtype="movie")) - for m in all_movies: - if m.ratingKey == m_key: - movie = m - break - if not movie: + # Perform a global search for the movie + movies = await asyncio.to_thread(lambda: plex.library.search(libtype="movie", ratingKey=m_key)) + if not movies: return f"No movie found with key {movie_key}." + movie = movies[0] # Since the search is scoped to the ratingKey, there should be at most one result + + # Add the movie to the playlist await asyncio.to_thread(lambda p=playlist, m=movie: p.addItems([m])) logger.info("Added movie '%s' to playlist '%s'", movie.title, playlist.title) return f"Successfully added '{movie.title}' to playlist '{playlist.title}'." - except NotFound as e: - return f"ERROR: Item not found. {str(e)}" + except ValueError: + return "ERROR: Invalid playlist or movie key. Please provide valid numbers." except Exception as e: logger.exception("Failed to add movie to playlist") return f"ERROR: Failed to add movie to playlist. {str(e)}" @@ -452,23 +530,22 @@ async def recent_movies(count: int = 5) -> str: Returns: A formatted string of recent movies or an error message. """ + if count <= 0: + return "ERROR: Count must be a positive integer." + try: plex = await get_plex_server() except Exception as e: return f"ERROR: Could not connect to Plex server. {str(e)}" try: - movie_sections = [section for section in plex.library.sections() if section.type == 'movie'] - if not movie_sections: - return "No movie libraries found in your Plex server." - all_recent = [] - for section in movie_sections: - recent = await asyncio.to_thread(section.recentlyAdded, maxresults=count) - all_recent.extend(recent) - all_recent.sort(key=lambda x: x.addedAt, reverse=True) + # Perform a global search for recently added movies + all_recent = await asyncio.to_thread(lambda: plex.library.search(libtype="movie", sort="addedAt:desc")) recent_movies_list = all_recent[:count] + if not recent_movies_list: return "No recent movies found in your Plex library." + formatted_movies = [] for i, movie in enumerate(recent_movies_list, 1): formatted_movies.append( @@ -497,30 +574,20 @@ async def get_movie_genres(movie_key: str) -> str: try: key = int(movie_key) - sections = await asyncio.to_thread(plex.library.sections) - movie = None - for section in sections: - try: - items = await asyncio.to_thread(lambda s=section, k=key: s.search(filters={"ratingKey": k})) - if items: - movie = items[0] - break - except Exception: - continue - if not movie: - all_movies = await asyncio.to_thread(lambda: plex.library.search(libtype="movie")) - for m in all_movies: - if m.ratingKey == key: - movie = m - break + + # Perform a global search for the movie + all_movies = await asyncio.to_thread(lambda: plex.library.search(libtype="movie")) + movie = next((m for m in all_movies if m.ratingKey == key), None) if not movie: return f"No movie found with key {movie_key}." + + # Extract genres genres = [genre.tag for genre in movie.genres] if hasattr(movie, 'genres') else [] if not genres: return f"No genres found for movie '{movie.title}'." return f"Genres for '{movie.title}':\n{', '.join(genres)}" - except NotFound: - return f"ERROR: Movie with key {movie_key} not found." + except ValueError: + return f"ERROR: Invalid movie key '{movie_key}'. Please provide a valid number." except Exception as e: logger.exception("Failed to fetch genres for movie with key '%s'", movie_key) return f"ERROR: Failed to fetch movie genres. {str(e)}" diff --git a/tests/test_app.py b/tests/test_app.py index 6401363..704b158 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -9,6 +9,7 @@ # --- Import the Module Under Test --- from plex_mcp import ( + MovieSearchParams, search_movies, get_movie_details, list_playlists, @@ -18,7 +19,6 @@ add_to_playlist, recent_movies, get_movie_genres, - get_plex_server, ) # --- Set Dummy Environment Variables --- @@ -29,28 +29,39 @@ def set_dummy_env(monkeypatch): # --- Dummy Classes to Simulate Plex Objects --- +class DummyTag: + def __init__(self, tag): + self.tag = tag + + class DummyMovie: - def __init__(self, ratingKey, title, year=2022, summary="A test movie", - duration=7200000, rating="PG", studio="Test Studio"): - self.ratingKey = ratingKey + def __init__( + self, + rating_key, + title, + year=2022, + duration=120 * 60_000, # in ms + studio="Test Studio", + summary="A test summary", + rating="PG", + directors=None, + roles=None, + genres=None, + type_="movie", + addedAt=None, # New parameter + ): + self.ratingKey = rating_key self.title = title self.year = year - self.summary = summary self.duration = duration - self.rating = rating self.studio = studio - self.directors = [] - self.roles = [] - # Add 'type' attribute required by get_playlist_items - self.type = "movie" - - @property - def addedAt(self): - return getattr(self, "_addedAt", datetime(2022, 1, 1)) - - @addedAt.setter - def addedAt(self, value): - self._addedAt = value + self.summary = summary + self.rating = rating + self.directors = [DummyTag(d) for d in (directors or [])] + self.roles = [DummyTag(r) for r in (roles or [])] + self.genres = [DummyTag(g) for g in (genres or [])] + self.type = type_ + self.addedAt = addedAt # New attribute # Subclass for movies with genres. class DummyMovieWithGenres(DummyMovie): @@ -81,8 +92,11 @@ def __init__(self, movies=None): self._movies = movies if movies is not None else [] def search(self, **kwargs): + title = kwargs.get("title") + if isinstance(title, MovieSearchParams): + title = title.title # Unwrap if passed improperly if kwargs.get("libtype") == "movie": - return self._movies + return [m for m in self._movies if title is None or title.lower() in m.title.lower()] return [] def sections(self): @@ -129,15 +143,21 @@ def patch_get_plex_server(monkeypatch): """Fixture to patch the get_plex_server function with a dummy Plex server.""" def _patch(movies=None, playlists=None): monkeypatch.setattr( - "plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServer(movies, playlists)) + "plex_mcp.plex_mcp.get_plex_server", + lambda: dummy_get_plex_server(movies, playlists) ) return _patch @pytest.fixture def dummy_movie(): - """Fixture that returns a default DummyMovie instance.""" - return DummyMovie(1, "Test Movie") + return DummyMovie( + rating_key=1, + title="Test Movie", + year=2022, + directors=["Jane Doe"], + roles=["Test Actor"], + genres=["Thriller"] + ) # --- Tests for search_movies --- @@ -145,7 +165,7 @@ def dummy_movie(): async def test_search_movies_found(patch_get_plex_server, dummy_movie): """Test that search_movies returns a formatted result when a movie is found.""" patch_get_plex_server([dummy_movie]) - result = await search_movies("Test") + result = await search_movies(MovieSearchParams(title="Test")) assert "Test Movie" in result assert "more results" not in result @@ -154,7 +174,7 @@ async def test_search_movies_multiple_results(patch_get_plex_server): """Test that search_movies shows an extra results message when more than 5 movies are found.""" movies = [DummyMovie(i, f"Test Movie {i}") for i in range(1, 8)] patch_get_plex_server(movies) - result = await search_movies("Test") + result = await search_movies(MovieSearchParams(title="Test")) for i in range(1, 6): assert f"Test Movie {i}" in result assert "and 2 more results" in result @@ -163,49 +183,112 @@ async def test_search_movies_multiple_results(patch_get_plex_server): async def test_search_movies_not_found(monkeypatch, patch_get_plex_server): """Test that search_movies returns a 'not found' message when no movies match the query.""" patch_get_plex_server([]) - # Force DummySection.search to return an empty list. monkeypatch.setattr(DummySection, "search", lambda self, filters: []) - result = await search_movies("NonExisting") + result = await search_movies(MovieSearchParams(title="NonExisting")) assert "No movies found" in result @pytest.mark.asyncio async def test_search_movies_exception(monkeypatch): """Test that search_movies returns an error message when an exception occurs.""" + # Mock the Plex library to raise an exception during search dummy_server = DummyPlexServer([DummyMovie(1, "Test Movie")]) dummy_server.library.search = MagicMock(side_effect=Exception("Search error")) - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=dummy_server)) - result = await search_movies("Test") - assert "ERROR: Failed to search movies" in result + + # Patch get_plex_server to return the dummy server + async def mock_get_plex_server(): + return dummy_server + + monkeypatch.setattr("plex_mcp.plex_mcp.get_plex_server", mock_get_plex_server) + + # Call the function under test + result = await search_movies(MovieSearchParams(title="Test")) + + # Assert the error message is returned + assert "ERROR: Could not search Plex" in result assert "Search error" in result @pytest.mark.asyncio async def test_search_movies_empty_string(patch_get_plex_server): """Test search_movies with an empty string returns the not-found message.""" patch_get_plex_server([]) - result = await search_movies("") - assert result == "No movies found matching ''." + result = await search_movies(MovieSearchParams(title="")) + assert result.startswith("No movies found") @pytest.mark.asyncio async def test_search_movies_none_input(patch_get_plex_server, dummy_movie): - """Test that search_movies returns an error when None is provided as input.""" + """Test that search_movies with None input returns results (treated as unfiltered search).""" patch_get_plex_server([dummy_movie]) - try: - result = await search_movies(None) - except Exception as e: - result = str(e) - assert "ERROR" in result + result = await search_movies(None) + assert "Test Movie" in result @pytest.mark.asyncio async def test_search_movies_large_dataset(patch_get_plex_server): """Test that search_movies correctly handles a large dataset of movies.""" movies = [DummyMovie(i, f"Test Movie {i}") for i in range(1, 201)] patch_get_plex_server(movies) - result = await search_movies("Test") + result = await search_movies(MovieSearchParams(title="Test")) for i in range(1, 6): assert f"Test Movie {i}" in result assert "and 195 more results" in result +@pytest.mark.asyncio +async def test_search_movies_with_default_limit(patch_get_plex_server): + """Test that search_movies respects the default limit of 5 results.""" + movies = [DummyMovie(i, f"Test Movie {i}") for i in range(1, 11)] + patch_get_plex_server(movies) + + result = await search_movies(MovieSearchParams(title="Test")) + assert "Result #1" in result + assert "Result #5" in result + assert "... and 5 more results." in result + assert "Result #6" not in result # Ensure only 5 results are shown + +@pytest.mark.asyncio +async def test_search_movies_with_custom_limit(patch_get_plex_server): + """Test that search_movies respects a custom limit parameter.""" + # Mock the Plex library search to return 10 dummy movies + movies = [DummyMovie(i, f"Test Movie {i}") for i in range(1, 11)] + patch_get_plex_server(movies) + + result = await search_movies(MovieSearchParams(title="Test"), limit=8) + assert "Result #1" in result + assert "Result #8" in result + assert "... and 2 more results." in result + assert "Result #9" not in result # Ensure only 8 results are shown + +@pytest.mark.asyncio +async def test_search_movies_with_limit_exceeding_results(patch_get_plex_server): + """Test that search_movies handles a limit larger than the number of results.""" + movies = [DummyMovie(i, f"Test Movie {i}") for i in range(1, 4)] + patch_get_plex_server(movies) + + result = await search_movies(MovieSearchParams(title="Test"), limit=10) + assert "Result #1" in result + assert "Result #3" in result + assert "... and" not in result # Ensure no "and more results" message is shown + assert "Result #4" not in result # Ensure no extra results are shown + +@pytest.mark.asyncio +async def test_search_movies_with_invalid_limit(patch_get_plex_server): + """Test that search_movies handles an invalid limit (e.g., 0 or negative).""" + # Mock the Plex library search to return 10 dummy movies + movies = [DummyMovie(i, f"Test Movie {i}") for i in range(1, 11)] + patch_get_plex_server(movies) + + result = await search_movies(MovieSearchParams(title="Test"), limit=0) + assert "Result #1" in result + assert "Result #5" in result + assert "... and 5 more results." in result + assert "Result #6" not in result # Ensure only 5 results are shown (default behavior) + +@pytest.mark.asyncio +async def test_search_movies_no_results(patch_get_plex_server): + """Test that search_movies returns an appropriate message when no results are found.""" + patch_get_plex_server([]) + + result = await search_movies(MovieSearchParams(title="Nonexistent")) + assert "No movies found" in result + # --- Tests for get_movie_details --- @pytest.mark.asyncio @@ -224,35 +307,29 @@ async def test_get_movie_details_invalid_key(patch_get_plex_server, dummy_movie) assert "ERROR" in result @pytest.mark.asyncio -async def test_get_movie_details_not_found(monkeypatch, patch_get_plex_server): +async def test_get_movie_details_not_found(patch_get_plex_server): """Test that get_movie_details returns a 'not found' message when the movie is missing.""" patch_get_plex_server([]) - monkeypatch.setattr(DummySection, "search", lambda self, filters: []) + result = await get_movie_details("1") assert "No movie found with key 1" in result # --- Tests for list_playlists --- @pytest.mark.asyncio -async def test_list_playlists_empty(monkeypatch): +async def test_list_playlists_empty(patch_get_plex_server): """Test that list_playlists returns a message when there are no playlists.""" - class DummyPlexServerNoPlaylists(DummyPlexServer): - def playlists(self): - return [] - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServerNoPlaylists())) + patch_get_plex_server(playlists=[]) + result = await list_playlists() assert "No playlists found" in result @pytest.mark.asyncio -async def test_list_playlists_found(monkeypatch, dummy_movie): +async def test_list_playlists_found(patch_get_plex_server, dummy_movie): """Test that list_playlists returns a formatted list when playlists exist.""" dummy_playlist = DummyPlaylist(1, "My Playlist", [dummy_movie]) - class DummyPlexServerWithPlaylists(DummyPlexServer): - def playlists(self): - return [dummy_playlist] - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServerWithPlaylists())) + patch_get_plex_server(playlists=[dummy_playlist]) + result = await list_playlists() assert "My Playlist" in result assert "Playlist #1" in result @@ -260,126 +337,95 @@ def playlists(self): # --- Tests for get_playlist_items --- @pytest.mark.asyncio -async def test_get_playlist_items_found(monkeypatch, dummy_movie): +async def test_get_playlist_items_found(patch_get_plex_server, dummy_movie): """Test that get_playlist_items returns the items of a found playlist.""" dummy_playlist = DummyPlaylist(2, "My Playlist", [dummy_movie]) - class DummyPlexServerWithPlaylists(DummyPlexServer): - def playlists(self): - return [dummy_playlist] - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServerWithPlaylists())) + patch_get_plex_server(playlists=[dummy_playlist]) + result = await get_playlist_items("2") assert "Test Movie" in result @pytest.mark.asyncio -async def test_get_playlist_items_not_found(monkeypatch): +async def test_get_playlist_items_not_found(patch_get_plex_server): """Test that get_playlist_items returns an error when the playlist is not found.""" - class DummyPlexServerNoPlaylists(DummyPlexServer): - def playlists(self): - return [] - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServerNoPlaylists())) + patch_get_plex_server(playlists=[]) + result = await get_playlist_items("99") assert "No playlist found with key 99" in result # --- Tests for create_playlist --- @pytest.mark.asyncio -async def test_create_playlist_success(monkeypatch, dummy_movie): +async def test_create_playlist_success(patch_get_plex_server, dummy_movie): """Test that create_playlist returns a success message on valid input.""" - class DummyPlexServerWithCreate(DummyPlexServer): - def createPlaylist(self, name, items): - return DummyPlaylist(1, name, items) - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServerWithCreate([dummy_movie]))) + patch_get_plex_server([dummy_movie]) + result = await create_playlist("My Playlist", "1") assert "Successfully created playlist 'My Playlist'" in result @pytest.mark.asyncio -async def test_create_playlist_no_valid_movies(monkeypatch): +async def test_create_playlist_no_valid_movies(patch_get_plex_server): """Test that create_playlist returns an error when no valid movies are provided.""" - class DummyPlexServerWithSearch(DummyPlexServer): - def createPlaylist(self, name, items): - return DummyPlaylist(1, name, items) - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServerWithSearch([]))) + patch_get_plex_server([]) + result = await create_playlist("My Playlist", "1,2") assert "ERROR:" in result # --- Tests for delete_playlist --- @pytest.mark.asyncio -async def test_delete_playlist_success(monkeypatch, dummy_movie): +async def test_delete_playlist_success(patch_get_plex_server, dummy_movie): """Test that delete_playlist returns a success message when deletion is successful.""" dummy_playlist = DummyPlaylist(3, "Delete Me", [dummy_movie]) - class DummyPlexServerWithPlaylist(DummyPlexServer): - def playlists(self): - return [dummy_playlist] - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServerWithPlaylist())) + patch_get_plex_server(playlists=[dummy_playlist]) + result = await delete_playlist("3") assert "Successfully deleted playlist" in result @pytest.mark.asyncio -async def test_delete_playlist_not_found(monkeypatch): +async def test_delete_playlist_not_found(patch_get_plex_server): """Test that delete_playlist returns an error when no matching playlist is found.""" - class DummyPlexServerNoPlaylists(DummyPlexServer): - def playlists(self): - return [] - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServerNoPlaylists())) + patch_get_plex_server(playlists=[]) + result = await delete_playlist("99") assert "No playlist found with key 99" in result # --- Tests for add_to_playlist --- @pytest.mark.asyncio -async def test_add_to_playlist_success(monkeypatch): +async def test_add_to_playlist_success(patch_get_plex_server): """Test that add_to_playlist returns a success message when a movie is added.""" dummy_playlist = DummyPlaylist(4, "My Playlist", []) - class DummyPlexServerWithPlaylist(DummyPlexServer): - def playlists(self): - return [dummy_playlist] - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServerWithPlaylist())) - # Override DummySection.search so that it returns a dummy movie with key 5. - monkeypatch.setattr(DummySection, "search", - lambda self, filters: [DummyMovie(5, "Added Movie")] if filters.get("ratingKey") == 5 else []) + dummy_movie = DummyMovie(5, "Added Movie") + patch_get_plex_server([dummy_movie], playlists=[dummy_playlist]) + result = await add_to_playlist("4", "5") assert "Successfully added 'Added Movie' to playlist" in result @pytest.mark.asyncio -async def test_add_to_playlist_playlist_not_found(monkeypatch): +async def test_add_to_playlist_playlist_not_found(patch_get_plex_server): """Test that add_to_playlist returns an error when the specified playlist is not found.""" - class DummyPlexServerNoPlaylist(DummyPlexServer): - def playlists(self): - return [] - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServerNoPlaylist())) + patch_get_plex_server(playlists=[]) + result = await add_to_playlist("999", "5") assert "No playlist found with key 999" in result # --- Tests for recent_movies --- @pytest.mark.asyncio -async def test_recent_movies_found(monkeypatch): +async def test_recent_movies_found(patch_get_plex_server): """Test that recent_movies returns recent movie information when available.""" - class DummySectionWithRecent(DummySection): - def recentlyAdded(self, maxresults): - m = DummyMovie(1, "Recent Movie") - m.addedAt = datetime(2022, 5, 1) - return [m] - monkeypatch.setattr(DummyLibrary, "sections", lambda self: [DummySectionWithRecent("movie")]) - monkeypatch.setattr("plex_mcp.get_plex_server", - lambda: asyncio.sleep(0, result=DummyPlexServer())) + recent_movie = DummyMovie(1, "Recent Movie", addedAt=datetime(2022, 5, 1)) + patch_get_plex_server([recent_movie]) + result = await recent_movies(5) assert "Recent Movie" in result @pytest.mark.asyncio -async def test_recent_movies_not_found(monkeypatch, patch_get_plex_server): +async def test_recent_movies_not_found(patch_get_plex_server): """Test that recent_movies returns an error message when no recent movies are found.""" - patch_get_plex_server() - monkeypatch.setattr(DummySection, "recentlyAdded", lambda self, maxresults: []) + patch_get_plex_server([]) + result = await recent_movies(5) assert "No recent movies found" in result @@ -388,18 +434,21 @@ async def test_recent_movies_not_found(monkeypatch, patch_get_plex_server): @pytest.mark.asyncio async def test_get_movie_genres_found(monkeypatch, patch_get_plex_server): """Test that get_movie_genres returns the correct genres for a movie.""" - class DummyMovieWithGenres(DummyMovie): - def __init__(self, ratingKey, title, genres, **kwargs): - super().__init__(ratingKey, title, **kwargs) - self.genres = genres - class DummyGenre: - def __init__(self, tag): - self.tag = tag - # Patch DummySection.search to return a movie with genres. - monkeypatch.setattr(DummySection, "search", - lambda self, filters: [DummyMovieWithGenres(1, "Test Movie", [DummyGenre("Action"), DummyGenre("Thriller")])] - if filters.get("ratingKey") == 1 else []) - patch_get_plex_server([DummyMovieWithGenres(1, "Test Movie", [DummyGenre("Action"), DummyGenre("Thriller")])]) + # Create a dummy movie with genre tags + movie_with_genres = DummyMovie( + rating_key=1, + title="Test Movie", + genres=["Action", "Thriller"] + ) + + # Patch DummySection.search to return our dummy movie when the ratingKey matches + monkeypatch.setattr( + DummySection, + "search", + lambda self, filters: [movie_with_genres] if filters.get("ratingKey") == 1 else [] + ) + + patch_get_plex_server([movie_with_genres]) result = await get_movie_genres("1") assert "Action" in result assert "Thriller" in result diff --git a/tests/test_integration.py b/tests/test_integration.py index 6f1bb14..27586db 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -27,17 +27,74 @@ async def test_integration_get_plex_server(): plex = await get_plex_server() assert plex is not None +# tests/test_integration.py + +import os +import asyncio +import pytest +from dotenv import load_dotenv + +# Load environment variables from .env +load_dotenv() + +from plex_mcp import search_movies + +pytestmark = pytest.mark.integration + @pytest.mark.asyncio -async def test_integration_search_movies(): +async def test_integration_search_movies_title(): """ - Integration test: Search for movies on the real Plex server. - This test uses a sample query and asserts that the result is a non-empty string. - Adjust the query as needed to match expected results. + Search by title keyword. """ result = await search_movies("Avengers") - # Expect a response that either contains movies matching the search or a 'no movies found' message. assert isinstance(result, str) - assert ("Avengers" in result) or ("No movies found" in result) + assert ("Result #" in result) or ("No movies found" in result) + +@pytest.mark.asyncio +async def test_integration_search_movies_by_year(): + """ + Search by release year. + """ + # find all movies from 1999 + result = await search_movies(year=1999) + assert isinstance(result, str) + assert ("Result #" in result) or ("No movies found" in result) + +@pytest.mark.asyncio +async def test_integration_search_movies_by_director(): + """ + Search by director name. + """ + result = await search_movies(director="Christopher Nolan") + assert isinstance(result, str) + assert ("Result #" in result) or ("No movies found" in result) + +@pytest.mark.asyncio +async def test_integration_search_movies_unwatched(): + """ + Search for only unwatched movies. + """ + result = await search_movies(watched=False) + assert isinstance(result, str) + assert ("Result #" in result) or ("No movies found" in result) + +@pytest.mark.asyncio +async def test_integration_search_movies_min_duration(): + """ + Search for movies at least 120 minutes long. + """ + result = await search_movies(min_duration=120) + assert isinstance(result, str) + assert ("Result #" in result) or ("No movies found" in result) + +@pytest.mark.asyncio +async def test_integration_search_movies_multiple_filters(): + """ + Search combining year, genre, and minimum duration. + """ + result = await search_movies(year=2020, genre="Drama", min_duration=100) + assert isinstance(result, str) + assert ("Result #" in result) or ("No movies found" in result) @pytest.mark.asyncio async def test_integration_list_playlists():