import os import tempfile import unittest from pathlib import Path try: from fastapi.testclient import TestClient except ModuleNotFoundError: TestClient = None os.environ['MUSIC_WORKSHOP_DB_PATH'] = str( Path(tempfile.gettempdir()) / f'music_workshop_exception_api_{next(tempfile._get_candidate_names())}.db' ) try: from backend.app.exception_service import ExceptionItemNotFoundError from backend.app.schemas import ( ExceptionDetailPayload, ExceptionListResponse, ExceptionSummaryPayload ) import backend.app.main as main_module except ModuleNotFoundError as error: main_module = None ExceptionItemNotFoundError = None ExceptionDetailPayload = None ExceptionListResponse = None ExceptionSummaryPayload = None FASTAPI_IMPORT_ERROR = error else: FASTAPI_IMPORT_ERROR = None @unittest.skipIf(main_module is None, f'api deps unavailable: {FASTAPI_IMPORT_ERROR}') class ExceptionApiTests(unittest.TestCase): def setUp(self): self.previous_service = main_module.exception_service self.fake_service = _FakeExceptionService() main_module.exception_service = self.fake_service self.client = TestClient(main_module.app) if TestClient else None def tearDown(self): main_module.exception_service = self.previous_service def test_get_exception_summary_serializes_payload(self): response = main_module.get_exception_summary() payload = ExceptionSummaryPayload.model_validate(response) self.assertEqual(payload.total, 6) self.assertEqual(payload.counts_by_type['duplicates'], 2) self.assertEqual(self.fake_service.summary_calls, 1) def test_get_exception_items_passes_filters_and_pagination(self): response = main_module.get_exception_items( type='duplicates', resolution_status='resolved', page=2, page_size=25 ) payload = ExceptionListResponse.model_validate(response) self.assertEqual(payload.page, 2) self.assertEqual(payload.page_size, 25) self.assertEqual(payload.total, 1) self.assertEqual(payload.items[0].exception_id, 101) self.assertEqual(payload.items[0].exception_type, 'duplicates') self.assertEqual( self.fake_service.list_calls, [{'type': 'duplicates', 'resolution_status': 'resolved', 'page': 2, 'page_size': 25}] ) def test_get_exception_item_serializes_detail_payload(self): response = main_module.get_exception_item(101) payload = ExceptionDetailPayload.model_validate(response) self.assertEqual(payload.exception_id, 101) self.assertEqual(payload.filename, 'duplicate.flac') self.assertEqual(payload.dedupe_decision_json['comparison_scope'], 'library') self.assertEqual(self.fake_service.detail_calls, [101]) def test_get_exception_item_not_found_raises_service_error(self): with self.assertRaises(ExceptionItemNotFoundError): main_module.get_exception_item(999) response = main_module.exception_item_not_found_error_handler( None, ExceptionItemNotFoundError(999) ) self.assertEqual(response.status_code, 404) def test_streams_exception_audio_with_range_support(self): if self.client is None: self.skipTest('fastapi test client unavailable') audio_path = Path(tempfile.gettempdir()) / f'exception-audio-{next(tempfile._get_candidate_names())}.mp3' audio_path.write_bytes(b'0123456789abcdef') self.fake_service.audio_path = audio_path try: full_response = self.client.get('/api/exceptions/items/101/audio') self.assertEqual(full_response.status_code, 200) self.assertEqual(full_response.content, b'0123456789abcdef') self.assertEqual(full_response.headers['accept-ranges'], 'bytes') range_response = self.client.get( '/api/exceptions/items/101/audio', headers={'Range': 'bytes=4-7'} ) self.assertEqual(range_response.status_code, 206) self.assertEqual(range_response.content, b'4567') self.assertEqual(range_response.headers['content-range'], 'bytes 4-7/16') finally: if audio_path.exists(): audio_path.unlink() class _FakeExceptionService: def __init__(self): self.summary_calls = 0 self.list_calls: list[dict] = [] self.detail_calls: list[int] = [] self.audio_path: Path | None = None def get_summary(self) -> dict: self.summary_calls += 1 return { 'total': 6, 'counts_by_type': { 'missing_tags': 1, 'duplicates': 2, 'match_failed': 1, 'low_score': 1, 'convert_failed': 0, 'organize_failed': 1 }, 'scanned_at': '2024-01-03T12:00:00Z' } def get_items( self, exception_type: str = 'all', page: int = 1, page_size: int = 50, resolution_status: str = 'open' ) -> dict: self.list_calls.append( { 'type': exception_type, 'resolution_status': resolution_status, 'page': page, 'page_size': page_size } ) return { 'items': [self._detail_payload()], 'page': page, 'page_size': page_size, 'total': 1 } def get_item(self, exception_id: int) -> dict: self.detail_calls.append(exception_id) if exception_id != 101: raise ExceptionItemNotFoundError(exception_id) return self._detail_payload() def resolve_audio_path(self, exception_id: int) -> Path: self.detail_calls.append(exception_id) if exception_id != 101 or self.audio_path is None: raise FileNotFoundError(f'No playable audio found for exception item: {exception_id}') return self.audio_path def _detail_payload(self) -> dict: return { 'exception_id': 101, 'task_id': 'task-123', 'task_started_at': '2024-01-01T08:00:00Z', 'exception_type': 'duplicates', 'exception_stage': 'dedupe', 'exception_reason_code': 'library_duplicate', 'exception_message': '输出库中已存在重复文件，保留库内文件', 'captured_at': '2024-01-03T12:00:00Z', 'filename': 'duplicate.flac', 'relative_path': 'Artist/Album/duplicate.flac', 'original_path': '/tmp/input/duplicate.flac', 'current_file_path': '/tmp/trash/duplicate.flac', 'trash_file_path': '/tmp/trash/duplicate.flac', 'audio_props_json': {'codec': 'FLAC'}, 'original_tags_json': {'title': 'Song'}, 'matched_metadata_json': {'title': 'Song'}, 'duplicate_of_path': '/tmp/output/Artist/Old.flac', 'dedupe_decision_json': {'comparison_scope': 'library'}, 'library_relative_path': None, 'library_file_path': None, 'match_source': 'musicbrainz', 'match_confidence': 91.2, 'preview_available': False, 'available_actions': [], 'exception_resolution_status': 'open', 'exception_resolution_json': None, 'workflow_state': 'open', 'raw_metadata': {'title': 'Song'}, 'metadata_draft': {'title': 'Song'}, 'effective_metadata': {'title': 'Song', 'artist': 'Artist', 'album_artist': 'Artist'}, 'can_ingest': True, 'pending_ingest': False, 'display_title': 'Song', 'display_reason': '输出库中已存在重复文件，保留库内文件', 'type_label': '文件重复', 'preprocess_artifacts_json': None, 'match_candidates_json': None, 'match_enrichment_json': None, 'organize_decision_json': None } if __name__ == '__main__': unittest.main()