diff --git a/core/hifi_client.py b/core/hifi_client.py index c1209716..ea9876de 100644 --- a/core/hifi_client.py +++ b/core/hifi_client.py @@ -110,9 +110,7 @@ class HiFiClient: 'Accept': 'application/json', }) - self.active_downloads: Dict[str, Dict[str, Any]] = {} - self._download_lock = threading.Lock() - + self._engine = None self.shutdown_check = None self._last_api_call = 0 @@ -125,6 +123,9 @@ class HiFiClient: def set_shutdown_check(self, check_callable): self.shutdown_check = check_callable + def set_engine(self, engine): + self._engine = engine + def _load_instances_from_db(self): try: from database.music_database import get_database @@ -571,71 +572,31 @@ class HiFiClient: ) async def download(self, username: str, filename: str, file_size: int = 0) -> Optional[str]: - try: - if '||' not in filename: - logger.error(f"Invalid filename format: {filename}") - return None - - track_id_str, display_name = filename.split('||', 1) - try: - track_id = int(track_id_str) - except ValueError: - logger.error(f"Invalid track ID: {track_id_str}") - return None - - download_id = str(uuid.uuid4()) - - with self._download_lock: - self.active_downloads[download_id] = { - 'id': download_id, - 'filename': filename, - 'username': 'hifi', - 'state': 'Initializing', - 'progress': 0.0, - 'size': 0, - 'transferred': 0, - 'speed': 0, - 'time_remaining': None, - 'track_id': track_id, - 'display_name': display_name, - 'file_path': None, - } - - thread = threading.Thread( - target=self._download_worker, - args=(download_id, track_id, display_name), - daemon=True, - ) - thread.start() - - return download_id - - except Exception as e: - logger.error(f"Failed to start HiFi download: {e}") + if '||' not in filename: + logger.error(f"Invalid filename format: {filename}") + return None + if self._engine is None: + logger.error("HiFi client has no engine reference — cannot dispatch download") return None - def _download_worker(self, download_id: str, track_id: int, display_name: str): + track_id_str, display_name = filename.split('||', 1) try: - with self._download_lock: - if download_id in self.active_downloads: - self.active_downloads[download_id]['state'] = 'InProgress, Downloading' + track_id = int(track_id_str) + except ValueError: + logger.error(f"Invalid track ID: {track_id_str}") + return None - file_path = self._download_sync(download_id, track_id, display_name) - - with self._download_lock: - if download_id in self.active_downloads: - if file_path: - self.active_downloads[download_id]['state'] = 'Completed, Succeeded' - self.active_downloads[download_id]['progress'] = 100.0 - self.active_downloads[download_id]['file_path'] = file_path - else: - self.active_downloads[download_id]['state'] = 'Errored' - - except Exception as e: - logger.error(f"HiFi download worker failed for {download_id}: {e}") - with self._download_lock: - if download_id in self.active_downloads: - self.active_downloads[download_id]['state'] = 'Errored' + return self._engine.worker.dispatch( + source_name='hifi', + target_id=track_id, + display_name=display_name, + original_filename=filename, + impl_callable=self._download_sync, + extra_record_fields={ + 'track_id': track_id, + 'display_name': display_name, + }, + ) def _download_sync(self, download_id: str, track_id: int, display_name: str) -> Optional[str]: quality_key = config_manager.get('hifi_download.quality', 'lossless') @@ -676,9 +637,8 @@ class HiFiClient: speed_start = time.time() segments_completed = 0 - with self._download_lock: - if download_id in self.active_downloads: - self.active_downloads[download_id]['size'] = 0 + if self._engine is not None: + self._engine.update_record('hifi', download_id, {'size': 0}) with intermediate_path.open('wb') as output_file: if init_uri: @@ -777,80 +737,77 @@ class HiFiClient: def _update_download_progress(self, download_id: str, downloaded: int, segments_completed: int, total_segments: int, speed_start: float): - with self._download_lock: - if download_id not in self.active_downloads: - return - info = self.active_downloads[download_id] - info['transferred'] = downloaded + if self._engine is None: + return + record = self._engine.get_record('hifi', download_id) + if record is None: + return - now = time.time() - elapsed_total = now - speed_start - speed = int(downloaded / elapsed_total) if elapsed_total > 0 else 0 - info['speed'] = speed + now = time.time() + elapsed_total = now - speed_start + speed = int(downloaded / elapsed_total) if elapsed_total > 0 else 0 - if total_segments > 0: - progress = (segments_completed / total_segments) * 100 - info['progress'] = round(min(progress, 99.9), 1) + progress = record.get('progress', 0.0) + if total_segments > 0: + progress = round(min((segments_completed / total_segments) * 100, 99.9), 1) - time_remaining = None - if speed > 0: - remaining_bytes = downloaded * (total_segments / max(segments_completed, 1)) - downloaded - if remaining_bytes > 0: - time_remaining = int(remaining_bytes / speed) - info['time_remaining'] = time_remaining + time_remaining = None + if speed > 0: + remaining_bytes = downloaded * (total_segments / max(segments_completed, 1)) - downloaded + if remaining_bytes > 0: + time_remaining = int(remaining_bytes / speed) + + self._engine.update_record('hifi', download_id, { + 'transferred': downloaded, + 'speed': speed, + 'progress': progress, + 'time_remaining': time_remaining, + }) + + def _record_to_status(self, record): + return DownloadStatus( + id=record['id'], + filename=record['filename'], + username=record['username'], + state=record['state'], + progress=record['progress'], + size=record.get('size', 0), + transferred=record.get('transferred', 0), + speed=record.get('speed', 0), + time_remaining=record.get('time_remaining'), + file_path=record.get('file_path'), + ) async def get_all_downloads(self) -> List[DownloadStatus]: - statuses = [] - with self._download_lock: - for _dl_id, info in self.active_downloads.items(): - statuses.append(DownloadStatus( - id=info['id'], - filename=info['filename'], - username=info['username'], - state=info['state'], - progress=info['progress'], - size=info['size'], - transferred=info['transferred'], - speed=info['speed'], - time_remaining=info.get('time_remaining'), - file_path=info.get('file_path'), - )) - return statuses + if self._engine is None: + return [] + return [ + self._record_to_status(record) + for record in self._engine.iter_records_for_source('hifi') + ] async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]: - with self._download_lock: - info = self.active_downloads.get(download_id) - if not info: - return None - return DownloadStatus( - id=info['id'], - filename=info['filename'], - username=info['username'], - state=info['state'], - progress=info['progress'], - size=info['size'], - transferred=info['transferred'], - speed=info['speed'], - time_remaining=info.get('time_remaining'), - file_path=info.get('file_path'), - ) + if self._engine is None: + return None + record = self._engine.get_record('hifi', download_id) + return self._record_to_status(record) if record is not None else None async def cancel_download(self, download_id: str, username: str = None, remove: bool = False) -> bool: - with self._download_lock: - if download_id not in self.active_downloads: - return False - self.active_downloads[download_id]['state'] = 'Cancelled' - if remove: - del self.active_downloads[download_id] + if self._engine is None: + return False + if self._engine.get_record('hifi', download_id) is None: + return False + self._engine.update_record('hifi', download_id, {'state': 'Cancelled'}) + if remove: + self._engine.remove_record('hifi', download_id) return True async def clear_all_completed_downloads(self) -> bool: - with self._download_lock: - to_remove = [ - did for did, info in self.active_downloads.items() - if info.get('state', '') in ('Completed, Succeeded', 'Cancelled', 'Errored', 'Aborted') - ] - for did in to_remove: - del self.active_downloads[did] + if self._engine is None: + return True + terminal = {'Completed, Succeeded', 'Cancelled', 'Errored', 'Aborted'} + for record in list(self._engine.iter_records_for_source('hifi')): + if record.get('state') in terminal: + self._engine.remove_record('hifi', record['id']) return True diff --git a/tests/downloads/test_hifi_pinning.py b/tests/downloads/test_hifi_pinning.py index d06594e0..8111423f 100644 --- a/tests/downloads/test_hifi_pinning.py +++ b/tests/downloads/test_hifi_pinning.py @@ -1,10 +1,4 @@ -"""Phase A pinning tests for HiFiClient's download lifecycle. - -HiFi uses public hifi-api instances backed by Tidal-sourced metadata. -Same int track_id + thread-worker + state-dict pattern as Tidal/Qobuz, -EXCEPT the worker method is named `_download_worker` (no `_thread_`). -Engine refactor must preserve the worker target signature. -""" +"""Phase A pinning tests for HiFiClient — UPDATED for Phase C5.""" from __future__ import annotations @@ -15,6 +9,7 @@ from unittest.mock import patch import pytest +from core.download_engine import DownloadEngine from core.hifi_client import HiFiClient @@ -27,65 +22,78 @@ def _run_async(coro): @pytest.fixture -def hifi_client(): +def hifi_client_with_engine(): client = HiFiClient.__new__(HiFiClient) client.download_path = Path('./test_hifi_downloads') client.shutdown_check = None - client.active_downloads = {} - client._download_lock = threading.Lock() - return client + client._engine = None + engine = DownloadEngine() + client.set_engine(engine) + return client, engine -def test_download_returns_none_for_invalid_filename_format(hifi_client): - result = _run_async(hifi_client.download('hifi', 'no-separator', 0)) +def test_download_returns_none_for_invalid_filename_format(hifi_client_with_engine): + client, _ = hifi_client_with_engine + result = _run_async(client.download('hifi', 'no-separator', 0)) assert result is None -def test_download_returns_none_for_non_integer_track_id(hifi_client): - result = _run_async(hifi_client.download('hifi', 'not-int||title', 0)) +def test_download_returns_none_for_non_integer_track_id(hifi_client_with_engine): + client, _ = hifi_client_with_engine + result = _run_async(client.download('hifi', 'not-int||title', 0)) assert result is None -def test_download_returns_uuid_for_valid_filename(hifi_client): - with patch('core.hifi_client.threading.Thread') as fake: - fake.return_value.start = lambda: None - result = _run_async(hifi_client.download('hifi', '12345||Some Song', 0)) +def test_download_returns_none_when_engine_not_wired(): + client = HiFiClient.__new__(HiFiClient) + client._engine = None + result = _run_async(client.download('hifi', '12345||x', 0)) + assert result is None + + +def test_download_returns_uuid_for_valid_filename(hifi_client_with_engine): + client, _ = hifi_client_with_engine + with patch.object(client, '_download_sync', return_value='/tmp/x.flac'): + result = _run_async(client.download('hifi', '12345||Some Song', 0)) assert result is not None assert len(result) == 36 -def test_download_populates_active_downloads_with_initial_state(hifi_client): - with patch('core.hifi_client.threading.Thread') as fake: - fake.return_value.start = lambda: None - download_id = _run_async(hifi_client.download('hifi', '999||My HiFi Song', 0)) +def test_download_populates_engine_record_with_initial_state(hifi_client_with_engine): + client, engine = hifi_client_with_engine + started = threading.Event() + release = threading.Event() - record = hifi_client.active_downloads[download_id] - assert record['id'] == download_id - assert record['filename'] == '999||My HiFi Song' - assert record['username'] == 'hifi' - assert record['state'] == 'Initializing' - assert record['track_id'] == 999 - assert record['display_name'] == 'My HiFi Song' + def slow_impl(*args, **kwargs): + started.set() + release.wait(timeout=1.0) + return '/tmp/done.flac' + + with patch.object(client, '_download_sync', side_effect=slow_impl): + download_id = _run_async(client.download('hifi', '999||My HiFi Song', 0)) + started.wait(timeout=1.0) + record = engine.get_record('hifi', download_id) + assert record['filename'] == '999||My HiFi Song' + assert record['username'] == 'hifi' + assert record['track_id'] == 999 + assert record['display_name'] == 'My HiFi Song' + release.set() -def test_download_spawns_daemon_thread_targeting_download_worker(hifi_client): - """Pinning: target is `_download_worker` (NOT `_thread_worker` like - Tidal/Qobuz). Engine refactor's plugin contract must accommodate - this naming variance OR force a rename — pinned here so the - decision is conscious.""" - captured_kwargs = {} +def test_get_all_downloads_reads_engine_records(hifi_client_with_engine): + client, engine = hifi_client_with_engine + engine.add_record('hifi', 'dl-1', { + 'id': 'dl-1', 'filename': '111||A', 'username': 'hifi', + 'state': 'InProgress, Downloading', 'progress': 50.0, + }) + result = _run_async(client.get_all_downloads()) + assert len(result) == 1 + assert result[0].id == 'dl-1' - def capture_thread(*args, **kwargs): - captured_kwargs.update(kwargs) - return type('FakeThread', (), {'start': lambda self: None})() - with patch('core.hifi_client.threading.Thread', side_effect=capture_thread): - _run_async(hifi_client.download('hifi', '777||Title', 0)) - - assert captured_kwargs.get('daemon') is True - assert captured_kwargs.get('target') == hifi_client._download_worker - args = captured_kwargs.get('args', ()) - # HiFi's worker signature: (download_id, track_id, display_name) — 3 args, not 4 - assert len(args) == 3 - assert args[1] == 777 - assert args[2] == 'Title' +def test_cancel_download_marks_cancelled(hifi_client_with_engine): + client, engine = hifi_client_with_engine + engine.add_record('hifi', 'dl-1', {'id': 'dl-1', 'state': 'InProgress, Downloading'}) + ok = _run_async(client.cancel_download('dl-1', None, remove=False)) + assert ok is True + assert engine.get_record('hifi', 'dl-1')['state'] == 'Cancelled'