luzadev/musicdownload
1
0 Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 2 Wiki Activity
Files
ddacd050d82f76951062118110adcc08d1415c76
musicdownload/tests/test_royalty_check.py
T
lucianoandClaude Opus 4.7 018d310f53 royalty: modulo check ccMixter + Jamendo con cache SQLite 
Nuova feature "Diritti": verifica royalty-free status di brani locali
contro DB CC/open (ccMixter sempre attivo, Jamendo opzionale via
jamendo_client_id in config). Approccio honest-only:

- status 'royalty_free' SOLO con match Levenshtein combinato >= 0.80
- status 'unknown' NON implica automaticamente SIAE (= da verificare)
- status 'error' per fingerprint fail / network / tag illeggibili

Pipeline:
- Scan AUDIO_EXTENSIONS da core.upgrader (ricorsivo opzionale)
- Legge artist/title da ID3 (core.metadata.read_metadata), fallback
  a core.catalog._parse_filename sul nome file
- Query parallele ccMixter + Jamendo con throttle 0.3s
- Cache SQLite in <config_dir>/royalty_cache.db

21 test verdi (mock requests via responses, cache isolata in tmp_path).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-10-09 14:25:10 +02:00

374 lines
14 KiB
Python
Raw Blame History

"""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
Reference in new issue View Git Blame Copy Permalink