plex-playlist-old/tests/plex/test_server.py

"""Tests for the playlist.plex.server module."""
import asyncio
import datetime
import unittest.mock

import pytest

from playlist.data import models
from playlist.data import settings
from playlist.plex import server


def test_calc_delay(mocker):  # type: ignore [no-untyped-def]
    """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: list[float] = [100, 100, 100]
    weights: list[float] = [1, 1, 1]

    server.calc_delay(times, weights)

    result = mock_s.download.process_delay

    assert result == 100


def test_calc_duration(mocker):  # type: ignore [no-untyped-def]
    """Test the calc_duration function."""
    mock_settings = mocker.patch("playlist.plex.server.settings")
    durations = [100, 100, 100]
    mock_s = mock_settings.modify.return_value.__enter__.return_value
    mock_s.playlist.playtime.total_seconds.return_value = 300
    server.calc_duration(durations)
    result = mock_s.playlist.max_tracks
    assert result == 2


@pytest.mark.asyncio
async def test_gen_tracks(mocker, fake_track):  # type: ignore [no-untyped-def]
    """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, [fake_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
async def test_total_track_count(mocker):  # type: ignore [no-untyped-def]
    """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):  # type: ignore [no-untyped-def]
    """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(
        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
async def test_downloader(mocker):  # type: ignore [no-untyped-def]
    """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


@pytest.fixture
def fake_track(fake_plextrack):  # type: ignore [no-untyped-def]
    """Make a fake track dict object."""
    return models.Track(
        plex_id=fake_plextrack.ratingKey,
        track_num=fake_plextrack.index,
        title=fake_plextrack.title,
        artist=fake_plextrack.artist().title,
        album_num=fake_plextrack.parentIndex,
        album=fake_plextrack.parentTitle,
        album_artist=fake_plextrack.grandparentTitle,
        duration=fake_plextrack.duration,
        rating=fake_plextrack.userRating,
        comments=fake_plextrack.summary,
        added=fake_plextrack.addedAt,
        play_count=fake_plextrack.viewCount,
        played=fake_plextrack.lastViewedAt,
    )


def test_get_track_batch(mocker, fake_track):  # type: ignore [no-untyped-def]
    """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 = fake_track

    result = server._get_track_batch(1, 1)

    assert len(result)
    assert mock_track_dump.called


@pytest.fixture
def fake_plextrack():  # type: ignore [no-untyped-def]
    """Mock up a fake Plex track for testing."""
    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=datetime.datetime.now(),
        viewCount=1,
        lastViewedAt=None,
    )
    mock_track.artist.return_value.title = "Nothing"
    return mock_track


def test_track_dump(fake_plextrack, fake_track):  # type: ignore [no-untyped-def]
    """Test the _track_dump function."""
    result = server._track_dump(fake_plextrack)

    assert result == fake_track


@pytest.mark.asyncio
async def test_gen_batch_params(mocker):  # type: ignore [no-untyped-def]
    """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 = 10
    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