From d812559b15e9937cfaf070e38fcc88fc19bd6642 Mon Sep 17 00:00:00 2001 From: jrgutier Date: Fri, 31 Jul 2026 15:19:24 -0500 Subject: [PATCH] Refuse to publish a database that disagrees with its own source MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gate added in 0.3.2 compares the written database against what insert_playlists intended, and both sides of that comparison derive from track_id_map. A mapping fault is therefore invisible to it: the intent and the rows agree with each other while both disagree with the source. The playlist is internally coherent and points at the wrong track. This adds a second, independent check at the same pre-publish point. It recomputes every expected Engine track id from the source track itself, through map_track and the database's own path index, and never consults track_id_map — so it fails exactly where the intent gate cannot. Demonstrated, not asserted: with insert_tracks returning a swapped map, the intent gate passes and the build publishes. Stub out only the new check and the same corrupt build publishes again; leave it in and the conversion is refused with "1 entries written, 1 expected from the source (unexpected track ids [2]; absent track ids [1])". Placement and blast radius -------------------------- It runs on the staged copy before os.replace, so a refusal leaves the previous m.db byte-for-byte intact — asserted on the bytes, not on the absence of an exception. build_library wraps the failure as FatalError, which convert maps to exit 2. One implementation ------------------ The comparison moved to playlist_check.py, which verify.py also calls; verify records findings and keeps checking, the writer aborts. They can no longer disagree about what is wrong. map_track is resolved at call time, matching build.py, because binding it at import made the checker able to run a different mapper than the writer just used — that showed up as an order-dependent test failure and would have made the comparison meaningless rather than independent. What it cannot do ----------------- It compares a parse against itself, so it can never detect that the source file was misread. That is the reader's job (G1d). It is playlist-scoped; verify remains the field-level check. Verified on the real 3,673-track stick, read-only: 45 playlists, 0 problems, 1.4 s. 704 tests, 88% branch coverage, playlist_check at 100%, ruff + mypy clean. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01V3aF4y4GdsuJta2bfYUZ8w --- CHANGELOG.md | 19 +++ src/rb2engine/playlist_check.py | 221 ++++++++++++++++++++++++++++++ src/rb2engine/verify.py | 156 ++++----------------- src/rb2engine/writer/build.py | 31 ++++- tests/unit/test_build.py | 141 +++++++++++++++++++ tests/unit/test_playlist_check.py | 57 ++++++++ tests/unit/test_verify.py | 2 +- 7 files changed, 492 insertions(+), 135 deletions(-) create mode 100644 src/rb2engine/playlist_check.py create mode 100644 tests/unit/test_playlist_check.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ed8762..2f563a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **`convert` now refuses to publish a database that disagrees with its own + source.** Before the new `m.db` is swapped into place, every playlist's + membership, order and entry chain is recomputed from the source and compared + against what was actually written. On any disagreement the conversion fails + with exit 2 and your existing library is left byte-for-byte intact. + + This is deliberately *not* the check added in 0.3.2. That one compares the + database against what the writer intended, and both sides of it derive from + the same track id map — so a mapping fault agrees with itself and passes. The + new check recomputes each expected track id from the source track through the + mapper and the database's own path index, never consulting that map, and so + fails exactly where the older gate cannot. + + Scope, stated plainly: it is playlist-scoped, not a full verify, and it cannot + tell you the source *file* was misread — both sides descend from the same + parse. That is the reader's job (see G1d above). `rb2engine verify` remains + the field-level check. + ### Fixed - **G1d — refuse a torn `export.pdb` instead of converting it.** A conversion on a real stick published two playlist entries that the settled `export.pdb` does diff --git a/src/rb2engine/playlist_check.py b/src/rb2engine/playlist_check.py new file mode 100644 index 0000000..ddaa258 --- /dev/null +++ b/src/rb2engine/playlist_check.py @@ -0,0 +1,221 @@ +"""Compare an Engine database's playlists against the source they came from. + +Shared by ``verify`` and by the writer's pre-publish gate. The two react +differently — verify records a finding and keeps checking, the writer refuses to +publish — but they must never disagree about what is wrong, which is why the +comparison itself lives here. + +Why this exists separately from ``assert_entities_match_intent`` +--------------------------------------------------------------- +That gate compares the database against what ``insert_playlists`` *intended* to +write, and both sides of it derive from ``track_id_map``. A mapping fault is +therefore invisible to it: the intent and the rows agree with each other while +both disagree with the source. + +This module never receives ``track_id_map``. It recomputes each expected Engine +track id from the source track itself, through ``map_track`` and the database's +own ``Track.path`` index — the same route Engine will use to find the file. So +it fails exactly where the intent gate cannot. + +What it cannot do: it compares a parse against itself. Run inline during a +conversion it can never detect that the source *file* was misread, because both +sides descend from the same parse. Torn-source detection is the reader's job +(see ``reader/pdb.py`` G1d). +""" + +from __future__ import annotations + +import sqlite3 +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import cast + +from rb2engine.chain import ChainInconsistent, walk_entity_chain +from rb2engine.ir import SourceLibrary, SourcePlaylist +from rb2engine.playlist_naming import format_path, resolve_paths + +# Discrepancy kinds, kept as constants so verify's field names and the writer's +# error text cannot drift apart. +MISSING = "missing" +CHAIN = "chain" +TRACK_ORDER = "track_order" + + +@dataclass(frozen=True) +class PlaylistProblem: + """One disagreement between a database playlist and its source.""" + + path: tuple[str, ...] + kind: str + expected: object + actual: object + + @property + def label(self) -> str: + return format_path(self.path) + + def describe(self) -> str: + if self.kind == MISSING: + return f"playlist {self.label!r} is absent from the database" + if self.kind == CHAIN: + return f"playlist {self.label!r} has a broken entry chain: {self.actual}" + exp = cast(list[int], self.expected) + act = cast(list[int], self.actual) + extra = sorted(set(act) - set(exp)) + missing = sorted(set(exp) - set(act)) + return ( + f"playlist {self.label!r}: {len(act)} entries written, " + f"{len(exp)} expected from the source " + f"(unexpected track ids {extra or 'none'}; " + f"absent track ids {missing or 'none'})" + ) + + +def db_playlist_paths(conn: sqlite3.Connection) -> dict[tuple[str, ...], int]: + """Engine playlist path (root first) → list id. + + Built by walking ``parentListId`` to the root, so a title is only ever + matched within its own folder. + """ + rows: dict[int, tuple[str, int]] = { + int(r[0]): (str(r[1]), int(r[2])) + for r in conn.execute("SELECT id, title, parentListId FROM Playlist") + } + paths: dict[tuple[str, ...], int] = {} + for list_id in rows: + parts: list[str] = [] + cur = list_id + walked: set[int] = set() + # Guard against a cyclic parent chain in a database we did not write. + while cur != 0 and cur in rows and cur not in walked: + walked.add(cur) + title, parent = rows[cur] + parts.append(title) + cur = parent + paths[tuple(reversed(parts))] = list_id + return paths + + +def db_track_ids_by_path(conn: sqlite3.Connection) -> dict[str, int]: + """Engine ``Track.path`` → ``Track.id``.""" + return { + str(path): int(tid) for tid, path in conn.execute("SELECT id, path FROM Track") + } + + +def expected_entity_track_ids( + pl: SourcePlaylist, + source: SourceLibrary, + *, + track_id_by_path: Mapping[str, int], + drive_root: Path, + engine_lib: Path, +) -> list[int]: + """Engine track ids this playlist should hold, derived from the source. + + Order is preserved. Tracks that were skipped during conversion have no + database row and drop out here the same way, and a track repeated inside one + playlist keeps only its first occurrence — Engine's uniqueness constraint + does not permit the repeat. + """ + # Resolved at call time, exactly as writer/build.py does. Binding it at + # import would let this check run a different mapper than the writer just + # used, which would make the comparison meaningless rather than independent. + from rb2engine.mapper.track import map_track + + out: list[int] = [] + seen: set[int] = set() + for rb in pl.track_rb_ids: + src = source.tracks.get(rb) + if src is None or src.resolved_path is None: + continue + et = map_track(src, drive_root=drive_root, engine_library_dir=engine_lib) + db_id = track_id_by_path.get(et.path) + if db_id is None or db_id in seen: + continue + seen.add(db_id) + out.append(db_id) + return out + + +def entity_track_order( + conn: sqlite3.Connection, list_id: int +) -> tuple[list[int], str | None]: + """Track order from the nextEntityId chain, plus any inconsistency found. + + Returns ``(order, problem)``. ``problem`` is None when the chain accounts + for every row; otherwise it describes what is wrong and ``order`` holds the + rows in id order as a best effort. + + The writer must abort on a problem and verify must record it and keep going, + which is why this reports rather than decides. + """ + rows = [ + (int(eid), int(track_id), int(next_id)) + for eid, track_id, next_id in conn.execute( + "SELECT id, trackId, nextEntityId FROM PlaylistEntity WHERE listId = ?", + (list_id,), + ) + ] + try: + return walk_entity_chain(list_id, rows), None + except ChainInconsistent as exc: + return [track_id for _, track_id, _ in rows], str(exc) + + +def compare_playlists( + source: SourceLibrary, + conn: sqlite3.Connection, + *, + drive_root: Path, + engine_lib: Path, + track_id_by_path: Mapping[str, int] | None = None, +) -> list[PlaylistProblem]: + """Every disagreement between *conn*'s playlists and *source*. + + Empty means the database's playlist membership, order and chain integrity + all match what the source implies. ``track_id_by_path`` is optional purely + to let a caller that already loaded the Track table avoid a second query. + """ + if track_id_by_path is None: + track_id_by_path = db_track_ids_by_path(conn) + + problems: list[PlaylistProblem] = [] + path_to_id = db_playlist_paths(conn) + source_paths = resolve_paths(source.playlists) + + for pl in source.playlists: + path = source_paths[pl.rb_id] + list_id = path_to_id.get(path) + if list_id is None: + problems.append( + PlaylistProblem(path, MISSING, expected="present", actual="absent") + ) + continue + + expected = expected_entity_track_ids( + pl, + source, + track_id_by_path=track_id_by_path, + drive_root=drive_root, + engine_lib=engine_lib, + ) + actual, chain_problem = entity_track_order(conn, list_id) + if chain_problem is not None: + # Reported in its own right: a broken chain is a defect even when + # the set of tracks happens to match what the source expected. + problems.append( + PlaylistProblem( + path, + CHAIN, + expected="every row reachable from the nextEntityId chain", + actual=chain_problem, + ) + ) + if expected != actual: + problems.append( + PlaylistProblem(path, TRACK_ORDER, expected=expected, actual=actual) + ) + + return problems diff --git a/src/rb2engine/verify.py b/src/rb2engine/verify.py index de018ed..747f659 100644 --- a/src/rb2engine/verify.py +++ b/src/rb2engine/verify.py @@ -14,12 +14,11 @@ from pathlib import Path from typing import Any -from rb2engine.chain import ChainInconsistent, walk_entity_chain from rb2engine.errors import FatalError -from rb2engine.ir import SourceLibrary, SourcePlaylist, SourceTrack +from rb2engine.ir import SourceLibrary, SourceTrack from rb2engine.ir_engine import artwork_content_hash from rb2engine.mapper.track import map_track -from rb2engine.playlist_naming import format_path, resolve_paths +from rb2engine.playlist_check import compare_playlists from rb2engine.reader.library import read_library from rb2engine.writer.blobs import ( decode_beat_data, @@ -66,7 +65,13 @@ def ok(self) -> bool: return len(self.discrepancies) == 0 def render_text(self) -> str: - """Human-readable summary for CLI / convert post-pass.""" + """Human-readable summary for the ``verify`` command. + + ``convert`` does not render this. It runs its own playlist-scoped + recheck before publishing (``playlist_check.compare_playlists``) and + refuses rather than reports; a full field-level verify stays an explicit + second step. + """ status = "OK" if self.ok else "FAILED" lines = [ "rb2engine verify", @@ -538,136 +543,25 @@ def _compare_playlists( ) ) - # Pair on the whole path, not the title: the same title may legally exist - # under several folders (Engine's constraint is per-parent), and duplicates - # within one folder are renamed by the writer. Both are predicted by - # playlist_naming, which the writer uses to assign the names in the first - # place. - path_to_id = _db_playlist_paths(conn) - source_paths = resolve_paths(source.playlists) - - for pl in source.playlists: - path = source_paths[pl.rb_id] - label = format_path(path) - list_id = path_to_id.get(path) - if list_id is None: - discrepancies.append( - Discrepancy( - track_id=None, - field=f"playlist[{label}].missing", - expected="present", - actual="absent", - ) + # The comparison itself lives in playlist_check, which the writer's + # pre-publish gate also calls. verify records findings and keeps checking; + # the writer aborts. They must not disagree about what is wrong. + for problem in compare_playlists( + source, + conn, + drive_root=drive_root, + engine_lib=engine_lib, + track_id_by_path={path: t.id for path, t in by_path.items()}, + ): + discrepancies.append( + Discrepancy( + track_id=None, + field=f"playlist[{problem.label}].{problem.kind}", + expected=problem.expected, + actual=problem.actual, ) - continue - - expected_track_ids = _expected_entity_track_ids( - pl, source, by_path=by_path, drive_root=drive_root, engine_lib=engine_lib ) - actual_track_ids, chain_problem = _entity_track_order(conn, list_id) - if chain_problem is not None: - # Report it in its own right: a broken chain is a defect even when - # the set of tracks happens to match what the source expected. - discrepancies.append( - Discrepancy( - track_id=None, - field=f"playlist[{label}].chain", - expected="every row reachable from the nextEntityId chain", - actual=chain_problem, - ) - ) - if expected_track_ids != actual_track_ids: - discrepancies.append( - Discrepancy( - track_id=None, - field=f"playlist[{label}].track_order", - expected=expected_track_ids, - actual=actual_track_ids, - ) - ) - - -def _db_playlist_paths( - conn: sqlite3.Connection, -) -> dict[tuple[str, ...], int]: - """Engine playlist path (root first) → list id. - - Built by walking ``parentListId`` to the root, so a title is only ever - matched within its own folder. - """ - rows: dict[int, tuple[str, int]] = { - int(r[0]): (str(r[1]), int(r[2])) - for r in conn.execute("SELECT id, title, parentListId FROM Playlist") - } - paths: dict[tuple[str, ...], int] = {} - for list_id in rows: - parts: list[str] = [] - cur = list_id - walked: set[int] = set() - # Guard against a cyclic parent chain in a database we did not write. - while cur != 0 and cur in rows and cur not in walked: - walked.add(cur) - title, parent = rows[cur] - parts.append(title) - cur = parent - paths[tuple(reversed(parts))] = list_id - return paths - - -def _expected_entity_track_ids( - pl: SourcePlaylist, - source: SourceLibrary, - *, - by_path: dict[str, _DbTrack], - drive_root: Path, - engine_lib: Path, -) -> list[int]: - """Map source track_rb_ids → Engine Track.id via expected path (order preserved).""" - out: list[int] = [] - seen: set[int] = set() - for rb in pl.track_rb_ids: - src = source.tracks.get(rb) - if src is None: - continue - if src.resolved_path is None: - continue - et = map_track(src, drive_root=drive_root, engine_library_dir=engine_lib) - db = by_path.get(et.path) - if db is None: - continue - if db.id in seen: - continue # Engine de-dupes within a playlist (first occurrence wins) - seen.add(db.id) - out.append(db.id) - return out - -def _entity_track_order( - conn: sqlite3.Connection, list_id: int -) -> tuple[list[int], str | None]: - """Track order from the nextEntityId chain, plus any inconsistency found. - - Returns ``(order, problem)``. ``problem`` is None when the chain accounts - for every row; otherwise it describes what is wrong and ``order`` holds the - rows in id order as a best effort. - - This used to swallow an inconsistent chain and return the shorter, tidier - list, which reported a clean library while the writer's own gate would - refuse to publish that exact database. Both now walk the same code - (``rb2engine.chain``); they differ only in how they react, because ``verify`` - must record the finding and keep checking rather than abort. - """ - rows = [ - (int(eid), int(track_id), int(next_id)) - for eid, track_id, next_id in conn.execute( - "SELECT id, trackId, nextEntityId FROM PlaylistEntity WHERE listId = ?", - (list_id,), - ) - ] - try: - return walk_entity_chain(list_id, rows), None - except ChainInconsistent as exc: - return [track_id for _, track_id, _ in rows], str(exc) def _compare_artwork( diff --git a/src/rb2engine/writer/build.py b/src/rb2engine/writer/build.py index 46fff6e..5f11d80 100644 --- a/src/rb2engine/writer/build.py +++ b/src/rb2engine/writer/build.py @@ -29,6 +29,7 @@ from rb2engine.errors import FatalError from rb2engine.ir import SourceArtwork, SourceLibrary from rb2engine.ir_engine import EngineTrack +from rb2engine.playlist_check import compare_playlists from rb2engine.progress import ProgressCallback, phase_callback from rb2engine.report import ConversionReport @@ -359,12 +360,36 @@ def build_library( # discarded before anyone can compare it, so this is the last point # where that class of corruption is still catchable. Running it before # os.replace means a failure leaves the user's previous m.db in place. - if intended_playlists: + if intended_playlists or lib.playlists: check_conn = sqlite3.connect(f"file:{tmp_path}?mode=ro", uri=True) try: - playlists_mod.assert_entities_match_intent( - check_conn, intended_playlists + if intended_playlists: + playlists_mod.assert_entities_match_intent( + check_conn, intended_playlists + ) + + # Independent oracle. The check above compares the database + # against what insert_playlists *intended*, and both sides of it + # descend from track_id_map — so a mapping fault agrees with + # itself and passes. This one recomputes every expected track id + # from the source through map_track and the database's own path + # index, never touching that map, and therefore fails where the + # intent check cannot. + # + # It cannot detect a misread source: both sides descend from the + # same parse. That is the reader's job (pdb G1d). + problems = compare_playlists( + lib, check_conn, drive_root=drive_root, engine_lib=engine_lib ) + if problems: + detail = "; ".join(p.describe() for p in problems[:5]) + more = ( + f" (+{len(problems) - 5} more)" if len(problems) > 5 else "" + ) + raise RuntimeError( + "playlist-scoped recheck against the source failed, so " + f"nothing was published: {detail}{more}" + ) finally: check_conn.close() diff --git a/tests/unit/test_build.py b/tests/unit/test_build.py index 8a21600..c1ba00e 100644 --- a/tests/unit/test_build.py +++ b/tests/unit/test_build.py @@ -1737,3 +1737,144 @@ def fail_unlink(self: Path, *a: Any, **k: Any) -> None: build_mod._remove_appledouble_sidecar_for(mdb2) + + +# --------------------------------------------------------------------------- +# W1d — independent pre-publish recheck +# +# assert_entities_match_intent compares the database against what +# insert_playlists intended, and both sides descend from track_id_map. A +# mapping fault therefore agrees with itself and passes. These tests pin the +# source-derived recheck that does not consult that map. +# --------------------------------------------------------------------------- + + +def _swapping_insert_tracks(swap: bool): + """insert_tracks double that inserts correct rows but may return a bad map.""" + + def insert_tracks( + conn: sqlite3.Connection, + tracks: Sequence[EngineTrack], + *, + art_ids: Mapping[str, int] | None = None, + on_progress: Any = None, + ) -> dict[int, int]: + row_ids: list[int] = [] + for et in tracks: + cur = conn.execute( + "INSERT INTO Track (path, title, artist, originDatabaseUuid, " + "originTrackId) VALUES (?, ?, ?, " + "(SELECT uuid FROM Information LIMIT 1), ?)", + (et.path, et.title, et.artist, 0), + ) + row_ids.append(int(cur.lastrowid)) + # rb_ids are 1..N in the fixtures below, in the same order as `tracks`. + rb_ids = list(range(1, len(tracks) + 1)) + if swap and len(row_ids) >= 2: + row_ids[0], row_ids[1] = row_ids[1], row_ids[0] + return dict(zip(rb_ids, row_ids, strict=True)) + + return insert_tracks + + +def _two_track_lib(drive: Path) -> SourceLibrary: + return SourceLibrary( + drive_root=drive, + tracks={1: _source_track(1, drive=drive), 2: _source_track(2, drive=drive)}, + playlists=[ + SourcePlaylist( + rb_id=1, + parent_rb_id=0, + name="Main", + sort_order=0, + is_folder=False, + track_rb_ids=[1], + ) + ], + warnings=[], + ) + + +def _stick(tmp_path: Path) -> Path: + drive = tmp_path / "stick" + drive.mkdir() + (drive / "Contents").mkdir() + (drive / "PIONEER").mkdir() + (drive / "Contents" / "1.mp3").write_bytes(b"a") + (drive / "Contents" / "2.mp3").write_bytes(b"b") + (drive / "PIONEER" / "export.pdb").write_bytes(b"pdb") + return drive + + +def test_recheck_catches_a_mapping_fault_the_intent_gate_cannot( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A wrong track_id_map is self-consistent, so only a source-derived check sees it. + + WHY: this is the blind spot W1d exists for. insert_playlists writes the ids + the map gave it and then confirms the database holds exactly those, which it + does — the playlist is internally coherent and points at the wrong track. + Only recomputing from the source catches it, and it must be caught before + the database is published rather than by a verify run days later. + """ + import rb2engine.writer.tracks as tracks_mod + from rb2engine.writer.build import build_library + + _install_database_fakes(monkeypatch) + _install_pipeline_fakes(monkeypatch) + monkeypatch.setattr(tracks_mod, "insert_tracks", _swapping_insert_tracks(swap=True)) + + drive = _stick(tmp_path) + + with pytest.raises(FatalError) as exc: + build_library( + _two_track_lib(drive), + drive_root=drive, + report=ConversionReport(), + with_artwork=False, + ) + + msg = str(exc.value) + assert "playlist-scoped recheck" in msg, msg + # The intent gate is what did NOT fire: its message names intended entries. + assert "intended" not in msg, msg + + +def test_failed_recheck_leaves_the_previous_database_untouched( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A refused build must not disturb the m.db the DJ already has. + + WHY: the check runs before os.replace precisely so failure is survivable. A + gate that detected the fault after publishing would only be able to + apologise, with the previous library already gone. + """ + import rb2engine.writer.tracks as tracks_mod + from rb2engine.writer.build import build_library + + _install_database_fakes(monkeypatch) + _install_pipeline_fakes(monkeypatch) + monkeypatch.setattr( + tracks_mod, "insert_tracks", _swapping_insert_tracks(swap=False) + ) + + drive = _stick(tmp_path) + lib = _two_track_lib(drive) + + m_db = build_library( + lib, drive_root=drive, report=ConversionReport(), with_artwork=False + ) + before = m_db.read_bytes() + assert before + + # Second conversion of the same library, now with a corrupted mapping. + monkeypatch.setattr( + tracks_mod, "insert_tracks", _swapping_insert_tracks(swap=True) + ) + with pytest.raises(FatalError): + build_library( + lib, drive_root=drive, report=ConversionReport(), with_artwork=False + ) + + assert m_db.read_bytes() == before, "published database was modified by a refused build" + assert not Path(str(m_db) + ".tmp").exists() diff --git a/tests/unit/test_playlist_check.py b/tests/unit/test_playlist_check.py new file mode 100644 index 0000000..ad8634d --- /dev/null +++ b/tests/unit/test_playlist_check.py @@ -0,0 +1,57 @@ +"""Problem reporting for the shared playlist comparison. + +WHY these are worth their own test: ``describe()`` is what a DJ actually sees +when a conversion is refused. Every branch of it is reachable from the writer's +pre-publish gate, but only the track-order one is exercised by the integration +tests, so the other two could rot into something unreadable without any test +noticing. +""" + +from __future__ import annotations + +from rb2engine.playlist_check import ( + CHAIN, + MISSING, + TRACK_ORDER, + PlaylistProblem, +) + + +def test_missing_playlist_reads_as_absent() -> None: + p = PlaylistProblem( + ("Sets", "Warmup"), MISSING, expected="present", actual="absent" + ) + + assert p.label == "Sets/Warmup" + assert p.describe() == "playlist 'Sets/Warmup' is absent from the database" + + +def test_broken_chain_names_the_chain_problem() -> None: + p = PlaylistProblem( + ("Main",), + CHAIN, + expected="every row reachable from the nextEntityId chain", + actual="row 7 is unreachable", + ) + + assert "broken entry chain" in p.describe() + assert "row 7 is unreachable" in p.describe() + + +def test_track_order_names_both_directions_of_the_difference() -> None: + """The message must say what is extra AND what is absent. + + WHY: the incident this check exists for was an *extra* entry with nothing + missing. A message that only reported absences would have described that + database as fine. + """ + p = PlaylistProblem( + ("Organic House",), TRACK_ORDER, expected=[1, 2, 3], actual=[1, 2, 9, 3] + ) + + text = p.describe() + + assert "4 entries written" in text + assert "3 expected" in text + assert "unexpected track ids [9]" in text + assert "absent track ids none" in text diff --git a/tests/unit/test_verify.py b/tests/unit/test_verify.py index b9b144a..cb74f8e 100644 --- a/tests/unit/test_verify.py +++ b/tests/unit/test_verify.py @@ -1568,7 +1568,7 @@ def test_db_playlist_paths_survives_a_malformed_parent_chain() -> None: library is to report, so the walk has to terminate on structures the writer would have refused to create. """ - from rb2engine.verify import _db_playlist_paths + from rb2engine.playlist_check import db_playlist_paths as _db_playlist_paths conn = sqlite3.connect(":memory:") conn.execute(