soulsync/tests/test_orphan_history.py
dev 0be1952222 downloads(#934): opt-in "Clean orphaned" action for dead review-queue rows
The reconcile heals rows whose file is still in the library; it deliberately
leaves ORPHANS — history rows whose file is gone (deleted / replaced /
re-downloaded elsewhere). Those can never be healed (no file left to confirm)
and linger in the Unverified list forever. This adds an explicit, user-initiated
cleanup for them.

- core/downloads/orphan_history.py: pure, tested rule. A row is an orphan when
  its file resolves nowhere; flags `suspicious` when EVERY reviewed file is
  unreachable (the mount-down signature) so the caller refuses rather than
  mass-delete a healthy log during an outage.
- POST /api/verification/clean-orphans (admin-only): runs it against
  _resolve_history_audio_path (raw path -> prefix-swap resolver -> tracks-table
  title fallback), refuses on the suspicious signature, and deletes only history
  ROWS — never a file (the files are already gone).
- UI: "🧹 Clean orphaned" button in the Unverified bulk-actions row, with a
  confirm dialog spelling out that it removes log rows only and refuses if the
  library looks offline.

NEVER automatic / never at boot — a filesystem check during a mount outage would
otherwise wipe good history. 5 pure-rule tests + safety-gate coverage.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LWJk7EuM7YktQeNyqQwTZY
2026-06-28 02:04:08 +02:00

49 lines
1.8 KiB
Python

"""Pure orphan-detection rules for the review-queue cleanup (#934 follow-up)."""
from core.downloads.orphan_history import find_orphan_history_ids
def _rows(*specs):
# specs: (id, file_path, exists?)
return [{'id': i, 'file_path': p, '_exists': e} for i, p, e in specs]
def _resolve(row):
# stand-in for _resolve_history_audio_path: returns a path or None
return '/on/disk' if row.get('_exists') else None
def test_only_missing_files_are_orphans():
rows = _rows((1, '/a.flac', True), (2, '/b.flac', False), (3, '/c.flac', True))
out = find_orphan_history_ids(rows, _resolve)
assert out['orphan_ids'] == [2]
assert out['checked'] == 3
assert out['suspicious'] is False
def test_rows_without_path_are_skipped():
rows = _rows((1, '', False), (2, None, False), (3, '/c.flac', False))
out = find_orphan_history_ids(rows, _resolve)
assert out['checked'] == 1
assert out['orphan_ids'] == [3]
def test_all_missing_with_enough_rows_is_suspicious():
rows = _rows(*[(i, f'/x{i}.flac', False) for i in range(6)])
out = find_orphan_history_ids(rows, _resolve)
assert out['suspicious'] is True # mount-down signature -> caller refuses
assert len(out['orphan_ids']) == 6
def test_all_missing_but_few_rows_is_not_suspicious():
rows = _rows((1, '/x.flac', False), (2, '/y.flac', False))
out = find_orphan_history_ids(rows, _resolve)
assert out['suspicious'] is False # too few to suspect an outage
assert out['orphan_ids'] == [1, 2]
def test_some_present_is_never_suspicious():
rows = _rows(*[(i, f'/x{i}.flac', False) for i in range(8)], (99, '/real.flac', True))
out = find_orphan_history_ids(rows, _resolve)
assert out['suspicious'] is False # at least one file exists -> library is up
assert 99 not in out['orphan_ids']