import shutil import tempfile import unittest from datetime import datetime, timezone import os from pathlib import Path from backend.app.exception_service import ExceptionService from backend.app.library_service import LibraryService, LibraryTrackNotFoundError from backend.app.task_store import TaskStore class LibraryServiceTests(unittest.TestCase): def setUp(self): self.root = Path(tempfile.mkdtemp()) self.output_dir = self.root / 'output' self.output_dir.mkdir() self.task_store = TaskStore(self.root / 'music_workshop.db') self.preprocessor = _FakePreprocessor() self.service = LibraryService( self.task_store, self.preprocessor, read_tags=self.preprocessor.read_tags ) def tearDown(self): shutil.rmtree(self.root) def test_empty_output_dir_returns_empty_summary_and_tracks(self): summary = self.service.get_summary(str(self.output_dir)) page = self.service.get_tracks_page(str(self.output_dir)) self.assertEqual(summary['total_tracks'], 0) self.assertEqual(summary['total_albums'], 0) self.assertEqual(summary['total_artists'], 0) self.assertEqual(summary['suspected_duplicates'], 0) self.assertEqual(page['items'], []) self.assertEqual(page['total'], 0) def test_scans_metadata_audio_and_filters_tracks(self): first_path = self._write_library_file('A/Artist A/Album A/01 - Echoes.flac', _timestamp(2024, 1, 1)) second_path = self._write_library_file('B/Artist B/Album B/03 - Neon.mp3', _timestamp(2024, 1, 2)) self.preprocessor.audio_props[str(first_path)] = { 'format': 'FLAC', 'codec': 'FLAC', 'bitrate': 980000, 'sample_rate': 96000, 'bit_depth': 24, 'channels': 2, 'duration_seconds': 301.4 } self.preprocessor.tags[str(first_path)] = { 'title': 'Echoes', 'artist': 'Artist A', 'album': 'Album A', 'albumartist': 'Artist A', 'tracknumber': '1', 'discnumber': '1', 'date': '2024-01-01' } self.preprocessor.audio_props[str(second_path)] = { 'format': 'MP3', 'codec': 'MP3', 'bitrate': 320000, 'sample_rate': 44100, 'bit_depth': 16, 'channels': 2, 'duration_seconds': 240.1 } self.preprocessor.tags[str(second_path)] = { 'title': 'Neon', 'artist': 'Artist B', 'album': 'Album B', 'albumartist': 'Artist B', 'tracknumber': '3', 'discnumber': '1', 'date': '2023-12-01' } task = self._create_completed_task() self._insert_provenance_item( task['task_id'], library_file_path=str(first_path), library_relative_path='A/Artist A/Album A/01 - Echoes.flac', updated_at='2024-01-03T09:00:00Z', match_source='musicbrainz', match_confidence=94.5, dedupe_status='unique' ) page = self.service.get_tracks_page( str(self.output_dir), q='echo', artist='Artist A', album='Album A', format='flac', has_provenance=True ) self.assertEqual(page['total'], 1) track = page['items'][0] self.assertEqual(track['filename'], '01 - Echoes.flac') self.assertEqual(track['title'], 'Echoes') self.assertEqual(track['artist'], 'Artist A') self.assertEqual(track['album'], 'Album A') self.assertEqual(track['format'], 'FLAC') self.assertEqual(track['codec'], 'FLAC') self.assertEqual(track['bit_depth'], 24) self.assertEqual(track['sample_rate'], 96000) self.assertEqual(track['ingest_provenance']['task_id'], task['task_id']) self.assertEqual(track['ingest_provenance']['match_source'], 'musicbrainz') self.assertEqual(track['ingest_provenance']['dedupe_status'], 'unique') def test_default_sort_prefers_organized_at_and_falls_back_to_modified_at(self): newest_path = self._write_library_file('N/Newest/Album/01 - Fresh.flac', _timestamp(2024, 1, 3)) organized_path = self._write_library_file('O/Organized/Album/01 - Sorted.flac', _timestamp(2024, 1, 1)) oldest_path = self._write_library_file('Z/Oldest/Album/01 - Archive.flac', _timestamp(2023, 12, 31)) for path, title, artist in ( (newest_path, 'Fresh', 'Newest'), (organized_path, 'Sorted', 'Organized'), (oldest_path, 'Archive', 'Oldest') ): self.preprocessor.audio_props[str(path)] = {'format': 'FLAC', 'codec': 'FLAC'} self.preprocessor.tags[str(path)] = { 'title': title, 'artist': artist, 'album': 'Album', 'albumartist': artist, 'tracknumber': '1', 'discnumber': '1' } task = self._create_completed_task() self._insert_provenance_item( task['task_id'], library_file_path=str(organized_path), library_relative_path='O/Organized/Album/01 - Sorted.flac', updated_at='2024-01-02T12:00:00Z' ) page = self.service.get_tracks_page(str(self.output_dir)) ordered_titles = [item['title'] for item in page['items']] self.assertEqual(ordered_titles, ['Fresh', 'Sorted', 'Archive']) def test_provenance_prefers_absolute_path_then_falls_back_to_relative_path(self): exact_path = self._write_library_file('A/Artist/Album/01 - Exact.flac', _timestamp(2024, 1, 1)) fallback_path = self._write_library_file('B/Artist/Album/02 - Fallback.flac', _timestamp(2024, 1, 1)) for path, title in ((exact_path, 'Exact'), (fallback_path, 'Fallback')): self.preprocessor.audio_props[str(path)] = {'format': 'FLAC', 'codec': 'FLAC'} self.preprocessor.tags[str(path)] = { 'title': title, 'artist': 'Artist', 'album': 'Album', 'albumartist': 'Artist', 'tracknumber': '1', 'discnumber': '1' } old_task = self._create_completed_task() self._insert_provenance_item( old_task['task_id'], library_file_path=str(exact_path), library_relative_path='A/Artist/Album/01 - Exact.flac', updated_at='2024-01-01T08:00:00Z', match_source='exact-source' ) newer_task = self._create_completed_task() self._insert_provenance_item( newer_task['task_id'], library_file_path='/legacy/output/A/Artist/Album/01 - Exact.flac', library_relative_path='A/Artist/Album/01 - Exact.flac', updated_at='2024-01-03T08:00:00Z', match_source='relative-source' ) fallback_task = self._create_completed_task() self._insert_provenance_item( fallback_task['task_id'], library_file_path='/legacy/output/B/Artist/Album/02 - Fallback.flac', library_relative_path='B/Artist/Album/02 - Fallback.flac', updated_at='2024-01-04T08:00:00Z', match_source='fallback-source' ) page = self.service.get_tracks_page(str(self.output_dir), sort_by='filename', sort_order='asc') exact_track = next(item for item in page['items'] if item['title'] == 'Exact') fallback_track = next(item for item in page['items'] if item['title'] == 'Fallback') self.assertEqual(exact_track['ingest_provenance']['match_source'], 'exact-source') self.assertEqual(fallback_track['ingest_provenance']['match_source'], 'fallback-source') def test_summary_counts_suspected_duplicates_without_false_live_match(self): duplicate_one = self._write_library_file('A/Artist/Album/01 - Song.flac', _timestamp(2024, 1, 1)) duplicate_two = self._write_library_file('A/Artist/Album/01 - Song Copy.flac', _timestamp(2024, 1, 2)) studio = self._write_library_file('S/Artist/Singles/2024 - Ballad/01 - Ballad.flac', _timestamp(2024, 1, 3)) live = self._write_library_file('S/Artist/Singles/2024 - Ballad Live/01 - Ballad Live.flac', _timestamp(2024, 1, 4)) for path in (duplicate_one, duplicate_two, studio, live): self.preprocessor.audio_props[str(path)] = { 'format': 'FLAC', 'codec': 'FLAC', 'duration_seconds': 201 } self.preprocessor.tags[str(duplicate_one)] = { 'title': 'Song', 'artist': 'Artist', 'album': 'Album', 'albumartist': 'Artist', 'tracknumber': '1', 'discnumber': '1', 'musicbrainzrecordingid': 'recording-1' } self.preprocessor.tags[str(duplicate_two)] = { 'title': 'Song', 'artist': 'Artist', 'album': 'Album', 'albumartist': 'Artist', 'tracknumber': '1', 'discnumber': '1', 'musicbrainzrecordingid': 'recording-1' } self.preprocessor.tags[str(studio)] = { 'title': 'Ballad', 'artist': 'Artist', 'albumartist': 'Artist' } self.preprocessor.tags[str(live)] = { 'title': 'Ballad (Live)', 'artist': 'Artist', 'albumartist': 'Artist' } summary = self.service.get_summary(str(self.output_dir)) self.assertEqual(summary['total_tracks'], 4) self.assertEqual(summary['suspected_duplicates'], 1) def test_move_track_to_exception_moves_file_and_creates_match_failed_item(self): trash_dir = self.root / 'trash' library_path = self._write_library_file('A/Artist/Album/01 - Song.flac', _timestamp(2024, 1, 5)) self.preprocessor.audio_props[str(library_path)] = { 'format': 'FLAC', 'codec': 'FLAC', 'duration_seconds': 180.25 } self.preprocessor.tags[str(library_path)] = { 'title': 'Song', 'artist': 'Artist', 'album': 'Album', 'albumartist': 'Artist', 'tracknumber': '1' } self.preprocessor.fingerprints[str(library_path)] = { 'fingerprint': 'abc123', 'duration_seconds': 180.0 } track = self.service.get_tracks_page(str(self.output_dir))['items'][0] response = self.service.move_track_to_exception( { 'input': str(self.root / 'input'), 'output': str(self.output_dir), 'trash': str(trash_dir) }, track['track_id'] ) trash_path = Path(response['trash_file_path']) self.assertFalse(library_path.exists()) self.assertTrue(trash_path.exists()) self.assertEqual(response['library_relative_path'], 'A/Artist/Album/01 - Song.flac') self.assertEqual(self.service.get_tracks_page(str(self.output_dir))['total'], 0) exception_service = ExceptionService(self.task_store) page = exception_service.get_items('match_failed') self.assertEqual(page['total'], 1) exception_item = page['items'][0] self.assertEqual(exception_item['exception_id'], response['exception_id']) self.assertEqual(exception_item['exception_type'], 'match_failed') self.assertEqual(exception_item['exception_reason_code'], 'manual_library_requeue') self.assertEqual(exception_item['trash_file_path'], str(trash_path)) self.assertEqual(exception_item['library_file_path'], str(library_path)) self.assertIn('retry_match', exception_item['available_actions']) source_item = self.task_store.get_exception_source_item(response['exception_id']) self.assertEqual(source_item['current_file_path'], str(trash_path)) self.assertEqual(source_item['original_tags_json']['title'], 'Song') self.assertEqual(source_item['audio_props_json']['codec'], 'FLAC') self.assertEqual(source_item['acoustic_fingerprint'], 'abc123') def test_move_track_to_exception_rejects_unknown_track_id(self): with self.assertRaises(LibraryTrackNotFoundError): self.service.move_track_to_exception( { 'input': str(self.root / 'input'), 'output': str(self.output_dir), 'trash': str(self.root / 'trash') }, 'unknown-track-id' ) def _write_library_file(self, relative_path: str, modified_at_timestamp: int) -> Path: path = self.output_dir / relative_path path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(b'audio-data') os.utime(path, (float(modified_at_timestamp), float(modified_at_timestamp))) return path def _create_completed_task(self) -> dict: task = self.task_store.create_task_if_idle( {'input': '', 'output': str(self.output_dir), 'trash': ''} ) self.task_store.update_task( task['task_id'], status='completed', completed_at='2024-01-01T00:00:00Z' ) return task def _insert_provenance_item( self, task_id: str, *, library_file_path: str, library_relative_path: str, updated_at: str, match_source: str | None = None, match_confidence: float | None = None, dedupe_status: str = 'unique' ): item = self.task_store.insert_task_item( task_id, original_path=library_file_path, current_file_path=library_file_path, relative_path=library_relative_path, filename=Path(library_file_path).name, extension=Path(library_file_path).suffix.lower(), size_bytes=123, modified_at='2024-01-01T00:00:00Z', local_cover=None, local_lyric=None, scan_status='queued', scan_reason=None, scan_message=None, preprocess_status='completed', match_status='matched', match_source=match_source, match_confidence=match_confidence, dedupe_status=dedupe_status, organize_status='organized', library_relative_path=library_relative_path, library_file_path=library_file_path ) with self.task_store._connect() as connection: connection.execute( 'UPDATE task_items SET updated_at = ? WHERE id = ?', (updated_at, item['id']) ) connection.commit() class _FakePreprocessor: def __init__(self): self.audio_props: dict[str, dict] = {} self.tags: dict[str, dict] = {} self.fingerprints: dict[str, dict] = {} def probe_audio(self, file_path: str) -> dict: return self.audio_props.get(file_path, {}) def read_tags(self, file_path: str) -> dict: return self.tags.get(file_path, {}) def calculate_fingerprint(self, file_path: str) -> dict: return self.fingerprints.get(file_path, {}) def _timestamp(year: int, month: int, day: int) -> int: return int(datetime(year, month, day, tzinfo=timezone.utc).timestamp()) if __name__ == '__main__': unittest.main()