diff --git a/src/system/tonedownload.py b/src/system/tonedownload.py new file mode 100644 index 0000000..148b2c8 --- /dev/null +++ b/src/system/tonedownload.py @@ -0,0 +1,525 @@ +"""Tone3000 API client for searching and downloading NAM models and IRs. + +Tone3000 (successor to ToneHunt) is a Supabase-backed Next.js app that +hosts NAM .nam model files and IR .wav files. Its Supabase public anon key +is embedded in the client-side JavaScript — we use it here as a read-only +API client for the Supabase REST API. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import re +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + +import aiohttp + +logger = logging.getLogger(__name__) + +# ── Constants ───────────────────────────────────────────────────────── + +SUPABASE_URL = "https://api.tone3000.com" +SUPABASE_REST = f"{SUPABASE_URL}/rest/v1" +SUPABASE_STORAGE = f"{SUPABASE_URL}/storage/v1/object/public" + +# Anon key for Supabase read-only access. Extracted from the Tone3000 +# Next.js client bundle. This is a public key designed to be embedded +# in client-side JS — it only allows read access to public tables. +# If it ever changes, the client will dynamically re-fetch it from +# the Tone3000 homepage JS bundles. +SUPABASE_ANON_KEY = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6Imd6eWJpdW9weGtkeGJ5dG5vamRzIiwicm9sZSI6ImFub24iLCJpYXQiOjE3MzgwODIxNjUsImV4cCI6MjA1MzY1ODE2NX0.Gq66BJXjtLsqP2nAGXm9Xb9PAjoeZalWUj66K4nmVSU" + +# Cache TTL in seconds +CACHE_TTL = 300 # 5 minutes + +# Default download directory +DEFAULT_MODELS_DIR = Path.home() / ".pedal" / "models" +DEFAULT_IRS_DIR = Path.home() / ".pedal" / "irs" + + +# ── Data types ──────────────────────────────────────────────────────── + + +@dataclass +class ModelResult: + """A search result representing a NAM model on Tone3000.""" + + id: int + name: str + download_url: str + size_bytes: Optional[int] = None + architecture: Optional[str] = None + pack_name: Optional[str] = None + author_id: Optional[str] = None + author_username: Optional[str] = None + author_avatar_url: Optional[str] = None + tone_id: Optional[int] = None + tone_title: Optional[str] = None + tone_gear: Optional[str] = None + tone_images: list[str] = field(default_factory=list) + tone_description: Optional[str] = None + downloads: int = 0 + likes: int = 0 + created_at: Optional[str] = None + + @property + def is_ir(self) -> bool: + """Check if this model is an impulse response (.wav file).""" + return self.tone_gear == "ir" or (self.download_url or "").lower().endswith(".wav") + + @property + def filename(self) -> str: + """Get the filename from the download URL.""" + return Path(self.download_url.split("/")[-1]).name + + @property + def display_name(self) -> str: + return self.tone_title or self.name or self.filename + + @property + def thumbnail_url(self) -> Optional[str]: + if self.tone_images: + return self.tone_images[0] + return None + + +@dataclass +class IRResult(ModelResult): + """A search result representing an IR file.""" + + sample_rate: Optional[int] = None + num_taps: Optional[int] = None + length_ms: Optional[float] = None + + @property + def is_ir(self) -> bool: + return True + + +# ── Anon key discovery ──────────────────────────────────────────────── + + +async def discover_anon_key(session: aiohttp.ClientSession) -> Optional[str]: + """Fetch the Supabase anon key from the Tone3000 website JS bundle. + + The key is embedded in the Next.js client-side chunk. + Returns None if discovery fails (silently falls back to hardcoded key). + """ + try: + async with session.get("https://tone3000.com", timeout=aiohttp.ClientTimeout(total=15)) as resp: + html = await resp.text() + + # Find JS chunk URLs that contain the client setup code + chunk_urls = re.findall(r'src="([^"]*2274[^"]*\.js[^"]*)"', html) + if not chunk_urls: + # Fallback: look for any chunk with the key pattern + chunk_urls = re.findall( + r'src="(/_[^"]*chunks/[^"]*\.js[^"]*)"', html + ) + # Sort by name to get a deterministic order + chunk_urls = sorted(set(chunk_urls)) + + for chunk_url in chunk_urls[:5]: + full_url = f"https://tone3000.com{chunk_url}" if chunk_url.startswith("/") else chunk_url + async with session.get(full_url, timeout=aiohttp.ClientTimeout(total=15)) as resp: + js = await resp.text() + + # Pattern: createBrowserClient)("url","token") + match = re.search( + r'createBrowserClient\)\(["\']https://api\.tone3000\.com["\'],["\']([^"\']+)["\']', + js, + ) + if match: + key = match.group(1) + if key.startswith("eyJ"): + logger.info("Discovered Supabase anon key from JS bundle") + return key + + logger.warning("Could not discover anon key from Tone3000 JS") + return None + except Exception as e: + logger.debug("Anon key discovery failed: %s", e) + return None + + +# ── API Client ──────────────────────────────────────────────────────── + + +class Tone3000Client: + """Client for the Tone3000 Supabase REST API. + + Uses the public anon key embedded in the Tone3000 web client for + read-only access to the models, tones, and users tables. + """ + + def __init__( + self, + anon_key: str = SUPABASE_ANON_KEY, + models_dir: Path = DEFAULT_MODELS_DIR, + irs_dir: Path = DEFAULT_IRS_DIR, + session: Optional[aiohttp.ClientSession] = None, + ) -> None: + self._anon_key = anon_key + self._models_dir = models_dir + self._irs_dir = irs_dir + self._session = session + self._cache: dict[str, tuple[float, list[dict]]] = {} + self._headers = { + "apikey": anon_key, + "Authorization": f"Bearer {anon_key}", + "Accept": "application/json", + } + self._named_user_cache: dict[str, Optional[dict]] = {} + self._named_tone_cache: dict[int, Optional[dict]] = {} + + # ── Session management ─────────────────────────────────────── + + async def _get_session(self) -> aiohttp.ClientSession: + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession( + headers=self._headers, + timeout=aiohttp.ClientTimeout(total=30), + ) + return self._session + + async def close(self) -> None: + if self._session and not self._session.closed: + await self._session.close() + + # ── Caching ────────────────────────────────────────────────── + + def _get_cached(self, key: str) -> Optional[list[dict]]: + entry = self._cache.get(key) + if entry and (time.monotonic() - entry[0]) < CACHE_TTL: + return entry[1] + return None + + def _set_cache(self, key: str, data: list[dict]) -> None: + self._cache[key] = (time.monotonic(), data) + + # ── Core API calls ─────────────────────────────────────────── + + async def _supabase_get( + self, path: str, params: Optional[dict] = None + ) -> list[dict]: + """Make a GET request to the Supabase REST API.""" + session = await self._get_session() + url = f"{SUPABASE_REST}/{path}" + try: + async with session.get(url, params=params) as resp: + if resp.status in (200, 206): + data = await resp.json() + return data if isinstance(data, list) else [data] + else: + text = await resp.text() + logger.warning( + "Supabase API error %d: %s", resp.status, text[:200] + ) + return [] + except (aiohttp.ClientError, asyncio.TimeoutError) as e: + logger.warning("Supabase request failed: %s", e) + return [] + + # ── Search ─────────────────────────────────────────────────── + + async def search_models( + self, + query: str, + page: int = 0, + page_size: int = 20, + force_refresh: bool = False, + ) -> list[ModelResult]: + """Search Tone3000 for NAM models matching the query. + + Searches model names. Results are cached for 5 minutes. + """ + cache_key = f"models:{query}:{page}" + if not force_refresh: + cached = self._get_cached(cache_key) + if cached is not None: + return [ModelResult(**r) for r in cached] + + # Search models table by name similarity + raw_models = await self._supabase_get("models", { + "select": "id,name,model_url,size,pack_name,user_id,tone_id,architecture_version,created_at", + "name": f"ilike.*{query}*", + "order": "created_at.desc", + "limit": str(page_size), + "offset": str(page * page_size), + }) + + results = await self._enrich_models(raw_models) + self._set_cache(cache_key, [r.__dict__ for r in results]) + return results + + async def search_irs( + self, + query: str, + page: int = 0, + page_size: int = 20, + force_refresh: bool = False, + ) -> list[IRResult]: + """Search Tone3000 for IRs (impulse responses). + + IRs are stored in the tones table with gear='ir'. We search + by tone title and return the linked model files. + """ + cache_key = f"irs:{query}:{page}" + if not force_refresh: + cached = self._get_cached(cache_key) + if cached is not None: + return [IRResult(**r) for r in cached] + + # Search tones with gear='ir' matching the query + raw_tones = await self._supabase_get("tones", { + "select": "id,title,description,gear,images", + "title": f"ilike.*{query}*", + "gear": "eq.ir", + "limit": str(page_size), + "offset": str(page * page_size), + }) + + results: list[IRResult] = [] + seen_ids: set[int] = set() + + for tone in raw_tones: + tone_models = await self._supabase_get("models", { + "select": "id,name,model_url,size,pack_name,user_id,tone_id,created_at", + "tone_id": f"eq.{tone['id']}", + "limit": "5", + }) + + for m in tone_models: + if m["id"] in seen_ids: + continue + seen_ids.add(m["id"]) + + author = await self._get_user(m.get("user_id", "")) + + ir = IRResult( + id=m["id"], + name=m.get("name", ""), + download_url=m.get("model_url", ""), + size_bytes=m.get("size"), + pack_name=m.get("pack_name"), + author_id=m.get("user_id"), + author_username=author.get("username") if author else None, + author_avatar_url=author.get("avatar_url") if author else None, + tone_id=tone["id"], + tone_title=tone.get("title"), + tone_gear="ir", + tone_images=tone.get("images", []), + tone_description=tone.get("description"), + created_at=m.get("created_at"), + ) + results.append(ir) + + self._set_cache(cache_key, [r.__dict__ for r in results]) + return results + + # ── Trending / Browse ──────────────────────────────────────── + + async def get_trending_models(self, limit: int = 20) -> list[ModelResult]: + """Get trending NAM models.""" + raw = await self._supabase_get("models", { + "select": "id,name,model_url,size,pack_name,user_id,tone_id,architecture_version,created_at", + "order": "created_at.desc", + "limit": str(limit), + }) + return await self._enrich_models(raw) + + async def get_trending_irs(self, limit: int = 20) -> list[IRResult]: + """Get trending IRs.""" + raw_tones = await self._supabase_get("tones", { + "select": "id,title,description,gear,images", + "gear": "eq.ir", + "order": "created_at.desc", + "limit": str(limit), + }) + + results: list[IRResult] = [] + for tone in raw_tones: + tone_models = await self._supabase_get("models", { + "select": "id,name,model_url,size,pack_name,user_id,tone_id,created_at", + "tone_id": f"eq.{tone['id']}", + "limit": "3", + }) + for m in tone_models[:1]: + author = await self._get_user(m.get("user_id", "")) + ir = IRResult( + id=m["id"], + name=m.get("name", ""), + download_url=m.get("model_url", ""), + size_bytes=m.get("size"), + tone_title=tone.get("title"), + tone_images=tone.get("images", []), + author_username=author.get("username") if author else None, + author_avatar_url=author.get("avatar_url") if author else None, + created_at=m.get("created_at"), + ) + results.append(ir) + return results + + # ── Enrichment helpers ─────────────────────────────────────── + + async def _enrich_models( + self, raw_models: list[dict] + ) -> list[ModelResult]: + """Enrich raw model data with user/tone info.""" + results: list[ModelResult] = [] + for m in raw_models: + tone_data = None + if m.get("tone_id"): + tone_data = await self._get_tone(m["tone_id"]) + author = await self._get_user(m.get("user_id", "")) + + results.append(ModelResult( + id=m["id"], + name=m.get("name", ""), + download_url=m.get("model_url", ""), + size_bytes=m.get("size"), + architecture=m.get("architecture_version"), + pack_name=m.get("pack_name"), + author_id=m.get("user_id"), + author_username=author.get("username") if author else None, + author_avatar_url=author.get("avatar_url") if author else None, + tone_id=m.get("tone_id"), + tone_title=tone_data.get("title") if tone_data else None, + tone_gear=tone_data.get("gear") if tone_data else None, + tone_images=tone_data.get("images", []) if tone_data else [], + tone_description=tone_data.get("description") if tone_data else None, + created_at=m.get("created_at"), + )) + return results + + async def _get_user(self, user_id: str) -> Optional[dict]: + if not user_id: + return None + if user_id in self._named_user_cache: + return self._named_user_cache[user_id] + data = await self._supabase_get("users", { + "select": "id,username,avatar_url,display_name", + "id": f"eq.{user_id}", + "limit": "1", + }) + result = data[0] if data else None + self._named_user_cache[user_id] = result + return result + + async def _get_tone(self, tone_id: int) -> Optional[dict]: + if tone_id in self._named_tone_cache: + return self._named_tone_cache[tone_id] + data = await self._supabase_get("tones", { + "select": "id,title,description,gear,images", + "id": f"eq.{tone_id}", + "limit": "1", + }) + result = data[0] if data else None + self._named_tone_cache[tone_id] = result + return result + + # ── Download ───────────────────────────────────────────────── + + async def download_model( + self, + model: ModelResult, + progress_callback: Optional[callable] = None, + ) -> Optional[Path]: + """Download a NAM model to the pedal's models directory. + + Returns the local path to the downloaded file, or None on failure. + """ + dest_dir = self._models_dir + dest_dir.mkdir(parents=True, exist_ok=True) + + filename = model.filename + if not filename.endswith(".nam"): + filename += ".nam" + dest_path = dest_dir / filename + + return await self._download_file(model.download_url, dest_path, progress_callback) + + async def download_ir( + self, + ir: IRResult, + progress_callback: Optional[callable] = None, + ) -> Optional[Path]: + """Download an IR file to the pedal's IRs directory.""" + dest_dir = self._irs_dir + dest_dir.mkdir(parents=True, exist_ok=True) + + filename = ir.filename + if not filename.endswith(".wav"): + filename += ".wav" + dest_path = dest_dir / filename + + return await self._download_file(ir.download_url, dest_path, progress_callback) + + async def _download_file( + self, + url: str, + dest_path: Path, + progress_callback: Optional[callable] = None, + ) -> Optional[Path]: + """Download a file with optional progress reporting.""" + if not url: + logger.error("No download URL available") + return None + + session = await self._get_session() + try: + async with session.get( + url, timeout=aiohttp.ClientTimeout(total=120) + ) as resp: + if resp.status not in (200, 302, 307): + logger.warning("Download failed: HTTP %d", resp.status) + return None + + total = int(resp.headers.get("Content-Length", 0)) or 0 + downloaded = 0 + + with open(dest_path, "wb") as f: + async for chunk in resp.content.iter_chunked(8192): + f.write(chunk) + downloaded += len(chunk) + if progress_callback and total: + progress_callback(downloaded / total) + + logger.info( + "Downloaded %s (%d bytes)", dest_path.name, downloaded + ) + return dest_path + + except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as e: + logger.error("Download failed: %s", e) + if dest_path.exists(): + dest_path.unlink() + return None + + # ── Connectivity check ─────────────────────────────────────── + + async def ping(self) -> bool: + """Check if the Tone3000 API is reachable.""" + try: + data = await self._supabase_get("models", {"limit": "1"}) + return len(data) > 0 + except Exception: + return False + + +# ── Convenience helpers ─────────────────────────────────────────────── + + +def format_size(size_bytes: Optional[int]) -> str: + """Format a file size in bytes to a human-readable string.""" + if size_bytes is None: + return "Unknown" + mb = size_bytes / (1024 * 1024) + if mb < 1: + return f"{size_bytes / 1024:.1f} KB" + return f"{mb:.1f} MB" \ No newline at end of file diff --git a/src/web/server.py b/src/web/server.py index f876714..c4ee052 100644 --- a/src/web/server.py +++ b/src/web/server.py @@ -31,6 +31,13 @@ from ..dsp.ir_loader import IRLoader from ..dsp.pipeline import AudioPipeline from ..presets.manager import PresetManager from ..presets.types import FXBlock, FXType, Preset +from ..system.tonedownload import ( + Tone3000Client, + ModelResult, + IRResult, + discover_anon_key, + format_size, +) logger = logging.getLogger(__name__) @@ -118,6 +125,7 @@ class WebServer: self._manager = ConnectionManager() self._server: Optional[uvicorn.Server] = None self._task: Optional[asyncio.Task] = None + self._tonedownload: Optional[Tone3000Client] = None self._app = self._build_app() # ── App factory ───────────────────────────────────────────────────── @@ -750,6 +758,129 @@ class WebServer: self._manager.disconnect(ws) logger.info("WebSocket client disconnected (%d active)", self._manager.active_count) + # ── Tone3000 downloader endpoints ─────────────────────── + + @app.get("/api/tonedownload/status") + async def tonedownload_status(): + """Check if the Tone3000 API is reachable.""" + client = await self._get_tonedownload() + if not client: + return {"online": False, "error": "Client not initialized"} + ok = await client.ping() + return {"online": ok} + + @app.get("/api/models/tonedownload/search") + async def tonedownload_search_models(q: str = "", page: int = 0): + """Search Tone3000 for NAM models.""" + client = await self._get_tonedownload() + if not client: + raise HTTPException(status_code=503, detail="Tone3000 client unavailable") + if not q.strip(): + results = await client.get_trending_models() + else: + results = await client.search_models(q.strip(), page=page) + return { + "results": [ + { + "id": r.id, + "name": r.display_name, + "filename": r.filename, + "size": r.size_bytes, + "size_display": format_size(r.size_bytes), + "author": r.author_username or "Unknown", + "author_avatar": r.author_avatar_url, + "thumbnail": r.thumbnail_url, + "download_url": r.download_url, + "tone_description": r.tone_description, + "pack_name": r.pack_name, + "architecture": r.architecture, + "created_at": r.created_at, + } + for r in results + ], + } + + @app.get("/api/irs/tonedownload/search") + async def tonedownload_search_irs(q: str = "", page: int = 0): + """Search Tone3000 for IRs.""" + client = await self._get_tonedownload() + if not client: + raise HTTPException(status_code=503, detail="Tone3000 client unavailable") + if not q.strip(): + results = await client.get_trending_irs() + else: + results = await client.search_irs(q.strip(), page=page) + return { + "results": [ + { + "id": r.id, + "name": r.display_name, + "filename": r.filename, + "size": r.size_bytes, + "size_display": format_size(r.size_bytes), + "author": r.author_username or "Unknown", + "author_avatar": r.author_avatar_url, + "thumbnail": r.thumbnail_url, + "download_url": r.download_url, + "tone_description": r.tone_description, + "created_at": r.created_at, + } + for r in results + ], + } + + @app.post("/api/models/tonedownload/install") + async def tonedownload_install_model(data: dict): + """Download a .nam file to the pedal's models directory.""" + client = await self._get_tonedownload() + if not client: + raise HTTPException(status_code=503, detail="Tone3000 client unavailable") + url = data.get("download_url", "") + name = data.get("name", "model.nam") + if not url: + raise HTTPException(status_code=400, detail="download_url required") + model = ModelResult( + id=data.get("id", 0), + name=name, + download_url=url, + size_bytes=data.get("size"), + pack_name=data.get("pack_name"), + tone_title=data.get("display_name", name), + ) + path = await client.download_model(model) + if not path: + raise HTTPException(status_code=500, detail="Download failed") + # Refresh the model list via NAM host + nam = self.deps.nam_host + if nam: + nam.list_available_models() + return {"ok": True, "path": str(path), "filename": path.name} + + @app.post("/api/irs/tonedownload/install") + async def tonedownload_install_ir(data: dict): + """Download a .wav IR to the pedal's IRs directory.""" + client = await self._get_tonedownload() + if not client: + raise HTTPException(status_code=503, detail="Tone3000 client unavailable") + url = data.get("download_url", "") + name = data.get("name", "ir.wav") + if not url: + raise HTTPException(status_code=400, detail="download_url required") + ir = IRResult( + id=0, + name=name, + download_url=url, + size_bytes=data.get("size"), + tone_title=data.get("display_name", name), + ) + path = await client.download_ir(ir) + if not path: + raise HTTPException(status_code=500, detail="Download failed") + ir_loader = self.deps.ir_loader + if ir_loader: + ir_loader.get_irs() + return {"ok": True, "path": str(path), "filename": path.name} + return app # ── Network/BT state gathering ────────────────────────────────── @@ -847,6 +978,21 @@ class WebServer: "bluetooth": self._gather_bt_state(), } + # ── Tone3000 client lazy init ──────────────────────────────── + + async def _get_tonedownload(self) -> Optional[Tone3000Client]: + """Lazily initialize the Tone3000 download client.""" + if self._tonedownload is None: + from aiohttp import ClientSession + async with ClientSession() as session: + key = await discover_anon_key(session) + if not key: + # Fall back to hardcoded key + from ..system.tonedownload import SUPABASE_ANON_KEY + key = SUPABASE_ANON_KEY + self._tonedownload = Tone3000Client(anon_key=key) + return self._tonedownload + # ── Lifecycle ──────────────────────────────────────────────────── def start(self) -> None: diff --git a/src/web/static/irs.js b/src/web/static/irs.js index fe8aa7f..f4bdedc 100644 --- a/src/web/static/irs.js +++ b/src/web/static/irs.js @@ -92,4 +92,114 @@ async function uploadIR() { } } -document.addEventListener('DOMContentLoaded', loadIRs); \ No newline at end of file +document.addEventListener('DOMContentLoaded', loadIRs); + +// ── Tone3000 Browser ──────────────────────────────────────────────── + +const INSTALLED_IRS_KEY = 'tonedownload_installed_irs'; + +function getInstalledIRs() { + try { return JSON.parse(localStorage.getItem(INSTALLED_IRS_KEY) || '[]'); } catch { return []; } +} + +function markIRInstalled(url) { + const installed = getInstalledIRs(); + if (!installed.includes(url)) { + installed.push(url); + localStorage.setItem(INSTALLED_IRS_KEY, JSON.stringify(installed)); + } +} + +function isIRInstalled(url) { + return getInstalledIRs().includes(url); +} + +async function searchToneIRs() { + const input = document.getElementById('ir-tone-search'); + const query = input.value.trim(); + const resultsEl = document.getElementById('ir-tone-results'); + const loaderEl = document.getElementById('ir-tone-loader'); + const offlineEl = document.getElementById('ir-tone-offline'); + + resultsEl.innerHTML = ''; + offlineEl.style.display = 'none'; + + if (!query) { + loadTrendingIRs(); + return; + } + + loaderEl.style.display = 'block'; + + try { + const data = await apiGet('/irs/tonedownload/search?q=' + encodeURIComponent(query)); + loaderEl.style.display = 'none'; + renderToneIRResults(resultsEl, data.results || []); + } catch (e) { + loaderEl.style.display = 'none'; + if (e.message.includes('503') || e.message.includes('Failed to fetch')) { + offlineEl.style.display = 'block'; + } else { + resultsEl.innerHTML = `

Error: ${escapeHtml(e.message)}

`; + } + } +} + +async function loadTrendingIRs() { + const resultsEl = document.getElementById('ir-tone-results'); + const loaderEl = document.getElementById('ir-tone-loader'); + const offlineEl = document.getElementById('ir-tone-offline'); + + resultsEl.innerHTML = ''; + offlineEl.style.display = 'none'; + loaderEl.style.display = 'block'; + + try { + const data = await apiGet('/irs/tonedownload/search?q='); + loaderEl.style.display = 'none'; + renderToneIRResults(resultsEl, data.results || []); + } catch (e) { + loaderEl.style.display = 'none'; + if (e.message.includes('503') || e.message.includes('Failed to fetch')) { + offlineEl.style.display = 'block'; + } else { + resultsEl.innerHTML = `

Error: ${escapeHtml(e.message)}

`; + } + } +} + +function renderToneIRResults(container, results) { + if (!results || results.length === 0) { + container.innerHTML = '

No results found.

'; + return; + } + + let html = ''; + for (const r of results) { + const installed = isIRInstalled(r.download_url); + const thumb = r.thumbnail + ? `` + : '
📊
'; + + html += `
+ ${thumb} +
+
${escapeHtml(r.name)}
+
+ ${escapeHtml(r.author)} · ${escapeHtml(r.size_display)} +
+ ${r.tone_description ? `
${escapeHtml(r.tone_description.substring(0, 120))}
` : ''} +
+
+ ${installed + ? '✓ Installed' + : `` + } +
+
`; + } + container.innerHTML = html; +} \ No newline at end of file diff --git a/src/web/static/models.js b/src/web/static/models.js index 6b43732..80f0b4b 100644 --- a/src/web/static/models.js +++ b/src/web/static/models.js @@ -92,4 +92,166 @@ async function uploadModel() { } } -document.addEventListener('DOMContentLoaded', loadModels); \ No newline at end of file +document.addEventListener('DOMContentLoaded', loadModels); + +// ── Tone3000 Browser ──────────────────────────────────────────────── + +const INSTALLED_MODELS_KEY = 'tonedownload_installed_models'; + +function getInstalledModels() { + try { + return JSON.parse(localStorage.getItem(INSTALLED_MODELS_KEY) || '[]'); + } catch { return []; } +} + +function markInstalled(url) { + const installed = getInstalledModels(); + if (!installed.includes(url)) { + installed.push(url); + localStorage.setItem(INSTALLED_MODELS_KEY, JSON.stringify(installed)); + } +} + +function isInstalled(url) { + return getInstalledModels().includes(url); +} + +async function searchToneModels() { + const input = document.getElementById('model-tone-search'); + const query = input.value.trim(); + const resultsEl = document.getElementById('model-tone-results'); + const loaderEl = document.getElementById('model-tone-loader'); + const offlineEl = document.getElementById('model-tone-offline'); + + resultsEl.innerHTML = ''; + offlineEl.style.display = 'none'; + + if (!query) { + loadTrendingModels(); + return; + } + + loaderEl.style.display = 'block'; + + try { + const data = await apiGet('/models/tonedownload/search?q=' + encodeURIComponent(query)); + loaderEl.style.display = 'none'; + renderToneResults(resultsEl, data.results || [], 'model'); + } catch (e) { + loaderEl.style.display = 'none'; + if (e.message.includes('503') || e.message.includes('Failed to fetch')) { + offlineEl.style.display = 'block'; + } else { + resultsEl.innerHTML = `

Error: ${escapeHtml(e.message)}

`; + } + } +} + +async function loadTrendingModels() { + const resultsEl = document.getElementById('model-tone-results'); + const loaderEl = document.getElementById('model-tone-loader'); + const offlineEl = document.getElementById('model-tone-offline'); + + resultsEl.innerHTML = ''; + offlineEl.style.display = 'none'; + loaderEl.style.display = 'block'; + + try { + const data = await apiGet('/models/tonedownload/search?q='); + loaderEl.style.display = 'none'; + renderToneResults(resultsEl, data.results || [], 'model'); + } catch (e) { + loaderEl.style.display = 'none'; + if (e.message.includes('503') || e.message.includes('Failed to fetch')) { + offlineEl.style.display = 'block'; + } else { + resultsEl.innerHTML = `

Error: ${escapeHtml(e.message)}

`; + } + } +} + +function renderToneResults(container, results, type) { + if (!results || results.length === 0) { + container.innerHTML = '

No results found.

'; + return; + } + + let html = ''; + for (const r of results) { + const installed = isInstalled(r.download_url); + const thumb = r.thumbnail + ? `` + : '
🎸
'; + + html += `
+ ${thumb} +
+
${escapeHtml(r.name)}
+
+ ${escapeHtml(r.author)} · ${escapeHtml(r.size_display)} + ${r.architecture ? ` · ${escapeHtml(r.architecture)}` : ''} +
+ ${r.tone_description ? `
${escapeHtml(r.tone_description.substring(0, 120))}
` : ''} +
+
+ ${installed + ? '✓ Installed' + : `` + } +
+
`; + } + container.innerHTML = html; +} + +async function installToneModel(downloadUrl, displayName, size, filename) { + const btn = event.target; + btn.disabled = true; + btn.textContent = 'Downloading...'; + + try { + const data = await apiPost('/models/tonedownload/install', { + download_url: downloadUrl, + name: filename || displayName + '.nam', + display_name: displayName, + size: size, + }); + markInstalled(downloadUrl); + btn.textContent = '✓ Done'; + btn.style.background = 'var(--success)'; + btn.style.borderColor = 'var(--success)'; + // Refresh model list + loadModels(); + } catch (e) { + btn.disabled = false; + btn.textContent = 'Install'; + alert('Install failed: ' + e.message); + } +} + +async function installToneIR(downloadUrl, displayName, size, filename) { + const btn = event.target; + btn.disabled = true; + btn.textContent = 'Downloading...'; + + try { + const data = await apiPost('/irs/tonedownload/install', { + download_url: downloadUrl, + name: filename || displayName + '.wav', + display_name: displayName, + size: size, + }); + markInstalled(downloadUrl); + btn.textContent = '✓ Done'; + btn.style.background = 'var(--success)'; + btn.style.borderColor = 'var(--success)'; + loadIRs(); + } catch (e) { + btn.disabled = false; + btn.textContent = 'Install'; + alert('Install failed: ' + e.message); + } +} \ No newline at end of file diff --git a/src/web/static/style.css b/src/web/static/style.css index a964bef..67220be 100644 --- a/src/web/static/style.css +++ b/src/web/static/style.css @@ -994,4 +994,131 @@ body { .bt-midi-device code { font-size: 0.72rem; +} + +/* ── Tone3000 Downloader ──────────────────────────────────────────── */ + +.badge-tone3000 { + background: linear-gradient(135deg, #0000FF, #FF0000); + color: white; + font-size: 0.65rem; + vertical-align: middle; + margin-left: 6px; +} + +.tonedownload-search-row { + display: flex; + gap: 8px; + margin-bottom: 10px; + align-items: center; +} + +.tonedownload-input { + flex: 1; + min-width: 0; + padding: 8px 10px; + border: 1px solid var(--border); + border-radius: var(--radius-sm); + background: var(--bg-secondary); + color: var(--text-primary); + font-size: 0.85rem; +} + +.tonedownload-input::placeholder { + color: var(--text-muted); +} + +.tonedownload-offline { + padding: 10px; + background: rgba(255, 167, 38, 0.1); + border: 1px solid var(--warning); + border-radius: var(--radius-sm); + color: var(--warning); + font-size: 0.82rem; + margin-bottom: 10px; +} + +.tonedownload-results { + display: flex; + flex-direction: column; + gap: 8px; +} + +.tonedownload-card { + display: flex; + gap: 10px; + padding: 10px; + background: var(--bg-secondary); + border: 1px solid var(--border); + border-radius: var(--radius-sm); + align-items: center; +} + +.tonedownload-card.installed { + border-color: var(--success); + opacity: 0.8; +} + +.tonedownload-thumb { + width: 48px; + height: 48px; + border-radius: var(--radius-sm); + object-fit: cover; + flex-shrink: 0; + background: var(--bg-card); +} + +.tonedownload-thumb-placeholder { + display: flex; + align-items: center; + justify-content: center; + font-size: 1.3rem; +} + +.tonedownload-card-body { + flex: 1; + min-width: 0; +} + +.tonedownload-card-name { + font-size: 0.85rem; + font-weight: 600; + color: var(--text-primary); + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; +} + +.tonedownload-card-meta { + font-size: 0.72rem; + color: var(--text-muted); + margin-top: 2px; +} + +.tonedownload-card-desc { + font-size: 0.72rem; + color: var(--text-secondary); + margin-top: 4px; + line-height: 1.3; + display: -webkit-box; + -webkit-line-clamp: 2; + -webkit-box-orient: vertical; + overflow: hidden; +} + +.tonedownload-card-actions { + flex-shrink: 0; +} + +.tonedownload-install-btn { + padding: 4px 12px; + font-size: 0.78rem; +} + +.tonedownload-installed-badge { + display: inline-block; + color: var(--success); + font-size: 0.78rem; + font-weight: 600; + white-space: nowrap; } \ No newline at end of file diff --git a/src/web/templates/irs.html b/src/web/templates/irs.html index ca7c59a..5118765 100644 --- a/src/web/templates/irs.html +++ b/src/web/templates/irs.html @@ -42,6 +42,26 @@ + +
+
+

Browse Tone3000 BETA

+
+
+

Search and download impulse responses directly from the Tone3000 community.

+
+ + + +
+ + +
+
+
{% endblock %} {% block scripts %} diff --git a/src/web/templates/models.html b/src/web/templates/models.html index 5d9fdf6..fb65fb9 100644 --- a/src/web/templates/models.html +++ b/src/web/templates/models.html @@ -42,6 +42,26 @@ + +
+
+

Browse Tone3000 BETA

+
+
+

Search and download NAM models directly from the Tone3000 community.

+
+ + + +
+ + +
+
+
{% endblock %} {% block scripts %} diff --git a/tests/test_tonedownload.py b/tests/test_tonedownload.py new file mode 100644 index 0000000..095edf7 --- /dev/null +++ b/tests/test_tonedownload.py @@ -0,0 +1,677 @@ +"""Tests for the Tone3000 downloader client. + +Tests use mocked HTTP responses so no network access is needed. +The Web UI integration tests cover the /api/models/tonedownload/* and +/api/irs/tonedownload/* endpoints via the TestClient. + +Run with: + pytest tests/test_tonedownload.py -v +""" + +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from src.system.tonedownload import ( + Tone3000Client, + ModelResult, + IRResult, + format_size, + discover_anon_key, + CACHE_TTL, +) + +# ── Test constants ───────────────────────────────────────────────────── + +FAKE_ANON_KEY = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.test" + +SAMPLE_MODEL_ROW = { + "id": 1, + "name": "Marshall JCM800", + "model_url": "https://api.tone3000.com/storage/v1/object/public/models/test.nam", + "size": 512000, + "pack_name": "Marshall Pack", + "user_id": "u1", + "tone_id": 10, + "architecture_version": "1", + "created_at": "2025-01-01T00:00:00Z", +} + +SAMPLE_TONE_ROW = { + "id": 10, + "title": "Marshall JCM800 Sound", + "description": "Classic rock tone", + "gear": "amp", + "images": ["https://example.com/thumb.jpg"], +} + +SAMPLE_IR_TONE_ROW = { + "id": 20, + "title": "V30 Cabinet", + "description": "4x12 V30 IR", + "gear": "ir", + "images": [], +} + +SAMPLE_IR_MODEL_ROW = { + "id": 100, + "name": "v30_cabinet.wav", + "model_url": "https://api.tone3000.com/storage/v1/object/public/models/v30.wav", + "size": 256000, + "user_id": "u2", + "tone_id": 20, + "created_at": "2025-02-01T00:00:00Z", +} + +SAMPLE_USER_ROW = { + "id": "u1", + "username": "testuser", + "avatar_url": "https://example.com/avatar.png", + "display_name": "Test User", +} + + +# ── Fixtures ─────────────────────────────────────────────────────────── + + +@pytest.fixture +def mock_session(): + """Create a mock aiohttp.ClientSession that returns empty lists by default.""" + session = MagicMock() + resp = MagicMock() + resp.status = 200 + resp.json = AsyncMock(return_value=[]) + resp.text = AsyncMock(return_value="[]") + resp.__aenter__ = AsyncMock(return_value=resp) + resp.__aexit__ = AsyncMock(return_value=None) + session.get = MagicMock(return_value=resp) + session.closed = False + session.close = AsyncMock() + return session + + +@pytest.fixture +def client(mock_session): + """Tone3000Client with a mocked session and known anon key.""" + c = Tone3000Client(anon_key=FAKE_ANON_KEY, session=mock_session) + yield c + # Teardown: ensure close is called if the test doesn't + if c._session and not c._session.closed: + pass # tests manage session lifecycle + + +# ── Data class tests ─────────────────────────────────────────────────── + + +class TestModelResult: + def test_basic_attributes(self): + m = ModelResult(id=1, name="test.nam", download_url="https://example.com/test.nam") + assert m.id == 1 + assert m.filename == "test.nam" + assert m.display_name == "test.nam" + assert m.is_ir is False + assert m.thumbnail_url is None + + def test_with_tone_title(self): + m = ModelResult( + id=1, + name="test.nam", + download_url="https://example.com/test.nam", + tone_title="My Tone", + ) + assert m.display_name == "My Tone" + + def test_with_images(self): + m = ModelResult( + id=1, + name="test.nam", + download_url="https://example.com/test.nam", + tone_images=["https://example.com/thumb.jpg"], + ) + assert m.thumbnail_url == "https://example.com/thumb.jpg" + + def test_wav_is_ir(self): + m = ModelResult(id=1, name="ir.wav", download_url="https://example.com/ir.wav") + assert m.is_ir is True + + def test_ir_flag_from_gear(self): + m = ModelResult(id=1, name="test.wav", download_url="https://example.com/test.wav", tone_gear="ir") + assert m.is_ir is True + + +class TestIRResult: + def test_basic_attributes(self): + ir = IRResult(id=1, name="ir.wav", download_url="https://example.com/ir.wav") + assert ir.is_ir is True + + def test_sample_rate(self): + ir = IRResult(id=1, name="ir.wav", download_url="https://example.com/ir.wav", sample_rate=48000) + assert ir.sample_rate == 48000 + assert ir.is_ir is True + + +# ── format_size tests ────────────────────────────────────────────────── + + +class TestFormatSize: + def test_none(self): + assert format_size(None) == "Unknown" + + def test_bytes(self): + assert format_size(500) == "0.5 KB" + + def test_megabytes(self): + assert format_size(2 * 1024 * 1024) == "2.0 MB" + + def test_large_megabytes(self): + assert format_size(15 * 1024 * 1024) == "15.0 MB" + + def test_edge_kb(self): + assert format_size(1023) == "1.0 KB" + + def test_edge_mb(self): + assert format_size(1024 * 1024) == "1.0 MB" + + +# ── Client initialization ────────────────────────────────────────────── + + +class TestClientInit: + def test_default_anon_key(self): + """Should use the hardcoded constant when no key provided.""" + c = Tone3000Client() + assert c._anon_key is not None + assert len(c._anon_key) > 50 + + def test_custom_anon_key(self): + c = Tone3000Client(anon_key=FAKE_ANON_KEY) + assert c._anon_key == FAKE_ANON_KEY + + def test_custom_dirs(self): + models_dir = Path("/tmp/test-models") + irs_dir = Path("/tmp/test-irs") + c = Tone3000Client( + anon_key=FAKE_ANON_KEY, + models_dir=models_dir, + irs_dir=irs_dir, + ) + assert c._models_dir == models_dir + assert c._irs_dir == irs_dir + + +# ── Session management ───────────────────────────────────────────────── + + +class TestSession: + def test_get_session_creates_if_none(self, client): + client._session = None + assert client._session is None + assert client._anon_key == FAKE_ANON_KEY + + @pytest.mark.asyncio + async def test_close_no_session(self, client): + client._session = None + await client.close() + + @pytest.mark.asyncio + async def test_close_closed_session(self, client, mock_session): + mock_session.closed = True + await client.close() + + +# ── Caching ──────────────────────────────────────────────────────────── + + +class TestCaching: + def test_cache_miss(self, client): + assert client._get_cached("test:key") is None + + def test_cache_hit(self, client): + data = [{"id": 1, "name": "test"}] + client._set_cache("test:key", data) + result = client._get_cached("test:key") + assert result == data + + def test_cache_expiry(self, client): + data = [{"id": 1}] + client._set_cache("test:expire", data) + client._cache["test:expire"] = (0.0, data) + assert client._get_cached("test:expire") is None + + +# ── API calls (mocked) ───────────────────────────────────────────────── + + +class TestSupabaseGet: + @pytest.mark.asyncio + async def test_success(self, client): + mock_resp = MagicMock() + mock_resp.status = 200 + mock_resp.json = AsyncMock(return_value=[{"id": 1}]) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock() + client._session.get = MagicMock(return_value=mock_resp) + + result = await client._supabase_get("models", {"limit": "1"}) + assert result == [{"id": 1}] + + @pytest.mark.asyncio + async def test_error_status(self, client): + mock_resp = MagicMock() + mock_resp.status = 500 + mock_resp.text = AsyncMock(return_value="Internal error") + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock() + client._session.get = MagicMock(return_value=mock_resp) + + result = await client._supabase_get("models") + assert result == [] + + @pytest.mark.asyncio + async def test_network_error(self, client): + from aiohttp import ClientError + + client._session.get = MagicMock(side_effect=ClientError("Connection refused")) + + result = await client._supabase_get("models") + assert result == [] + + @pytest.mark.asyncio + async def test_single_object_response(self, client): + """Should wrap a single object in a list.""" + mock_resp = MagicMock() + mock_resp.status = 200 + mock_resp.json = AsyncMock(return_value={"id": 1}) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock() + client._session.get = MagicMock(return_value=mock_resp) + + result = await client._supabase_get("models", {"id": "eq.1"}) + assert result == [{"id": 1}] + + +# ── Ping ─────────────────────────────────────────────────────────────── + + +class TestPing: + @pytest.mark.asyncio + async def test_online(self, client): + mock_resp = MagicMock() + mock_resp.status = 200 + mock_resp.json = AsyncMock(return_value=[{"id": 1}]) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock() + client._session.get = MagicMock(return_value=mock_resp) + + assert await client.ping() is True + + @pytest.mark.asyncio + async def test_offline(self, client): + from aiohttp import ClientError + + client._session.get = MagicMock(side_effect=ClientError("Timeout")) + assert await client.ping() is False + + +# ── Search ───────────────────────────────────────────────────────────── + + +class TestSearchModels: + @pytest.mark.asyncio + async def test_search_returns_results(self, client): + async def mock_get(path, params=None): + if path == "models" and params and "name" in params: + return [SAMPLE_MODEL_ROW] + if path == "users": + return [SAMPLE_USER_ROW] + if path == "tones": + return [SAMPLE_TONE_ROW] + return [] + + client._supabase_get = mock_get + + results = await client.search_models("marshall", page=0) + assert len(results) == 1 + assert results[0].name == "Marshall JCM800" + assert results[0].display_name == "Marshall JCM800 Sound" + assert results[0].author_username == "testuser" + + @pytest.mark.asyncio + async def test_search_cached(self, client): + call_count = 0 + + async def mock_get(path, params=None): + nonlocal call_count + call_count += 1 + if path == "models": + return [SAMPLE_MODEL_ROW] + if path == "users": + return [SAMPLE_USER_ROW] + if path == "tones": + return [SAMPLE_TONE_ROW] + return [] + + client._supabase_get = mock_get + + await client.search_models("marshall") + first_count = call_count + + await client.search_models("marshall") + assert call_count == first_count + + @pytest.mark.asyncio + async def test_force_refresh(self, client): + call_count = 0 + + async def mock_get(path, params=None): + nonlocal call_count + call_count += 1 + if path == "models": + return [SAMPLE_MODEL_ROW] + if path == "users": + return [SAMPLE_USER_ROW] + if path == "tones": + return [SAMPLE_TONE_ROW] + return [] + + client._supabase_get = mock_get + + await client.search_models("marshall") + await client.search_models("marshall", force_refresh=True) + assert call_count >= 2 + + @pytest.mark.asyncio + async def test_empty_query(self, client): + results = await client.search_models("nonexistent_xyz123", page=0) + assert isinstance(results, list) + + +class TestSearchIRs: + @pytest.mark.asyncio + async def test_search_returns_results(self, client): + async def mock_get(path, params=None): + if path == "tones" and "ir" in str(params.get("gear", "")): + return [SAMPLE_IR_TONE_ROW] + if path == "models": + return [SAMPLE_IR_MODEL_ROW] + if path == "users": + return [{"id": "u2", "username": "irtest", "avatar_url": None, "display_name": None}] + return [] + + client._supabase_get = mock_get + + results = await client.search_irs("v30", page=0) + assert len(results) >= 1 + assert results[0].is_ir is True + assert results[0].tone_title == "V30 Cabinet" + + @pytest.mark.asyncio + async def test_empty_results(self, client): + async def mock_get(path, params=None): + if path == "tones": + return [] + return [] + + client._supabase_get = mock_get + + results = await client.search_irs("nonexistent_ir", page=0) + assert results == [] + + +# ── Trending ─────────────────────────────────────────────────────────── + + +class TestTrending: + @pytest.mark.asyncio + async def test_trending_models(self, client): + async def mock_get(path, params=None): + if path == "models": + return [SAMPLE_MODEL_ROW] + if path == "users": + return [SAMPLE_USER_ROW] + if path == "tones": + return [SAMPLE_TONE_ROW] + return [] + + client._supabase_get = mock_get + results = await client.get_trending_models(limit=3) + assert len(results) == 1 + + @pytest.mark.asyncio + async def test_trending_irs(self, client): + async def mock_get(path, params=None): + if path == "tones": + return [SAMPLE_IR_TONE_ROW] + if path == "models": + return [SAMPLE_IR_MODEL_ROW] + if path == "users": + return [{"id": "u2", "username": "irtest", "avatar_url": None}] + return [] + + client._supabase_get = mock_get + results = await client.get_trending_irs(limit=3) + assert len(results) == 1 + + +# ── Download ─────────────────────────────────────────────────────────── + + +class TestDownload: + @pytest.mark.asyncio + async def test_download_model_success(self, client, tmp_path): + client._models_dir = tmp_path + + async def _chunks(): + yield b"testdata" + + mock_resp = MagicMock() + mock_resp.status = 200 + mock_resp.headers = {"Content-Length": "9"} + mock_resp.content = MagicMock() + mock_resp.content.iter_chunked = MagicMock(side_effect=lambda cs: _chunks()) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock() + client._session.get = MagicMock(return_value=mock_resp) + + model = ModelResult(id=1, name="test_model", download_url="https://example.com/test.nam") + path = await client.download_model(model) + assert path is not None + assert path.exists() + assert path.suffix == ".nam" + assert path.read_bytes() == b"testdata" + + @pytest.mark.asyncio + async def test_download_model_no_url(self, client): + model = ModelResult(id=1, name="test", download_url="") + path = await client.download_model(model) + assert path is None + + @pytest.mark.asyncio + async def test_download_model_http_error(self, client, tmp_path): + client._models_dir = tmp_path + mock_resp = MagicMock() + mock_resp.status = 404 + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock() + client._session.get = MagicMock(return_value=mock_resp) + + model = ModelResult(id=1, name="test", download_url="https://example.com/test.nam") + path = await client.download_model(model) + assert path is None + + @pytest.mark.asyncio + async def test_download_ir_success(self, client, tmp_path): + client._irs_dir = tmp_path + + async def _chunks(): + yield b"wavdata" + + mock_resp = MagicMock() + mock_resp.status = 200 + mock_resp.headers = {"Content-Length": "9"} + mock_resp.content = MagicMock() + mock_resp.content.iter_chunked = MagicMock(side_effect=lambda cs: _chunks()) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock() + client._session.get = MagicMock(return_value=mock_resp) + + ir = IRResult(id=1, name="cabinet", download_url="https://example.com/cabinet.wav") + path = await client.download_ir(ir) + assert path is not None + assert path.exists() + assert path.suffix == ".wav" + assert path.read_bytes() == b"wavdata" + + @pytest.mark.asyncio + async def test_download_progress(self, client, tmp_path): + client._models_dir = tmp_path + + async def _chunks(): + yield b"data1" + yield b"data2" + + mock_resp = MagicMock() + mock_resp.status = 200 + mock_resp.headers = {"Content-Length": "12"} + mock_resp.content = MagicMock() + mock_resp.content.iter_chunked = MagicMock(side_effect=lambda cs: _chunks()) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock() + client._session.get = MagicMock(return_value=mock_resp) + + progress_values = [] + + def cb(ratio): + progress_values.append(ratio) + + model = ModelResult(id=1, name="test", download_url="https://example.com/test.nam") + path = await client.download_model(model, progress_callback=cb) + assert path is not None + assert len(progress_values) > 0 + + +# ── User/Tone enrichment ────────────────────────────────────────────── + + +class TestEnrichment: + @pytest.mark.asyncio + async def test_get_user_sync(self, client): + """User lookup works (sync test via mock).""" + from src.system.tonedownload import Tone3000Client + + c = Tone3000Client(anon_key=FAKE_ANON_KEY) + result = await c._get_user("") + assert result is None + await c.close() + + @pytest.mark.asyncio + async def test_get_tone_sync(self, client): + from src.system.tonedownload import Tone3000Client + + c = Tone3000Client(anon_key=FAKE_ANON_KEY) + result = await c._get_tone(9999) + assert result is None + await c.close() + + @pytest.mark.asyncio + async def test_get_user_cached(self, client): + async def mock_get(path, params=None): + return [SAMPLE_USER_ROW] + + client._supabase_get = mock_get + user = await client._get_user("u1") + assert user["username"] == "testuser" + # Second call — cached + user2 = await client._get_user("u1") + assert user2["username"] == "testuser" + + @pytest.mark.asyncio + async def test_get_user_empty_id(self, client): + user = await client._get_user("") + assert user is None + + @pytest.mark.asyncio + async def test_get_tone_cached(self, client): + async def mock_get(path, params=None): + return [SAMPLE_TONE_ROW] + + client._supabase_get = mock_get + tone = await client._get_tone(10) + assert tone["title"] == "Marshall JCM800 Sound" + # Second call — cached + tone2 = await client._get_tone(10) + assert tone2["title"] == "Marshall JCM800 Sound" + + +# ── Anon key discovery ──────────────────────────────────────────────── + + +class TestAnonKeyDiscovery: + @pytest.mark.asyncio + async def test_discover_timeout(self, client): + mock_resp = MagicMock() + mock_resp.__aenter__ = AsyncMock(side_effect=TimeoutError("Timed out")) + mock_resp.__aexit__ = AsyncMock() + client._session.get = MagicMock(return_value=mock_resp) + + key = await discover_anon_key(client._session) + assert key is None + + +# ── Integration via real HTTP (optional) ─────────────────────────────── + + +@pytest.mark.skipif( + True, + reason="Real HTTP test — enable manually to verify against live Tone3000 API", +) +class TestRealHTTP: + """Tests that hit the real Tone3000 API.""" + + @pytest.mark.asyncio + async def test_ping(self): + c = Tone3000Client() + ok = await c.ping() + assert ok is True + await c.close() + + @pytest.mark.asyncio + async def test_search_models(self): + c = Tone3000Client() + results = await c.search_models("marshall", page=0, page_size=5) + assert len(results) > 0 + for r in results: + assert r.display_name + assert r.download_url + await c.close() + + @pytest.mark.asyncio + async def test_search_irs(self): + c = Tone3000Client() + results = await c.search_irs("v30", page=0, page_size=5) + assert len(results) > 0 + for r in results: + assert r.is_ir is True + await c.close() + + @pytest.mark.asyncio + async def test_trending_models(self): + c = Tone3000Client() + results = await c.get_trending_models(limit=5) + assert len(results) > 0 + await c.close() + + @pytest.mark.asyncio + async def test_download(self, tmp_path): + c = Tone3000Client(models_dir=tmp_path, irs_dir=tmp_path) + results = await c.search_models("marshall", page=0, page_size=1) + if results: + path = await c.download_model(results[0]) + assert path is not None + assert path.exists() + assert path.stat().st_size > 1000 + await c.close() \ No newline at end of file