DarkHelm.org/plex-playlist-old
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

server.py tests completed.

Browse Source 
Signed-off-by: Cliff Hill <xlorep@darkhelm.org>
This commit is contained in:
Xlorep DarkHelm
2021-10-01 15:37:17 -04:00
parent 96641858ef
commit 3c87b49554
3 changed files with 180 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/playlist/data/settings.py
View File

@@ -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,

55
src/playlist/plex/server.py
View File

@@ -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(

153
tests/plex/test_server.py
View File

@@ -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