fix(quality): .aiff probe class + refresh tests drifted by the 2.7.4 merge (#896)

- file_ops.probe_audio_quality: .aiff/.aif were opened with mutagen.wave.WAVE,
  which can't parse AIFF — it raised, failed open, and let AIFF silently bypass
  the quality filter. Route aiff/aif to mutagen.aiff.AIFF (still the 'wav'
  lossless tier).
- test_hifi_preview_guard: _get_hls_manifest gained an expected_duration_s kwarg
  and the start tier now comes from quality_tier_for_source (default profile ->
  'hires'); accept the kwarg and pin the tier so the chain is deterministic.
- test_quarantine_management: quarantine_group_key intentionally no longer uses
  source-specific ids/uri (they break cross-batch sibling matching); assert the
  isrc -> normalized-name contract instead.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
pull/896/head
dev 4 days ago
parent 2ae9ad3c6f
commit 17137cea5b

@ -286,8 +286,15 @@ def probe_audio_quality(file_path: str):
)
if ext in ('wav', 'aiff', 'aif'):
from mutagen.wave import WAVE
audio = WAVE(file_path)
# AIFF must use mutagen.aiff.AIFF — WAVE() can't parse it and would
# raise, making the file fail open and silently bypass the quality
# filter. Both are uncompressed PCM, so they share the 'wav' tier.
if ext == 'wav':
from mutagen.wave import WAVE
audio = WAVE(file_path)
else:
from mutagen.aiff import AIFF
audio = AIFF(file_path)
return AudioQuality(
format='wav',
bitrate=audio.info.bitrate // 1000 if audio.info.bitrate else None,

@ -470,9 +470,14 @@ def test_group_key_prefers_isrc_over_everything():
assert quarantine_group_key("Artist", "Track", ctx) == "isrc:usrc12345678"
def test_group_key_falls_back_to_source_id_then_uri():
assert quarantine_group_key("A", "T", {"track_info": {"id": "abc123"}}) == "id:abc123"
assert quarantine_group_key("A", "T", {"track_info": {"uri": "spotify:track:z"}}) == "uri:spotify:track:z"
def test_group_key_uses_isrc_and_ignores_source_specific_ids():
# ISRC is the universal target identity → wins.
assert quarantine_group_key("A", "T", {"track_info": {"isrc": "USABC1234567"}}) == "isrc:usabc1234567"
# Source-specific ids / uris are intentionally NOT used (they differ across
# sources/batches and break cross-batch sibling matching) — with no ISRC the
# key falls back to the normalized artist|track name, NOT the id/uri.
assert quarantine_group_key("A", "T", {"track_info": {"id": "abc123"}}) == "nm:a|t"
assert quarantine_group_key("A", "T", {"track_info": {"uri": "spotify:track:z"}}) == "nm:a|t"
def test_group_key_falls_back_to_normalized_name_without_context():

@ -85,10 +85,13 @@ def _bare_client(tmp_path):
def test_download_sync_skips_preview_manifests_and_never_downloads(tmp_path, monkeypatch):
monkeypatch.setattr(hc, 'config_manager', _Cfg())
# Pin the starting tier so the chain is deterministic and independent of the
# global quality profile (which now drives quality_tier_for_source).
monkeypatch.setattr(hc, 'quality_tier_for_source', lambda *a, **k: 'lossless')
c = _bare_client(tmp_path)
c.get_track_info = lambda tid: {'duration_s': 215} # real track length
tiers = []
c._get_hls_manifest = lambda tid, quality='lossless': (
c._get_hls_manifest = lambda tid, quality='lossless', **kw: (
tiers.append(quality) or
{'segment_uris': ['seg'], 'init_uri': None, 'extension': 'flac',
'manifest_duration': 30.0}) # preview at EVERY tier
@ -105,7 +108,7 @@ def test_download_sync_proceeds_past_the_gate_for_a_full_manifest(tmp_path, monk
monkeypatch.setattr(hc, 'config_manager', _Cfg())
c = _bare_client(tmp_path)
c.get_track_info = lambda tid: {'duration_s': 215}
c._get_hls_manifest = lambda tid, quality='lossless': {
c._get_hls_manifest = lambda tid, quality='lossless', **kw: {
'segment_uris': ['seg'], 'init_uri': None, 'extension': 'flac',
'manifest_duration': 215.0} # full length → must NOT skip
seg_calls = []
@ -124,10 +127,11 @@ def test_download_sync_aborts_on_a_faked_full_length_file_no_tier_cascade(tmp_pa
# decodes to 30s. It must abort HiFi (return None) on the first tier — NOT drop to the
# lossy 'high' tier (the same 30s preview, which dodges the bitrate check).
monkeypatch.setattr(hc, 'config_manager', _Cfg())
monkeypatch.setattr(hc, 'quality_tier_for_source', lambda *a, **k: 'lossless')
c = _bare_client(tmp_path)
c.get_track_info = lambda tid: {'duration_s': 215}
tiers = []
c._get_hls_manifest = lambda tid, quality='lossless': (
c._get_hls_manifest = lambda tid, quality='lossless', **kw: (
tiers.append(quality) or
{'segment_uris': ['seg'], 'init_uri': None, 'extension': 'flac', 'manifest_duration': 215.0})
c._download_segment_with_retry = lambda url: b'\x00' * 200_000 # > MIN_AUDIO_SIZE
@ -144,7 +148,7 @@ def test_download_sync_does_not_reject_when_track_length_unknown(tmp_path, monke
monkeypatch.setattr(hc, 'config_manager', _Cfg())
c = _bare_client(tmp_path)
c.get_track_info = lambda tid: {'duration_s': 0} # expected unknown
c._get_hls_manifest = lambda tid, quality='lossless': {
c._get_hls_manifest = lambda tid, quality='lossless', **kw: {
'segment_uris': ['seg'], 'init_uri': None, 'extension': 'flac',
'manifest_duration': 30.0} # short, but expected is unknown
seg_calls = []

Loading…
Cancel
Save