diff --git a/codebeaver.yml b/codebeaver.yml new file mode 100644 index 0000000..68c1df6 --- /dev/null +++ b/codebeaver.yml @@ -0,0 +1,2 @@ +from:python-pytest-poetry +# This file was generated automatically by CodeBeaver based on your repository. Learn how to customize it here: https://docs.codebeaver.ai/configuration \ No newline at end of file diff --git a/tests/test_deezer.py b/tests/test_deezer.py new file mode 100644 index 0000000..b5aceba --- /dev/null +++ b/tests/test_deezer.py @@ -0,0 +1,285 @@ +import pytest +from deezer_downloader import deezer +import re +import json +import struct +import urllib.parse +import html.parser +import requests +import io +from binascii import a2b_hex, b2a_hex +from Crypto.Cipher import Blowfish +from Crypto.Hash import MD5 + +# No additional imports needed beyond what already exist in the test file. +# No additional imports needed beyond what already exist in the test file. +def test_calcbfkey_known_value(): + """ + Test that calcbfkey returns the expected Blowfish decryption key + for a known song id ('123456'). This increases coverage for the crypto helper function. + """ + # The MD5 hash of "123456" is "e10adc3949ba59abbe56e057f20f883e". + # Then, calcbfkey computes the key by XORing each character of the first 16 characters + # with the corresponding character in the last 16 characters and the static key b"g4el58wc0zvf9na1". + # The expected result from these operations is " +def test_get_song_infos_not_logged_in(monkeypatch): + """ + Test that get_song_infos_from_deezer_website raises Deezer403Exception + when the response from Deezer does not contain the expected "MD5_ORIGIN" + (simulating a situation where we are not logged in). + """ + # Define a dummy response that does not contain "MD5_ORIGIN" + class DummyResponse: + def __init__(self, status_code, text): + self.status_code = status_code + self.text = text + def raise_for_status(self): + # Assume response is OK so do nothing + pass + # Define a dummy session where get() returns our dummy response + class DummySession: + def get(self, url): + # Returning a page without "MD5_ORIGIN" + return DummyResponse(200, "Not logged in") + # Monkeypatch the global session in the deezer module + monkeypatch.setattr(deezer, "session", DummySession()) + # This call should raise a Deezer403Exception because "MD5_ORIGIN" is missing. + with pytest.raises(deezer.Deezer403Exception) as excinfo: + deezer.get_song_infos_from_deezer_website(deezer.TYPE_TRACK, "dummy_id") + + # Optionally, check that the exception message mentions login issues. + assert "not logged in" in str(excinfo.value).lower() +def test_downloadpicture_returns_dummy_content(monkeypatch): + """ + Test that downloadpicture returns the expected dummy image content. + This test monkeypatches the session.get method to simulate a network response. + """ + dummy_content = b"dummyimage" + + class DummyResponse: + def __init__(self, content): + self.content = content + class DummySession: + def get(self, url): + # Verify that the URL is constructed correctly for a cover image. + assert "cover" in url + return DummyResponse(dummy_content) + # Monkeypatch the global session in the deezer module with our dummy session. + monkeypatch.setattr(deezer, "session", DummySession()) + + # Call downloadpicture and verify that it returns the dummy content. + result = deezer.downloadpicture("dummy_pic_id") + assert result == dummy_content +def test_writeid3v1_tag_creates_tag(): + """ + Test that writeid3v1_1 writes an ID3v1.1 tag that starts with the 'TAG' header + and that the tag length matches the expected size. + """ + # Create a dummy song dictionary with necessary keys + dummy_song = { + "SNG_TITLE": "Test Song", + "ART_NAME": "Test Artist", + "ALB_TITLE": "Test Album", + "TRACK_NUMBER": "1" + } + # Set the global album_Data variable with dummy values required by album_get + deezer.album_Data = { + "PHYSICAL_RELEASE_DATE": "2020", + "LABEL_NAME": "Test Label", + "TRACKS": "10" + } + # Use BytesIO to capture the output written by writeid3v1_1 + output_file = io.BytesIO() + deezer.writeid3v1_1(output_file, dummy_song) + result = output_file.getvalue() + # Check that the tag starts with the "TAG" header (first 3 bytes) + assert result.startswith(b"TAG") + # Calculate expected size using the same struct format used in writeid3v1_1: + # The format string is "3s30s30s30s4s28sBHB" (combining the pieces from the code) + expected_size = struct.calcsize("3s30s30s30s4s28sBHB") + # Check that the written tag length equals the expected size + assert len(result) == expected_size +def test_set_song_quality(monkeypatch, capsys): + """ + Test that set_song_quality correctly sets the global sound_format: + - When lossless is supported and quality_config is "flac", it should set sound_format to "FLAC". + - When lossless is not supported, it should fallback to "MP3_128" and print a warning. + """ + # Case 1: flac quality_config and premium lossless support => should set to FLAC. + deezer.set_song_quality("flac", {"lossless": True}) + assert deezer.sound_format == "FLAC", "Expected sound_format to be FLAC when lossless is supported." + # Case 2: flac quality_config but lossless not supported => should fallback to MP3_128 and print warning. + deezer.set_song_quality("flac", {"lossless": False}) + assert deezer.sound_format == "MP3_128", "Expected sound_format to fallback to MP3_128 when lossless is not supported." + captured = capsys.readouterr().out + assert "WARNING: flac quality is configured" in captured, "Expected a warning message when flac is not supported." +def test_parse_deezer_playlist_returns_playlist_data(monkeypatch): + """ + Test that parse_deezer_playlist correctly processes a dummy API response and returns the expected + playlist name and list of songs. + """ + class DummyResponse: + def __init__(self, json_data): + self._json = json_data + self.status_code = 200 + def json(self): + return self._json + class DummySession: + def post(self, url, json=None): + if "method=deezer.getUserData" in url: + # Simulate CSRF token retrieval + return DummyResponse({"results": {"checkForm": "dummycsrf"}}) + elif "method=deezer.pagePlaylist" in url: + # Simulate playlist request returning dummy data + return DummyResponse({ + "error": "", + "results": { + "DATA": {"TITLE": "My Playlist", "NB_SONG": 2}, + "SONGS": {"count": 2, "data": [ + {"SNG_ID": "1", "SNG_TITLE": "Song1"}, + {"SNG_ID": "2", "SNG_TITLE": "Song2"} + ]} + } + }) + else: + return DummyResponse({}) + + # Monkeypatch the global session with DummySession + monkeypatch.setattr(deezer, "session", DummySession()) + + # Call parse_deezer_playlist with a dummy playlist_id (digits will be extracted) + playlist_name, songs = deezer.parse_deezer_playlist("12345") + + # Assert that the response matches our dummy data + assert playlist_name == "My Playlist" + assert isinstance(songs, list) + assert len(songs) == 2 + assert songs[0]["SNG_ID"] == "1" + assert songs[0]["SNG_TITLE"] == "Song1" + assert songs[1]["SNG_ID"] == "2" + assert songs[1]["SNG_TITLE"] == "Song2" +def test_blowfish_decrypt_returns_original(): + """ + Test that blowfishDecrypt correctly decrypts ciphertext produced by Blowfish encryption. + This increases test coverage for the crypto helper function by ensuring that encryption/decryption + with the fixed IV and key returns the original data. + """ + # Use the same key as the decryption function expects + key = "g4el58wc0zvf9na1" + # The initialization vector as a hex string used in blowfishDecrypt: "0001020304050607" + iv = a2b_hex("0001020304050607") + # Prepare a plaintext block that is exactly 8 bytes (Blowfish block size) + plaintext = b"ABCDEFGH" + # Create a Blowfish cipher with the same key and IV, using CBC mode + cipher = Blowfish.new(key.encode(), Blowfish.MODE_CBC, iv) + # Encrypt the plaintext to obtain ciphertext + ciphertext = cipher.encrypt(plaintext) + # Now use the function under test to decrypt the ciphertext + decrypted = deezer.blowfishDecrypt(ciphertext, key) + # Verify that the decrypted text matches the original plaintext + assert decrypted == plaintext +def test_script_extractor_extracts_scripts(): + """ + Test that ScriptExtractor correctly extracts all script contents from HTML with multiple ' + '' + '' + '
Not a script
' + '' + '' + '' + ) + # Initialize the ScriptExtractor from the original source code. + parser = deezer.ScriptExtractor() + parser.feed(html_input) + parser.close() + + # We expect the parser to extract the content of all three script tags. + expected_scripts = [ + 'console.log("Hello")', + 'var a = 1;', + '\nalert("Test");\n' + ] + assert parser.scripts == expected_scripts, "Extracted scripts do not match expected output." +def test_deezer_search_returns_expected_results(monkeypatch): + """ + Test that deezer_search returns correctly formatted results for a search query. + This test monkeypatches the 'session.get' method to simulate a dummy API response. + """ + # Define a dummy response with a json() method returning dummy data. + class DummyResponse: + def __init__(self, json_data): + self._json = json_data + self.status_code = 200 + def json(self): + return self._json + # Define a dummy session with a get method that returns the dummy response. + class DummySession: + def get(self, url): + # Check that the URL is constructed correctly for a search query. + # For TYPE_TRACK, expect the URL to contain "/search/track?q=" + assert "search/track" in url, f"URL '{url}' does not contain expected segment." + # Return dummy JSON that simulates one track. + dummy_json = { + "data": [{ + "id": "1", + "title": "Dummy Song", + "album": { + "cover_small": "http://dummy.url/cover.jpg", + "title": "Dummy Album", + "id": "101" + }, + "artist": {"name": "Dummy Artist"}, + "preview": "http://dummy.url/preview.mp3" + }] + } + return DummyResponse(dummy_json) + # Monkeypatch the global session in the deezer module with our dummy session. + monkeypatch.setattr(deezer, "session", DummySession()) + # Call the function under test with TYPE_TRACK and a dummy search term. + results = deezer.deezer_search("dummy search", deezer.TYPE_TRACK) + # Check that the result is a list with one track and that the keys match. + assert isinstance(results, list), "Expected results to be a list." + assert len(results) == 1, "Expected a single track in the search result." + track = results[0] + expected_keys = {"id", "id_type", "title", "img_url", "album", "album_id", "artist", "preview_url"} + assert expected_keys.issubset(track.keys()), f"Missing keys in returned result. Expected at least: {expected_keys}" + assert track["id"] == "1" + assert track["title"] == "Dummy Song" + assert track["img_url"] == "http://dummy.url/cover.jpg" + assert track["album"] == "Dummy Album" + assert track["album_id"] == "101" + assert track["artist"] == "Dummy Artist" + assert track["preview_url"] == "http://dummy.url/preview.mp3" +def test_get_song_url_api_error(monkeypatch): + """ + Test that get_song_url raises a RuntimeError when the API returns an error response. + This test simulates an API response containing an error message and verifies that + the exception is raised with the expected message. + """ + # Ensure the global license_token is set so that get_song_url can proceed. + deezer.license_token = {"dummy": "dummy"} + + # Define a dummy response that simulates an API error response. + class DummyResponse: + def __init__(self): + self.status_code = 200 + def raise_for_status(self): + pass + def json(self): + return {"data": [{"errors": [{"message": "Test error from API"}]}]} + + # Define a dummy requests.post to return our DummyResponse. + def dummy_post(*args, **kwargs): + return DummyResponse() + + # Monkeypatch the requests.post function used in get_song_url. + monkeypatch.setattr(requests, "post", dummy_post) + + # Verify that calling get_song_url raises a RuntimeError with the expected error message. + with pytest.raises(RuntimeError, match="Test error from API"): + deezer.get_song_url("dummy_track") \ No newline at end of file diff --git a/tests/test_runner.py b/tests/test_runner.py new file mode 100644 index 0000000..f94cd96 --- /dev/null +++ b/tests/test_runner.py @@ -0,0 +1,259 @@ +import sys +import pytest +from deezer_downloader.cli import runner +import types + +# No additional imports needed. +def test_version_flag(monkeypatch, capsys): + """ + Test that the '-v' flag prints the version string (via a patched importlib.metadata.version) + and exits with a zero status code. + """ + # Patch the version function to always return "1.2.3" + def fake_version(pkg): + if pkg == "deezer_downloader": + return "1.2.3" + raise Exception("Unexpected package") + monkeypatch.setattr("importlib.metadata.version", fake_version) + + # Set sys.argv to simulate calling the command with the -v flag. + test_argv = ["runner.py", "-v"] + monkeypatch.setattr(sys, "argv", test_argv) + + # Running main() should print the version and exit with status code 0. + with pytest.raises(SystemExit) as exit_info: + runner.main() + + # Ensure that the exit code is 0 (indicating success). + assert exit_info.value.code == 0 + + # Capture and assert that the version string appears in the output. + captured = capsys.readouterr().out + assert "1.2.3" in captured +def test_no_args_prints_help_and_exits(monkeypatch, capsys): + """ + Test that running the CLI without any arguments prints the help message and exits with status code 1. + """ + # Set sys.argv to simulate running the command without any arguments. + monkeypatch.setattr(sys, "argv", ["runner.py"]) + + with pytest.raises(SystemExit) as exit_info: + runner.main() + + # Assert that the exit code is 1 when no arguments are provided. + assert exit_info.value.code == 1 + + # Check that the output contains the help message (look for the usage line). + output = capsys.readouterr().out + assert "usage:" in output.lower() +def test_show_config_template_flag(monkeypatch, capsys): + """ + Test that using the '-t' flag prints the config template content and exits with status code 0. + This is done by patching the Path.read_text method so that it returns a fake template content. + """ + # Save the original read_text method. + original_read_text = runner.Path.read_text + # Define a fake read_text that returns fake content when the file name matches the template. + def fake_read_text(self, encoding=None): + if self.name == "deezer-downloader.ini.template": + return "fake config template content" + return original_read_text(self, encoding) + + # Monkeypatch the read_text method for Path so that any instance representing the config template file returns our fake content. + monkeypatch.setattr(runner.Path, "read_text", fake_read_text) + + # Set sys.argv to simulate calling the command with the '-t' flag. + monkeypatch.setattr(sys, "argv", ["runner.py", "-t"]) + + # Running main() should detect the -t flag, print the template content, and exit with code 0. + with pytest.raises(SystemExit) as exit_info: + runner.main() + + # Capture the output and ensure that the exit status is 0. + captured = capsys.readouterr().out + assert exit_info.value.code == 0 + assert "fake config template content" in captured +def test_config_flag_calls_load_config_and_run_backend(monkeypatch): + """ + Test that passing the '-c' flag with a value calls the load_config function with the provided argument + and then calls run_backend. + """ + # Create a fake configuration module with a fake load_config function. + fake_conf_module = types.ModuleType("deezer_downloader.configuration") + config_called = {} + def fake_load_config(config_path): + config_called['config'] = config_path + fake_conf_module.load_config = fake_load_config + # Inject the fake module into sys.modules so it is imported in runner.main(). + monkeypatch.setitem(sys.modules, "deezer_downloader.configuration", fake_conf_module) + # Simulate passing the '-c' flag with a dummy configuration file. + test_argv = ["runner.py", "-c", "dummy_config.ini"] + monkeypatch.setattr(sys, "argv", test_argv) + # Override run_backend in runner to simply record that it was called. + run_backend_called = [False] + def fake_run_backend(): + run_backend_called[0] = True + monkeypatch.setattr(runner, 'run_backend', fake_run_backend) + # Call main() and verify that it processes the -c flag correctly. + # Note: main() does not exit when a -c flag is provided, it calls run_backend. + runner.main() + # Check that load_config was called with the provided config filename. + assert config_called.get('config') == "dummy_config.ini" + # Check that run_backend was called. + assert run_backend_called[0] is Trueimport sys +import types + + +def test_run_backend_waitress_called(monkeypatch): + """ + Test that run_backend calls waitress.serve with the fake app and correct listen string + when __name__ is not '__main__'. + """ + # Create a fake configuration with a fake http object. + class FakeHTTP: + host = "127.0.0.1" + def getint(self, key): + return 8080 + + fake_config = types.SimpleNamespace(http=FakeHTTP()) + + # Create a fake configuration module. + fake_conf_module = types.ModuleType("deezer_downloader.configuration") + fake_conf_module.config = fake_config + # Provide a dummy load_config to bypass configuration loading. + fake_conf_module.load_config = lambda config_path=None: None + monkeypatch.setitem(sys.modules, "deezer_downloader.configuration", fake_conf_module) + + # Create a fake web.app module with a dummy WSGI app. + fake_app = lambda environ, start_response: None # a dummy WSGI application + fake_web_app_module = types.ModuleType("deezer_downloader.web.app") + fake_web_app_module.app = fake_app + monkeypatch.setitem(sys.modules, "deezer_downloader.web.app", fake_web_app_module) + + # Create a flag dictionary to record the call to waitress.serve. + called = {"called": False, "app": None, "listen": None} + def fake_waitress_serve(app_arg, listen): + called["called"] = True + called["app"] = app_arg + called["listen"] = listen + + # Patch waitress.serve in the runner module. + monkeypatch.setattr(runner.waitress, "serve", fake_waitress_serve) + + # Call run_backend. Since __name__ in runner is not '__main__', it should call waitress.serve. + runner.run_backend() + + # Verify that waitress.serve was called with the fake app and correct listen string. + assert called["called"] is True + assert called["app"] == fake_app + assert called["listen"] == "127.0.0.1:8080" +import sys +import types + + +from deezer_downloader.cli import runner + +def test_run_backend_waitress_called(monkeypatch): + """ + Test that run_backend calls waitress.serve with the fake app and correct listen string + when __name__ is not '__main__'. The fake configuration uses a dictionary with an 'http' + key mapping to a FakeHTTP instance that mimics subscript access and a getint method. + """ + # Create a fake configuration with a fake http object. + class FakeHTTP: + def __init__(self): + self.host = "127.0.0.1" + def getint(self, key): + return 8080 + def __getitem__(self, key): + if key == "host": + return self.host + raise KeyError(key) + + fake_config = {'http': FakeHTTP()} + + # Create a fake configuration module. + fake_conf_module = types.ModuleType("deezer_downloader.configuration") + fake_conf_module.config = fake_config + # Provide a dummy load_config to bypass configuration loading. + fake_conf_module.load_config = lambda config_path=None: None + monkeypatch.setitem(sys.modules, "deezer_downloader.configuration", fake_conf_module) + + # Create a fake web.app module with a dummy WSGI app. + fake_app = lambda environ, start_response: None # a dummy WSGI application + fake_web_app_module = types.ModuleType("deezer_downloader.web.app") + fake_web_app_module.app = fake_app + monkeypatch.setitem(sys.modules, "deezer_downloader.web.app", fake_web_app_module) + + # Create a flag dictionary to record the call to waitress.serve. + called = {"called": False, "app": None, "listen": None} + def fake_waitress_serve(app_arg, listen): + called["called"] = True + called["app"] = app_arg + called["listen"] = listen + + # Patch waitress.serve in the runner module. + monkeypatch.setattr(runner.waitress, "serve", fake_waitress_serve) + + # Call run_backend. Since __name__ in runner is not '__main__', it should call waitress.serve. + runner.run_backend() + + # Verify that waitress.serve was called with the fake app and correct listen string. + assert called["called"] is True + assert called["app"] == fake_app + assert called["listen"] == "127.0.0.1:8080" +import sys +import types +from deezer_downloader.cli import runner + + +def test_run_backend_waitress_called(monkeypatch): + """ + Test that run_backend calls waitress.serve with the fake app and correct listen string + when __name__ is not '__main__'. The fake configuration uses a FakeHTTP instance that mimics + the http configuration. This test ensures that waitress.serve is correctly called when the + runner is imported as a module (and not run as a script). + """ + # Create a fake configuration with a fake http object. + class FakeHTTP: + def __init__(self): + self.host = "127.0.0.1" + def getint(self, key): + return 8080 + def __getitem__(self, key): + if key == "host": + return self.host + raise KeyError(key) + + fake_config = {'http': FakeHTTP()} + + # Create a fake configuration module. + fake_conf_module = types.ModuleType("deezer_downloader.configuration") + fake_conf_module.config = fake_config + # Provide a dummy load_config to bypass configuration loading. + fake_conf_module.load_config = lambda config_path=None: None + monkeypatch.setitem(sys.modules, "deezer_downloader.configuration", fake_conf_module) + + # Create a fake web.app module with a dummy WSGI app. + fake_app = lambda environ, start_response: None # a dummy WSGI application + fake_web_app_module = types.ModuleType("deezer_downloader.web.app") + fake_web_app_module.app = fake_app + monkeypatch.setitem(sys.modules, "deezer_downloader.web.app", fake_web_app_module) + + # Create a flag dictionary to record the call to waitress.serve. + called = {"called": False, "app": None, "listen": None} + def fake_waitress_serve(app_arg, listen): + called["called"] = True + called["app"] = app_arg + called["listen"] = listen + + # Patch waitress.serve in the runner module. + monkeypatch.setattr(runner.waitress, "serve", fake_waitress_serve) + + # Call run_backend. Since __name__ in runner is not '__main__', it should call waitress.serve. + runner.run_backend() + + # Verify that waitress.serve was called with the fake app and correct listen string. + assert called["called"] is True + assert called["app"] == fake_app + assert called["listen"] == "127.0.0.1:8080" diff --git a/tests/test_spotify.py b/tests/test_spotify.py new file mode 100644 index 0000000..ab0218e --- /dev/null +++ b/tests/test_spotify.py @@ -0,0 +1,238 @@ +import pytest +from deezer_downloader import spotify +from time import sleep + +def test_get_json_rate_limit(monkeypatch, capsys): + """ + Test get_json_from_api when the API returns a 429 rate limit response. + This test verifies that the function prints the rate limiting message, + sleeps for the designated time (patched to avoid delay), and returns None. + """ + class FakeResponse: + def __init__(self, status_code, headers): + self.status_code = status_code + self.headers = headers + def json(self): + return {} + def fake_requests_get(url, headers, proxies): + return FakeResponse(429, {"Retry-After": "1"}) + monkeypatch.setattr(spotify.requests, "get", fake_requests_get) + monkeypatch.setattr(spotify, "sleep", lambda seconds: None) + result = spotify.get_json_from_api("http://dummy_url", "dummy_token", proxy="dummy_proxy") + output = capsys.readouterr().out + assert "rate limited" in output + assert result is None + +def test_parse_uri_with_invalid_netloc(): + """ + Test parse_uri with a URL having an unsupported netloc. + This should raise SpotifyInvalidUrlException since the URL's domain is not supported by the parser. + """ + invalid_url = "https://www.example.com/playlist/12345" + with pytest.raises(spotify.SpotifyInvalidUrlException): + spotify.parse_uri(invalid_url) + +def test_get_json_success(monkeypatch): + """ + Test get_json_from_api when the API returns a 200 status code. + This verifies that the function correctly returns the JSON response payload. + """ + class FakeResponse: + def __init__(self): + self.status_code = 200 + self.headers = {} + def json(self): + return {"key": "value"} + def fake_requests_get(url, headers, proxies): + return FakeResponse() + monkeypatch.setattr(spotify.requests, "get", fake_requests_get) + result = spotify.get_json_from_api("http://dummy_url", "dummy_token", proxy="dummy_proxy") + assert result == {"key": "value"} + +def test_parse_uri_embed(): + """ + Test parse_uri with a Spotify embed URL. + Verifies that an embed URL in the form "https://embed.spotify.com/?uri=spotify:track:..." + correctly returns the corresponding type and id. + """ + embed_url = "https://embed.spotify.com/?uri=spotify:track:6piFKF6WvM6ZZLmi2Vz8Vt" + expected = {"type": "track", "id": "6piFKF6WvM6ZZLmi2Vz8Vt"} + result = spotify.parse_uri(embed_url) + assert result == expected + +def test_parse_uri_spotify_uri(): + """ + Test parse_uri with Spotify URIs in the format "spotify::". + This verifies that the parser correctly extracts the type and id for both album and track URIs. + """ + album_uri = "spotify:album:7zCODUHkfuRxsUjtuzNqbd" + expected_album = {"type": "album", "id": "7zCODUHkfuRxsUjtuzNqbd"} + result_album = spotify.parse_uri(album_uri) + assert result_album == expected_album + + track_uri = "spotify:track:6piFKF6WvM6ZZLmi2Vz8Vt" + expected_track = {"type": "track", "id": "6piFKF6WvM6ZZLmi2Vz8Vt"} + result_track = spotify.parse_uri(track_uri) + assert result_track == expected_track + +def test_get_songs_from_spotify_album(monkeypatch): + """ + Test get_songs_from_spotify_website for album type. + This test simulates the API's token endpoint and album endpoint responses. + It verifies that the function correctly returns the list of parsed tracks from an album. + """ + token_response = {"accessToken": "dummy_access_token"} + album_response_data = { + "items": [ + {"artists": [{"name": "Artist1"}], "name": "Song1 (Remastered)"}, + {"artists": [{"name": "Artist2"}], "name": "Song2"} + ] + } + class FakeTokenResponse: + def __init__(self): + self.status_code = 200 + self.headers = {} + def json(self): + return token_response + class FakeAlbumResponse: + def __init__(self): + self.status_code = 200 + self.headers = {} + def json(self): + return album_response_data + def fake_requests_get(url, headers, proxies): + if url == spotify.token_url: + return FakeTokenResponse() + elif url.startswith("https://api.spotify.com/v1/albums/"): + return FakeAlbumResponse() + else: + raise ValueError("Unexpected URL: " + url) + monkeypatch.setattr(spotify.requests, "get", fake_requests_get) + album_uri = "spotify:album:7zCODUHkfuRxsUjtuzNqbd" + tracks = spotify.get_songs_from_spotify_website(album_uri, proxy="dummy_proxy") + expected_tracks = ["Artist1 Song1 ", "Artist2 Song2"] + assert tracks == expected_tracks + +def test_get_songs_from_spotify_playlist_pagination(monkeypatch): + """ + Test get_songs_from_spotify_website for playlist pagination. + This test simulates a playlist API response with two paginated pages. + It verifies that the function collects and processes tracks from both pages. + """ + class FakeResponse: + def __init__(self, status_code, json_data, headers=None): + self.status_code = status_code + self._json = json_data + self.headers = headers or {} + def json(self): + return self._json + token_response = {"accessToken": "dummy_token"} + playlist_page1 = { + "items": [ + {"track": {"artists": [{"name": "Artist1"}], "name": "Song1 (Radio Edit)"}} + ], + "next": "http://dummy_next_page" + } + playlist_page2 = { + "items": [ + {"track": {"artists": [{"name": "Artist2"}], "name": "Song2"}} + ], + "next": None + } + playlist_id = "dummy_playlist_id" + first_page_url = spotify.playlist_base_url.format(playlist_id) + def fake_requests_get(url, headers, proxies): + if url == spotify.token_url: + return FakeResponse(200, token_response) + elif url == first_page_url: + return FakeResponse(200, playlist_page1) + elif url == "http://dummy_next_page": + return FakeResponse(200, playlist_page2) + else: + raise ValueError("Unexpected URL: " + url) + monkeypatch.setattr(spotify.requests, "get", fake_requests_get) + tracks = spotify.get_songs_from_spotify_website(playlist_id, proxy="dummy_proxy") + expected_tracks = ["Artist1 Song1 ", "Artist2 Song2"] + assert tracks == expected_tracks + +def test_get_songs_from_spotify_track(monkeypatch): + """ + Test get_songs_from_spotify_website for track type. + This test simulates the token API and track endpoint responses and verifies that + a single track is correctly processed by the parser. + """ + class FakeResponse: + def __init__(self, status_code, json_data, headers=None): + self.status_code = status_code + self._json = json_data + self.headers = headers or {} + def json(self): + return self._json + token_response = {"accessToken": "dummy_access_token"} + track_response = {"artists": [{"name": "TestArtist"}], "name": "TestSong (Live)"} + def fake_requests_get(url, headers, proxies): + if url == spotify.token_url: + return FakeResponse(200, token_response) + elif url == spotify.track_base_url.format("testtrackid"): + return FakeResponse(200, track_response) + else: + raise ValueError("Unexpected URL: " + url) + monkeypatch.setattr(spotify.requests, "get", fake_requests_get) + track_uri = "spotify:track:testtrackid" + tracks = spotify.get_songs_from_spotify_website(track_uri, proxy="dummy_proxy") + expected_tracks = ["TestArtist TestSong "] + assert tracks == expected_tracks + +def test_get_songs_from_spotify_album_retry(monkeypatch): + """ + Test get_songs_from_spotify_website for album type when the album endpoint first returns a + rate limiting response (429) and then succeeds upon retry. This test simulates the retry logic. + """ + token_response = {"accessToken": "dummy_access_token"} + album_response_data = { + "items": [ + {"artists": [{"name": "ArtistRetry"}], "name": "SongRetry (Live)"} + ] + } + call_counter = {"album_calls": 0} + class FakeTokenResponse: + def __init__(self): + self.status_code = 200 + self.headers = {} + def json(self): + return token_response + class FakeAlbumResponse: + def __init__(self, status_code, json_data, headers=None): + self.status_code = status_code + self._json = json_data + self.headers = headers or {} + def json(self): + return self._json + def fake_requests_get(url, headers, proxies): + if url == spotify.token_url: + return FakeTokenResponse() + elif url == spotify.album_base_url.format("retry_album_id"): + if call_counter["album_calls"] == 0: + call_counter["album_calls"] += 1 + return FakeAlbumResponse(429, {}, headers={"Retry-After": "1"}) + else: + return FakeAlbumResponse(200, album_response_data) + else: + raise ValueError("Unexpected URL: " + url) + monkeypatch.setattr(spotify.requests, "get", fake_requests_get) + monkeypatch.setattr(spotify, "sleep", lambda seconds: None) + album_uri = "spotify:album:retry_album_id" + tracks = spotify.get_songs_from_spotify_website(album_uri, proxy="dummy_proxy") + expected_tracks = ["ArtistRetry SongRetry "] + assert tracks == expected_tracks + +def test_parse_uri_backwards_compatibility(): + """ + Test parse_uri with a string that has no scheme and no netloc. + This verifies the backwards compatibility branch where the input (a raw playlist id) + is returned as a playlist type with the id equal to the original string. + """ + simple_input = "simplePlaylistID" + expected = {"type": "playlist", "id": "simplePlaylistID"} + result = spotify.parse_uri(simple_input) + assert result == expected diff --git a/tests/test_threadpool_queue.py b/tests/test_threadpool_queue.py new file mode 100644 index 0000000..5006902 --- /dev/null +++ b/tests/test_threadpool_queue.py @@ -0,0 +1,59 @@ +import time +import threading +import pytest +from deezer_downloader.threadpool_queue import ThreadpoolScheduler, QueuedTask, report_progress + +def test_task_exception_handling(): + """ + Test that a task raising an exception is caught by the scheduler. + The test verifies that the task's state becomes "failed" + and the exception is recorded. + """ + # Initialize the thread pool scheduler + scheduler = ThreadpoolScheduler() + # Define a dummy command that fails intentionally. + @scheduler.register_command() + def failing_task(x): + raise ValueError("Intentional Failure") + # Start the worker threads + scheduler.run_workers(1) + # Enqueue a task that will run the failing_task command. + task = scheduler.enqueue_task("failing task", "failing_task", x=42) + # Stop all workers (placing stop tokens in the queue) + scheduler.stop_workers() + # Check that the task failed and the exception is of type ValueError. + assert task.state == "failed" + assert isinstance(task.exception, ValueError) +def test_successful_task(): + """ + Test that a successful task: + - Completes normally with state "mission accomplished" + - Returns the expected result (sum of two numbers) + - Updates progress values via the report_progress function. + """ + scheduler = ThreadpoolScheduler() + @scheduler.register_command() + def successful_task(a, b): + # Update progress to simulate work being done. + report_progress(1, 10) + time.sleep(0.1) + report_progress(10, 10) + return a + b + scheduler.run_workers(1) + task = scheduler.enqueue_task("addition task", "successful_task", a=3, b=4) + scheduler.stop_workers() + # Validate that the task completed as expected. + assert task.state == "mission accomplished" + assert task.result == 7 + assert task.progress == 10 + assert task.progress_maximum == 10 +def test_enqueue_invalid_command_raises_key_error(): + """ + Test that enqueuing a task with an unregistered command raises a KeyError. + This verifies that the scheduler correctly errors out when a command is missing. + """ + scheduler = ThreadpoolScheduler() + # Attempting to enqueue a task with a command that was never registered should + # immediately raise a KeyError. + with pytest.raises(KeyError): + scheduler.enqueue_task("invalid task", "non_existent_command", foo=123) \ No newline at end of file