mirrors/streamrip
1
0
Fork 0
mirror of https://github.com/nathom/streamrip.git synced 2025-05-09 14:11:55 -04:00
Code Issues Projects Releases Packages Wiki Activity

Fix format and lint checks

Browse source
This commit is contained in:
Nathan Thomas 2025-03-10 08:46:44 -07:00
parent 7697ae7b6e
commit d0dfab82ab
4 changed files with 56 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

6
streamrip/media/album.py
View file

@ -40,8 +40,10 @@ class Album(Media):
except Exception as e:
logger.error(f"Error downloading track: {e}")
results = await asyncio.gather(*[_resolve_and_download(p) for p in self.tracks], return_exceptions=True)
results = await asyncio.gather(
*[_resolve_and_download(p) for p in self.tracks], return_exceptions=True
)
for result in results:
if isinstance(result, Exception):
logger.error(f"Album track processing error: {result}")

4
streamrip/media/playlist.py
View file

@ -133,10 +133,10 @@ class Playlist(Media):
[_resolve_download(track) for track in self.tracks],
track_resolve_chunk_size,
)
for batch in batches:
results = await asyncio.gather(*batch, return_exceptions=True)
for result in results:
if isinstance(result, Exception):
logger.error(f"Batch processing error: {result}")

15
streamrip/rip/main.py
View file

@ -28,7 +28,8 @@ from .prompter import get_prompter
logger = logging.getLogger("streamrip")
if platform.system() == "Windows":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
class Main:
"""Provides all of the functionality called into by the CLI.
@ -164,17 +165,21 @@ class Main:
async def rip(self):
"""Download all resolved items."""
results = await asyncio.gather(*[item.rip() for item in self.media], return_exceptions=True)
results = await asyncio.gather(
*[item.rip() for item in self.media], return_exceptions=True
)
failed_items = 0
for result in results:
if isinstance(result, Exception):
logger.error(f"Error processing media item: {result}")
failed_items += 1
if failed_items > 0:
total_items = len(self.media)
logger.info(f"Download completed with {failed_items} failed items out of {total_items} total items.")
logger.info(
f"Download completed with {failed_items} failed items out of {total_items} total items."
)
async def search_interactive(self, source: str, media_type: str, query: str):
client = await self.get_logged_in_client(source)

76
tests/test_error_handling.py
View file

@ -1,11 +1,10 @@
import asyncio
import json
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from streamrip.media.playlist import Playlist, PendingPlaylistTrack
from streamrip.media.album import Album, PendingTrack
from streamrip.exceptions import NonStreamableError
import pytest
from streamrip.media.album import Album
from streamrip.media.playlist import Playlist
class TestErrorHandling:
@ -16,24 +15,25 @@ class TestErrorHandling:
"""Test that a playlist download continues even if one track fails."""
mock_config = MagicMock()
mock_client = MagicMock()
mock_db = MagicMock()
mock_track_success = MagicMock()
mock_track_success.resolve = AsyncMock(return_value=MagicMock())
mock_track_success.resolve.return_value.rip = AsyncMock()
mock_track_failure = MagicMock()
mock_track_failure.resolve = AsyncMock(side_effect=json.JSONDecodeError("Expecting value", "", 0))
mock_track_failure.resolve = AsyncMock(
side_effect=json.JSONDecodeError("Expecting value", "", 0)
)
playlist = Playlist(
name="Test Playlist",
config=mock_config,
client=mock_client,
tracks=[mock_track_success, mock_track_failure]
tracks=[mock_track_success, mock_track_failure],
)
await playlist.download()
mock_track_success.resolve.assert_called_once()
mock_track_success.resolve.return_value.rip.assert_called_once()
mock_track_failure.resolve.assert_called_once()
@ -42,29 +42,30 @@ class TestErrorHandling:
async def test_album_handles_failed_track(self):
"""Test that an album download continues even if one track fails."""
mock_config = MagicMock()
mock_client = MagicMock()
mock_db = MagicMock()
mock_meta = MagicMock()
# Create a list of mock tracks - one will succeed, one will fail
mock_track_success = MagicMock()
mock_track_success.resolve = AsyncMock(return_value=MagicMock())
mock_track_success.resolve.return_value.rip = AsyncMock()
# This track will raise a JSONDecodeError when resolved
mock_track_failure = MagicMock()
mock_track_failure.resolve = AsyncMock(side_effect=json.JSONDecodeError("Expecting value", "", 0))
mock_track_failure.resolve = AsyncMock(
side_effect=json.JSONDecodeError("Expecting value", "", 0)
)
album = Album(
meta=mock_meta,
config=mock_config,
tracks=[mock_track_success, mock_track_failure],
folder="/test/folder",
db=mock_db
db=mock_db,
)
await album.download()
mock_track_success.resolve.assert_called_once()
mock_track_success.resolve.return_value.rip.assert_called_once()
mock_track_failure.resolve.assert_called_once()
@ -73,29 +74,32 @@ class TestErrorHandling:
async def test_main_rip_handles_failed_media(self):
"""Test that the Main.rip method handles failed media items."""
from streamrip.rip.main import Main
mock_config = MagicMock()
mock_config.session.downloads.requests_per_minute = 0
mock_config.session.database.downloads_enabled = False
mock_config.session.database.failed_downloads_enabled = False
with patch('streamrip.rip.main.QobuzClient'), \
patch('streamrip.rip.main.TidalClient'), \
patch('streamrip.rip.main.DeezerClient'), \
patch('streamrip.rip.main.SoundcloudClient'):
with (
patch("streamrip.rip.main.QobuzClient"),
patch("streamrip.rip.main.TidalClient"),
patch("streamrip.rip.main.DeezerClient"),
patch("streamrip.rip.main.SoundcloudClient"),
):
main = Main(mock_config)
mock_media_success = MagicMock()
mock_media_success.rip = AsyncMock()
mock_media_failure = MagicMock()
mock_media_failure.rip = AsyncMock(side_effect=Exception("Media download failed"))
mock_media_failure.rip = AsyncMock(
side_effect=Exception("Media download failed")
)
main.media = [mock_media_success, mock_media_failure]
await main.rip()
mock_media_success.rip.assert_called_once()
mock_media_failure.rip.assert_called_once()
mock_media_failure.rip.assert_called_once()