MusicWorkshop/backend/tests/test_library_api.py

import os
import tempfile
import unittest
from pathlib import Path

os.environ['MUSIC_WORKSHOP_DB_PATH'] = str(
  Path(tempfile.gettempdir()) / f'music_workshop_library_api_{next(tempfile._get_candidate_names())}.db'
)

from backend.app.storage import ConfigStore
from backend.app.task_store import TaskConflictError

try:
  from backend.app.library_service import LibraryTrackNotFoundError
  from backend.app.schemas import (
    LibraryMoveToExceptionResponse,
    LibrarySummaryPayload,
    LibraryTracksPageResponse
  )
  import backend.app.main as main_module
except ModuleNotFoundError as error:
  main_module = None
  LibraryTrackNotFoundError = None
  LibraryMoveToExceptionResponse = None
  LibrarySummaryPayload = None
  LibraryTracksPageResponse = None
  FASTAPI_IMPORT_ERROR = error
else:
  FASTAPI_IMPORT_ERROR = None


@unittest.skipIf(main_module is None, f'api deps unavailable: {FASTAPI_IMPORT_ERROR}')
class LibraryApiTests(unittest.TestCase):
  def setUp(self):
    self.db_path = Path(os.environ['MUSIC_WORKSHOP_DB_PATH'])
    if self.db_path.exists():
      self.db_path.unlink()
    self.store = ConfigStore(self.db_path)
    config = self.store.get_config()
    config['output'] = '/tmp/library-output'
    self.store.save_config(config)
    self.previous_store = main_module.store
    self.previous_service = main_module.library_service
    self.fake_service = _FakeLibraryService()
    main_module.store = self.store
    main_module.library_service = self.fake_service

  def tearDown(self):
    main_module.store = self.previous_store
    main_module.library_service = self.previous_service

  def test_get_library_summary_uses_current_output_config(self):
    response = main_module.get_library_summary()
    payload = LibrarySummaryPayload.model_validate(response)

    self.assertEqual(payload.total_tracks, 12)
    self.assertEqual(payload.total_albums, 3)
    self.assertEqual(payload.total_artists, 2)
    self.assertEqual(self.fake_service.summary_calls, ['/tmp/library-output'])

  def test_get_library_tracks_passes_pagination_filters_and_serializes_provenance(self):
    response = main_module.get_library_tracks(
      q='echoes',
      artist='Artist A',
      album='Album A',
      format='FLAC',
      has_provenance=True,
      page=2,
      page_size=25,
      sort_by='filename',
      sort_order='asc'
    )
    payload = LibraryTracksPageResponse.model_validate(response)

    self.assertEqual(payload.page, 2)
    self.assertEqual(payload.page_size, 25)
    self.assertEqual(payload.total, 1)
    self.assertEqual(payload.items[0].track_id, 'track-1')
    self.assertEqual(payload.items[0].ingest_provenance.task_id, 'task-123')
    self.assertEqual(
      self.fake_service.track_calls,
      [
        {
          'output_dir': '/tmp/library-output',
          'q': 'echoes',
          'artist': 'Artist A',
          'album': 'Album A',
          'format': 'FLAC',
          'has_provenance': True,
          'page': 2,
          'page_size': 25,
          'sort_by': 'filename',
          'sort_order': 'asc'
        }
      ]
    )

  def test_move_library_track_to_exception_uses_current_config(self):
    response = main_module.move_library_track_to_exception('track-1')
    payload = LibraryMoveToExceptionResponse.model_validate(response)

    self.assertEqual(payload.exception_id, 123)
    self.assertEqual(payload.library_relative_path, 'A/Artist A/Album A/01 - Echoes.flac')
    self.assertEqual(
      self.fake_service.move_calls,
      [{'output': '/tmp/library-output', 'trash': '/volume1/docker/navidrome/trash', 'track_id': 'track-1'}]
    )

  def test_move_library_track_to_exception_maps_conflict_to_409(self):
    self.fake_service.move_error = TaskConflictError('active-task')

    response = main_module.move_library_track_to_exception('track-1')

    self.assertEqual(response.status_code, 409)

  def test_library_track_not_found_handler_returns_404(self):
    response = main_module.library_track_not_found_error_handler(
      None,
      LibraryTrackNotFoundError('missing-track')
    )

    self.assertEqual(response.status_code, 404)


class _FakeLibraryService:
  def __init__(self):
    self.summary_calls: list[str] = []
    self.track_calls: list[dict] = []
    self.move_calls: list[dict] = []
    self.move_error = None

  def get_summary(self, output_dir: str) -> dict:
    self.summary_calls.append(output_dir)
    return {
      'total_tracks': 12,
      'total_albums': 3,
      'total_artists': 2,
      'suspected_duplicates': 1,
      'scanned_at': '2024-01-03T12:00:00Z'
    }

  def get_tracks_page(self, output_dir: str, **kwargs) -> dict:
    self.track_calls.append({'output_dir': output_dir, **kwargs})
    return {
      'items': [
        {
          'track_id': 'track-1',
          'library_relative_path': 'A/Artist A/Album A/01 - Echoes.flac',
          'library_file_path': '/tmp/library-output/A/Artist A/Album A/01 - Echoes.flac',
          'filename': '01 - Echoes.flac',
          'title': 'Echoes',
          'artist': 'Artist A',
          'album': 'Album A',
          'album_artist': 'Artist A',
          'track_number': 1,
          'disc_number': 1,
          'year': 2024,
          'duration_seconds': 301.4,
          'format': 'FLAC',
          'codec': 'FLAC',
          'bitrate': 980000,
          'sample_rate': 96000,
          'bit_depth': 24,
          'channels': 2,
          'size_bytes': 12345678,
          'modified_at': '2024-01-02T12:00:00Z',
          'ingest_provenance': {
            'task_id': 'task-123',
            'organized_at': '2024-01-03T12:00:00Z',
            'match_source': 'musicbrainz',
            'match_confidence': 95.2,
            'dedupe_status': 'unique'
          }
        }
      ],
      'page': kwargs['page'],
      'page_size': kwargs['page_size'],
      'total': 1
    }

  def move_track_to_exception(self, config_snapshot: dict, track_id: str) -> dict:
    if self.move_error:
      raise self.move_error
    self.move_calls.append(
      {
        'output': config_snapshot['output'],
        'trash': config_snapshot['trash'],
        'track_id': track_id
      }
    )
    return {
      'exception_id': 123,
      'library_relative_path': 'A/Artist A/Album A/01 - Echoes.flac',
      'trash_file_path': '/tmp/trash/match_failed/task-1/01 - Echoes.flac',
      'message': '已移入异常中心，等待重新匹配'
    }


if __name__ == '__main__':
  unittest.main()