lossy copy: support all lossless formats, not just FLAC (#941)
radoslav-orlov: "create lossy copies of lossless tracks" only recognized FLAC, even though ALAC/ WAV/AIFF/DSD are now quality-profile formats. the FLAC knowledge was hardcoded in 3 separate places (the import path, the Lossy Converter scan, and the fix executor) — exactly how a format gets added in one spot but not another. kettui-style fix — one canonical seam both sites route through, instead of 3 more string edits: - new core/quality/lossless.py: is_lossless_format / is_lossless_audio_path (pure; injects a codec probe for the ambiguous .m4a/.mp4 — ALAC vs AAC — so the decision stays testable with no I/O), LOSSLESS_FORMATS (single source of truth, derived-consistent with model.tier_score), and the lossy_output_would_overwrite_source safety invariant. - create_lossy_copy + the Lossy Converter scan + repair_worker._fix_missing_lossy_copy all route through it. SQL pre-filters by candidate extensions, then each file is confirmed (probing .m4a). - SAFETY: a lossy copy must never be written over its own source — an .m4a ALAC source + AAC target lands on the same .m4a path, and ffmpeg runs with -y. all three sites now bail on the overwrite case BEFORE ffmpeg (the existing delete-original guard was too late — the source was already clobbered). dropped a vestigial mutagen FLAC import; updated FLAC-only UI strings. 19 tests: full seam coverage (formats, the .m4a ALAC/AAC probe branch, candidate extensions, the overwrite guard), a tier-model consistency test that fails if the lossless set drifts, and import- site wiring tests — WAV now converts (was rejected), and the .m4a-ALAC+AAC overwrite case proves ffmpeg NEVER runs. 286 quality/import/repair tests green, ruff clean.
This commit is contained in:
BoulderBadgeDad
parent
b62d9b5b08
commit
4af7600fd5
6 changed files with 355 additions and 15 deletions
|
|
@ -524,14 +524,29 @@ def downsample_hires_flac(final_path, context):
|
|||
return None
|
||||
|
||||
|
||||
def m4a_codec(path):
|
||||
"""Codec of an .m4a/.mp4 container ('alac', 'aac', …) or None — lets the
|
||||
lossless check tell ALAC (lossless) from AAC (lossy), since both are .m4a."""
|
||||
try:
|
||||
from mutagen.mp4 import MP4
|
||||
return (getattr(MP4(path).info, 'codec', '') or '').lower() or None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def create_lossy_copy(final_path):
|
||||
"""Convert a FLAC file to a lossy copy using the configured codec."""
|
||||
from mutagen.flac import FLAC
|
||||
"""Convert a lossless file (FLAC / ALAC / WAV / AIFF / DSD) to a lossy copy
|
||||
using the configured codec. Non-lossless inputs are skipped (#941)."""
|
||||
from core.quality.lossless import (
|
||||
is_lossless_audio_path,
|
||||
lossy_output_would_overwrite_source,
|
||||
)
|
||||
|
||||
if not config_manager.get("lossy_copy.enabled", False):
|
||||
return None
|
||||
|
||||
if os.path.splitext(final_path)[1].lower() != ".flac":
|
||||
# Was FLAC-only; now any lossless source. .m4a is probed (ALAC vs AAC).
|
||||
if not is_lossless_audio_path(final_path, probe_codec=m4a_codec):
|
||||
return None
|
||||
|
||||
codec = config_manager.get("lossy_copy.codec", "mp3").lower()
|
||||
|
|
@ -560,6 +575,16 @@ def create_lossy_copy(final_path):
|
|||
out_basename = out_basename.replace(original_quality, quality_label)
|
||||
out_path = os.path.join(os.path.dirname(out_path), out_basename)
|
||||
|
||||
# Safety invariant: never write the lossy copy over its own source (an .m4a
|
||||
# ALAC source + AAC target lands on the same .m4a path). ffmpeg runs with -y,
|
||||
# so this guard MUST precede it — the later delete-original guard is too late.
|
||||
if lossy_output_would_overwrite_source(final_path, out_path):
|
||||
logger.info(
|
||||
f"[Lossy Copy] Skipping — {codec.upper()} output would overwrite the "
|
||||
f"source: {os.path.basename(final_path)}"
|
||||
)
|
||||
return None
|
||||
|
||||
ffmpeg_bin = shutil.which("ffmpeg")
|
||||
if not ffmpeg_bin:
|
||||
local = os.path.join(os.path.dirname(__file__), "tools", "ffmpeg")
|
||||
|
|
|
|||
103
core/quality/lossless.py
Normal file
103
core/quality/lossless.py
Normal file
|
|
@ -0,0 +1,103 @@
|
|||
"""Single source of truth for "is this audio lossless?" + the lossy-copy
|
||||
overwrite invariant.
|
||||
|
||||
The "create a lossy copy of lossless tracks" feature lives in two places (the
|
||||
import post-processing path and the Lossy Converter repair job). Both used to
|
||||
hardcode ``.flac``, which is exactly how ALAC/WAV/DSD ended up being quality-
|
||||
profile options but NOT lossy-copy sources (#941). The knowledge of which
|
||||
formats are lossless now lives HERE, derived from the same format names the
|
||||
quality model ranks, so adding a format lights it up in both sites at once and
|
||||
they can never drift.
|
||||
|
||||
Two seams, both pure (no I/O) so they're unit-testable without real files:
|
||||
|
||||
* :func:`is_lossless_format` / :func:`is_lossless_audio_path` — eligibility.
|
||||
``.m4a``/``.mp4`` are ambiguous (ALAC=lossless, AAC=lossy) and can only be told
|
||||
apart by codec, so the path check delegates them to an *injected* ``probe_codec``
|
||||
callable — the file I/O stays at the edge, the decision stays pure.
|
||||
* :func:`lossy_output_would_overwrite_source` — the safety invariant: a lossy copy
|
||||
must NEVER be written over its own source (which becomes possible once ``.m4a``
|
||||
is eligible and the target codec is AAC → same ``.m4a`` path).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
from core.quality.source_map import AUDIO_EXTENSIONS, format_from_extension
|
||||
|
||||
# Canonical lossless format set. MUST stay in sync with the lossless tiers in
|
||||
# core.quality.model.tier_score and the frontend RT_LOSSLESS_FORMATS — the
|
||||
# consistency test in tests/quality/test_lossless.py pins the tier agreement.
|
||||
LOSSLESS_FORMATS = frozenset({'flac', 'alac', 'wav', 'dsf'})
|
||||
|
||||
# Container extensions that may hold EITHER lossless (ALAC) or lossy (AAC) audio —
|
||||
# only a codec probe can decide, so they're never lossless on extension alone.
|
||||
_AMBIGUOUS_EXTS = frozenset({'m4a', 'mp4'})
|
||||
|
||||
|
||||
def is_lossless_format(fmt: Any) -> bool:
|
||||
"""True when a unified format name (as returned by ``format_from_extension``)
|
||||
is lossless. ``'aac'`` is False here — an ALAC-in-m4a file reports format
|
||||
``'aac'`` by extension and must be resolved by codec, not by name."""
|
||||
return str(fmt or '').lower() in LOSSLESS_FORMATS
|
||||
|
||||
|
||||
# Extensions worth checking as possibly-lossless (a caller can pre-filter SQL by
|
||||
# these, then confirm each via is_lossless_audio_path). Derived from the format
|
||||
# map so it can't drift: every extension whose format is lossless, plus the
|
||||
# ambiguous ALAC containers. Leading dot included (e.g. '.flac', '.m4a').
|
||||
LOSSLESS_CANDIDATE_EXTENSIONS = frozenset(
|
||||
e for e in AUDIO_EXTENSIONS
|
||||
if is_lossless_format(format_from_extension(e)) or e.lstrip('.') in _AMBIGUOUS_EXTS
|
||||
)
|
||||
|
||||
|
||||
def _ext(path: Any) -> str:
|
||||
return os.path.splitext(str(path or ''))[1].lower().lstrip('.')
|
||||
|
||||
|
||||
def is_lossless_audio_path(
|
||||
path: Any,
|
||||
*,
|
||||
probe_codec: Optional[Callable[[str], Optional[str]]] = None,
|
||||
) -> bool:
|
||||
"""True when the file at ``path`` is lossless.
|
||||
|
||||
Unambiguous extensions (flac/wav/aiff/dsf/dff/alac) are decided by extension.
|
||||
``.m4a``/``.mp4`` are decided by ``probe_codec(path)`` (returns the codec, e.g.
|
||||
``'alac'`` or ``'aac'``) — without a probe they're treated as NOT lossless, so
|
||||
a missing probe can never misclassify an AAC file as lossless.
|
||||
"""
|
||||
ext = _ext(path)
|
||||
if is_lossless_format(format_from_extension(ext)):
|
||||
return True
|
||||
if ext in _AMBIGUOUS_EXTS and probe_codec is not None:
|
||||
try:
|
||||
codec = (probe_codec(str(path)) or '').lower()
|
||||
except Exception:
|
||||
return False
|
||||
return 'alac' in codec
|
||||
return False
|
||||
|
||||
|
||||
def lossy_output_would_overwrite_source(source_path: Any, output_path: Any) -> bool:
|
||||
"""True when the computed lossy-copy output path is the source file itself.
|
||||
|
||||
Safety invariant: the converter must skip (never run ffmpeg with ``-y``) when
|
||||
this is True, or it would destroy the original. Happens when an ``.m4a`` ALAC
|
||||
source is converted with the AAC codec (output is also ``.m4a``)."""
|
||||
if not source_path or not output_path:
|
||||
return False
|
||||
a = os.path.normcase(os.path.normpath(str(source_path)))
|
||||
b = os.path.normcase(os.path.normpath(str(output_path)))
|
||||
return a == b
|
||||
|
||||
|
||||
__all__ = [
|
||||
'LOSSLESS_FORMATS',
|
||||
'is_lossless_format',
|
||||
'is_lossless_audio_path',
|
||||
'lossy_output_would_overwrite_source',
|
||||
]
|
||||
|
|
@ -1,13 +1,19 @@
|
|||
"""Lossy Converter Job — finds FLAC files that don't have a lossy copy.
|
||||
"""Lossy Converter Job — finds lossless files that don't have a lossy copy.
|
||||
|
||||
Scans the library for FLAC files without a corresponding lossy copy alongside
|
||||
Scans the library for lossless files without a corresponding lossy copy alongside
|
||||
them, and creates a finding for each. The fix action converts the file using
|
||||
ffmpeg with the user's configured codec/bitrate settings.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
from core.imports.file_ops import m4a_codec
|
||||
from core.library.path_resolver import resolve_library_file_path
|
||||
from core.quality.lossless import (
|
||||
LOSSLESS_CANDIDATE_EXTENSIONS,
|
||||
is_lossless_audio_path,
|
||||
lossy_output_would_overwrite_source,
|
||||
)
|
||||
from core.repair_jobs import register_job
|
||||
from core.repair_jobs.base import JobContext, JobResult, RepairJob
|
||||
from utils.logging_config import get_logger
|
||||
|
|
@ -21,6 +27,16 @@ CODEC_MAP = {
|
|||
}
|
||||
|
||||
|
||||
def _lossless_ext_where(col: str) -> str:
|
||||
"""SQL pre-filter matching files whose extension *might* be lossless. The
|
||||
final decision (including ALAC-in-.m4a, which needs a codec probe) is made
|
||||
per-file by is_lossless_audio_path. Extensions are trusted constants from the
|
||||
quality model, never user input — safe to interpolate."""
|
||||
return '(' + ' OR '.join(
|
||||
f"LOWER({col}) LIKE '%{ext}'" for ext in sorted(LOSSLESS_CANDIDATE_EXTENSIONS)
|
||||
) + ')'
|
||||
|
||||
|
||||
def _resolve_file_path(file_path, transfer_folder, download_folder=None, config_manager=None):
|
||||
"""Backwards-compat wrapper. Use ``resolve_library_file_path`` directly."""
|
||||
return resolve_library_file_path(
|
||||
|
|
@ -35,15 +51,15 @@ def _resolve_file_path(file_path, transfer_folder, download_folder=None, config_
|
|||
class LossyConverterJob(RepairJob):
|
||||
job_id = 'lossy_converter'
|
||||
display_name = 'Lossy Converter'
|
||||
description = 'Finds FLAC files without a lossy copy'
|
||||
description = 'Finds lossless files without a lossy copy'
|
||||
help_text = (
|
||||
'Scans your library for FLAC files that don\'t already have a lossy copy '
|
||||
'Scans your library for lossless files (FLAC/ALAC/WAV/AIFF/DSD) that don\'t already have a lossy copy '
|
||||
'(MP3, Opus, or AAC) alongside them.\n\n'
|
||||
'Uses the codec setting from your Lossy Copy configuration on the Settings '
|
||||
'page. Enable Lossy Copy in Settings first, then run this job to find FLAC '
|
||||
'files missing a lossy copy.\n\n'
|
||||
'Each finding can be fixed individually or in bulk — the fix action converts '
|
||||
'the FLAC file using ffmpeg at your configured bitrate.\n\n'
|
||||
'the lossless file using ffmpeg at your configured bitrate.\n\n'
|
||||
'Requires ffmpeg to be installed.'
|
||||
)
|
||||
icon = 'repair-icon-lossy'
|
||||
|
|
@ -81,14 +97,14 @@ class LossyConverterJob(RepairJob):
|
|||
try:
|
||||
conn = context.db._get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
cursor.execute(f"""
|
||||
SELECT t.id, t.title, ar.name, al.title, t.file_path,
|
||||
al.thumb_url, ar.thumb_url
|
||||
FROM tracks t
|
||||
LEFT JOIN artists ar ON ar.id = t.artist_id
|
||||
LEFT JOIN albums al ON al.id = t.album_id
|
||||
WHERE t.file_path IS NOT NULL AND t.file_path != ''
|
||||
AND LOWER(t.file_path) LIKE '%.flac'
|
||||
AND {_lossless_ext_where('t.file_path')}
|
||||
""")
|
||||
tracks = cursor.fetchall()
|
||||
except Exception as e:
|
||||
|
|
@ -104,7 +120,7 @@ class LossyConverterJob(RepairJob):
|
|||
context.update_progress(0, total)
|
||||
if context.report_progress:
|
||||
context.report_progress(
|
||||
phase=f'Scanning {total} FLAC files for missing {quality_label} copies...',
|
||||
phase=f'Scanning {total} lossless files for missing {quality_label} copies...',
|
||||
total=total
|
||||
)
|
||||
|
||||
|
|
@ -135,8 +151,17 @@ class LossyConverterJob(RepairJob):
|
|||
if not resolved or not os.path.exists(resolved):
|
||||
continue
|
||||
|
||||
# Confirm it's actually lossless — the SQL pre-filter lets .m4a through,
|
||||
# which is ALAC (lossless) OR AAC (lossy); only a codec probe decides.
|
||||
if not is_lossless_audio_path(resolved, probe_codec=m4a_codec):
|
||||
continue
|
||||
|
||||
# Check if lossy copy already exists
|
||||
out_path = os.path.splitext(resolved)[0] + out_ext
|
||||
# Never offer to convert a file onto itself (e.g. .m4a ALAC + AAC target
|
||||
# lands on the same path) — that conversion would destroy the original.
|
||||
if lossy_output_would_overwrite_source(resolved, out_path):
|
||||
continue
|
||||
if os.path.exists(out_path):
|
||||
continue
|
||||
|
||||
|
|
@ -159,7 +184,7 @@ class LossyConverterJob(RepairJob):
|
|||
file_path=file_path,
|
||||
title=f'No {quality_label} copy: {title or "Unknown"}',
|
||||
description=(
|
||||
f'FLAC file "{title}" by {artist_name or "Unknown"} does not have '
|
||||
f'Lossless file "{title}" by {artist_name or "Unknown"} does not have '
|
||||
f'a {quality_label} copy alongside it'
|
||||
),
|
||||
details={
|
||||
|
|
@ -195,7 +220,7 @@ class LossyConverterJob(RepairJob):
|
|||
context.report_progress(
|
||||
scanned=total, total=total,
|
||||
phase='Complete',
|
||||
log_line=f'Found {result.findings_created} FLAC files without {quality_label} copies',
|
||||
log_line=f'Found {result.findings_created} lossless files without {quality_label} copies',
|
||||
log_type='success' if result.findings_created == 0 else 'info'
|
||||
)
|
||||
|
||||
|
|
@ -208,10 +233,10 @@ class LossyConverterJob(RepairJob):
|
|||
try:
|
||||
conn = context.db._get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
cursor.execute(f"""
|
||||
SELECT COUNT(*) FROM tracks
|
||||
WHERE file_path IS NOT NULL AND file_path != ''
|
||||
AND LOWER(file_path) LIKE '%.flac'
|
||||
AND {_lossless_ext_where('file_path')}
|
||||
""")
|
||||
row = cursor.fetchone()
|
||||
return row[0] if row else 0
|
||||
|
|
|
|||
|
|
@ -3373,6 +3373,14 @@ class RepairWorker:
|
|||
return {'success': False, 'error': f'Source file not found: {file_path}'}
|
||||
|
||||
out_path = os.path.splitext(resolved)[0] + out_ext
|
||||
# Safety invariant: ffmpeg runs with -y, so refuse to convert a file onto
|
||||
# itself (an .m4a ALAC source + AAC target shares the .m4a path) — that
|
||||
# would destroy the original lossless file (#941).
|
||||
from core.quality.lossless import lossy_output_would_overwrite_source
|
||||
if lossy_output_would_overwrite_source(resolved, out_path):
|
||||
return {'success': False,
|
||||
'error': f'{codec.upper()} output would overwrite the source file; '
|
||||
f'choose a different lossy codec'}
|
||||
if os.path.exists(out_path):
|
||||
return {'success': True, 'action': 'already_exists',
|
||||
'message': f'{quality_label} copy already exists'}
|
||||
|
|
|
|||
|
|
@ -244,3 +244,61 @@ def test_atomic_helper_cleans_temp_and_keeps_source_on_failure(tmp_path, monkeyp
|
|||
assert src.exists() # source preserved on failure
|
||||
assert not dst.exists() # no partial final file
|
||||
assert not list(dstdir.glob(".*ssync-tmp")) # temp cleaned up
|
||||
|
||||
|
||||
# ── #941: create_lossy_copy now accepts all lossless sources + never overwrites them ──
|
||||
|
||||
import core.imports.file_ops as _fo
|
||||
|
||||
|
||||
def _enable_lossy(monkeypatch, codec="mp3", bitrate="320"):
|
||||
cfg = {"lossy_copy.enabled": True, "lossy_copy.codec": codec,
|
||||
"lossy_copy.bitrate": bitrate, "lossy_copy.delete_original": False}
|
||||
monkeypatch.setattr(_fo.config_manager, "get", lambda k, d=None: cfg.get(k, d))
|
||||
monkeypatch.setattr(_fo, "get_audio_quality_string", lambda _p: None)
|
||||
|
||||
|
||||
def test_create_lossy_copy_rejects_non_lossless(monkeypatch, tmp_path):
|
||||
_enable_lossy(monkeypatch)
|
||||
src = tmp_path / "song.mp3"
|
||||
src.write_bytes(b"id3")
|
||||
assert _fo.create_lossy_copy(str(src)) is None # lossy input → nothing to do
|
||||
|
||||
|
||||
def test_create_lossy_copy_now_accepts_wav(monkeypatch, tmp_path):
|
||||
"""Was FLAC-only; a WAV must now pass the gate and convert (#941)."""
|
||||
_enable_lossy(monkeypatch, codec="mp3")
|
||||
monkeypatch.setattr(_fo.shutil, "which", lambda _name: "/usr/bin/ffmpeg")
|
||||
monkeypatch.setattr("mutagen.File", lambda *_a, **_k: None) # skip tag write
|
||||
|
||||
seen = {}
|
||||
|
||||
def _fake_run(cmd, **_kw):
|
||||
seen["cmd"] = cmd
|
||||
open(cmd[-1], "wb").write(b"fake-mp3") # ffmpeg "writes" the output
|
||||
return types.SimpleNamespace(returncode=0, stderr="")
|
||||
|
||||
monkeypatch.setattr(_fo.subprocess, "run", _fake_run)
|
||||
|
||||
src = tmp_path / "song.wav"
|
||||
src.write_bytes(b"RIFF....WAVE")
|
||||
out = _fo.create_lossy_copy(str(src))
|
||||
assert out and out.endswith(".mp3") # WAV passed the gate + converted
|
||||
assert str(src) in seen["cmd"] # ffmpeg got the .wav input
|
||||
|
||||
|
||||
def test_create_lossy_copy_skips_when_output_would_overwrite_source(monkeypatch, tmp_path):
|
||||
"""REGRESSION: .m4a ALAC source + AAC codec → output is the same .m4a path.
|
||||
ffmpeg (-y) must NEVER run, or it would destroy the original lossless file."""
|
||||
_enable_lossy(monkeypatch, codec="aac", bitrate="256")
|
||||
monkeypatch.setattr(_fo, "m4a_codec", lambda _p: "alac") # source IS ALAC (lossless)
|
||||
|
||||
ran = {"called": False}
|
||||
monkeypatch.setattr(_fo.subprocess, "run",
|
||||
lambda *_a, **_k: ran.__setitem__("called", True))
|
||||
|
||||
src = tmp_path / "track.m4a"
|
||||
src.write_bytes(b"....ALAC....")
|
||||
out = _fo.create_lossy_copy(str(src))
|
||||
assert out is None # skipped — output would overwrite source
|
||||
assert ran["called"] is False # the original was never touched
|
||||
|
|
|
|||
121
tests/quality/test_lossless.py
Normal file
121
tests/quality/test_lossless.py
Normal file
|
|
@ -0,0 +1,121 @@
|
|||
"""The canonical "is this lossless?" seam + the lossy-copy overwrite invariant
|
||||
(#941). All pure — no files, no ffmpeg — so the decision and the safety guard are
|
||||
unit-testable in isolation. Both the import path (create_lossy_copy) and the Lossy
|
||||
Converter repair job route through these, so the same rules drive both."""
|
||||
|
||||
from core.quality.lossless import (
|
||||
LOSSLESS_FORMATS,
|
||||
LOSSLESS_CANDIDATE_EXTENSIONS,
|
||||
is_lossless_format,
|
||||
is_lossless_audio_path,
|
||||
lossy_output_would_overwrite_source,
|
||||
)
|
||||
from core.quality.model import AudioQuality
|
||||
from core.repair_jobs.lossy_converter import _lossless_ext_where
|
||||
|
||||
|
||||
# ── is_lossless_format ──
|
||||
|
||||
def test_lossless_formats_recognized():
|
||||
for fmt in ('flac', 'alac', 'wav', 'dsf', 'FLAC', 'Dsf'):
|
||||
assert is_lossless_format(fmt) is True
|
||||
|
||||
|
||||
def test_lossy_formats_not_lossless():
|
||||
for fmt in ('mp3', 'aac', 'ogg', 'opus', 'wma', 'unknown', '', None):
|
||||
assert is_lossless_format(fmt) is False
|
||||
|
||||
|
||||
# ── is_lossless_audio_path (the ambiguity is the whole point) ──
|
||||
|
||||
def test_unambiguous_extensions_decided_by_extension():
|
||||
for path in ('/m/a.flac', '/m/a.wav', '/m/a.wave', '/m/a.aiff', '/m/a.aif',
|
||||
'/m/a.dsf', '/m/a.dff', '/m/a.alac', '/m/A.FLAC'):
|
||||
assert is_lossless_audio_path(path) is True
|
||||
|
||||
|
||||
def test_lossy_extensions_are_not_lossless():
|
||||
for path in ('/m/a.mp3', '/m/a.ogg', '/m/a.opus', '/m/a.wma', '/m/a.aac'):
|
||||
assert is_lossless_audio_path(path) is False
|
||||
|
||||
|
||||
def test_m4a_without_probe_is_not_lossless():
|
||||
# The safe default: with no codec probe, an .m4a can't be proven lossless, so
|
||||
# an AAC file is never misclassified as lossless and converted/deleted.
|
||||
assert is_lossless_audio_path('/m/a.m4a') is False
|
||||
assert is_lossless_audio_path('/m/a.mp4') is False
|
||||
|
||||
|
||||
def test_m4a_alac_is_lossless_via_probe():
|
||||
assert is_lossless_audio_path('/m/a.m4a', probe_codec=lambda _p: 'alac') is True
|
||||
assert is_lossless_audio_path('/m/a.mp4', probe_codec=lambda _p: 'ALAC') is True
|
||||
|
||||
|
||||
def test_m4a_aac_is_not_lossless_via_probe():
|
||||
assert is_lossless_audio_path('/m/a.m4a', probe_codec=lambda _p: 'mp4a.40.2') is False
|
||||
|
||||
|
||||
def test_probe_exception_is_not_lossless():
|
||||
def _boom(_p):
|
||||
raise RuntimeError("probe failed")
|
||||
assert is_lossless_audio_path('/m/a.m4a', probe_codec=_boom) is False
|
||||
|
||||
|
||||
# ── LOSSLESS_CANDIDATE_EXTENSIONS (the SQL pre-filter set) ──
|
||||
|
||||
def test_candidate_extensions_cover_lossless_plus_ambiguous():
|
||||
for e in ('.flac', '.wav', '.aiff', '.dsf', '.dff', '.alac', '.m4a', '.mp4'):
|
||||
assert e in LOSSLESS_CANDIDATE_EXTENSIONS
|
||||
# raw lossy extensions must NOT be candidates
|
||||
for e in ('.mp3', '.aac', '.ogg', '.opus', '.wma'):
|
||||
assert e not in LOSSLESS_CANDIDATE_EXTENSIONS
|
||||
|
||||
|
||||
def test_sql_where_clause_matches_candidates_only():
|
||||
where = _lossless_ext_where('t.file_path')
|
||||
assert "LIKE '%.flac'" in where and "LIKE '%.dsf'" in where and "LIKE '%.m4a'" in where
|
||||
assert "LIKE '%.mp3'" not in where and "LIKE '%.aac'" not in where
|
||||
|
||||
|
||||
# ── lossy_output_would_overwrite_source (the safety invariant) ──
|
||||
|
||||
def test_overwrite_detected_when_paths_equal():
|
||||
assert lossy_output_would_overwrite_source('/m/Album/01.m4a', '/m/Album/01.m4a') is True
|
||||
|
||||
|
||||
def test_overwrite_detected_after_normalization():
|
||||
assert lossy_output_would_overwrite_source('/m/Album/../Album/01.m4a', '/m/Album/01.m4a') is True
|
||||
|
||||
|
||||
def test_no_overwrite_for_different_extension():
|
||||
assert lossy_output_would_overwrite_source('/m/Album/01.flac', '/m/Album/01.mp3') is False
|
||||
assert lossy_output_would_overwrite_source('/m/Album/01.m4a', '/m/Album/01.mp3') is False
|
||||
|
||||
|
||||
def test_overwrite_guard_handles_empty():
|
||||
assert lossy_output_would_overwrite_source('', '/x.mp3') is False
|
||||
assert lossy_output_would_overwrite_source('/x.flac', '') is False
|
||||
|
||||
|
||||
# ── anti-drift: the seam must agree with the quality tier model ──
|
||||
|
||||
def test_lossless_set_consistent_with_tier_model():
|
||||
"""If a format is in LOSSLESS_FORMATS it must out-rank every lossy format in
|
||||
tier_score — guards against the two lists drifting apart (the whole reason
|
||||
this seam exists)."""
|
||||
lossy = ('mp3', 'aac', 'ogg', 'opus', 'wma')
|
||||
worst_lossless = min(AudioQuality(f, bitrate=11290, sample_rate=44100, bit_depth=16).tier_score()
|
||||
for f in LOSSLESS_FORMATS)
|
||||
best_lossy = max(AudioQuality(f, bitrate=320).tier_score() for f in lossy)
|
||||
assert worst_lossless > best_lossy
|
||||
|
||||
|
||||
# ── regression: the exact bug class this guards (overwrite the original) ──
|
||||
|
||||
def test_regression_m4a_alac_to_aac_would_overwrite_and_is_blocked():
|
||||
"""An .m4a ALAC source converted with the AAC codec lands on the SAME .m4a
|
||||
path. The guard must catch it so ffmpeg -y never destroys the original."""
|
||||
src = '/library/Sade/Diamond Life/01. Smooth Operator.m4a' # ALAC
|
||||
out = src # AAC target → .m4a
|
||||
assert is_lossless_audio_path(src, probe_codec=lambda _p: 'alac') is True
|
||||
assert lossy_output_would_overwrite_source(src, out) is True # → callers skip
|
||||
Loading…
Reference in a new issue