Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feat/refactorize file downloading #33

Merged
merged 2 commits into from
Sep 30, 2023
Merged

Feat/refactorize file downloading #33

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c466c1f
Refactorize the tests for file downloading
kutu-dev Sep 30, 2023
de1cf6b
Refactorize media retrieval methods
kutu-dev Sep 30, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 41 additions & 59 deletions src/knuckles/media_retrieval.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
from pathlib import Path
from typing import Any

from requests import Response
from requests.models import PreparedRequest

from .api import Api
Expand All @@ -26,6 +27,25 @@ def _generate_url(self, endpoint: str, params: dict[str, Any]) -> str:
# as the prepare_url method always set it to a string.
return prepared_request.url # type: ignore [return-value]

def _download_file(
self, response: Response, file_or_directory_path: Path, directory_filename: str
) -> Path:
response.raise_for_status()

if file_or_directory_path.is_dir():
download_path = Path(
file_or_directory_path,
directory_filename,
)
else:
download_path = file_or_directory_path

with open(download_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)

return download_path

def stream(self, id: str) -> str:
"""Returns a valid url for streaming the requested song

Expand All @@ -51,33 +71,18 @@ def download(self, id: str, file_or_directory_path: Path) -> Path:
"""

response = self.api.raw_request("download", {"id": id})
response.raise_for_status()

if file_or_directory_path.is_dir():
filename = (
response.headers["Content-Disposition"].split("filename=")[1].strip()
)

# Remove leading quote char
if filename[0] == '"':
filename = filename[1:]
filename = response.headers["Content-Disposition"].split("filename=")[1].strip()

# Remove trailing quote char
if filename[-1] == '"':
filename = filename[:-1]

download_path = Path(
file_or_directory_path,
filename,
)
else:
download_path = file_or_directory_path
# Remove leading quote char
if filename[0] == '"':
filename = filename[1:]

with open(download_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
# Remove trailing quote char
if filename[-1] == '"':
filename = filename[:-1]

return download_path
return self._download_file(response, file_or_directory_path, filename)

def hls(self, id: str) -> str:
"""Returns a valid url for streaming the requested song with hls.m3u8
Expand All @@ -93,7 +98,9 @@ def hls(self, id: str) -> str:
def get_captions(self) -> None:
...

def get_cover_art(self, id: str, file_or_directory_path: Path, size: int) -> Path:
def get_cover_art(
self, id: str, file_or_directory_path: Path, size: int | None = None
) -> Path:
"""Calls the "getCoverArt" endpoint of the API.

:param id: The id of the cover art to download.
Expand All @@ -110,27 +117,14 @@ def get_cover_art(self, id: str, file_or_directory_path: Path, size: int) -> Pat
"""

response = self.api.raw_request("getCoverArt", {"id": id, "size": size})
response.raise_for_status()

if file_or_directory_path.is_dir():
file_extension = guess_extension(
response.headers["content-type"].partition(";")[0].strip()
)

filename = id + file_extension if file_extension else id

download_path = Path(
file_or_directory_path,
filename,
)
else:
download_path = file_or_directory_path
file_extension = guess_extension(
response.headers["content-type"].partition(";")[0].strip()
)

with open(download_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
filename = id + file_extension if file_extension else id

return download_path
return self._download_file(response, file_or_directory_path, filename)

def get_lyrics(self) -> None:
...
Expand All @@ -152,22 +146,10 @@ def get_avatar(self, username: str, file_or_directory_path: Path) -> Path:
response = self.api.raw_request("getAvatar", {"username": username})
response.raise_for_status()

if file_or_directory_path.is_dir():
file_extension = guess_extension(
response.headers["content-type"].partition(";")[0].strip()
)

filename = username + file_extension if file_extension else username

download_path = Path(
file_or_directory_path,
filename,
)
else:
download_path = file_or_directory_path
file_extension = guess_extension(
response.headers["content-type"].partition(";")[0].strip()
)

with open(download_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
filename = username + file_extension if file_extension else username

return download_path
return self._download_file(response, file_or_directory_path, filename)
122 changes: 41 additions & 81 deletions tests/api/test_media_retrieval.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,11 @@
from pathlib import Path
from typing import Any


from responses import Response
import responses
from knuckles import Subsonic

from tests.mocks.media_retrieval import MockDownload
from tests.mocks.media_retrieval import FileMetadata


def test_stream(subsonic: Subsonic, song: dict[str, Any]) -> None:
Expand All @@ -18,58 +18,44 @@ def test_stream(subsonic: Subsonic, song: dict[str, Any]) -> None:

@responses.activate
def test_download_with_a_given_filename(
subsonic: Subsonic,
mock_download: Response,
tmp_path: Path,
output_song_filename: str,
placeholder_data: str,
mock_download_file: MockDownload,
subsonic: Subsonic,
song: dict[str, Any],
song_content_type: str,
download_metadata: FileMetadata,
) -> None:
responses.add(
mock_download_file("download", {"id": song["id"]}, tmp_path, song_content_type)
)
responses.add(mock_download)

download_path = subsonic.media_retrieval.download(
song["id"], tmp_path / output_song_filename
song["id"], tmp_path / download_metadata.output_filename
)

# Check if the file data has been unaltered
with open(tmp_path / output_song_filename, "r") as file:
with open(tmp_path / download_metadata.output_filename, "r") as file:
assert placeholder_data == file.read()

assert download_path == tmp_path / output_song_filename
assert download_path == tmp_path / download_metadata.output_filename


@responses.activate
def test_download_without_a_given_filename(
subsonic: Subsonic,
mock_download: Response,
tmp_path: Path,
default_song_filename: str,
placeholder_data: str,
mock_download_file: MockDownload,
subsonic: Subsonic,
song: dict[str, Any],
song_content_type: str,
download_metadata: FileMetadata,
) -> None:
responses.add(
mock_download_file(
"download",
{"id": song["id"]},
tmp_path,
song_content_type,
headers={
"Content-Disposition": f'attachment; filename="{default_song_filename}"'
},
)
)
responses.add(mock_download)

download_path = subsonic.media_retrieval.download(song["id"], tmp_path)

# Check if the file data has been unaltered
with open(tmp_path / default_song_filename, "r") as file:
with open(tmp_path / download_metadata.default_filename, "r") as file:
assert placeholder_data == file.read()

assert download_path == tmp_path / default_song_filename
assert download_path == tmp_path / download_metadata.default_filename


def test_hls(subsonic: Subsonic, song: dict[str, Any]) -> None:
Expand All @@ -81,113 +67,87 @@ def test_hls(subsonic: Subsonic, song: dict[str, Any]) -> None:

@responses.activate
def test_get_cover_art_with_a_given_filename(
subsonic: Subsonic,
mock_cover_art: Response,
tmp_path: Path,
output_cover_art_filename: str,
placeholder_data: str,
mock_download_file: MockDownload,
subsonic: Subsonic,
song: dict[str, Any],
cover_art_content_type: str,
cover_art_metadata: FileMetadata,
cover_art_size: int,
) -> None:
responses.add(
mock_download_file(
"getCoverArt",
{"id": song["coverArt"], "size": cover_art_size},
tmp_path,
cover_art_content_type,
)
)
responses.add(mock_cover_art)

download_path = subsonic.media_retrieval.get_cover_art(
song["coverArt"], tmp_path / output_cover_art_filename, cover_art_size
song["coverArt"], tmp_path / cover_art_metadata.output_filename, cover_art_size
)

# Check if the file data has been unaltered
with open(tmp_path / output_cover_art_filename, "r") as file:
with open(tmp_path / cover_art_metadata.output_filename, "r") as file:
assert placeholder_data == file.read()

assert download_path == tmp_path / output_cover_art_filename
assert download_path == tmp_path / cover_art_metadata.output_filename


@responses.activate
def test_get_cover_art_without_a_given_filename(
subsonic: Subsonic,
mock_cover_art: Response,
tmp_path: Path,
default_cover_art_filename: str,
placeholder_data: str,
mock_download_file: MockDownload,
subsonic: Subsonic,
song: dict[str, Any],
cover_art_content_type: str,
cover_art_size: int,
cover_art_metadata: FileMetadata,
) -> None:
responses.add(
mock_download_file(
"getCoverArt",
{"id": song["coverArt"], "size": cover_art_size},
tmp_path,
cover_art_content_type,
)
)
responses.add(mock_cover_art)

download_path = subsonic.media_retrieval.get_cover_art(
song["coverArt"], tmp_path, cover_art_size
)

# Check if the file data has been unaltered
with open(tmp_path / default_cover_art_filename, "r") as file:
with open(tmp_path / cover_art_metadata.default_filename, "r") as file:
assert placeholder_data == file.read()

assert download_path == tmp_path / default_cover_art_filename
assert download_path == tmp_path / cover_art_metadata.default_filename


@responses.activate
def test_get_avatar_with_a_given_filename(
subsonic: Subsonic,
mock_avatar: Response,
tmp_path: Path,
output_avatar_filename: str,
placeholder_data: str,
mock_download_file: MockDownload,
subsonic: Subsonic,
username: str,
avatar_content_type: str,
avatar_metadata: FileMetadata,
) -> None:
responses.add(
mock_download_file(
"getAvatar", {"username": username}, tmp_path, avatar_content_type
)
)
responses.add(mock_avatar)

download_path = subsonic.media_retrieval.get_avatar(
username, tmp_path / output_avatar_filename
username, tmp_path / avatar_metadata.output_filename
)

# Check if the file data has been unaltered
with open(tmp_path / output_avatar_filename, "r") as file:
with open(tmp_path / avatar_metadata.output_filename, "r") as file:
assert placeholder_data == file.read()

assert download_path == tmp_path / output_avatar_filename
assert download_path == tmp_path / avatar_metadata.output_filename


@responses.activate
def test_get_avatar_without_a_given_filename(
subsonic: Subsonic,
mock_avatar: Response,
tmp_path: Path,
default_avatar_filename: str,
placeholder_data: str,
mock_download_file: MockDownload,
subsonic: Subsonic,
username: str,
avatar_content_type: str,
avatar_metadata: FileMetadata,
) -> None:
responses.add(
mock_download_file(
"getAvatar", {"username": username}, tmp_path, avatar_content_type
)
)
responses.add(mock_avatar)

download_path = subsonic.media_retrieval.get_avatar(username, tmp_path)

# Check if the file data has been unaltered
with open(tmp_path / default_avatar_filename, "r") as file:
with open(tmp_path / avatar_metadata.default_filename, "r") as file:
assert placeholder_data == file.read()

assert download_path == tmp_path / default_avatar_filename
assert download_path == tmp_path / avatar_metadata.default_filename
Loading