From 3c87b4955459042958b14618edcd2fb5f8c14424 Mon Sep 17 00:00:00 2001 From: Cliff Hill Date: Fri, 1 Oct 2021 15:37:17 -0400 Subject: [PATCH] server.py tests completed. Signed-off-by: Cliff Hill --- src/playlist/data/settings.py | 13 +-- src/playlist/plex/server.py | 55 ++++++------ tests/plex/test_server.py | 153 +++++++++++++++++++++++++++++++++- 3 files changed, 180 insertions(+), 41 deletions(-) diff --git a/src/playlist/data/settings.py b/src/playlist/data/settings.py index 0f15284..77ba1c8 100644 --- a/src/playlist/data/settings.py +++ b/src/playlist/data/settings.py @@ -6,7 +6,6 @@ import contextlib import dataclasses import datetime import functools -import getpass import pathlib import yaml @@ -28,11 +27,7 @@ class CredentialSettings(base.BaseData["CredentialSettings"]): def create(cls: type[CredentialSettings]) -> CredentialSettings: # pragma: no cover from playlist.plex import server - username = input("Plex Username: ") - password = getpass.getpass("Plex Password: ") - servername = input("Plex Server: ") - - data = server.get_creds(username, password, servername) + data = server.get_creds() return cls.load(data) @@ -44,7 +39,7 @@ class DownloadSettings(base.BaseData["DownloadSettings"]): process_delay: float @classmethod - def create(cls: type[DownloadSettings]) -> DownloadSettings: # pragma: no cover + def create(cls: type[DownloadSettings]) -> DownloadSettings: return cls( batch_size=const.DEFAULTS.BATCH_SIZE, process_delay=const.DEFAULTS.PROCESS_DELAY, @@ -58,7 +53,7 @@ class TrackSettings(base.BaseData["TrackSettings"]): duration: float @classmethod - def create(cls: type[TrackSettings]) -> TrackSettings: # pragma: no cover + def create(cls: type[TrackSettings]) -> TrackSettings: return cls(duration=const.DEFAULTS.DURATION) @@ -70,7 +65,7 @@ class PlaylistSettings(base.BaseData["PlaylistSettings"]): max_tracks: int @classmethod - def create(cls: type[PlaylistSettings]) -> PlaylistSettings: # pragma: no cover + def create(cls: type[PlaylistSettings]) -> PlaylistSettings: return cls( playtime=const.DEFAULTS.PLAYTIME, max_tracks=const.DEFAULTS.MAX_TRACKS, diff --git a/src/playlist/plex/server.py b/src/playlist/plex/server.py index 5426188..63c02f2 100644 --- a/src/playlist/plex/server.py +++ b/src/playlist/plex/server.py @@ -1,6 +1,7 @@ """Contains the code to communicate to the Plex server.""" import asyncio import collections.abc +import getpass import statistics import time import typing @@ -57,15 +58,14 @@ async def gen_tracks( Yields: models.Track: The Track that was pulled from Plex. """ - quotient, remainder = divmod(await total_track_count(), batch_size) loop = asyncio.get_running_loop() + times = [] + weights = [] + durations = [] batches = [ asyncio.create_task(_downloader(ndx, size, loop)) async for ndx, size in _gen_batch_params(batch_size) ] - times = [] - weights = [] - durations = [] for batch_task in asyncio.as_completed(batches): process_time, batch = await batch_task weight = len(batch) / batch_size @@ -100,24 +100,19 @@ async def total_track_count() -> int: return num_tracks -def get_creds( - username: str, - password: str, - server: str, -) -> settings.CredentialSettings.Dict: # type: ignore [name-defined] +def get_creds() -> settings.CredentialSettings.Dict: # type: ignore [name-defined] """Get the credentials to store in the Settings instance. - Args: - username: The username to log in as. - password: The password to log in with. - server: The server alias in Plex to use to connect to. - Returns: settings.CredentialSettings.Dict: The dictionary containing the baseurl and token string to connect with. """ + username = input("Plex Username: ") + password = getpass.getpass("Plex Password: ") + server = input("Plex Server: ") + account = plexapi.myplex.MyPlexAccount(username, password) - plex = account.reource(server).connect() + plex = account.resource(server).connect() return {"baseurl": plex._baseurl, "token": plex._token} @@ -182,21 +177,21 @@ def _track_dump(track: plexapi.audio.Track) -> models.Track.Dict: # type: ignor Returns: models.Track.Dict: The dictionary form for the Track object converted to. """ - return { - "plex_id": track.ratingKey, - "track_num": track.index, - "title": track.title, - "artist": track.artist().title, - "album_num": track.parentIndex, - "album": track.parentTitle, - "album_artist": track.grandparentTitle, - "duration": round(track.duration / 1000, ndigits=6), - "rating": track.userRating, - "comments": track.summary, - "added": str(track.addedAt), - "play_count": track.viewCount, - "played": str(track.lastViewedAt) if track.lastViewedAt is not None else None, - } + return models.Track.Dict( + plex_id=track.ratingKey, + track_num=track.index, + title=track.title, + artist=track.artist().title, + album_num=track.parentIndex, + album=track.parentTitle, + album_artist=track.grandparentTitle, + duration=round(track.duration / 1000, ndigits=6), + rating=track.userRating, + comments=track.summary, + added=str(track.addedAt), + play_count=track.viewCount, + played=str(track.lastViewedAt) if track.lastViewedAt is not None else None, + ) async def _gen_batch_params( diff --git a/tests/plex/test_server.py b/tests/plex/test_server.py index d7d5c9a..0fd3f3f 100644 --- a/tests/plex/test_server.py +++ b/tests/plex/test_server.py @@ -1,6 +1,10 @@ """Tests for the playlist.plex.server module.""" +import asyncio +import unittest.mock + import pytest # type: ignore [import] +from playlist.data import settings # type: ignore [import] from playlist.plex import server # type: ignore [import] @@ -8,12 +12,16 @@ def test_calc_delay(mocker: pytest.fixture) -> None: """Test the calc_delay function.""" mock_settings = mocker.patch("playlist.plex.server.settings") mock_const = mocker.patch("playlist.plex.server.const") + mock_s = mock_settings.modify.return_value.__enter__.return_value + mock_const.MAX_PROCESSES = 1 + times = [100, 100, 100] weights = [100, 100, 100] - mock_const.MAX_PROCESSES = 1 + server.calc_delay(times, weights) - mock_s = mock_settings.modify.return_value.__enter__.return_value + result = mock_s.download.process_delay + assert result == 100 @@ -26,3 +34,144 @@ def test_calc_duration(mocker: pytest.fixture) -> None: server.calc_duration(durations) result = mock_s.playlist.max_tracks assert result == 2 + + +@pytest.mark.asyncio # type: ignore [misc] +async def test_gen_tracks(mocker: pytest.fixture) -> None: + """Test the gen_tracks asynchronous generator.""" + mock_downloader = mocker.patch( + "playlist.plex.server._downloader", + new_callable=unittest.mock.AsyncMock, + ) + mock_gen_batch_params = mocker.patch("playlist.plex.server._gen_batch_params") + mock_models = mocker.patch("playlist.plex.server.models") + mock_calc_delay = mocker.patch("playlist.plex.server.calc_delay") + mock_calc_duration = mocker.patch("playlist.plex.server.calc_duration") + mock_downloader.return_value = (1, ["Not a real track"]) + + async def fake_gen_batch_params(*args): # type: ignore + for item in [(0, 1), (1, 1)]: + yield item + + mock_gen_batch_params.side_effect = fake_gen_batch_params + mock_models.Track.load.return_value.duration = 1 + + results = [item async for item in server.gen_tracks(batch_size=1)] + + assert len(results) == 2 + assert mock_calc_delay.called + assert mock_calc_duration.called + + +@pytest.mark.asyncio # type: ignore [misc] +async def test_total_track_count(mocker: pytest.fixture) -> None: + """Test the total_track_count coroutine.""" + mock_plexapi_server = mocker.patch("playlist.plex.server.plexapi.server") + mock_server = mock_plexapi_server.PlexServer.return_value + mock_total_view_size = mock_server.library.section.return_value.totalViewSize + mock_total_view_size.return_value = 1 + + result = await server.total_track_count() + + assert result == 1 + + +def test_get_creds(mocker: pytest.fixture) -> None: + """Test the get_creds function.""" + mock_plexapi_myplex = mocker.patch("playlist.plex.server.plexapi.myplex") + mock_input = mocker.patch("builtins.input") + mock_getpass = mocker.patch("playlist.plex.server.getpass") + mock_account = mock_plexapi_myplex.MyPlexAccount.return_value + mock_plex = mock_account.resource.return_value.connect.return_value + mock_plex._baseurl = "Not a valid URL" + mock_plex._token = "Fake token" + + expected = settings.CredentialSettings.Dict( + baseurl=mock_plex._baseurl, + token=mock_plex._token, + ) + + result = server.get_creds() + + assert result == expected + assert mock_input.called + assert mock_getpass.getpass.called + + +@pytest.mark.asyncio # type: ignore [misc] +async def test_downloader(mocker: pytest.fixture) -> None: + """Test the _downloader coroutine.""" + + async def fake_run_in_executor(*args): # type: ignore + await asyncio.sleep(0.1) + return "Fake batch" + + mock_loop = unittest.mock.MagicMock() + mock_loop.run_in_executor.side_effect = fake_run_in_executor + + result = await server._downloader(1, 1, mock_loop) + + process_time, batch = result + + assert batch + assert process_time >= 0.1 + + +def test_get_track_batch(mocker: pytest.fixture) -> None: + """Test _get_track_batch function.""" + mock_plexapi_server = mocker.patch("playlist.plex.server.plexapi.server") + mock_server = mock_plexapi_server.PlexServer.return_value + mock_search_tracks = mock_server.library.section.return_value.searchTracks + mock_search_tracks.return_value = ["Not a track."] + mock_track_dump = mocker.patch("playlist.plex.server._track_dump") + mock_track_dump.return_value = "Also not a track." + + result = server._get_track_batch(1, 1) + + assert len(result) + assert mock_track_dump.called + + +def test_track_dump(mocker: pytest.fixture) -> None: + """Test the _track_dump function.""" + mock_track = unittest.mock.MagicMock() + mock_track.configure_mock( + ratingKey=1, + index=1, + title="Nothing", + parentIndex=1, + parentTile="Nothing", + grandparentTitle="Nothing", + duration=1000, + userRating=1, + summary="Whatever", + addedAt="Some time", + viewCount=1, + lastViewedAt=None, + ) + mock_track.artist.return_value.title = "Nothing" + + result = server._track_dump(mock_track) + + assert isinstance(result, dict) + + +@pytest.mark.asyncio # type: ignore [misc] +async def test_gen_batch_params(mocker: pytest.fixture) -> None: + """Test the _gen_batch_params asynchronous generator.""" + mock_total_track_count = mocker.patch( + "playlist.plex.server.total_track_count", + new_callable=unittest.mock.AsyncMock, + ) + mock_total_track_count.return_value = 12 + mock_const = mocker.patch("playlist.plex.server.const") + mock_const.MAX_PROCESSES = 3 + mocker.patch("playlist.plex.server.settings") + mocker.patch( + "playlist.plex.server.asyncio", + new_callable=unittest.mock.AsyncMock, + ) + + result = [item async for item in server._gen_batch_params(5)] + + assert len(result) == 3