shawn
3e55ea8a9d
Replace spaces with PostgREST wildcards in ilike filter values so
multi-word queries like 'vox ac30' produce valid URLs.
Root cause: PostgREST filter parser doesn't handle spaces URL-encoded
as '+' in query parameter values, returning 400 Bad Request.
Fix: query.replace(' ', '*') converts spaces to PostgREST wildcards
before inserting into ilike.* pattern. 'vox ac30' becomes
ilike.*vox*ac30* -> ILIKE '%vox%ac30%'.
Changes:
- src/system/tonedownload.py: fix search_models() and search_irs()
- tests/test_tonedownload.py: update test_multi_word_query assertion
717 lines
24 KiB
Python
717 lines
24 KiB
Python
"""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() |