Something went wrong. Try again.
A local, collaborative music player (main-branch-only mirror of forge.ejuarezg.com/ejuarezg/jam-session)
Something went wrong. Try again.
6.2 kB · 166 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167import jsonfrom pathlib import Pathfrom unittest.mock import MagicMock, patch
import pytest
from jam_session.cache import AudioCacheManagerfrom jam_session.models import Track
@pytest.fixturedef tmp_cache_dir(tmp_path): return tmp_path / "cache"
@pytest.fixturedef cache_mgr(tmp_cache_dir): return AudioCacheManager(max_size_gb=0.01, cache_dir=tmp_cache_dir, prefetch_count=5)
class TestLoadIndex: def test_load_existing_index(self, tmp_cache_dir): tmp_cache_dir.mkdir(parents=True, exist_ok=True) idx_path = tmp_cache_dir / "cache_index.json" idx_path.write_text(json.dumps({"vid1": {"title": "Test"}}))
mgr = AudioCacheManager(cache_dir=tmp_cache_dir) assert mgr._index == {"vid1": {"title": "Test"}}
def test_load_missing_returns_empty(self, tmp_cache_dir): mgr = AudioCacheManager(cache_dir=tmp_cache_dir) assert mgr._index == {}
def test_load_corrupt_json_returns_empty(self, tmp_cache_dir): tmp_cache_dir.mkdir(parents=True, exist_ok=True) idx_path = tmp_cache_dir / "cache_index.json" idx_path.write_text("{corrupt")
mgr = AudioCacheManager(cache_dir=tmp_cache_dir) assert mgr._index == {}
class TestGetCached: def test_hit_returns_path(self, cache_mgr): file_path = cache_mgr.cache_dir / "test.opus" file_path.write_text("audio data") cache_mgr._index["test"] = {"file_path": str(file_path), "last_accessed": "old", "file_size": 100}
result = cache_mgr.get_cached("test") assert result == file_path assert cache_mgr._index["test"]["last_accessed"] != "old"
def test_miss_returns_none(self, cache_mgr): result = cache_mgr.get_cached("nope") assert result is None
def test_file_gone_removes_from_index(self, cache_mgr): cache_mgr._index["gone"] = {"file_path": str(cache_mgr.cache_dir / "gone.opus"), "file_size": 100}
result = cache_mgr.get_cached("gone") assert result is None assert "gone" not in cache_mgr._index
class TestAddToCache: def test_creates_opus_entry(self, cache_mgr): cache_mgr.cache_dir.mkdir(parents=True, exist_ok=True) opus_path = cache_mgr.cache_dir / "abc123.opus" opus_path.write_text("dummy opus data")
import yt_dlp
with patch.object(yt_dlp.YoutubeDL, 'download') as mock_dl: result = cache_mgr.add_to_cache("abc123", "https://youtube.com/watch?v=abc123", "Test", 180) mock_dl.assert_called_once() assert result == opus_path assert "abc123" in cache_mgr._index assert cache_mgr._index["abc123"]["title"] == "Test" assert cache_mgr._index["abc123"]["file_size"] == opus_path.stat().st_size
def test_evicts_after_adding(self, cache_mgr): cache_mgr.cache_dir.mkdir(parents=True, exist_ok=True)
for i in range(20): path = cache_mgr.cache_dir / f"vid{i}.opus" path.write_text("x" * 500000) cache_mgr._index[f"vid{i}"] = { "file_path": str(path), "file_size": 500000, "last_accessed": f"2020-01-{i+1:02d}T00:00:00", }
opus_path = cache_mgr.cache_dir / "new.opus" opus_path.write_text("x" * 200000)
import yt_dlp with patch.object(yt_dlp.YoutubeDL, 'download'): result = cache_mgr.add_to_cache("new", "https://youtube.com/watch?v=new", "New", 60)
total_after = sum(e["file_size"] for e in cache_mgr._index.values()) assert total_after <= cache_mgr.max_size_bytes
class TestLRUEviction: def test_noop_when_under_limit(self, cache_mgr): cache_mgr.cache_dir.mkdir(parents=True, exist_ok=True) path = cache_mgr.cache_dir / "small.opus" path.write_text("x" * 100) cache_mgr._index["small"] = {"file_path": str(path), "file_size": 100}
cache_mgr._evict_lru() assert "small" in cache_mgr._index
def test_removes_oldest_when_over_limit(self, cache_mgr): cache_mgr.cache_dir.mkdir(parents=True, exist_ok=True) path1 = cache_mgr.cache_dir / "old.opus" path2 = cache_mgr.cache_dir / "new.opus" path1.write_text("x" * 8000000) path2.write_text("x" * 5000000)
cache_mgr._index["old"] = {"file_path": str(path1), "file_size": 8000000, "last_accessed": "2020-01-01T00:00:00"} cache_mgr._index["new"] = {"file_path": str(path2), "file_size": 5000000, "last_accessed": "2025-01-01T00:00:00"}
cache_mgr._evict_lru() assert "old" not in cache_mgr._index assert "new" in cache_mgr._index
class TestPreFetch: @pytest.mark.asyncio async def test_pre_fetch_skips_already_cached(self, cache_mgr): cache_mgr._index["cached-id"] = {"file_path": "/tmp/x", "file_size": 100}
track = Track( title="Test", duration=60, stream_url="https://x", source_id="cached-id", webpage_url="https://youtube.com/watch?v=cached-id" )
with patch.object(cache_mgr, 'add_to_cache') as mock_add: await cache_mgr.pre_fetch_upcoming([track]) mock_add.assert_not_called()
@pytest.mark.asyncio async def test_pre_fetch_adds_uncached(self, cache_mgr): cache_mgr.cache_dir.mkdir(parents=True, exist_ok=True) (cache_mgr.cache_dir / "uncached.opus").write_text("audio")
track = Track( title="New", duration=60, stream_url="https://x", source_id="uncached", webpage_url="https://youtube.com/watch?v=uncached" )
import yt_dlp with patch.object(yt_dlp.YoutubeDL, 'download'): await cache_mgr.pre_fetch_upcoming([track]) assert "uncached" in cache_mgr._index
class TestSizeBytes: def test_returns_zero_for_empty(self, cache_mgr): assert cache_mgr.size_bytes == 0
def test_sums_all_entries(self, cache_mgr): cache_mgr._index["a"] = {"file_size": 100, "file_path": "/tmp/a"} cache_mgr._index["b"] = {"file_size": 200, "file_path": "/tmp/b"} assert cache_mgr.size_bytes == 300