From a207bd943bca82692eda0a8970f9bf591446f864 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Thu, 11 Jun 2026 11:07:25 -0700 Subject: [PATCH] #845 tests: lift history-path resolver to core/ + seam-test the delete-safety MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- core/matching/history_paths.py | 65 +++++++++++++++++++++ tests/matching/test_history_paths.py | 87 ++++++++++++++++++++++++++++ web_server.py | 55 +++++++----------- 3 files changed, 173 insertions(+), 34 deletions(-) create mode 100644 core/matching/history_paths.py create mode 100644 tests/matching/test_history_paths.py diff --git a/core/matching/history_paths.py b/core/matching/history_paths.py new file mode 100644 index 00000000..8d3c0509 --- /dev/null +++ b/core/matching/history_paths.py @@ -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//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 (Docker↔host / transfer→library), + 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"] diff --git a/tests/matching/test_history_paths.py b/tests/matching/test_history_paths.py new file mode 100644 index 00000000..140abe91 --- /dev/null +++ b/tests/matching/test_history_paths.py @@ -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 diff --git a/web_server.py b/web_server.py index 8d4e147c..dc5b2e36 100644 --- a/web_server.py +++ b/web_server.py @@ -7598,40 +7598,27 @@ def _resolve_history_audio_path(row): 3. the tracks table — the media-server mirror knows the CURRENT path for this title+artist even after a rename — resolved the same way. """ - raw_path = (row.get('file_path') or '').strip() - if raw_path and os.path.exists(raw_path): - return raw_path - resolved = _resolve_library_file_path(raw_path) if raw_path else None - if resolved and os.path.exists(resolved): - return resolved - title = (row.get('title') or '').strip() - if not title: - return None - artist = (row.get('artist_name') or '').strip() - try: - conn = get_database()._get_connection() - cursor = conn.cursor() - cursor.execute( - "SELECT file_path FROM tracks WHERE file_path IS NOT NULL AND LOWER(title) = LOWER(?)", - (title,)) - candidates = [r[0] for r in cursor.fetchall() if r[0]] - except Exception as e: - logger.debug(f"[Verification] tracks-table path fallback failed: {e}") - return None - # 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 + def _lookup_titled_paths(title): + # tracks-table mirror: paths for this title (knows the CURRENT path + # after a media-server rename). [] on any DB error → resolver returns None. + try: + conn = get_database()._get_connection() + cursor = conn.cursor() + cursor.execute( + "SELECT file_path FROM tracks WHERE file_path IS NOT NULL AND LOWER(title) = LOWER(?)", + (title,)) + return [r[0] for r in cursor.fetchall() if r[0]] + except Exception as e: + logger.debug(f"[Verification] tracks-table path fallback failed: {e}") + return [] + + from core.matching.history_paths import resolve_history_audio_path + return resolve_history_audio_path( + row, + exists=os.path.exists, + resolve_library_path=_resolve_library_file_path, + lookup_titled_paths=_lookup_titled_paths, + ) @app.route('/api/verification//stream', methods=['GET'])