mirror of
https://github.com/Benexl/FastAnime.git
synced 2026-07-28 22:40:52 -07:00
test: Add comprehensive test coverage across modules
Adds extensive test suite covering CLI notifications, auth service, file utilities, networking, AniList API integration, mappers, API factory, MPV player, and selector implementations. Includes 900+ lines of test code covering main workflows and edge cases. Minor refactor to improve dict type safety in AniList API client. - CLI and service layer tests - Utility function tests - Media API and mapper tests - Player and selector factory tests
This commit is contained in:
Type-Delta
@@ -0,0 +1,101 @@
|
||||
from contextlib import nullcontext
|
||||
from datetime import datetime
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from viu_media.cli.commands.anilist.commands.notifications import notifications
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def runner():
|
||||
return CliRunner()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config():
|
||||
return SimpleNamespace(general=SimpleNamespace(media_api="anilist"))
|
||||
|
||||
|
||||
def test_notifications_requires_authentication(runner, mock_config):
|
||||
with (
|
||||
patch("viu_media.cli.service.feedback.FeedbackService") as mock_feedback,
|
||||
patch("viu_media.libs.media_api.api.create_api_client") as mock_create_api,
|
||||
patch("viu_media.cli.service.auth.AuthService") as mock_auth,
|
||||
):
|
||||
feedback_instance = mock_feedback.return_value
|
||||
feedback_instance.progress.return_value = nullcontext()
|
||||
|
||||
auth_instance = mock_auth.return_value
|
||||
auth_instance.get_auth.return_value = None
|
||||
|
||||
api_client = mock_create_api.return_value
|
||||
api_client.is_authenticated.return_value = False
|
||||
|
||||
result = runner.invoke(notifications, [], obj=mock_config)
|
||||
|
||||
assert result.exit_code == 0
|
||||
feedback_instance.error.assert_called_with(
|
||||
"Authentication Required", "Please log in with 'viu anilist auth'."
|
||||
)
|
||||
|
||||
|
||||
def test_notifications_shows_all_caught_up_message(runner, mock_config):
|
||||
with (
|
||||
patch("viu_media.cli.service.feedback.FeedbackService") as mock_feedback,
|
||||
patch("viu_media.libs.media_api.api.create_api_client") as mock_create_api,
|
||||
patch("viu_media.cli.service.auth.AuthService") as mock_auth,
|
||||
):
|
||||
feedback_instance = mock_feedback.return_value
|
||||
feedback_instance.progress.return_value = nullcontext()
|
||||
|
||||
auth_instance = mock_auth.return_value
|
||||
auth_instance.get_auth.return_value = SimpleNamespace(token="token")
|
||||
|
||||
api_client = mock_create_api.return_value
|
||||
api_client.is_authenticated.return_value = True
|
||||
api_client.get_notifications.return_value = []
|
||||
|
||||
result = runner.invoke(notifications, [], obj=mock_config)
|
||||
|
||||
assert result.exit_code == 0
|
||||
api_client.authenticate.assert_called_once_with("token")
|
||||
feedback_instance.success.assert_called_with(
|
||||
"All caught up!", "You have no new notifications."
|
||||
)
|
||||
|
||||
|
||||
def test_notifications_prints_table_and_mark_read_info(runner, mock_config):
|
||||
notification_item = SimpleNamespace(
|
||||
created_at=datetime(2025, 1, 1, 10, 0, 0),
|
||||
media=SimpleNamespace(title=SimpleNamespace(english="Blue Lock", romaji=None)),
|
||||
episode=12,
|
||||
)
|
||||
|
||||
with (
|
||||
patch("viu_media.cli.service.feedback.FeedbackService") as mock_feedback,
|
||||
patch("viu_media.libs.media_api.api.create_api_client") as mock_create_api,
|
||||
patch("viu_media.cli.service.auth.AuthService") as mock_auth,
|
||||
patch("viu_media.cli.commands.anilist.commands.notifications.Console") as mock_console,
|
||||
):
|
||||
feedback_instance = mock_feedback.return_value
|
||||
feedback_instance.progress.return_value = nullcontext()
|
||||
|
||||
auth_instance = mock_auth.return_value
|
||||
auth_instance.get_auth.return_value = SimpleNamespace(token="token")
|
||||
|
||||
api_client = mock_create_api.return_value
|
||||
api_client.is_authenticated.return_value = True
|
||||
api_client.get_notifications.return_value = [notification_item]
|
||||
|
||||
console_instance = mock_console.return_value
|
||||
|
||||
result = runner.invoke(notifications, [], obj=mock_config)
|
||||
|
||||
assert result.exit_code == 0
|
||||
console_instance.print.assert_called_once()
|
||||
feedback_instance.info.assert_called_once_with(
|
||||
"Notifications have been marked as read on AniList.",
|
||||
)
|
||||
@@ -0,0 +1,42 @@
|
||||
from viu_media.cli.service.auth.service import AuthService
|
||||
from viu_media.libs.media_api.types import UserProfile
|
||||
|
||||
|
||||
def test_load_auth_creates_file_when_missing(tmp_path, monkeypatch):
|
||||
auth_file = tmp_path / "auth.json"
|
||||
monkeypatch.setattr("viu_media.cli.service.auth.service.AUTH_FILE", auth_file)
|
||||
|
||||
service = AuthService(media_api="anilist")
|
||||
profile = service.get_auth()
|
||||
|
||||
assert profile is None
|
||||
assert auth_file.exists()
|
||||
|
||||
|
||||
def test_save_and_get_auth_roundtrip(tmp_path, monkeypatch):
|
||||
auth_file = tmp_path / "auth.json"
|
||||
monkeypatch.setattr("viu_media.cli.service.auth.service.AUTH_FILE", auth_file)
|
||||
|
||||
service = AuthService(media_api="anilist")
|
||||
user = UserProfile(id=1, name="test-user", avatar_url="https://img/avatar.png")
|
||||
|
||||
service.save_user_profile(user, "token-abc")
|
||||
auth = service.get_auth()
|
||||
|
||||
assert auth is not None
|
||||
assert auth.token == "token-abc"
|
||||
assert auth.user_profile.name == "test-user"
|
||||
|
||||
|
||||
def test_clear_user_profile_deletes_auth_file(tmp_path, monkeypatch):
|
||||
auth_file = tmp_path / "auth.json"
|
||||
monkeypatch.setattr("viu_media.cli.service.auth.service.AUTH_FILE", auth_file)
|
||||
|
||||
service = AuthService(media_api="anilist")
|
||||
user = UserProfile(id=2, name="clear-me")
|
||||
service.save_user_profile(user, "token")
|
||||
assert auth_file.exists()
|
||||
|
||||
service.clear_user_profile()
|
||||
|
||||
assert not auth_file.exists()
|
||||
@@ -0,0 +1,76 @@
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from viu_media.core.utils.file import AtomicWriter, FileLock, check_file_modified, sanitize_filename
|
||||
|
||||
|
||||
def test_atomic_writer_writes_atomically(tmp_path: Path):
|
||||
target = tmp_path / "data.txt"
|
||||
|
||||
with AtomicWriter(target, mode="w", encoding="utf-8") as handle:
|
||||
handle.write("hello world")
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "hello world"
|
||||
|
||||
|
||||
def test_atomic_writer_cleans_temp_file_on_exception(tmp_path: Path):
|
||||
target = tmp_path / "data.txt"
|
||||
target.write_text("original", encoding="utf-8")
|
||||
|
||||
with pytest.raises(RuntimeError, match="boom"):
|
||||
with AtomicWriter(target, mode="w", encoding="utf-8") as handle:
|
||||
handle.write("new")
|
||||
raise RuntimeError("boom")
|
||||
|
||||
assert target.read_text(encoding="utf-8") == "original"
|
||||
assert list(tmp_path.glob("*.tmp")) == []
|
||||
|
||||
|
||||
def test_file_lock_acquire_release_non_blocking(tmp_path: Path):
|
||||
lock_path = tmp_path / "my.lock"
|
||||
lock = FileLock(lock_path, timeout=0, stale_timeout=10)
|
||||
|
||||
lock.acquire()
|
||||
assert lock_path.exists()
|
||||
|
||||
lock.release()
|
||||
assert not lock_path.exists()
|
||||
|
||||
|
||||
def test_file_lock_breaks_stale_lock(tmp_path: Path):
|
||||
lock_path = tmp_path / "stale.lock"
|
||||
lock_path.write_text("999999\n0", encoding="utf-8")
|
||||
stale_timestamp = time.time() - 100
|
||||
os.utime(lock_path, (stale_timestamp, stale_timestamp))
|
||||
|
||||
lock = FileLock(lock_path, timeout=0.5, stale_timeout=0.01)
|
||||
lock.acquire()
|
||||
|
||||
assert lock_path.exists()
|
||||
lock.release()
|
||||
assert not lock_path.exists()
|
||||
|
||||
|
||||
def test_check_file_modified_detects_changes(tmp_path: Path):
|
||||
target = tmp_path / "track.txt"
|
||||
target.write_text("a", encoding="utf-8")
|
||||
previous_mtime = target.stat().st_mtime
|
||||
|
||||
time.sleep(0.01)
|
||||
target.write_text("b", encoding="utf-8")
|
||||
|
||||
current_mtime, modified = check_file_modified(target, previous_mtime)
|
||||
|
||||
assert modified is True
|
||||
assert current_mtime > previous_mtime
|
||||
|
||||
|
||||
def test_sanitize_filename_removes_invalid_characters():
|
||||
result = sanitize_filename('My:/Invalid*"Name?', restricted=False)
|
||||
|
||||
assert "?" not in result
|
||||
assert "*" not in result
|
||||
assert result
|
||||
@@ -0,0 +1,56 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import httpx
|
||||
|
||||
from viu_media.core.utils.networking import get_remote_filename, random_user_agent
|
||||
|
||||
|
||||
def _make_response(url: str, headers: dict[str, str] | None = None) -> httpx.Response:
|
||||
request = httpx.Request("GET", url)
|
||||
return httpx.Response(200, headers=headers, request=request)
|
||||
|
||||
|
||||
def test_random_user_agent_uses_selected_chrome_version():
|
||||
with patch("viu_media.core.utils.networking.random.choice", return_value="97.0.4692.20"):
|
||||
user_agent = random_user_agent()
|
||||
|
||||
assert "Chrome/97.0.4692.20" in user_agent
|
||||
assert user_agent.startswith("Mozilla/5.0")
|
||||
|
||||
|
||||
def test_get_remote_filename_prefers_filename_star_header():
|
||||
response = _make_response(
|
||||
"https://example.com/download",
|
||||
{"Content-Disposition": "attachment; filename*=UTF-8''my%20anime%20file.mkv"},
|
||||
)
|
||||
|
||||
filename = get_remote_filename(response)
|
||||
|
||||
assert filename == "my anime file.mkv"
|
||||
|
||||
|
||||
def test_get_remote_filename_uses_regular_filename_header():
|
||||
response = _make_response(
|
||||
"https://example.com/download",
|
||||
{"Content-Disposition": 'attachment; filename="episode-01.mp4"'},
|
||||
)
|
||||
|
||||
filename = get_remote_filename(response)
|
||||
|
||||
assert filename == "episode-01.mp4"
|
||||
|
||||
|
||||
def test_get_remote_filename_falls_back_to_url_path():
|
||||
response = _make_response("https://example.com/files/ep%2001.mp4?token=abc")
|
||||
|
||||
filename = get_remote_filename(response)
|
||||
|
||||
assert filename == "ep 01.mp4"
|
||||
|
||||
|
||||
def test_get_remote_filename_returns_none_when_no_candidate_found():
|
||||
response = _make_response("https://example.com/")
|
||||
|
||||
filename = get_remote_filename(response)
|
||||
|
||||
assert filename is None
|
||||
@@ -0,0 +1,187 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from httpx import Client
|
||||
|
||||
from viu_media.core.config import AnilistConfig
|
||||
from viu_media.libs.media_api.anilist.api import AniListApi
|
||||
from viu_media.libs.media_api.params import MediaSearchParams, UpdateUserMediaListEntryParams
|
||||
from viu_media.libs.media_api.types import MediaGenre, MediaStatus, UserMediaListStatus
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(self, payload, text=""):
|
||||
self._payload = payload
|
||||
self.text = text
|
||||
|
||||
def json(self):
|
||||
return self._payload
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api_client():
|
||||
api = AniListApi(AnilistConfig(per_page=7), Client())
|
||||
try:
|
||||
yield api
|
||||
finally:
|
||||
api.http_client.close()
|
||||
|
||||
|
||||
def test_authenticate_sets_profile_and_authorization_header(api_client):
|
||||
viewer_payload = {
|
||||
"data": {
|
||||
"Viewer": {
|
||||
"id": 1,
|
||||
"name": "ash",
|
||||
"avatar": {"large": "https://img/avatar.jpg"},
|
||||
"bannerImage": "https://img/banner.jpg",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
with patch(
|
||||
"viu_media.libs.media_api.anilist.api.execute_graphql",
|
||||
return_value=FakeResponse(viewer_payload),
|
||||
):
|
||||
profile = api_client.authenticate("token-123")
|
||||
|
||||
assert profile is not None
|
||||
assert profile.name == "ash"
|
||||
assert api_client.http_client.headers["Authorization"] == "Bearer token-123"
|
||||
assert api_client.is_authenticated() is True
|
||||
|
||||
|
||||
def test_authenticate_clears_token_when_profile_fetch_fails(api_client):
|
||||
with patch.object(api_client, "get_viewer_profile", return_value=None):
|
||||
profile = api_client.authenticate("bad-token")
|
||||
|
||||
assert profile is None
|
||||
assert api_client.token is None
|
||||
assert "Authorization" not in api_client.http_client.headers
|
||||
assert api_client.is_authenticated() is False
|
||||
|
||||
|
||||
def test_get_viewer_profile_without_token_returns_none(api_client):
|
||||
with patch("viu_media.libs.media_api.anilist.api.execute_graphql") as mock_execute:
|
||||
profile = api_client.get_viewer_profile()
|
||||
|
||||
assert profile is None
|
||||
mock_execute.assert_not_called()
|
||||
|
||||
|
||||
def test_search_media_builds_expected_graphql_variables(api_client):
|
||||
params = MediaSearchParams(
|
||||
query="naruto",
|
||||
page=2,
|
||||
status=MediaStatus.RELEASING,
|
||||
genre_in=[MediaGenre.ACTION],
|
||||
)
|
||||
|
||||
with (
|
||||
patch("viu_media.libs.media_api.anilist.api.execute_graphql") as mock_execute,
|
||||
patch(
|
||||
"viu_media.libs.media_api.anilist.api.mapper.to_generic_search_result",
|
||||
return_value="mapped",
|
||||
),
|
||||
):
|
||||
mock_execute.return_value = FakeResponse({"data": {"Page": {}}})
|
||||
result = api_client.search_media(params)
|
||||
|
||||
assert result == "mapped"
|
||||
called_variables = mock_execute.call_args.args[3]
|
||||
assert called_variables["query"] == "naruto"
|
||||
assert called_variables["page"] == 2
|
||||
assert called_variables["status"] == "RELEASING"
|
||||
assert called_variables["genre_in"] == ["Action"]
|
||||
assert called_variables["genre_not_in"] == ["Hentai"]
|
||||
assert called_variables["type"] == "ANIME"
|
||||
assert called_variables["per_page"] == 7
|
||||
|
||||
|
||||
def test_update_list_entry_requires_authentication(api_client):
|
||||
params = UpdateUserMediaListEntryParams(media_id=99)
|
||||
|
||||
with patch("viu_media.libs.media_api.anilist.api.execute_graphql") as mock_execute:
|
||||
result = api_client.update_list_entry(params)
|
||||
|
||||
assert result is False
|
||||
mock_execute.assert_not_called()
|
||||
|
||||
|
||||
def test_update_list_entry_maps_score_and_progress_and_returns_true(api_client):
|
||||
api_client.token = "token"
|
||||
params = UpdateUserMediaListEntryParams(
|
||||
media_id=10,
|
||||
status=UserMediaListStatus.WATCHING,
|
||||
progress="3.0",
|
||||
score=8.5,
|
||||
)
|
||||
|
||||
with patch(
|
||||
"viu_media.libs.media_api.anilist.api.execute_graphql",
|
||||
return_value=FakeResponse({"data": {"SaveMediaListEntry": {"id": 10}}}),
|
||||
) as mock_execute:
|
||||
result = api_client.update_list_entry(params)
|
||||
|
||||
assert result is True
|
||||
called_variables = mock_execute.call_args.args[3]
|
||||
assert called_variables == {
|
||||
"mediaId": 10,
|
||||
"status": "CURRENT",
|
||||
"progress": 3,
|
||||
"scoreRaw": 85,
|
||||
}
|
||||
|
||||
|
||||
def test_update_list_entry_returns_false_on_api_errors(api_client):
|
||||
api_client.token = "token"
|
||||
params = UpdateUserMediaListEntryParams(media_id=10)
|
||||
|
||||
with patch(
|
||||
"viu_media.libs.media_api.anilist.api.execute_graphql",
|
||||
return_value=FakeResponse({"errors": [{"message": "boom"}]}),
|
||||
):
|
||||
result = api_client.update_list_entry(params)
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_delete_list_entry_requires_authentication(api_client):
|
||||
with patch("viu_media.libs.media_api.anilist.api.execute_graphql") as mock_execute:
|
||||
result = api_client.delete_list_entry(11)
|
||||
|
||||
assert result is False
|
||||
mock_execute.assert_not_called()
|
||||
|
||||
|
||||
def test_delete_list_entry_returns_false_when_item_not_found(api_client):
|
||||
api_client.token = "token"
|
||||
with patch(
|
||||
"viu_media.libs.media_api.anilist.api.execute_graphql",
|
||||
return_value=FakeResponse({"data": {"MediaList": None}}),
|
||||
) as mock_execute:
|
||||
result = api_client.delete_list_entry(11)
|
||||
|
||||
assert result is False
|
||||
assert mock_execute.call_count == 1
|
||||
|
||||
|
||||
def test_delete_list_entry_deletes_when_list_item_exists(api_client):
|
||||
api_client.token = "token"
|
||||
responses = [
|
||||
FakeResponse({"data": {"MediaList": {"id": 321}}}),
|
||||
FakeResponse({"data": {"DeleteMediaListEntry": {"deleted": True}}}),
|
||||
]
|
||||
|
||||
with patch(
|
||||
"viu_media.libs.media_api.anilist.api.execute_graphql",
|
||||
side_effect=responses,
|
||||
) as mock_execute:
|
||||
result = api_client.delete_list_entry(22)
|
||||
|
||||
assert result is True
|
||||
assert mock_execute.call_count == 2
|
||||
first_vars = mock_execute.call_args_list[0].args[3]
|
||||
second_vars = mock_execute.call_args_list[1].args[3]
|
||||
assert first_vars == {"mediaId": 22}
|
||||
assert second_vars == {"id": 321}
|
||||
@@ -0,0 +1,269 @@
|
||||
from datetime import datetime
|
||||
from typing import Any, cast
|
||||
|
||||
from viu_media.libs.media_api.anilist.mapper import (
|
||||
_to_generic_streaming_episodes,
|
||||
_to_generic_user_status,
|
||||
to_generic_airing_schedule_result,
|
||||
to_generic_characters_result,
|
||||
to_generic_notifications,
|
||||
to_generic_recommendations,
|
||||
to_generic_reviews_list,
|
||||
to_generic_search_result,
|
||||
)
|
||||
from viu_media.libs.media_api.types import UserMediaListStatus
|
||||
|
||||
|
||||
def _build_media_item(media_id: int = 1) -> dict:
|
||||
return {
|
||||
"id": media_id,
|
||||
"idMal": None,
|
||||
"type": "ANIME",
|
||||
"title": {"english": f"Title {media_id}", "romaji": "Romaji", "native": "Native"},
|
||||
"status": "RELEASING",
|
||||
"format": "TV",
|
||||
"coverImage": {"large": "https://img/large.jpg", "medium": "https://img/med.jpg"},
|
||||
"bannerImage": None,
|
||||
"trailer": None,
|
||||
"description": "description",
|
||||
"episodes": 12,
|
||||
"duration": 24,
|
||||
"genres": ["Action"],
|
||||
"tags": [{"name": "CGI", "rank": 88}],
|
||||
"studios": {
|
||||
"nodes": [
|
||||
{
|
||||
"name": "Studio One",
|
||||
"favourites": 111,
|
||||
"isAnimationStudio": True,
|
||||
}
|
||||
]
|
||||
},
|
||||
"synonyms": ["Alt title"],
|
||||
"averageScore": 80,
|
||||
"popularity": 9000,
|
||||
"favourites": 77,
|
||||
"nextAiringEpisode": {"airingAt": 1700000000, "episode": 5, "timeUntilAiring": 123},
|
||||
"startDate": {"year": 2024, "month": 1, "day": 1},
|
||||
"endDate": {"year": 2024, "month": 3, "day": 1},
|
||||
"streamingEpisodes": [
|
||||
{"title": "Episode 10 - Main", "thumbnail": "thumb1"},
|
||||
{"title": "Episode 10.5 - Special", "thumbnail": "thumb2"},
|
||||
],
|
||||
"mediaListEntry": {"id": 9, "status": "CURRENT", "progress": 2},
|
||||
}
|
||||
|
||||
|
||||
def test_to_generic_streaming_episodes_renumbers_titles():
|
||||
episodes = [
|
||||
{"title": "Episode 10 - Main", "thumbnail": "thumb1"},
|
||||
{"title": "Episode 10.5 - Special", "thumbnail": "thumb2"},
|
||||
]
|
||||
|
||||
result = _to_generic_streaming_episodes(cast(Any, episodes))
|
||||
|
||||
assert result["1"].title == "Episode 1 - Main"
|
||||
assert result["1.5"].title == "Episode 1.5 - Special"
|
||||
assert result["1"].thumbnail == "thumb1"
|
||||
|
||||
|
||||
def test_to_generic_user_status_with_explicit_list_entry():
|
||||
media = {"mediaListEntry": {"status": "CURRENT"}}
|
||||
list_entry = {
|
||||
"progress": 12,
|
||||
"score": 9.0,
|
||||
"repeat": 1,
|
||||
"notes": "finished",
|
||||
"startDate": {"year": 2024, "month": 1, "day": 2},
|
||||
"completedAt": {"year": 2024, "month": 4, "day": 2},
|
||||
"createdAt": 1700000000,
|
||||
}
|
||||
|
||||
result = _to_generic_user_status(cast(Any, media), cast(Any, list_entry))
|
||||
|
||||
assert result is not None
|
||||
assert result.status == UserMediaListStatus.WATCHING
|
||||
assert result.progress == 12
|
||||
assert result.created_at == "1700000000"
|
||||
|
||||
|
||||
def test_to_generic_user_status_uses_media_entry_when_list_entry_is_none():
|
||||
media = {"mediaListEntry": {"id": 55, "status": "CURRENT", "progress": 7}}
|
||||
|
||||
result = _to_generic_user_status(cast(Any, media), None)
|
||||
|
||||
assert result is not None
|
||||
assert result.id == 55
|
||||
assert result.status == UserMediaListStatus.WATCHING
|
||||
assert result.progress == 7
|
||||
|
||||
|
||||
def test_to_generic_search_result_maps_page_and_media():
|
||||
payload = {
|
||||
"data": {
|
||||
"Page": {
|
||||
"media": [_build_media_item(1)],
|
||||
"pageInfo": {
|
||||
"total": 1,
|
||||
"currentPage": 1,
|
||||
"hasNextPage": False,
|
||||
"perPage": 10,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result = to_generic_search_result(cast(Any, payload))
|
||||
|
||||
assert result is not None
|
||||
assert result.page_info.total == 1
|
||||
assert len(result.media) == 1
|
||||
assert result.media[0].id == 1
|
||||
assert result.media[0].streaming_episodes["1"].title == "Episode 1 - Main"
|
||||
|
||||
|
||||
def test_to_generic_recommendations_skips_invalid_recommendation_item():
|
||||
good_media = _build_media_item(2)
|
||||
bad_media = {"id": 999, "title": {"english": "broken", "romaji": "broken"}}
|
||||
payload = {
|
||||
"data": {
|
||||
"Page": {
|
||||
"recommendations": [
|
||||
{"media": good_media},
|
||||
{"media": bad_media},
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result = to_generic_recommendations(payload)
|
||||
|
||||
assert result is not None
|
||||
assert len(result) == 1
|
||||
assert result[0].id == 2
|
||||
|
||||
|
||||
def test_to_generic_reviews_list_handles_invalid_and_empty_data():
|
||||
assert to_generic_reviews_list(cast(Any, {})) is None
|
||||
|
||||
empty_reviews = {"data": {"Page": {"reviews": []}}}
|
||||
assert to_generic_reviews_list(cast(Any, empty_reviews)) == []
|
||||
|
||||
|
||||
def test_to_generic_characters_result_maps_valid_payload():
|
||||
payload = {
|
||||
"data": {
|
||||
"Page": {
|
||||
"media": [
|
||||
{
|
||||
"characters": {
|
||||
"nodes": [
|
||||
{
|
||||
"id": 1,
|
||||
"name": {"full": "Naruto Uzumaki"},
|
||||
"image": {
|
||||
"medium": "https://img/char-med.jpg",
|
||||
"large": "https://img/char-large.jpg",
|
||||
},
|
||||
"description": "Main character",
|
||||
"gender": "Male",
|
||||
"age": "17",
|
||||
"bloodType": "B",
|
||||
"favourites": 100,
|
||||
"dateOfBirth": {
|
||||
"year": 2000,
|
||||
"month": 10,
|
||||
"day": 10,
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result = to_generic_characters_result(payload)
|
||||
|
||||
assert result is not None
|
||||
assert len(result.characters) == 1
|
||||
assert result.characters[0].name.full == "Naruto Uzumaki"
|
||||
assert result.characters[0].date_of_birth == datetime(2000, 10, 10)
|
||||
|
||||
|
||||
def test_to_generic_characters_result_returns_none_on_malformed_data():
|
||||
payload = {"data": {"Page": {"media": []}}}
|
||||
|
||||
result = to_generic_characters_result(payload)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_to_generic_airing_schedule_result_handles_invalid_timestamps():
|
||||
payload = {
|
||||
"data": {
|
||||
"Page": {
|
||||
"media": [
|
||||
{
|
||||
"airingSchedule": {
|
||||
"nodes": [
|
||||
{"episode": 1, "airingAt": 1700000000, "timeUntilAiring": 10},
|
||||
{"episode": 2, "airingAt": "invalid", "timeUntilAiring": 20},
|
||||
]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result = to_generic_airing_schedule_result(payload)
|
||||
|
||||
assert result is not None
|
||||
assert len(result.schedule_items) == 2
|
||||
assert result.schedule_items[0].airing_at is not None
|
||||
assert result.schedule_items[1].airing_at is None
|
||||
|
||||
|
||||
def test_to_generic_notifications_handles_empty_invalid_and_valid_payloads():
|
||||
assert to_generic_notifications(cast(Any, {})) is None
|
||||
|
||||
empty_payload = {"data": {"Page": {"notifications": []}}}
|
||||
assert to_generic_notifications(cast(Any, empty_payload)) == []
|
||||
|
||||
valid_payload = {
|
||||
"data": {
|
||||
"Page": {
|
||||
"notifications": [
|
||||
{
|
||||
"id": 44,
|
||||
"type": "AIRING",
|
||||
"episode": 7,
|
||||
"contexts": ["Aired"],
|
||||
"createdAt": 1700000000,
|
||||
"media": {
|
||||
"id": 12,
|
||||
"idMal": None,
|
||||
"title": {
|
||||
"english": "Title",
|
||||
"romaji": "Romaji",
|
||||
"native": "Native",
|
||||
},
|
||||
"coverImage": {
|
||||
"large": "https://img/cover.jpg",
|
||||
"medium": "https://img/cover-med.jpg",
|
||||
},
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result = to_generic_notifications(cast(Any, valid_payload))
|
||||
|
||||
assert result is not None
|
||||
assert len(result) == 1
|
||||
assert result[0].id == 44
|
||||
assert result[0].media.id == 12
|
||||
@@ -0,0 +1,58 @@
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from viu_media.core.config import AppConfig
|
||||
from viu_media.libs.media_api import api as api_module
|
||||
from viu_media.libs.media_api.api import create_api_client
|
||||
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self, scoped_config, http_client):
|
||||
self.scoped_config = scoped_config
|
||||
self.http_client = http_client
|
||||
|
||||
|
||||
def test_create_api_client_unsupported_client_raises_value_error():
|
||||
config = AppConfig()
|
||||
|
||||
with pytest.raises(ValueError, match="Unsupported API client"):
|
||||
create_api_client("unknown", config)
|
||||
|
||||
|
||||
def test_create_api_client_returns_instance_with_scoped_config(monkeypatch):
|
||||
config = AppConfig()
|
||||
monkeypatch.setattr(
|
||||
api_module,
|
||||
"API_CLIENTS",
|
||||
{"dummy": ("dummy.module.FakeClient", "anilist")},
|
||||
)
|
||||
|
||||
fake_module = SimpleNamespace(FakeClient=FakeClient)
|
||||
|
||||
monkeypatch.setattr(api_module.importlib, "import_module", lambda _name: fake_module)
|
||||
monkeypatch.setattr(api_module, "random_user_agent", lambda: "test-agent")
|
||||
|
||||
client = create_api_client("dummy", config)
|
||||
|
||||
assert isinstance(client, FakeClient)
|
||||
assert client.scoped_config == config.anilist
|
||||
assert client.http_client.headers["User-Agent"] == "test-agent"
|
||||
client.http_client.close()
|
||||
|
||||
|
||||
def test_create_api_client_import_failure_raises_import_error(monkeypatch):
|
||||
config = AppConfig()
|
||||
monkeypatch.setattr(
|
||||
api_module,
|
||||
"API_CLIENTS",
|
||||
{"dummy": ("dummy.module.FakeClient", "anilist")},
|
||||
)
|
||||
|
||||
with patch(
|
||||
"viu_media.libs.media_api.api.importlib.import_module",
|
||||
side_effect=ImportError("boom"),
|
||||
):
|
||||
with pytest.raises(ImportError, match="Could not load API client 'dummy'"):
|
||||
create_api_client("dummy", config)
|
||||
@@ -0,0 +1,130 @@
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from viu_media.core.config import MpvConfig
|
||||
from viu_media.core.exceptions import ViuError
|
||||
from viu_media.libs.player.mpv.player import MpvPlayer
|
||||
from viu_media.libs.player.params import PlayerParams
|
||||
|
||||
|
||||
def _make_config(args: str = "", pre_args: str = ""):
|
||||
return MpvConfig(args=args, pre_args=pre_args)
|
||||
|
||||
|
||||
def _make_params(url: str, **overrides):
|
||||
base = {
|
||||
"url": url,
|
||||
"title": "Episode Title",
|
||||
"query": "query",
|
||||
"episode": "1",
|
||||
"syncplay": False,
|
||||
"subtitles": None,
|
||||
"headers": None,
|
||||
"start_time": None,
|
||||
}
|
||||
base.update(overrides)
|
||||
return PlayerParams(**base)
|
||||
|
||||
|
||||
def test_create_mpv_cli_options_builds_all_supported_flags():
|
||||
with patch("viu_media.libs.player.mpv.player.shutil.which", return_value="mpv"):
|
||||
player = MpvPlayer(_make_config(args="--force-window=yes,--quiet"))
|
||||
|
||||
params = _make_params(
|
||||
"https://example.com/video.m3u8",
|
||||
headers={"Referer": "https://example.com", "User-Agent": "ua"},
|
||||
subtitles=["sub1.srt", "sub2.srt"],
|
||||
start_time="00:01:05",
|
||||
title="My Show",
|
||||
)
|
||||
|
||||
options = player._create_mpv_cli_options(params)
|
||||
|
||||
assert "--http-header-fields=Referer:https://example.com,User-Agent:ua" in options
|
||||
assert "--sub-file=sub1.srt" in options
|
||||
assert "--sub-file=sub2.srt" in options
|
||||
assert "--start=00:01:05" in options
|
||||
assert "--title=My Show" in options
|
||||
assert "--force-window=yes" in options
|
||||
assert "--quiet" in options
|
||||
|
||||
|
||||
def test_play_on_desktop_raises_when_mpv_missing():
|
||||
with patch("viu_media.libs.player.mpv.player.shutil.which", return_value=None):
|
||||
player = MpvPlayer(_make_config())
|
||||
|
||||
params = _make_params("https://example.com/video.mp4")
|
||||
with pytest.raises(ViuError, match="MPV executable not found"):
|
||||
player._play_on_desktop(params)
|
||||
|
||||
|
||||
def test_play_on_mobile_uses_youtube_intent_for_youtube_urls():
|
||||
with patch("viu_media.libs.player.mpv.player.shutil.which", return_value="mpv"):
|
||||
player = MpvPlayer(_make_config())
|
||||
|
||||
params = _make_params("https://youtu.be/abc123")
|
||||
with patch("viu_media.libs.player.mpv.player.subprocess.run") as mock_run:
|
||||
result = player._play_on_mobile(params)
|
||||
|
||||
args = mock_run.call_args.args[0]
|
||||
assert "com.google.android.youtube/.UrlActivity" in args
|
||||
assert result.episode == "1"
|
||||
|
||||
|
||||
def test_play_on_mobile_uses_mpv_activity_for_non_youtube_urls():
|
||||
with patch("viu_media.libs.player.mpv.player.shutil.which", return_value="mpv"):
|
||||
player = MpvPlayer(_make_config())
|
||||
|
||||
params = _make_params("https://example.com/video.mp4")
|
||||
with patch("viu_media.libs.player.mpv.player.subprocess.run") as mock_run:
|
||||
player._play_on_mobile(params)
|
||||
|
||||
args = mock_run.call_args.args[0]
|
||||
assert "is.xyz.mpv/.MPVActivity" in args
|
||||
|
||||
|
||||
def test_stream_on_desktop_with_subprocess_parses_av_time():
|
||||
with patch("viu_media.libs.player.mpv.player.shutil.which", return_value="mpv"):
|
||||
player = MpvPlayer(_make_config(pre_args="nohup"))
|
||||
|
||||
params = _make_params("https://example.com/video.m3u8")
|
||||
process_output = "line1\nAV: 00:10:03 / 00:24:00 (41%)\n"
|
||||
fake_proc = SimpleNamespace(stdout=process_output)
|
||||
|
||||
with patch(
|
||||
"viu_media.libs.player.mpv.player.subprocess.run",
|
||||
return_value=fake_proc,
|
||||
) as mock_run:
|
||||
result = player._stream_on_desktop_with_subprocess(params)
|
||||
|
||||
assert result.stop_time == "00:10:03"
|
||||
assert result.total_time == "00:24:00"
|
||||
assert mock_run.call_args.kwargs["capture_output"] is True
|
||||
|
||||
|
||||
def test_play_rejects_torrent_on_termux():
|
||||
with patch("viu_media.libs.player.mpv.player.shutil.which", return_value="mpv"):
|
||||
player = MpvPlayer(_make_config())
|
||||
|
||||
params = _make_params("magnet:?xt=urn:btih:1234567890123456789012345678901234567890")
|
||||
with patch(
|
||||
"viu_media.libs.player.mpv.player.detect.is_running_in_termux",
|
||||
return_value=True,
|
||||
):
|
||||
with pytest.raises(ViuError, match="Unable to play torrents on termux"):
|
||||
player.play(params)
|
||||
|
||||
|
||||
def test_play_rejects_syncplay_on_termux():
|
||||
with patch("viu_media.libs.player.mpv.player.shutil.which", return_value="mpv"):
|
||||
player = MpvPlayer(_make_config())
|
||||
|
||||
params = _make_params("https://example.com/video.mp4", syncplay=True)
|
||||
with patch(
|
||||
"viu_media.libs.player.mpv.player.detect.is_running_in_termux",
|
||||
return_value=True,
|
||||
):
|
||||
with pytest.raises(ViuError, match="Unable to play with syncplay on termux"):
|
||||
player.play(params)
|
||||
@@ -0,0 +1,53 @@
|
||||
from typing import Any, cast
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from viu_media.core.config import AppConfig
|
||||
from viu_media.libs.selectors.selector import SelectorFactory
|
||||
|
||||
|
||||
def _config_with_selector(selector_name: str) -> AppConfig:
|
||||
config = AppConfig()
|
||||
config.general.selector = cast(Any, selector_name)
|
||||
return config
|
||||
|
||||
|
||||
def test_selector_factory_creates_fzf_selector():
|
||||
config = _config_with_selector("fzf")
|
||||
|
||||
with patch("viu_media.libs.selectors.fzf.FzfSelector", return_value="fzf-selector"):
|
||||
result = SelectorFactory.create(config)
|
||||
|
||||
assert result == "fzf-selector"
|
||||
|
||||
|
||||
def test_selector_factory_creates_rofi_selector():
|
||||
config = _config_with_selector("rofi")
|
||||
|
||||
with patch(
|
||||
"viu_media.libs.selectors.rofi.RofiSelector",
|
||||
return_value="rofi-selector",
|
||||
):
|
||||
result = SelectorFactory.create(config)
|
||||
|
||||
assert result == "rofi-selector"
|
||||
|
||||
|
||||
def test_selector_factory_creates_default_inquirer_selector():
|
||||
config = _config_with_selector("default")
|
||||
|
||||
with patch(
|
||||
"viu_media.libs.selectors.inquirer.InquirerSelector",
|
||||
return_value="inquirer-selector",
|
||||
):
|
||||
result = SelectorFactory.create(config)
|
||||
|
||||
assert result == "inquirer-selector"
|
||||
|
||||
|
||||
def test_selector_factory_raises_for_unsupported_selector():
|
||||
config = _config_with_selector("unknown")
|
||||
|
||||
with pytest.raises(ValueError, match="Unsupported selector"):
|
||||
SelectorFactory.create(config)
|
||||
@@ -196,11 +196,8 @@ class AniListApi(BaseApiClient):
|
||||
)
|
||||
entry_data = response.json()
|
||||
|
||||
list_id = (
|
||||
entry_data.get("data", {}).get("MediaList", {}).get("id")
|
||||
if entry_data
|
||||
else None
|
||||
)
|
||||
media_list = entry_data.get("data", {}).get("MediaList") if entry_data else None
|
||||
list_id = media_list.get("id") if isinstance(media_list, dict) else None
|
||||
if not list_id:
|
||||
return False
|
||||
response = execute_graphql(
|
||||
|
||||
Reference in New Issue
Block a user