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

"""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]


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 = [100, 100, 100]
    weights = [100, 100, 100]

    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):  # 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, ["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 = 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