"""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" def test_string_input(self): """format_size should handle string inputs (Supabase can return string).""" assert format_size("500") == "0.5 KB" assert format_size("2097152") == "2.0 MB" assert format_size("1024") == "1.0 KB" def test_invalid_string(self): """format_size should handle unparseable string inputs.""" assert format_size("not-a-number") == "Unknown" assert format_size("") == "Unknown" # ── 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) @pytest.mark.asyncio async def test_multi_word_query(self, client): """Multi-word queries should convert spaces to wildcards.""" captured_params = [] async def mock_get(path, params=None): if path == "models" and params and "name" in params: captured_params.append(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("vox ac30", page=0) assert len(results) == 1 assert results[0].name == "Marshall JCM800" # Verify the Supabase param has spaces replaced with wildcards assert len(captured_params) >= 1 name_param = captured_params[0].get("name", "") assert "ilike.*vox*ac30*" in name_param # Should NOT contain the raw space (which PostgREST filter parser # can't handle in URL query values) assert "vox ac30" not in name_param 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()