danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

C4: Migrate Qobuz to engine.worker

Browse source 
Same migration pattern as C2/C3. QobuzClient drops
active_downloads + _download_lock + _download_thread_worker.
Mid-download chunk-progress mutations + cancel-state checks
flow through engine.update_record / get_record. Query/cancel
methods read engine state.

Pinning tests updated to assert engine state. Suite still green
(316 download tests).
This commit is contained in:
Broque Thomas 2026-05-04 13:52:55 -07:00
parent 73fb60a68a
commit 7944568c5c
2 changed files with 161 additions and 216 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

258
core/qobuz_client.py
View file

@ -136,13 +136,19 @@ class QobuzClient:
self.user_info: Optional[Dict] = None self.user_info: Optional[Dict] = None
self._auth_error: Optional[str] = None self._auth_error: Optional[str] = None
# Download queue management # Engine reference is populated by set_engine() at registration
self.active_downloads: Dict[str, Dict[str, Any]] = {} # time. None until orchestrator wires the registry.
self._download_lock = threading.Lock() self._engine = None
# Try to restore saved session # Try to restore saved session
self._restore_session() self._restore_session()
def set_engine(self, engine):
"""Engine callback — gives the client access to the central
thread worker + state store. Engine calls this during
``register_plugin`` if the plugin defines it."""
self._engine = engine
def set_shutdown_check(self, check_callable): def set_shutdown_check(self, check_callable):
"""Set a callback function to check for system shutdown""" """Set a callback function to check for system shutdown"""
self.shutdown_check = check_callable self.shutdown_check = check_callable
@ -884,97 +890,35 @@ class QobuzClient:
return None return None
async def download(self, username: str, filename: str, file_size: int = 0) -> Optional[str]: async def download(self, username: str, filename: str, file_size: int = 0) -> Optional[str]:
""" """Download a Qobuz track. Delegates to engine.worker which
Download a Qobuz track (async, Soulseek-compatible interface). spawns the background thread + manages state."""
if '||' not in filename:
Returns download_id immediately and runs download in background thread. logger.error(f"Invalid filename format: {filename}")
return None
Args: if self._engine is None:
username: Ignored for Qobuz (always "qobuz") logger.error("Qobuz client has no engine reference — cannot dispatch download")
filename: Encoded as "track_id||display_name"
file_size: Ignored
"""
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 Qobuz track ID: {track_id_str}")
return None
logger.info(f"Starting Qobuz download: {display_name}")
download_id = str(uuid.uuid4())
with self._download_lock:
self.active_downloads[download_id] = {
'id': download_id,
'filename': filename,
'username': 'qobuz',
'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,
}
# Start download in background thread
download_thread = threading.Thread(
target=self._download_thread_worker,
args=(download_id, track_id, display_name, filename),
daemon=True,
)
download_thread.start()
logger.info(f"Qobuz download {download_id} started in background")
return download_id
except Exception as e:
logger.error(f"Failed to start Qobuz download: {e}")
import traceback
traceback.print_exc()
return None return None
def _download_thread_worker(self, download_id: str, track_id: int, display_name: str, original_filename: str): track_id_str, display_name = filename.split('||', 1)
"""Background thread worker for downloading Qobuz tracks."""
try: try:
with self._download_lock: track_id = int(track_id_str)
if download_id in self.active_downloads: except ValueError:
self.active_downloads[download_id]['state'] = 'InProgress, Downloading' logger.error(f"Invalid Qobuz track ID: {track_id_str}")
return None
file_path = self._download_sync(download_id, track_id, display_name) logger.info(f"Starting Qobuz download: {display_name}")
if file_path: return self._engine.worker.dispatch(
with self._download_lock: source_name='qobuz',
if download_id in self.active_downloads: target_id=track_id,
self.active_downloads[download_id]['state'] = 'Completed, Succeeded' display_name=display_name,
self.active_downloads[download_id]['progress'] = 100.0 original_filename=filename,
self.active_downloads[download_id]['file_path'] = file_path impl_callable=self._download_sync,
extra_record_fields={
logger.info(f"Qobuz download {download_id} completed: {file_path}") 'track_id': track_id,
else: 'display_name': display_name,
with self._download_lock: },
if download_id in self.active_downloads: )
self.active_downloads[download_id]['state'] = 'Errored'
logger.error(f"Qobuz download {download_id} failed")
except Exception as e:
logger.error(f"Qobuz download thread failed for {download_id}: {e}")
import traceback
traceback.print_exc()
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['state'] = 'Errored'
def _download_sync(self, download_id: str, track_id: int, display_name: str) -> Optional[str]: def _download_sync(self, download_id: str, track_id: int, display_name: str) -> Optional[str]:
""" """
@ -1050,9 +994,8 @@ class QobuzClient:
chunk_size = 64 * 1024 # 64KB chunks chunk_size = 64 * 1024 # 64KB chunks
start_time = time.time() start_time = time.time()
with self._download_lock: if self._engine is not None:
if download_id in self.active_downloads: self._engine.update_record('qobuz', download_id, {'size': total_size})
self.active_downloads[download_id]['size'] = total_size
with open(out_path, 'wb') as f: with open(out_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=chunk_size): for chunk in response.iter_content(chunk_size=chunk_size):
@ -1061,9 +1004,10 @@ class QobuzClient:
# Check for shutdown or cancellation # Check for shutdown or cancellation
cancelled = False cancelled = False
with self._download_lock: if self._engine is not None:
if download_id in self.active_downloads: rec = self._engine.get_record('qobuz', download_id)
cancelled = self.active_downloads[download_id].get('state') == 'Cancelled' if rec is not None:
cancelled = rec.get('state') == 'Cancelled'
if cancelled or (self.shutdown_check and self.shutdown_check()): if cancelled or (self.shutdown_check and self.shutdown_check()):
reason = "cancelled" if cancelled else "server shutting down" reason = "cancelled" if cancelled else "server shutting down"
logger.info(f"Aborting Qobuz download mid-stream: {reason}") logger.info(f"Aborting Qobuz download mid-stream: {reason}")
@ -1084,18 +1028,20 @@ class QobuzClient:
progress = 0 progress = 0
time_remaining = None time_remaining = None
with self._download_lock: if self._engine is not None:
if download_id in self.active_downloads: self._engine.update_record('qobuz', download_id, {
self.active_downloads[download_id]['transferred'] = downloaded 'transferred': downloaded,
self.active_downloads[download_id]['progress'] = round(progress, 1) 'progress': round(progress, 1),
self.active_downloads[download_id]['speed'] = int(speed) 'speed': int(speed),
self.active_downloads[download_id]['time_remaining'] = time_remaining 'time_remaining': time_remaining,
})
# If download was aborted (shutdown/cancel), clean up partial file # If download was aborted (shutdown/cancel), clean up partial file
abort_check = False abort_check = False
with self._download_lock: if self._engine is not None:
if download_id in self.active_downloads: rec = self._engine.get_record('qobuz', download_id)
abort_check = self.active_downloads[download_id].get('state') == 'Cancelled' if rec is not None:
abort_check = rec.get('state') == 'Cancelled'
if abort_check or (self.shutdown_check and self.shutdown_check()): if abort_check or (self.shutdown_check and self.shutdown_check()):
out_path.unlink(missing_ok=True) out_path.unlink(missing_ok=True)
return None return None
@ -1139,79 +1085,55 @@ class QobuzClient:
# ===================== Status / Cancel / Clear ===================== # ===================== Status / Cancel / Clear =====================
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]: async def get_all_downloads(self) -> List[DownloadStatus]:
"""Get all active downloads (matches Soulseek interface).""" if self._engine is None:
download_statuses = [] return []
return [
with self._download_lock: self._record_to_status(record)
for _download_id, info in self.active_downloads.items(): for record in self._engine.iter_records_for_source('qobuz')
status = 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'),
)
download_statuses.append(status)
return download_statuses
async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]: async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]:
"""Get status of a specific download (matches Soulseek interface).""" if self._engine is None:
with self._download_lock: return None
if download_id not in self.active_downloads: record = self._engine.get_record('qobuz', download_id)
return None return self._record_to_status(record) if record is not None else None
info = self.active_downloads[download_id]
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'),
)
async def cancel_download(self, download_id: str, username: str = None, remove: bool = False) -> bool: async def cancel_download(self, download_id: str, username: str = None, remove: bool = False) -> bool:
"""Cancel an active download (matches Soulseek interface).""" if self._engine is None:
try:
with self._download_lock:
if download_id not in self.active_downloads:
logger.warning(f"Download {download_id} not found")
return False
self.active_downloads[download_id]['state'] = 'Cancelled'
logger.info(f"Marked Qobuz download {download_id} as cancelled")
if remove:
del self.active_downloads[download_id]
logger.info(f"Removed Qobuz download {download_id} from queue")
return True
except Exception as e:
logger.error(f"Failed to cancel download {download_id}: {e}")
return False return False
if self._engine.get_record('qobuz', download_id) is None:
logger.warning(f"Qobuz download {download_id} not found")
return False
self._engine.update_record('qobuz', download_id, {'state': 'Cancelled'})
logger.info(f"Marked Qobuz download {download_id} as cancelled")
if remove:
self._engine.remove_record('qobuz', download_id)
logger.info(f"Removed Qobuz download {download_id} from queue")
return True
async def clear_all_completed_downloads(self) -> bool: async def clear_all_completed_downloads(self) -> bool:
"""Clear all terminal downloads from the list (matches Soulseek interface).""" if self._engine is None:
return True
try: try:
with self._download_lock: terminal = {'Completed, Succeeded', 'Cancelled', 'Errored', 'Aborted'}
ids_to_remove = [ for record in list(self._engine.iter_records_for_source('qobuz')):
did for did, info in self.active_downloads.items() if record.get('state') in terminal:
if info.get('state', '') in ('Completed, Succeeded', 'Cancelled', 'Errored', 'Aborted') self._engine.remove_record('qobuz', record['id'])
]
for did in ids_to_remove:
del self.active_downloads[did]
return True return True
except Exception as e: except Exception as e:
logger.error(f"Error clearing downloads: {e}") logger.error(f"Error clearing downloads: {e}")

119
tests/downloads/test_qobuz_pinning.py
View file

@ -1,8 +1,6 @@
"""Phase A pinning tests for QobuzClient's download lifecycle. """Phase A pinning tests for QobuzClient — UPDATED for Phase C4.
Qobuz hits the Qobuz REST API + downloads HLS-segmented FLAC. Post-C4 the client uses engine.worker for thread + state management.
Same thread-worker + state-dict pattern as Tidal/HiFi — Phase C
will lift the threading. These tests pin the contract.
""" """
from __future__ import annotations from __future__ import annotations
@ -14,6 +12,7 @@ from unittest.mock import patch
import pytest import pytest
from core.download_engine import DownloadEngine
from core.qobuz_client import QobuzClient from core.qobuz_client import QobuzClient
@ -26,68 +25,92 @@ def _run_async(coro):
@pytest.fixture @pytest.fixture
def qobuz_client(): def qobuz_client_with_engine():
client = QobuzClient.__new__(QobuzClient) client = QobuzClient.__new__(QobuzClient)
client.download_path = Path('./test_qobuz_downloads') client.download_path = Path('./test_qobuz_downloads')
client.shutdown_check = None client.shutdown_check = None
client.active_downloads = {} client._engine = None
client._download_lock = threading.Lock() engine = DownloadEngine()
return client client.set_engine(engine)
return client, engine
def test_download_returns_none_for_invalid_filename_format(qobuz_client): def test_download_returns_none_for_invalid_filename_format(qobuz_client_with_engine):
"""Pinning: filename without `||` → None, not exception.""" client, _ = qobuz_client_with_engine
result = _run_async(qobuz_client.download('qobuz', 'no-separator', 0)) result = _run_async(client.download('qobuz', 'no-separator', 0))
assert result is None assert result is None
def test_download_returns_none_for_non_integer_track_id(qobuz_client): def test_download_returns_none_for_non_integer_track_id(qobuz_client_with_engine):
"""Pinning: Qobuz REST API uses int track IDs. Non-int → None.""" client, _ = qobuz_client_with_engine
result = _run_async(qobuz_client.download('qobuz', 'not-int||title', 0)) result = _run_async(client.download('qobuz', 'not-int||title', 0))
assert result is None assert result is None
def test_download_returns_uuid_for_valid_filename(qobuz_client): def test_download_returns_none_when_engine_not_wired():
"""Pinning: valid `<int>||display` returns UUID download_id.""" client = QobuzClient.__new__(QobuzClient)
with patch('core.qobuz_client.threading.Thread') as fake: client._engine = None
fake.return_value.start = lambda: None result = _run_async(client.download('qobuz', '12345||x', 0))
result = _run_async(qobuz_client.download('qobuz', '12345||Some Song', 0)) assert result is None
def test_download_returns_uuid_for_valid_filename(qobuz_client_with_engine):
client, _ = qobuz_client_with_engine
with patch.object(client, '_download_sync', return_value='/tmp/x.flac'):
result = _run_async(client.download('qobuz', '12345||Some Song', 0))
assert result is not None assert result is not None
assert len(result) == 36 assert len(result) == 36
def test_download_populates_active_downloads_with_initial_state(qobuz_client): def test_download_populates_engine_record_with_initial_state(qobuz_client_with_engine):
"""Pinning: per-download record schema for engine extraction.""" client, engine = qobuz_client_with_engine
with patch('core.qobuz_client.threading.Thread') as fake: started = threading.Event()
fake.return_value.start = lambda: None release = threading.Event()
download_id = _run_async(qobuz_client.download('qobuz', '999||My Qobuz Song', 0))
record = qobuz_client.active_downloads[download_id] def slow_impl(*args, **kwargs):
assert record['id'] == download_id started.set()
assert record['filename'] == '999||My Qobuz Song' release.wait(timeout=1.0)
assert record['username'] == 'qobuz' return '/tmp/done.flac'
assert record['state'] == 'Initializing'
assert record['progress'] == 0.0 with patch.object(client, '_download_sync', side_effect=slow_impl):
assert record['track_id'] == 999 download_id = _run_async(client.download('qobuz', '999||My Qobuz Song', 0))
assert record['display_name'] == 'My Qobuz Song' started.wait(timeout=1.0)
assert record['file_path'] is None record = engine.get_record('qobuz', download_id)
assert record is not None
assert record['filename'] == '999||My Qobuz Song'
assert record['username'] == 'qobuz'
assert record['track_id'] == 999
assert record['display_name'] == 'My Qobuz Song'
release.set()
def test_download_spawns_daemon_thread_targeting_worker(qobuz_client): def test_get_all_downloads_reads_engine_records(qobuz_client_with_engine):
"""Pinning: daemon thread → `_download_thread_worker(download_id, track_id, display_name, original_filename)`.""" client, engine = qobuz_client_with_engine
captured_kwargs = {} engine.add_record('qobuz', 'dl-1', {
'id': 'dl-1', 'filename': '111||A', 'username': 'qobuz',
'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.qobuz_client.threading.Thread', side_effect=capture_thread): def test_cancel_download_marks_cancelled(qobuz_client_with_engine):
_run_async(qobuz_client.download('qobuz', '777||Title', 0)) client, engine = qobuz_client_with_engine
engine.add_record('qobuz', 'dl-1', {'id': 'dl-1', 'state': 'InProgress, Downloading'})
assert captured_kwargs.get('daemon') is True ok = _run_async(client.cancel_download('dl-1', None, remove=False))
assert captured_kwargs.get('target') == qobuz_client._download_thread_worker assert ok is True
args = captured_kwargs.get('args', ()) assert engine.get_record('qobuz', 'dl-1')['state'] == 'Cancelled'
assert len(args) == 4
assert args[1] == 777
assert args[2] == 'Title' def test_clear_all_completed_drops_only_terminal_records(qobuz_client_with_engine):
assert args[3] == '777||Title' client, engine = qobuz_client_with_engine
engine.add_record('qobuz', 'done', {'id': 'done', 'state': 'Completed, Succeeded'})
engine.add_record('qobuz', 'live', {'id': 'live', 'state': 'InProgress, Downloading'})
_run_async(client.clear_all_completed_downloads())
assert engine.get_record('qobuz', 'done') is None
assert engine.get_record('qobuz', 'live') is not None