#845 tests: lift history-path resolver to core/ + seam-test the delete-safety

resolve_history_audio_path drives a DESTRUCTIVE delete (os.remove), but lived
endpoint-bound in web_server with zero tests. Lifted to core/matching/history_paths
with injected effects (exists / resolve_library_path / lookup_titled_paths) so the
fallback chain — and the collision-safety that stops delete() from removing the
wrong same-title file — is a clean importable seam. web_server now wraps it (DB
lookup + os.path.exists + prefix resolver injected); behavior preserved.

9 tests lock it: recorded-path hit, prefix-resolve fallback, single tracks-table
candidate, and the safety rules — multiple same-title candidates with NO artist ->
None (refuse to guess), artist filter picks only the matching path, artist named
but unmatched -> None, no-title/empty-lookup -> None. Full suite green (5906).
This commit is contained in:
BoulderBadgeDad 2026-06-11 11:07:25 -07:00
parent 89f843d223
commit a207bd943b
3 changed files with 173 additions and 34 deletions

View file

@ -0,0 +1,65 @@
"""Resolve a ``library_history`` row to a playable on-disk file.
Lifted out of ``web_server`` so the fallback chain and its collision-safety
is an importable, unit-tested seam. This matters because a *destructive* delete
(``/api/verification/<id>/delete`` ``os.remove``) trusts the path this returns:
if the tracks-table fallback guessed the wrong same-title file, delete would
remove the wrong track. The rules below are exactly that guard, and the tests
lock them.
Side effects are injected so the decision logic is pure:
- ``exists(path) -> bool`` (os.path.exists)
- ``resolve_library_path(raw) -> str | None`` (transfer/download/library prefix swap)
- ``lookup_titled_paths(title) -> list[str]`` (tracks.file_path WHERE LOWER(title)=LOWER(?))
"""
from __future__ import annotations
from typing import Any, Callable, Dict, List, Optional
def resolve_history_audio_path(
row: Dict[str, Any],
*,
exists: Callable[[str], bool],
resolve_library_path: Callable[[str], Optional[str]],
lookup_titled_paths: Callable[[str], List[str]],
) -> Optional[str]:
"""Return the on-disk path for a history row, or None. Fallback chain:
1. the recorded path as-is,
2. the prefix-swap resolver (Dockerhost / transferlibrary),
3. the tracks-table mirror by title (knows the CURRENT path after a rename),
resolved the same way but only when it can be picked UNAMBIGUOUSLY.
"""
raw_path = (row.get("file_path") or "").strip()
if raw_path and exists(raw_path):
return raw_path
resolved = resolve_library_path(raw_path) if raw_path else None
if resolved and exists(resolved):
return resolved
title = (row.get("title") or "").strip()
if not title:
return None
artist = (row.get("artist_name") or "").strip()
candidates = [p for p in (lookup_titled_paths(title) or []) if p]
# Same-title collisions across artists exist, and delete() trusts this path,
# so be strict: when the row names an artist, only accept candidates whose
# path mentions it; with no artist, only an unambiguous single candidate.
artist_l = artist.lower()
if artist_l:
candidates = [p for p in candidates if artist_l in p.lower()]
elif len(candidates) != 1:
return None
for cand in candidates:
cand_resolved = resolve_library_path(cand)
if cand_resolved and exists(cand_resolved):
return cand_resolved
return None
__all__ = ["resolve_history_audio_path"]

View file

@ -0,0 +1,87 @@
"""Seam tests for resolve_history_audio_path — the fallback chain a DESTRUCTIVE
delete trusts. The collision-safety rules (artist filter / single-candidate) are
what stop delete() from removing the wrong same-title file, so they're locked here."""
from __future__ import annotations
from core.matching.history_paths import resolve_history_audio_path
def _resolver(existing=(), resolve_map=None, titled=None):
existing = set(existing)
resolve_map = resolve_map or {}
titled = titled or {}
return dict(
exists=lambda p: p in existing,
resolve_library_path=lambda raw: resolve_map.get(raw),
lookup_titled_paths=lambda title: list(titled.get(title.lower(), [])),
)
def test_recorded_path_used_when_it_exists():
r = resolve_history_audio_path({'file_path': '/m/a.mp3'}, **_resolver(existing={'/m/a.mp3'}))
assert r == '/m/a.mp3'
def test_falls_back_to_prefix_resolved_path():
r = resolve_history_audio_path(
{'file_path': '/transfer/a.mp3'},
**_resolver(existing={'/library/a.mp3'}, resolve_map={'/transfer/a.mp3': '/library/a.mp3'}))
assert r == '/library/a.mp3'
def test_tracks_table_single_candidate_no_artist():
r = resolve_history_audio_path(
{'file_path': '/gone.mp3', 'title': 'Song'},
**_resolver(existing={'/library/song.mp3'},
resolve_map={'/lib/song.mp3': '/library/song.mp3'},
titled={'song': ['/lib/song.mp3']}))
assert r == '/library/song.mp3'
def test_collision_no_artist_multiple_candidates_returns_none():
# THE safety rule: same title, no artist to disambiguate -> refuse to guess
# (delete() must not remove an arbitrary one of two same-title files).
r = resolve_history_audio_path(
{'file_path': '', 'title': 'Intro'},
**_resolver(existing={'/library/a/intro.mp3', '/library/b/intro.mp3'},
resolve_map={'/a/intro.mp3': '/library/a/intro.mp3', '/b/intro.mp3': '/library/b/intro.mp3'},
titled={'intro': ['/a/intro.mp3', '/b/intro.mp3']}))
assert r is None
def test_artist_filter_picks_only_the_matching_path():
# Two same-title files by different artists -> only the one whose path
# mentions the row's artist is eligible (won't delete the other artist's file).
r = resolve_history_audio_path(
{'file_path': '', 'title': 'Intro', 'artist_name': 'Alpha'},
**_resolver(existing={'/music/Alpha/intro.mp3', '/music/Beta/intro.mp3'},
resolve_map={'/Alpha/intro.mp3': '/music/Alpha/intro.mp3',
'/Beta/intro.mp3': '/music/Beta/intro.mp3'},
titled={'intro': ['/Alpha/intro.mp3', '/Beta/intro.mp3']}))
assert r == '/music/Alpha/intro.mp3'
def test_artist_named_but_no_path_mentions_it_returns_none():
r = resolve_history_audio_path(
{'file_path': '', 'title': 'Intro', 'artist_name': 'Gamma'},
**_resolver(existing={'/music/Alpha/intro.mp3'},
resolve_map={'/Alpha/intro.mp3': '/music/Alpha/intro.mp3'},
titled={'intro': ['/Alpha/intro.mp3']}))
assert r is None
def test_no_title_returns_none():
assert resolve_history_audio_path({'file_path': '/gone.mp3'}, **_resolver()) is None
def test_nothing_resolves_returns_none():
assert resolve_history_audio_path(
{'file_path': '/gone.mp3', 'title': 'X'},
**_resolver(titled={'x': ['/also/gone.mp3']})) is None
def test_empty_lookup_returns_none():
# DB-error proxy: lookup yields [] -> None (never a stray delete target).
assert resolve_history_audio_path(
{'file_path': '', 'title': 'X'}, **_resolver(titled={})) is None

View file

@ -7598,40 +7598,27 @@ def _resolve_history_audio_path(row):
3. the tracks table the media-server mirror knows the CURRENT path for 3. the tracks table the media-server mirror knows the CURRENT path for
this title+artist even after a rename resolved the same way. this title+artist even after a rename resolved the same way.
""" """
raw_path = (row.get('file_path') or '').strip() def _lookup_titled_paths(title):
if raw_path and os.path.exists(raw_path): # tracks-table mirror: paths for this title (knows the CURRENT path
return raw_path # after a media-server rename). [] on any DB error → resolver returns None.
resolved = _resolve_library_file_path(raw_path) if raw_path else None try:
if resolved and os.path.exists(resolved): conn = get_database()._get_connection()
return resolved cursor = conn.cursor()
title = (row.get('title') or '').strip() cursor.execute(
if not title: "SELECT file_path FROM tracks WHERE file_path IS NOT NULL AND LOWER(title) = LOWER(?)",
return None (title,))
artist = (row.get('artist_name') or '').strip() return [r[0] for r in cursor.fetchall() if r[0]]
try: except Exception as e:
conn = get_database()._get_connection() logger.debug(f"[Verification] tracks-table path fallback failed: {e}")
cursor = conn.cursor() return []
cursor.execute(
"SELECT file_path FROM tracks WHERE file_path IS NOT NULL AND LOWER(title) = LOWER(?)", from core.matching.history_paths import resolve_history_audio_path
(title,)) return resolve_history_audio_path(
candidates = [r[0] for r in cursor.fetchall() if r[0]] row,
except Exception as e: exists=os.path.exists,
logger.debug(f"[Verification] tracks-table path fallback failed: {e}") resolve_library_path=_resolve_library_file_path,
return None lookup_titled_paths=_lookup_titled_paths,
# Same-title collisions across artists exist (and delete() trusts this )
# path), so be strict: when the row names an artist, only accept
# candidates whose path mentions it; otherwise only an unambiguous
# single candidate.
artist_l = artist.lower()
if artist_l:
candidates = [p for p in candidates if artist_l in p.lower()]
elif len(candidates) != 1:
return None
for cand in candidates:
cand_resolved = _resolve_library_file_path(cand)
if cand_resolved and os.path.exists(cand_resolved):
return cand_resolved
return None
@app.route('/api/verification/<int:history_id>/stream', methods=['GET']) @app.route('/api/verification/<int:history_id>/stream', methods=['GET'])