mirror of
https://github.com/oskvr37/tiddl.git
synced 2026-06-13 04:05:08 +03:00
✨ tiddl3 (#194)
This commit is contained in:
Oskar Dudziński
@@ -0,0 +1,200 @@
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from time import time
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from tiddl.core.auth import AuthClientError
|
||||
from tiddl.cli.commands.auth import auth_command
|
||||
from tiddl.cli.utils.auth import AuthData
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
|
||||
def test_login_already_logged(monkeypatch: pytest.MonkeyPatch):
|
||||
"""Should exit early if user is logged in."""
|
||||
|
||||
monkeypatch.setattr(
|
||||
"tiddl.cli.commands.auth.load_auth_data", lambda: AuthData(token="token")
|
||||
)
|
||||
|
||||
result = runner.invoke(auth_command, ["login"])
|
||||
|
||||
assert "Already logged in." in result.stdout
|
||||
assert result.exit_code == 0
|
||||
|
||||
|
||||
def test_login_success(monkeypatch: pytest.MonkeyPatch):
|
||||
"""Should save user token."""
|
||||
|
||||
monkeypatch.setattr(
|
||||
"tiddl.cli.commands.auth.load_auth_data", lambda: AuthData(token=None)
|
||||
)
|
||||
|
||||
device_auth_mock = MagicMock()
|
||||
device_auth_mock.verificationUriComplete = "verify.uri"
|
||||
device_auth_mock.deviceCode = "device123"
|
||||
device_auth_mock.expiresIn = 60
|
||||
device_auth_mock.interval = 1
|
||||
|
||||
auth_mock = MagicMock()
|
||||
auth_mock.access_token = "newtoken"
|
||||
auth_mock.refresh_token = "refreshtoken"
|
||||
auth_mock.expires_in = 3600
|
||||
auth_mock.user_id = 123
|
||||
auth_mock.user.countryCode = "US"
|
||||
|
||||
with (
|
||||
patch("tiddl.cli.commands.auth.AuthAPI") as MockAuthAPI,
|
||||
patch("tiddl.cli.commands.auth.typer.launch") as mock_launch,
|
||||
patch("tiddl.cli.commands.auth.save_auth_data") as mock_save,
|
||||
patch("tiddl.cli.commands.auth.time", side_effect=lambda: 1000),
|
||||
patch("tiddl.cli.commands.auth.sleep"),
|
||||
):
|
||||
|
||||
auth_api = MockAuthAPI.return_value
|
||||
auth_api.get_device_auth.return_value = device_auth_mock
|
||||
auth_api.get_auth.side_effect = [
|
||||
AuthClientError(error="authorization_pending"),
|
||||
auth_mock,
|
||||
]
|
||||
|
||||
result = runner.invoke(auth_command, ["login"])
|
||||
|
||||
auth_api.get_device_auth.assert_called_once()
|
||||
auth_api.get_auth.assert_called()
|
||||
mock_launch.assert_called_once_with("https://verify.uri")
|
||||
mock_save.assert_called_once()
|
||||
saved_data = mock_save.call_args[0][0]
|
||||
assert saved_data.token == "newtoken"
|
||||
assert "Logged in!" in result.stdout
|
||||
assert result.exit_code == 0
|
||||
|
||||
|
||||
def test_login_expired(monkeypatch: pytest.MonkeyPatch):
|
||||
"""Should not save token and exit."""
|
||||
|
||||
monkeypatch.setattr(
|
||||
"tiddl.cli.commands.auth.load_auth_data", lambda: AuthData(token=None)
|
||||
)
|
||||
|
||||
device_auth_mock = MagicMock()
|
||||
device_auth_mock.verificationUriComplete = "verify.uri"
|
||||
device_auth_mock.deviceCode = "device123"
|
||||
device_auth_mock.expiresIn = 60
|
||||
device_auth_mock.interval = 1
|
||||
|
||||
with (
|
||||
patch("tiddl.cli.commands.auth.AuthAPI") as MockAuthAPI,
|
||||
patch("tiddl.cli.commands.auth.typer.launch") as mock_launch,
|
||||
patch("tiddl.cli.commands.auth.save_auth_data") as mock_save,
|
||||
patch("tiddl.cli.commands.auth.time", side_effect=lambda: 1000),
|
||||
patch("tiddl.cli.commands.auth.sleep"),
|
||||
):
|
||||
|
||||
auth_api = MockAuthAPI.return_value
|
||||
auth_api.get_device_auth.return_value = device_auth_mock
|
||||
auth_api.get_auth.side_effect = [
|
||||
AuthClientError(error="expired_token"),
|
||||
]
|
||||
|
||||
result = runner.invoke(auth_command, ["login"])
|
||||
|
||||
auth_api.get_device_auth.assert_called_once()
|
||||
auth_api.get_auth.assert_called()
|
||||
mock_launch.assert_called_once_with("https://verify.uri")
|
||||
mock_save.assert_not_called()
|
||||
assert "Time for authentication has expired." in result.stdout
|
||||
assert result.exit_code == 0
|
||||
|
||||
|
||||
def test_logout_with_token(monkeypatch: pytest.MonkeyPatch):
|
||||
"""Should clear auth data and logout token in API."""
|
||||
|
||||
monkeypatch.setattr(
|
||||
"tiddl.cli.commands.auth.load_auth_data", lambda: AuthData(token="token")
|
||||
)
|
||||
|
||||
with (
|
||||
patch("tiddl.cli.commands.auth.AuthAPI") as MockAuthAPI,
|
||||
patch("tiddl.cli.commands.auth.save_auth_data") as mock_save,
|
||||
):
|
||||
mock_api_instance = MockAuthAPI.return_value
|
||||
result = runner.invoke(auth_command, ["logout"])
|
||||
|
||||
mock_api_instance.logout_token.assert_called_once_with("token")
|
||||
mock_save.assert_called_once_with(AuthData())
|
||||
|
||||
assert "Logged out!" in result.stdout
|
||||
assert result.exit_code == 0
|
||||
|
||||
|
||||
def test_logout_no_token(monkeypatch: pytest.MonkeyPatch):
|
||||
"""Should only clear auth data."""
|
||||
|
||||
monkeypatch.setattr(
|
||||
"tiddl.cli.commands.auth.load_auth_data", lambda: AuthData(token=None)
|
||||
)
|
||||
|
||||
with (
|
||||
patch("tiddl.cli.commands.auth.AuthAPI") as MockAuthAPI,
|
||||
patch("tiddl.cli.commands.auth.save_auth_data") as mock_save,
|
||||
):
|
||||
result = runner.invoke(auth_command, ["logout"])
|
||||
|
||||
mock_save.assert_called_once_with(AuthData())
|
||||
MockAuthAPI.assert_not_called()
|
||||
|
||||
assert "Logged out!" in result.stdout
|
||||
assert result.exit_code == 0
|
||||
|
||||
|
||||
def test_refresh_not_logged_in(monkeypatch: pytest.MonkeyPatch):
|
||||
"""Should exit early if refresh_token is missing."""
|
||||
|
||||
monkeypatch.setattr(
|
||||
"tiddl.cli.commands.auth.load_auth_data", lambda: AuthData(refresh_token=None)
|
||||
)
|
||||
result = runner.invoke(auth_command, ["refresh"])
|
||||
|
||||
assert "Not logged in." in result.stdout
|
||||
assert result.exit_code == 0
|
||||
|
||||
|
||||
def test_refresh_not_expired(monkeypatch: pytest.MonkeyPatch):
|
||||
"""Should exit early if token still valid."""
|
||||
|
||||
monkeypatch.setattr(
|
||||
"tiddl.cli.commands.auth.load_auth_data",
|
||||
lambda: AuthData(
|
||||
token="abc", refresh_token="ref", expires_at=int(time()) + 3600
|
||||
),
|
||||
)
|
||||
result = runner.invoke(auth_command, ["refresh"])
|
||||
|
||||
assert "Auth token expires in" in result.stdout
|
||||
assert result.exit_code == 0
|
||||
|
||||
|
||||
def test_refresh_success(monkeypatch: pytest.MonkeyPatch):
|
||||
"""Should refresh token if expired."""
|
||||
|
||||
expired_data = AuthData(
|
||||
token="oldtoken", refresh_token="refreshtoken", expires_at=0
|
||||
)
|
||||
monkeypatch.setattr("tiddl.cli.commands.auth.load_auth_data", lambda: expired_data)
|
||||
|
||||
mock_auth_response = MagicMock()
|
||||
mock_auth_response.access_token = "newtoken"
|
||||
|
||||
with (
|
||||
patch("tiddl.cli.commands.auth.AuthAPI") as MockAuthAPI,
|
||||
patch("tiddl.cli.commands.auth.save_auth_data") as mock_save,
|
||||
):
|
||||
|
||||
MockAuthAPI.return_value.refresh_token.return_value = mock_auth_response
|
||||
|
||||
result = runner.invoke(auth_command, ["refresh"])
|
||||
|
||||
mock_save.assert_called_once_with(expired_data)
|
||||
assert "Auth token has been refreshed!" in result.stdout
|
||||
assert result.exit_code == 0
|
||||
@@ -0,0 +1,41 @@
|
||||
from pathlib import Path
|
||||
|
||||
from tiddl.cli.utils.auth.core import load_auth_data, save_auth_data
|
||||
from tiddl.cli.utils.auth.models import AuthData
|
||||
|
||||
|
||||
def test_load_auth_data(tmp_path: Path):
|
||||
file = tmp_path / "auth.json"
|
||||
|
||||
auth_data = AuthData(
|
||||
token="token",
|
||||
refresh_token="refresh_token",
|
||||
expires_at=0,
|
||||
user_id="user_id",
|
||||
country_code="country_code",
|
||||
)
|
||||
|
||||
file.write_text(auth_data.model_dump_json())
|
||||
|
||||
loaded_auth_data = load_auth_data(file)
|
||||
|
||||
assert isinstance(loaded_auth_data, AuthData)
|
||||
assert loaded_auth_data.__dict__ == auth_data.__dict__
|
||||
|
||||
|
||||
def test_save_auth_data(tmp_path: Path):
|
||||
file = tmp_path / "auth.json"
|
||||
|
||||
auth_data = AuthData(
|
||||
token="token",
|
||||
refresh_token="refresh_token",
|
||||
expires_at=0,
|
||||
user_id="user_id",
|
||||
country_code="country_code",
|
||||
)
|
||||
|
||||
save_auth_data(auth_data=auth_data, file=file)
|
||||
|
||||
loaded_auth_data = AuthData.model_validate_json(file.read_text())
|
||||
|
||||
assert loaded_auth_data.__dict__ == auth_data.__dict__
|
||||
@@ -0,0 +1,13 @@
|
||||
import typer
|
||||
|
||||
from tiddl.cli.commands import register_commands, COMMANDS
|
||||
|
||||
|
||||
def test_register_commands_adds_typers():
|
||||
app = typer.Typer()
|
||||
register_commands(app)
|
||||
|
||||
registered_names = [cmd.name for cmd in app.registered_groups + app.registered_commands]
|
||||
|
||||
for command in COMMANDS:
|
||||
assert command.info.name in registered_names
|
||||
@@ -0,0 +1,63 @@
|
||||
from pathlib import Path
|
||||
from pytest import raises
|
||||
|
||||
from tiddl.cli.config import load_config_file, Config, CONFIG_FILENAME
|
||||
|
||||
|
||||
def write_config(tmp_path: Path, content: str) -> Path:
|
||||
cfg_path = tmp_path / CONFIG_FILENAME
|
||||
cfg_path.write_text(content)
|
||||
return cfg_path
|
||||
|
||||
|
||||
def test_missing_file_default_config(tmp_path: Path):
|
||||
cfg_file = tmp_path / "nonexistent.toml"
|
||||
cfg = load_config_file(cfg_file)
|
||||
|
||||
assert isinstance(cfg, Config)
|
||||
|
||||
|
||||
def test_valid_config_file(tmp_path: Path):
|
||||
cfg_file = write_config(
|
||||
tmp_path,
|
||||
"""
|
||||
enable_cache = false
|
||||
debug = true
|
||||
|
||||
[download]
|
||||
track_quality = "max"
|
||||
threads_count = 8
|
||||
""",
|
||||
)
|
||||
|
||||
cfg = load_config_file(cfg_file)
|
||||
|
||||
assert cfg.enable_cache is False
|
||||
assert cfg.debug is True
|
||||
assert cfg.download.track_quality == "max"
|
||||
assert cfg.download.threads_count == 8
|
||||
|
||||
|
||||
def test_invalid_type_raises(tmp_path: Path):
|
||||
cfg_file = write_config(
|
||||
tmp_path,
|
||||
"""
|
||||
enable_cache = "not_a_bool"
|
||||
""",
|
||||
)
|
||||
|
||||
with raises(Exception):
|
||||
load_config_file(cfg_file)
|
||||
|
||||
|
||||
def test_invalid_track_quality_raises(tmp_path: Path):
|
||||
cfg_file = write_config(
|
||||
tmp_path,
|
||||
"""
|
||||
[download]
|
||||
track_quality = "ultra"
|
||||
""",
|
||||
)
|
||||
|
||||
with raises(Exception):
|
||||
load_config_file(cfg_file)
|
||||
@@ -0,0 +1,20 @@
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
from tiddl.cli.const import get_app_path, APP_DIR_NAME, ENV_KEY
|
||||
|
||||
|
||||
def test_env_key_overrides(monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
|
||||
custom_path = tmp_path / "customdir"
|
||||
monkeypatch.setenv(ENV_KEY, str(custom_path))
|
||||
app_path = get_app_path(ENV_KEY)
|
||||
|
||||
assert app_path == custom_path
|
||||
|
||||
|
||||
def test_default_path_if_unset(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.delenv(ENV_KEY, raising=False)
|
||||
app_path = get_app_path(ENV_KEY)
|
||||
|
||||
assert str(Path.home()) in str(app_path)
|
||||
assert app_path.name == APP_DIR_NAME
|
||||
@@ -0,0 +1,68 @@
|
||||
import pytest
|
||||
from tiddl.cli.utils.resource import TidalResource, ResourceTypeLiteral
|
||||
|
||||
valid_test_data = [
|
||||
("track", "12345"),
|
||||
("album", "98765"),
|
||||
("video", "11111"),
|
||||
("artist", "22222"),
|
||||
("playlist", "abcde"),
|
||||
("mix", "xyz123"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("resource_type, resource_id", valid_test_data)
|
||||
def test_tidalresource_from_string_shorthand(
|
||||
resource_type: ResourceTypeLiteral, resource_id: str
|
||||
):
|
||||
string = f"{resource_type}/{resource_id}"
|
||||
res = TidalResource.from_string(string)
|
||||
|
||||
assert res.type == resource_type
|
||||
assert res.id == resource_id
|
||||
assert str(res) == string
|
||||
assert res.url == f"https://listen.tidal.com/{resource_type}/{resource_id}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("resource_type, resource_id", valid_test_data)
|
||||
def test_tidalresource_from_string_url(
|
||||
resource_type: ResourceTypeLiteral, resource_id: str
|
||||
):
|
||||
url = f"https://listen.tidal.com/{resource_type}/{resource_id}"
|
||||
res = TidalResource.from_string(url)
|
||||
|
||||
assert res.type == resource_type
|
||||
assert res.id == resource_id
|
||||
assert str(res) == f"{resource_type}/{resource_id}"
|
||||
assert res.url == url
|
||||
|
||||
|
||||
def test_from_string_invalid_type():
|
||||
with pytest.raises(ValueError, match="Invalid resource type"):
|
||||
TidalResource.from_string("invalid/123")
|
||||
|
||||
|
||||
invalid_test_data = [
|
||||
("track", "abc"),
|
||||
("album", "xyz"),
|
||||
("video", "id123"),
|
||||
("artist", "user1"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("resource_type, invalid_id", invalid_test_data)
|
||||
def test_from_string_invalid_digit_id(
|
||||
resource_type: ResourceTypeLiteral, invalid_id: str
|
||||
):
|
||||
with pytest.raises(ValueError, match="Invalid resource id"):
|
||||
TidalResource.from_string(f"{resource_type}/{invalid_id}")
|
||||
|
||||
|
||||
def test_url_property():
|
||||
res = TidalResource(type="track", id="12345")
|
||||
assert res.url == "https://listen.tidal.com/track/12345"
|
||||
|
||||
|
||||
def test_str_method():
|
||||
res = TidalResource(type="album", id="67890")
|
||||
assert str(res) == "album/67890"
|
||||
@@ -0,0 +1,206 @@
|
||||
import pytest
|
||||
from pytest_mock import MockerFixture, MockType
|
||||
|
||||
from tiddl.core.api.api import (
|
||||
TidalAPI,
|
||||
TidalClient,
|
||||
Limits,
|
||||
DO_NOT_CACHE,
|
||||
EXPIRE_IMMEDIATELY,
|
||||
)
|
||||
|
||||
from tiddl.core.api.models import (
|
||||
Album,
|
||||
Artist,
|
||||
Playlist,
|
||||
Track,
|
||||
Video,
|
||||
AlbumItems,
|
||||
AlbumItemsCredits,
|
||||
ArtistAlbumsItems,
|
||||
Favorites,
|
||||
TrackLyrics,
|
||||
PlaylistItems,
|
||||
MixItems,
|
||||
Search,
|
||||
SessionResponse,
|
||||
TrackStream,
|
||||
VideoStream,
|
||||
)
|
||||
|
||||
|
||||
def test_tidal_api_init(mocker: MockerFixture):
|
||||
mock_client = mocker.Mock(spec=TidalClient)
|
||||
|
||||
api = TidalAPI(client=mock_client, user_id="u123", country_code="US")
|
||||
|
||||
assert api.client is mock_client
|
||||
assert api.user_id == "u123"
|
||||
assert api.country_code == "US"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_client(mocker: MockerFixture):
|
||||
return mocker.Mock(spec=TidalClient)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api(mock_client: MockType):
|
||||
return TidalAPI(client=mock_client, user_id="u123", country_code="US")
|
||||
|
||||
|
||||
def test_get_album(api: TidalAPI, mock_client: MockType):
|
||||
api.get_album(album_id=1)
|
||||
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
Album, "albums/1", {"countryCode": "US"}, expire_after=3600
|
||||
)
|
||||
|
||||
|
||||
def test_get_album_items(api: TidalAPI, mock_client: MockType):
|
||||
api.get_album_items(1)
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
AlbumItems,
|
||||
"albums/1/items",
|
||||
{"countryCode": "US", "limit": Limits.ALBUM_ITEMS, "offset": 0},
|
||||
expire_after=3600,
|
||||
)
|
||||
|
||||
|
||||
def test_get_album_items_credits(api: TidalAPI, mock_client: MockType):
|
||||
api.get_album_items_credits(1)
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
AlbumItemsCredits,
|
||||
"albums/1/items/credits",
|
||||
{"countryCode": "US", "limit": Limits.ALBUM_ITEMS, "offset": 0},
|
||||
expire_after=3600,
|
||||
)
|
||||
|
||||
|
||||
def test_get_artist(api: TidalAPI, mock_client: MockType):
|
||||
api.get_artist(1)
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
Artist, "artists/1", {"countryCode": "US"}, expire_after=3600
|
||||
)
|
||||
|
||||
|
||||
def test_get_artist_albums(api: TidalAPI, mock_client: MockType):
|
||||
api.get_artist_albums(1)
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
ArtistAlbumsItems,
|
||||
"artists/1/albums",
|
||||
{
|
||||
"countryCode": "US",
|
||||
"limit": Limits.ARTIST_ALBUMS,
|
||||
"offset": 0,
|
||||
"filter": "ALBUMS",
|
||||
},
|
||||
expire_after=3600,
|
||||
)
|
||||
|
||||
|
||||
def test_get_mix(api: TidalAPI, mock_client: MockType):
|
||||
api.get_mix_items("abcd-1234")
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
MixItems,
|
||||
"mixes/abcd-1234/items",
|
||||
{"countryCode": "US", "limit": Limits.MIX_ITEMS, "offset": 0},
|
||||
expire_after=3600,
|
||||
)
|
||||
|
||||
|
||||
def test_get_favorites(api: TidalAPI, mock_client: MockType):
|
||||
api.get_favorites()
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
Favorites,
|
||||
"users/u123/favorites/ids",
|
||||
{"countryCode": "US"},
|
||||
expire_after=EXPIRE_IMMEDIATELY,
|
||||
)
|
||||
|
||||
|
||||
def test_get_playlist(api: TidalAPI, mock_client: MockType):
|
||||
api.get_playlist("uuid")
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
Playlist,
|
||||
"playlists/uuid",
|
||||
{"countryCode": "US"},
|
||||
expire_after=EXPIRE_IMMEDIATELY,
|
||||
)
|
||||
|
||||
|
||||
def test_get_playlist_items(api: TidalAPI, mock_client: MockType):
|
||||
api.get_playlist_items("uuid")
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
PlaylistItems,
|
||||
"playlists/uuid/items",
|
||||
{"countryCode": "US", "limit": Limits.PLAYLIST_ITEMS, "offset": 0},
|
||||
expire_after=EXPIRE_IMMEDIATELY,
|
||||
)
|
||||
|
||||
|
||||
def test_get_search(api: TidalAPI, mock_client: MockType):
|
||||
api.get_search("query")
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
Search,
|
||||
"search",
|
||||
{"countryCode": "US", "query": "query"},
|
||||
expire_after=DO_NOT_CACHE,
|
||||
)
|
||||
|
||||
|
||||
def test_get_session(api: TidalAPI, mock_client: MockType):
|
||||
api.get_session()
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
SessionResponse, "sessions", expire_after=DO_NOT_CACHE
|
||||
)
|
||||
|
||||
|
||||
def test_get_track_lyrics(api: TidalAPI, mock_client: MockType):
|
||||
api.get_track_lyrics(1)
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
TrackLyrics,
|
||||
"tracks/1/lyrics",
|
||||
{"countryCode": "US"},
|
||||
expire_after=3600,
|
||||
)
|
||||
|
||||
|
||||
def test_get_track(api: TidalAPI, mock_client: MockType):
|
||||
api.get_track(1)
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
Track,
|
||||
"tracks/1",
|
||||
{"countryCode": "US"},
|
||||
expire_after=3600,
|
||||
)
|
||||
|
||||
|
||||
def test_get_track_stream(api: TidalAPI, mock_client: MockType):
|
||||
api.get_track_stream(1, "HIGH")
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
TrackStream,
|
||||
"tracks/1/playbackinfopostpaywall",
|
||||
{"audioquality": "HIGH", "playbackmode": "STREAM", "assetpresentation": "FULL"},
|
||||
expire_after=DO_NOT_CACHE,
|
||||
)
|
||||
|
||||
|
||||
def test_get_video(api: TidalAPI, mock_client: MockType):
|
||||
api.get_video(1)
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
Video,
|
||||
"videos/1",
|
||||
{"countryCode": "US"},
|
||||
expire_after=3600,
|
||||
)
|
||||
|
||||
|
||||
def test_get_video_stream(api: TidalAPI, mock_client: MockType):
|
||||
api.get_video_stream(1, "HIGH")
|
||||
mock_client.fetch.assert_called_once_with(
|
||||
VideoStream,
|
||||
"videos/1/playbackinfopostpaywall",
|
||||
{"videoquality": "HIGH", "playbackmode": "STREAM", "assetpresentation": "FULL"},
|
||||
expire_after=DO_NOT_CACHE,
|
||||
)
|
||||
@@ -0,0 +1,93 @@
|
||||
import pytest
|
||||
import json
|
||||
|
||||
from pydantic import BaseModel
|
||||
from pytest_mock import MockerFixture
|
||||
from pathlib import Path
|
||||
|
||||
from tiddl.core.api.client import TidalClient, ApiError
|
||||
|
||||
|
||||
def test_tidal_client_init(mocker: MockerFixture):
|
||||
mock_cached_session = mocker.patch("tiddl.core.api.client.CachedSession")
|
||||
mock_session = mock_cached_session.return_value
|
||||
|
||||
client = TidalClient(
|
||||
token="test-token",
|
||||
cache_name="test_cache",
|
||||
omit_cache=True,
|
||||
debug_path=Path("/tmp/debug"),
|
||||
)
|
||||
|
||||
mock_cached_session.assert_called_once_with(
|
||||
cache_name="test_cache", always_revalidate=True
|
||||
)
|
||||
|
||||
assert client.token == "test-token"
|
||||
assert client.debug_path == Path("/tmp/debug")
|
||||
assert client.session is mock_session
|
||||
assert mock_session.headers["Authorization"] == "Bearer test-token"
|
||||
assert mock_session.headers["Accept"] == "application/json"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("omit_cache", [True, False])
|
||||
def test_omit_cache_flag(mocker: MockerFixture, omit_cache: bool):
|
||||
mock_cached_session = mocker.patch("tiddl.core.api.client.CachedSession")
|
||||
TidalClient("token", "cache", omit_cache=omit_cache)
|
||||
mock_cached_session.assert_called_once_with(
|
||||
cache_name="cache", always_revalidate=omit_cache
|
||||
)
|
||||
|
||||
|
||||
class DummyModel(BaseModel):
|
||||
foo: str
|
||||
|
||||
|
||||
def test_fetch_success(mocker: MockerFixture, tmp_path: Path):
|
||||
mock_session = mocker.Mock()
|
||||
mock_response = mocker.Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.from_cache = False
|
||||
mock_response.json.return_value = {"foo": "bar"}
|
||||
mock_session.get.return_value = mock_response
|
||||
|
||||
mocker.patch("tiddl.core.api.client.API_URL", "https://api.test")
|
||||
client = TidalClient("token", tmp_path / "cache", debug_path=tmp_path)
|
||||
client.session = mock_session
|
||||
|
||||
result = client.fetch(DummyModel, "albums/123", {"limit": 10}, expire_after=999)
|
||||
assert result.foo == "bar"
|
||||
|
||||
mock_session.get.assert_called_once_with(
|
||||
"https://api.test/albums/123",
|
||||
params={"limit": 10},
|
||||
expire_after=999,
|
||||
)
|
||||
|
||||
debug_file = tmp_path / "albums/123.json"
|
||||
assert debug_file.exists()
|
||||
|
||||
content = json.loads(debug_file.read_text())
|
||||
assert content["status_code"] == 200
|
||||
assert content["endpoint"] == "albums/123"
|
||||
assert content["params"]["limit"] == 10
|
||||
assert content["data"]["foo"] == "bar"
|
||||
|
||||
|
||||
def test_fetch_error_raises_api_error(mocker: MockerFixture, tmp_path: Path):
|
||||
mock_session = mocker.Mock()
|
||||
mock_response = mocker.Mock()
|
||||
mock_response.status_code = 400
|
||||
mock_response.from_cache = False
|
||||
mock_response.json.return_value = {
|
||||
"status": 400,
|
||||
"subStatus": "Bad request",
|
||||
"userMessage": "user_message",
|
||||
}
|
||||
mock_session.get.return_value = mock_response
|
||||
|
||||
client = TidalClient("token", tmp_path / "cache")
|
||||
client.session = mock_session
|
||||
|
||||
with pytest.raises(ApiError):
|
||||
client.fetch(DummyModel, "bad/endpoint")
|
||||
@@ -0,0 +1,38 @@
|
||||
import pytest
|
||||
from typing import Any
|
||||
from tiddl.core.api.exceptions import ApiError
|
||||
|
||||
|
||||
def test_api_error_attributes():
|
||||
data: dict[str, Any] = {
|
||||
"status": 1,
|
||||
"subStatus": "sub_status",
|
||||
"userMessage": "user_message",
|
||||
}
|
||||
|
||||
e = ApiError(**data)
|
||||
|
||||
assert isinstance(e, Exception)
|
||||
assert e.status == data["status"]
|
||||
assert e.sub_status == data["subStatus"]
|
||||
assert e.user_message == data["userMessage"]
|
||||
|
||||
|
||||
def test_api_error_raises():
|
||||
with pytest.raises(ApiError) as exc:
|
||||
raise ApiError(400, "bad_request", "invalid")
|
||||
|
||||
assert exc.value.status == 400
|
||||
assert exc.value.sub_status == "bad_request"
|
||||
|
||||
|
||||
def test_api_error_string():
|
||||
data: dict[str, Any] = {
|
||||
"status": 1,
|
||||
"subStatus": "sub_status",
|
||||
"userMessage": "user_message",
|
||||
}
|
||||
|
||||
e = ApiError(**data)
|
||||
|
||||
assert str(e) == f"{e.user_message}, {e.status}/{e.sub_status}"
|
||||
@@ -0,0 +1,105 @@
|
||||
from typing import Any
|
||||
import pytest
|
||||
from pytest_mock import MockerFixture
|
||||
from tiddl.core.auth.api import AuthAPI
|
||||
from tiddl.core.auth.models import (
|
||||
AuthDeviceResponse,
|
||||
AuthResponseWithRefresh,
|
||||
AuthResponse,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_auth_client(mocker: MockerFixture) -> Any:
|
||||
client = mocker.Mock()
|
||||
|
||||
client.get_device_auth.return_value = {
|
||||
"deviceCode": "abc",
|
||||
"userCode": "123",
|
||||
"verificationUri": "https://verify",
|
||||
"verificationUriComplete": "https://verify?code=123",
|
||||
"expiresIn": 300,
|
||||
"interval": 5,
|
||||
}
|
||||
|
||||
user_data: dict[str, Any] = {
|
||||
"userId": 1,
|
||||
"email": "test@example.com",
|
||||
"countryCode": "US",
|
||||
"fullName": None,
|
||||
"firstName": None,
|
||||
"lastName": None,
|
||||
"nickname": None,
|
||||
"username": "tester",
|
||||
"address": None,
|
||||
"city": None,
|
||||
"postalcode": None,
|
||||
"usState": None,
|
||||
"phoneNumber": None,
|
||||
"birthday": None,
|
||||
"channelId": 0,
|
||||
"parentId": 0,
|
||||
"acceptedEULA": True,
|
||||
"created": 0,
|
||||
"updated": 0,
|
||||
"facebookUid": 0,
|
||||
"appleUid": None,
|
||||
"googleUid": None,
|
||||
"accountLinkCreated": True,
|
||||
"emailVerified": True,
|
||||
"newUser": True,
|
||||
}
|
||||
|
||||
auth_base: dict[str, Any] = {
|
||||
"access_token": "token123",
|
||||
"refresh_token": "refresh123",
|
||||
"expires_in": 3600,
|
||||
"user_id": 1,
|
||||
"scope": "r_usr",
|
||||
"clientName": "tidal",
|
||||
"token_type": "Bearer",
|
||||
"user": user_data,
|
||||
}
|
||||
|
||||
client.get_auth.return_value = auth_base.copy()
|
||||
client.refresh_token.return_value = auth_base.copy()
|
||||
client.logout_token.return_value = None
|
||||
|
||||
return client
|
||||
|
||||
|
||||
def test_get_device_auth_returns_model(mock_auth_client: Any) -> None:
|
||||
api: AuthAPI = AuthAPI(client=mock_auth_client)
|
||||
result: AuthDeviceResponse = api.get_device_auth()
|
||||
|
||||
mock_auth_client.get_device_auth.assert_called_once()
|
||||
assert isinstance(result, AuthDeviceResponse)
|
||||
assert result.deviceCode == "abc"
|
||||
assert result.interval == 5
|
||||
|
||||
|
||||
def test_get_auth_returns_model(mock_auth_client: Any) -> None:
|
||||
api: AuthAPI = AuthAPI(client=mock_auth_client)
|
||||
result: AuthResponseWithRefresh = api.get_auth("device123")
|
||||
|
||||
mock_auth_client.get_auth.assert_called_once_with("device123")
|
||||
assert isinstance(result, AuthResponseWithRefresh)
|
||||
assert result.access_token == "token123"
|
||||
assert result.refresh_token == "refresh123"
|
||||
assert result.user.userId == 1
|
||||
|
||||
|
||||
def test_refresh_token_returns_model(mock_auth_client: Any) -> None:
|
||||
api: AuthAPI = AuthAPI(client=mock_auth_client)
|
||||
result: AuthResponse = api.refresh_token("refresh123")
|
||||
|
||||
mock_auth_client.refresh_token.assert_called_once_with("refresh123")
|
||||
assert isinstance(result, AuthResponse)
|
||||
assert result.access_token == "token123"
|
||||
|
||||
|
||||
def test_logout_token_calls_client(mock_auth_client: Any) -> None:
|
||||
api: AuthAPI = AuthAPI(client=mock_auth_client)
|
||||
api.logout_token("token123")
|
||||
|
||||
mock_auth_client.logout_token.assert_called_once_with("token123")
|
||||
@@ -0,0 +1,128 @@
|
||||
import pytest
|
||||
from pytest_mock import MockerFixture
|
||||
from tiddl.core.auth.client import AuthClient
|
||||
from tiddl.core.auth.exceptions import AuthClientError
|
||||
|
||||
|
||||
def test_get_device_auth_calls_request(mocker: MockerFixture):
|
||||
mock_request = mocker.patch("tiddl.core.auth.client.request")
|
||||
|
||||
data = {"device_code": "abc"}
|
||||
mock_response = mocker.Mock()
|
||||
mock_response.json.return_value = data
|
||||
mock_request.return_value = mock_response
|
||||
|
||||
client = AuthClient()
|
||||
result = client.get_device_auth()
|
||||
|
||||
mock_request.assert_called_once_with(
|
||||
"POST",
|
||||
"https://auth.tidal.com/v1/oauth2/device_authorization",
|
||||
data={"client_id": client.client_id, "scope": "r_usr+w_usr+w_sub"},
|
||||
)
|
||||
|
||||
assert result == data
|
||||
|
||||
|
||||
def test_get_auth_returns_json_on_200(mocker: MockerFixture):
|
||||
mock_request = mocker.patch("tiddl.core.auth.client.request")
|
||||
mock_response = mocker.Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {
|
||||
"access_token": "token123",
|
||||
"refresh_token": "refresh123",
|
||||
"expires_in": 3600,
|
||||
}
|
||||
mock_request.return_value = mock_response
|
||||
|
||||
client = AuthClient()
|
||||
result = client.get_auth("device123")
|
||||
|
||||
assert result["access_token"] == "token123"
|
||||
assert result["refresh_token"] == "refresh123"
|
||||
assert result["expires_in"] == 3600
|
||||
|
||||
mock_request.assert_called_once_with(
|
||||
"POST",
|
||||
"https://auth.tidal.com/v1/oauth2/token",
|
||||
data={
|
||||
"client_id": client.client_id,
|
||||
"device_code": "device123",
|
||||
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
|
||||
"scope": "r_usr+w_usr+w_sub",
|
||||
},
|
||||
auth=(client.client_id, client.client_secret),
|
||||
)
|
||||
|
||||
|
||||
def test_get_auth_raises_on_non_200(mocker: MockerFixture):
|
||||
mock_request = mocker.patch("tiddl.core.auth.client.request")
|
||||
mock_response = mocker.Mock()
|
||||
mock_response.status_code = 400
|
||||
mock_response.json.return_value = {
|
||||
"error": "error",
|
||||
"status": 400,
|
||||
"sub_status": 1001,
|
||||
"error_description": "invalid",
|
||||
}
|
||||
mock_request.return_value = mock_response
|
||||
|
||||
client = AuthClient()
|
||||
|
||||
with pytest.raises(AuthClientError):
|
||||
client.get_auth("device123")
|
||||
|
||||
mock_request.assert_called_once_with(
|
||||
"POST",
|
||||
"https://auth.tidal.com/v1/oauth2/token",
|
||||
data={
|
||||
"client_id": client.client_id,
|
||||
"device_code": "device123",
|
||||
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
|
||||
"scope": "r_usr+w_usr+w_sub",
|
||||
},
|
||||
auth=(client.client_id, client.client_secret),
|
||||
)
|
||||
|
||||
|
||||
def test_refresh_token(mocker: MockerFixture):
|
||||
mock_request = mocker.patch("tiddl.core.auth.client.request")
|
||||
|
||||
mock_response = mocker.Mock()
|
||||
mock_response.status_code = 400
|
||||
mock_response.json.return_value = {
|
||||
"token": "abc",
|
||||
}
|
||||
mock_request.return_value = mock_response
|
||||
|
||||
refresh_token = "token"
|
||||
|
||||
client = AuthClient()
|
||||
result = client.refresh_token(refresh_token)
|
||||
|
||||
mock_request.assert_called_once_with(
|
||||
"POST",
|
||||
"https://auth.tidal.com/v1/oauth2/token",
|
||||
data={
|
||||
"client_id": client.client_id,
|
||||
"refresh_token": refresh_token,
|
||||
"grant_type": "refresh_token",
|
||||
"scope": "r_usr+w_usr+w_sub",
|
||||
},
|
||||
auth=(client.client_id, client.client_secret),
|
||||
)
|
||||
|
||||
assert result["token"] == "abc"
|
||||
|
||||
|
||||
def test_logout_token(mocker: MockerFixture):
|
||||
mock_request = mocker.patch("tiddl.core.auth.client.request")
|
||||
|
||||
client = AuthClient()
|
||||
client.logout_token("token")
|
||||
|
||||
mock_request.assert_called_once_with(
|
||||
"POST",
|
||||
"https://api.tidal.com/v1/logout",
|
||||
headers={"authorization": "Bearer token"},
|
||||
)
|
||||
@@ -0,0 +1,41 @@
|
||||
import pytest
|
||||
from typing import Any
|
||||
from tiddl.core.auth.exceptions import AuthClientError
|
||||
|
||||
|
||||
def test_auth_client_error_attributes():
|
||||
data: dict[str, Any] = {
|
||||
"status": 1,
|
||||
"error": "error",
|
||||
"sub_status": "sub_status",
|
||||
"error_description": "error_description",
|
||||
}
|
||||
|
||||
e = AuthClientError(**data)
|
||||
|
||||
assert isinstance(e, Exception)
|
||||
assert e.status == data["status"]
|
||||
assert e.error == data["error"]
|
||||
assert e.sub_status == data["sub_status"]
|
||||
assert e.error_description == data["error_description"]
|
||||
|
||||
|
||||
def test_auth_client_error_raises():
|
||||
with pytest.raises(AuthClientError) as exc:
|
||||
raise AuthClientError(400, "bad_request", "invalid", "Malformed input")
|
||||
|
||||
assert exc.value.status == 400
|
||||
assert exc.value.error == "bad_request"
|
||||
|
||||
|
||||
def test_auth_client_error_string():
|
||||
data: dict[str, Any] = {
|
||||
"status": 1,
|
||||
"error": "error",
|
||||
"sub_status": "sub_status",
|
||||
"error_description": "error_description",
|
||||
}
|
||||
|
||||
e = AuthClientError(**data)
|
||||
|
||||
assert str(e) == f"{e.error}, {e.error_description}, {e.status}/{e.sub_status}"
|
||||
Reference in New Issue
Block a user