"""Test per core.royalty_check — ccMixter + Jamendo + cache SQLite. Nessuna integrazione reale: tutte le chiamate di rete sono mockate con `responses`. La cache SQLite e' isolata in tmp_path. """ from __future__ import annotations from pathlib import Path from unittest import mock import pytest import responses from core import royalty_check as rc # ------------------------------------------------------------------ # Fixture: cache isolata per test # ------------------------------------------------------------------ @pytest.fixture def patched_cache(tmp_path, monkeypatch): db = tmp_path / "royalty_cache_test.db" monkeypatch.setattr(rc, "_cache_db_path", lambda: db) # Azzera il throttle globale per non rallentare i test monkeypatch.setattr(rc, "_last_req", [0.0]) monkeypatch.setattr(rc, "_RATE_LIMIT_SEC", 0.0) return db # ================================================================== # _similarity # ================================================================== class TestSimilarity: def test_similarity_perfect_match(self): assert rc._similarity("hello world", "hello world") == pytest.approx(1.0) def test_similarity_case_insensitive(self): assert rc._similarity("Hello World", "hello world") == pytest.approx(1.0) def test_similarity_empty_returns_zero(self): assert rc._similarity("", "foo") == 0.0 assert rc._similarity("foo", "") == 0.0 assert rc._similarity("", "") == 0.0 def test_similarity_partial(self): s = rc._similarity("Daft Punk - Around The World", "Around The World") # SequenceMatcher usa chunks comuni: valore stabile e > 0.5 assert 0.5 < s < 1.0 # ================================================================== # _query_ccmixter # ================================================================== class TestQueryCcmixter: _SAMPLE = [ { "upload_name": "Around The World", "user_real_name": "Jane Doe", "user_name": "jane", "license_name": "Attribution Noncommercial (3.0)", "license_url": "https://creativecommons.org/licenses/by-nc/3.0/", "file_page_url": "https://ccmixter.org/files/jane/1", }, { "upload_name": "Other Song", "user_real_name": "", "user_name": "bob", "license_name": "CC BY 4.0", "license_url": "https://creativecommons.org/licenses/by/4.0/", "file_page_url": "https://ccmixter.org/files/bob/2", }, ] @responses.activate def test_query_ccmixter_parses_response(self, patched_cache): responses.add( responses.GET, rc._CCMIXTER_URL, json=self._SAMPLE, status=200, ) res = rc._query_ccmixter("Around The World") assert len(res) == 2 assert res[0]["source"] == "ccmixter" assert res[0]["title"] == "Around The World" assert res[0]["artist"] == "Jane Doe" assert "licenses" in res[0]["license_url"] # Fallback a user_name quando user_real_name e' vuoto assert res[1]["artist"] == "bob" @responses.activate def test_query_ccmixter_empty_on_http_error(self, patched_cache): responses.add( responses.GET, rc._CCMIXTER_URL, json={"error": "boom"}, status=500, ) assert rc._query_ccmixter("whatever") == [] @responses.activate def test_query_ccmixter_empty_on_bad_json(self, patched_cache): responses.add( responses.GET, rc._CCMIXTER_URL, json={"not": "a list"}, status=200, ) assert rc._query_ccmixter("whatever") == [] def test_query_ccmixter_empty_query(self, patched_cache): # Nessuna chiamata di rete attesa — se la facesse responses fallirebbe assert rc._query_ccmixter("") == [] # ================================================================== # _query_jamendo # ================================================================== class TestQueryJamendo: _SAMPLE = { "headers": {"status": "success"}, "results": [ { "id": "123", "name": "Freedom Song", "artist_name": "Open Artist", "license_name": "CC BY-SA", "license_ccurl": "https://creativecommons.org/licenses/by-sa/4.0/", "shareurl": "https://www.jamendo.com/track/123", "audio": "https://.../123.mp3", } ], } def test_query_jamendo_requires_client_id(self, patched_cache): # Nessun client_id -> ritorno vuoto senza chiamata di rete # (responses non e' attivato: una chiamata farebbe raise) assert rc._query_jamendo("Freedom Song", client_id="") == [] @responses.activate def test_query_jamendo_parses_response(self, patched_cache): responses.add( responses.GET, rc._JAMENDO_URL, json=self._SAMPLE, status=200, ) res = rc._query_jamendo("Freedom Song", client_id="abcd1234") assert len(res) == 1 assert res[0]["source"] == "jamendo" assert res[0]["title"] == "Freedom Song" assert res[0]["artist"] == "Open Artist" assert res[0]["license"] == "CC BY-SA" assert res[0]["url"].startswith("https://www.jamendo.com/") @responses.activate def test_query_jamendo_status_failure_returns_empty(self, patched_cache): responses.add( responses.GET, rc._JAMENDO_URL, json={"headers": {"status": "failed", "error_message": "bad key"}, "results": []}, status=200, ) assert rc._query_jamendo("Freedom Song", client_id="bad") == [] # ================================================================== # check_file # ================================================================== def _fake_metadata(artist: str, title: str): """Factory per un dict compatibile con core.metadata.read_metadata.""" return {"artist": artist, "title": title} class TestCheckFile: def test_check_file_matches_royalty_free(self, patched_cache, tmp_path): # Prepara un path "esistente" solo per plausibilita' (il # read_metadata e' mockato: non viene mai letto davvero). p = tmp_path / "track.mp3" p.write_bytes(b"\x00") cc_hit = [{ "upload_name": "Around The World", "user_real_name": "Jane Doe", "user_name": "jane", "license_name": "CC BY 3.0", "license_url": "https://creativecommons.org/licenses/by/3.0/", "file_page_url": "https://ccmixter.org/x/1", }] with mock.patch.object(rc, "_query_ccmixter", return_value=[{ "source": "ccmixter", "title": "Around The World", "artist": "Jane Doe", "license": "CC BY 3.0", "license_url": cc_hit[0]["license_url"], "url": cc_hit[0]["file_page_url"], }]), mock.patch.object(rc, "_query_jamendo", return_value=[]), \ mock.patch("core.metadata.read_metadata", return_value=_fake_metadata("Jane Doe", "Around The World")): res = rc.check_file(str(p)) assert res["status"] == "royalty_free" assert len(res["matches"]) == 1 assert res["matches"][0]["source"] == "ccmixter" assert res["matches"][0]["score"] >= 0.8 assert res["artist"] == "Jane Doe" assert res["title"] == "Around The World" def test_check_file_no_match_returns_unknown(self, patched_cache, tmp_path): p = tmp_path / "unknown.mp3" p.write_bytes(b"\x00") with mock.patch.object(rc, "_query_ccmixter", return_value=[{ "source": "ccmixter", "title": "Completamente Diverso", "artist": "Qualcun Altro", "license": "CC BY", "license_url": "", "url": "", }]), mock.patch.object(rc, "_query_jamendo", return_value=[]), \ mock.patch("core.metadata.read_metadata", return_value=_fake_metadata("Daft Punk", "Harder Better Faster")): res = rc.check_file(str(p)) assert res["status"] == "unknown" assert res["matches"] == [] # artist/title conservati dai tag assert res["artist"] == "Daft Punk" def test_check_file_uses_cache(self, patched_cache, tmp_path): p = tmp_path / "cached.mp3" p.write_bytes(b"\x00") cc_mock = mock.MagicMock(return_value=[{ "source": "ccmixter", "title": "Hello", "artist": "World", "license": "CC BY", "license_url": "", "url": "", }]) jm_mock = mock.MagicMock(return_value=[]) md_mock = mock.MagicMock(return_value=_fake_metadata("World", "Hello")) with mock.patch.object(rc, "_query_ccmixter", cc_mock), \ mock.patch.object(rc, "_query_jamendo", jm_mock), \ mock.patch("core.metadata.read_metadata", md_mock): res1 = rc.check_file(str(p)) res2 = rc.check_file(str(p)) assert res1["status"] == "royalty_free" assert res2["status"] == "royalty_free" # Seconda chiamata hitta la cache: le query API non vengono richiamate assert cc_mock.call_count == 1 assert jm_mock.call_count == 1 def test_check_file_fallback_to_filename_when_tags_missing( self, patched_cache, tmp_path): """Senza tag ID3, usa _parse_filename sul nome del file.""" # Filename stile "Artista - Titolo.mp3" p = tmp_path / "Jane Doe - Around The World.mp3" p.write_bytes(b"\x00") with mock.patch.object(rc, "_query_ccmixter", return_value=[{ "source": "ccmixter", "title": "Around The World", "artist": "Jane Doe", "license": "CC BY", "license_url": "", "url": "", }]), mock.patch.object(rc, "_query_jamendo", return_value=[]), \ mock.patch("core.metadata.read_metadata", return_value=_fake_metadata("", "")): res = rc.check_file(str(p)) assert res["status"] == "royalty_free" assert res["artist"] == "Jane Doe" assert res["title"] == "Around The World" def test_check_file_metadata_error_returns_error_status( self, patched_cache, tmp_path): p = tmp_path / "broken.mp3" p.write_bytes(b"\x00") with mock.patch("core.metadata.read_metadata", side_effect=ValueError("formato illeggibile")): res = rc.check_file(str(p)) assert res["status"] == "error" assert "formato illeggibile" in res.get("error", "") # ================================================================== # scan_folder # ================================================================== class TestScanFolder: def test_scan_folder_streams_entries(self, patched_cache, tmp_path): # Crea 3 file audio nella cartella for name in ("a.mp3", "b.mp3", "c.flac"): (tmp_path / name).write_bytes(b"\x00") fake_result = { "path": "", "status": "unknown", "matches": [], "query": "q", "artist": "A", "title": "T", } collected = [] def _entry_cb(entry): collected.append(entry) with mock.patch.object(rc, "check_file", return_value=fake_result): results = rc.scan_folder( str(tmp_path), recursive=False, entry_callback=_entry_cb, ) assert len(results) == 3 assert len(collected) == 3 def test_scan_folder_filters_non_audio(self, patched_cache, tmp_path): (tmp_path / "ok.mp3").write_bytes(b"\x00") (tmp_path / "note.txt").write_text("hello") (tmp_path / "image.png").write_bytes(b"\x00") with mock.patch.object(rc, "check_file", return_value={"path": "", "status": "unknown", "matches": [], "query": "", "artist": "", "title": ""}): results = rc.scan_folder(str(tmp_path), recursive=False) assert len(results) == 1 def test_scan_folder_empty_directory(self, patched_cache, tmp_path): empty = tmp_path / "empty" empty.mkdir() results = rc.scan_folder(str(empty), recursive=False) assert results == [] def test_scan_folder_nonexistent_directory(self, patched_cache, tmp_path): results = rc.scan_folder(str(tmp_path / "missing"), recursive=False) assert results == [] def test_scan_folder_progress_callback_called(self, patched_cache, tmp_path): (tmp_path / "a.mp3").write_bytes(b"\x00") (tmp_path / "b.mp3").write_bytes(b"\x00") events = [] def _progress(idx, total, name, status, err=""): events.append((idx, total, status)) fake_result = { "path": "", "status": "unknown", "matches": [], "query": "", "artist": "", "title": "", } with mock.patch.object(rc, "check_file", return_value=fake_result): rc.scan_folder(str(tmp_path), recursive=False, progress_callback=_progress) # Attendo almeno "checking" per ciascun file + "completed" finale assert any(e[2] == "completed" for e in events) assert sum(1 for e in events if e[2] == "checking") >= 2