andrew
e49d12a72b
First automated coverage. 38 tests, run against the built image (which has the app deps); tests/ and requirements-dev.txt are not baked into the runtime image. - tests/test_dedup_gating.py: the safety-critical one. Drives dedup-library.sh's process_group with --apply + --only-paths and asserts ONLY the confirmed path is deleted and the higher-ranked FLAC is kept. - tests/test_pipeline_runner.py: lock skip, use_lock=False bypass, lock release, success/failure/timeout recording. - tests/test_db.py: schema version + idempotent migrations, FK-safe job-run pruning, stale-run reconciliation. - tests/test_credential_service.py: conf-field substitution incl. newline- injection safety, secret/non-secret classification, cookie-expiry parsing. - tests/test_playlist_service.py: cron round-trips, name validation, render. - tests/test_diagnostics.py: /health + /setup checks. - conftest.py points every path setting at throwaway temp dirs and a temp DB, so tests never touch a real library, /config, or the network. See tests/README.md. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
54 lines
2.3 KiB
Python
54 lines
2.3 KiB
Python
"""The safety-critical test: dedup-library.sh must delete ONLY the files a
|
|
human confirmed via --only-paths, and must keep the higher-ranked copy.
|
|
|
|
Rather than stand up a full beets fixture, this sources the script's setup +
|
|
function definitions (everything before the Pass 1 top-level code) and drives
|
|
process_group directly with "ghost" (bare-path) entries, which skip the beets
|
|
path translation and are deleted with plain rm -- exactly the code path that
|
|
enforces --only-paths gating and FLAC-over-MP3 ranking."""
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parent.parent
|
|
SCRIPT = REPO / "pipeline" / "lib" / "dedup-library.sh"
|
|
|
|
|
|
def _drive_process_group(work: Path) -> subprocess.CompletedProcess:
|
|
keep = work / "keep.flac"
|
|
dupe = work / "dupe.mp3"
|
|
other = work / "other.mp3"
|
|
for f, size in ((keep, 5000), (dupe, 1000), (other, 1000)):
|
|
f.write_bytes(b"\0" * size)
|
|
confirm = work / "confirm.txt"
|
|
confirm.write_text(str(dupe) + "\n") # only dupe.mp3 is confirmed for deletion
|
|
|
|
script = r"""
|
|
set -euo pipefail
|
|
export ALEMBIC_CONFIG_DIR="{work}" MUSIC_DATA_DIR="{work}"
|
|
mkdir -p "{work}/logs"
|
|
# Source only setup + functions (stop before the Pass 1 top-level code).
|
|
end=$(grep -n '^# Pass 1: numbered siblings' "{script}" | head -1 | cut -d: -f1)
|
|
source <(head -n $((end - 2)) "{script}")
|
|
APPLY=1
|
|
ONLY_PATHS_FILE="{confirm}"
|
|
declare -A ONLY_PATHS=(["{dupe}"]=1)
|
|
# Group 1: keep.flac vs dupe.mp3 (dupe confirmed -> deleted)
|
|
process_group "P2 case-insensitive: g1" "{keep}" "{dupe}"
|
|
# Group 2: keep.flac vs other.mp3 (other NOT confirmed -> must survive)
|
|
process_group "P2 case-insensitive: g2" "{keep}" "{other}"
|
|
""".format(work=work, script=SCRIPT, confirm=confirm, dupe=dupe, keep=keep, other=other)
|
|
return subprocess.run(["bash", "-c", script], capture_output=True, text=True)
|
|
|
|
|
|
def test_only_confirmed_paths_are_deleted(tmp_path):
|
|
proc = _drive_process_group(tmp_path)
|
|
assert proc.returncode == 0, proc.stderr
|
|
|
|
keep = tmp_path / "keep.flac"
|
|
dupe = tmp_path / "dupe.mp3"
|
|
other = tmp_path / "other.mp3"
|
|
|
|
assert keep.exists(), "FLAC (higher rank) must be kept"
|
|
assert not dupe.exists(), "confirmed duplicate must be deleted"
|
|
assert other.exists(), "an unconfirmed delete-candidate must NOT be deleted"
|