#!/usr/bin/env python3 """ Test script for monitoring and debugging the SamCloud Music download functionality. This script connects to the Socket.IO server and simulates download requests while providing detailed logging of all communications. """ import asyncio import json import logging import signal import sys import time import socketio from datetime import datetime from urllib.parse import urlparse, urlunparse # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(), logging.FileHandler('download_test.log') ] ) logger = logging.getLogger('test_downloads') class DownloadTester: def __init__(self, url='http://localhost:6544'): self.url = self._ensure_http_url(url) self.sio = None self.running = False self.message_count = 0 self.downloads = {} self.connected = False def _ensure_http_url(self, url): """Ensure the URL is an HTTP URL.""" parsed = urlparse(url) if not parsed.scheme: parsed = parsed._replace(scheme='http') if not parsed.netloc: parsed = parsed._replace(netloc='localhost:6544') return urlunparse(parsed) async def connect(self): """Connect to the Socket.IO server.""" logger.info(f"Connecting to {self.url}...") # Create a new Socket.IO client self.sio = socketio.AsyncClient(logger=True, engineio_logger=True) # Set up event handlers @self.sio.event async def connect(): logger.info("Connected to Socket.IO server") self.connected = True # Request initial status await self.sio.emit('get_status') @self.sio.event async def connect_error(data): logger.error(f"Connection error: {data}") self.connected = False @self.sio.event async def disconnect(): logger.warning("Disconnected from server") self.connected = False @self.sio.on('download_progress') async def on_download_progress(data): await self.handle_download_progress(data) @self.sio.on('update_status') async def on_update_status(data): await self.handle_status_update(data) @self.sio.on('download_error') async def on_download_error(data): await self.handle_download_error(data) # Connect to the server try: await self.sio.connect(self.url, socketio_path='/socket.io') logger.info("Socket.IO connection established") except Exception as e: logger.error(f"Failed to connect to Socket.IO server: {e}") raise async def send_message(self, event, data=None): """Send a message to the Socket.IO server.""" if not self.sio or not self.connected: raise RuntimeError("Not connected to Socket.IO server") logger.debug(f"Emitting {event} with data: {data}") await self.sio.emit(event, data or {}) self.message_count += 1 async def start_download(self, url, item_type='track', name=None, artist=None): """Start a download.""" download_id = f'dl_{int(time.time())}' download_data = { 'url': url, 'type': item_type, 'name': name or f"Test {item_type.capitalize()} {time.time()}", 'artist': artist or "Test Artist", 'id': download_id, 'timestamp': datetime.utcnow().isoformat() } self.downloads[download_id] = { 'start_time': time.time(), 'status': 'queued', 'progress': 0, 'last_update': time.time(), 'data': download_data } logger.info(f"Starting download: {download_data['name']} ({download_id})") await self.send_message('download_item', download_data) return download_id async def listen(self): """Wait for messages from the Socket.IO server.""" self.running = True logger.info("Starting to wait for messages...") try: # The Socket.IO client handles messages in the background # We just need to keep the connection alive while self.running and self.connected: await asyncio.sleep(1) except asyncio.CancelledError: logger.info("Listen task cancelled") self.running = False except Exception as e: logger.error(f"Error in listener: {e}", exc_info=True) self.running = False finally: if self.sio: await self.sio.disconnect() async def handle_download_progress(self, data): """Handle download progress updates.""" if not isinstance(data, dict): logger.warning(f"Received invalid progress data: {data}") return download_id = data.get('id') if not download_id: logger.warning("Received progress update without download ID") return # Update download status if download_id in self.downloads: self.downloads[download_id].update({ 'status': data.get('status', 'unknown'), 'progress': data.get('progress', 0), 'message': data.get('message', ''), 'last_update': time.time(), 'data': {**self.downloads[download_id].get('data', {}), **data} }) # Log progress dl = self.downloads[download_id] logger.info( f"Download {download_id} - {dl['data'].get('name', 'Unknown')}: " f"{dl['status']} - {dl['progress']}% - {dl.get('message', '')}" ) # Log completion if dl['status'] == 'completed': duration = time.time() - dl['start_time'] logger.info(f"Download completed in {duration:.2f} seconds") elif dl['status'] == 'failed': logger.error(f"Download failed: {dl.get('message', 'Unknown error')}") async def handle_status_update(self, data): """Handle status updates from the server.""" if not isinstance(data, dict): logger.warning(f"Received invalid status update: {data}") return logger.info("Received status update from server") if 'history' in data and isinstance(data['history'], list): logger.info(f"Current download history: {len(data['history'])} items") for item in data['history']: if isinstance(item, dict): logger.debug(f"- {item.get('id')}: {item.get('status')} - {item.get('name')}") async def handle_download_error(self, data): """Handle download errors.""" if not isinstance(data, dict): logger.error(f"Received invalid error data: {data}") return error_msg = data.get('message', 'Unknown error') if isinstance(error_msg, dict): error_msg = error_msg.get('message', 'Unknown error') logger.error(f"Download error: {error_msg}") if 'details' in data: logger.debug(f"Error details: {data['details']}") async def run_test(self, test_urls): """Run a test with the given URLs.""" listen_task = None try: await self.connect() if not self.connected: logger.error("Failed to connect to server") return # Start listening for messages in the background listen_task = asyncio.create_task(self.listen()) # Start test downloads for i, url in enumerate(test_urls, 1): logger.info(f"\n=== Starting test download {i}/{len(test_urls)} ===") logger.info(f"URL: {url}") # Start the download download_id = await self.start_download( url=url, item_type='track', name=f"Test Track {i}", artist="Test Artist" ) # Wait a bit before starting the next download await asyncio.sleep(2) # Keep running until all downloads complete or timeout logger.info("\n=== All test downloads started. Monitoring progress... ===") # Monitor downloads until all complete or timeout start_time = time.time() timeout = 300 # 5 minutes timeout while self.running and (time.time() - start_time) < timeout: if not self.downloads: logger.info("No active downloads") break # Log current status active_downloads = [ dl for dl in self.downloads.values() if dl['status'] not in ['completed', 'failed', 'cancelled'] ] if not active_downloads: logger.info("All downloads completed or failed") break logger.info(f"Active downloads: {len(active_downloads)}") for dl_id, dl in self.downloads.items(): logger.info( f"- {dl_id}: {dl['status']} - " f"{dl['progress']}% - {dl.get('message', '')}" ) await asyncio.sleep(5) # Check every 5 seconds # Stop listening self.running = False if listen_task and not listen_task.done(): listen_task.cancel() try: await listen_task except asyncio.CancelledError: pass except Exception as e: logger.error(f"Test failed: {e}", exc_info=True) self.running = False if self.ws: await self.ws.close() async def main(): """Main function to run the test.""" # Test with some sample Spotify URLs test_urls = [ 'https://open.spotify.com/track/11dFghVXANMlKm9XsTMjGF', # Sample track 'https://open.spotify.com/album/6kZ42qRrzov54LcAk4onW9', # Sample album ] tester = DownloadTester() # Handle Ctrl+C def signal_handler(sig, frame): logger.info("Shutting down...") tester.running = False sys.exit(0) signal.signal(signal.SIGINT, signal_handler) # Run the test await tester.run_test(test_urls) if __name__ == "__main__": asyncio.run(main())