liumangmang
217 lines
7.2 KiB
Python
217 lines
7.2 KiB
Python
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()
|