147 lines
4.6 KiB
Python
147 lines
4.6 KiB
Python
"""
|
|
Basic cache layer tests to ensure CRUD operations and sync state behave.
|
|
|
|
These run against a temp SQLite file and a tiny fake client—no network or Qt.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
# Ensure repo root on path for `src` imports when running via python -m pytest
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.cache import CacheDatabase, CacheManager, CachedClient
|
|
from src.api.models import Artist, Album, Song, Playlist
|
|
|
|
|
|
class FakeClient:
|
|
"""Minimal fake SubsonicClient for incremental sync tests."""
|
|
|
|
def __init__(self):
|
|
self.serverUrl = "https://fake.example"
|
|
self._artists = [
|
|
Artist(id="a1", name="Artist One", albumCount=1),
|
|
Artist(id="a2", name="Artist Two", albumCount=0),
|
|
]
|
|
self._albums = {
|
|
"a1": [Album(id="al1", name="Album One", artist="Artist One", artistId="a1")],
|
|
}
|
|
self._songs = {
|
|
"al1": [
|
|
Song(
|
|
id="s1",
|
|
title="Song One",
|
|
album="Album One",
|
|
albumId="al1",
|
|
artist="Artist One",
|
|
artistId="a1",
|
|
)
|
|
]
|
|
}
|
|
self._playlists = [
|
|
Playlist(
|
|
id="p1",
|
|
name="PL1",
|
|
songCount=1,
|
|
songs=self._songs["al1"],
|
|
changed=datetime.now() - timedelta(hours=1),
|
|
)
|
|
]
|
|
|
|
# Methods used by CachedClient.syncIncremental
|
|
def getArtists(self, musicFolderId=None):
|
|
return self._artists
|
|
|
|
def getArtist(self, artistId):
|
|
return {"artist": next(a for a in self._artists if a.id == artistId), "albums": self._albums.get(artistId, [])}
|
|
|
|
def getAlbum(self, albumId):
|
|
songs = self._songs.get(albumId, [])
|
|
album = Album(id=albumId, name=f"Album {albumId}", artist="Artist One", artistId="a1")
|
|
return {"album": album, "songs": songs}
|
|
|
|
def getPlaylists(self):
|
|
return self._playlists
|
|
|
|
def getPlaylist(self, playlistId):
|
|
return next(p for p in self._playlists if p.id == playlistId)
|
|
|
|
def getGenres(self):
|
|
return []
|
|
|
|
def getStarred(self):
|
|
return {"artists": [], "albums": [], "songs": []}
|
|
|
|
# Passthroughs not used here
|
|
def ping(self):
|
|
return True
|
|
|
|
def clearCache(self):
|
|
pass
|
|
|
|
|
|
@pytest.fixture()
|
|
def temp_cache(tmp_path: Path):
|
|
data_dir = tmp_path / "data"
|
|
data_dir.mkdir()
|
|
db = CacheDatabase("https://fake.example", data_dir)
|
|
cache = CacheManager(db)
|
|
yield cache
|
|
db.close()
|
|
|
|
|
|
def test_artist_album_song_round_trip(temp_cache: CacheManager):
|
|
artist = Artist(id="a1", name="Artist One", albumCount=1)
|
|
album = Album(id="al1", name="Album One", artist="Artist One", artistId="a1")
|
|
song = Song(id="s1", title="Song One", album="Album One", albumId="al1", artist="Artist One", artistId="a1")
|
|
|
|
temp_cache.setArtist(artist)
|
|
temp_cache.setAlbum(album)
|
|
temp_cache.setSong(song)
|
|
|
|
assert temp_cache.getArtist("a1") == artist
|
|
assert temp_cache.getAlbum("al1") == album
|
|
assert temp_cache.getSong("s1") == song
|
|
|
|
|
|
def test_playlist_order_preserved(temp_cache: CacheManager):
|
|
songs = [
|
|
Song(id="s1", title="First", album="a", albumId="al", artist="x"),
|
|
Song(id="s2", title="Second", album="a", albumId="al", artist="x"),
|
|
Song(id="s3", title="Third", album="a", albumId="al", artist="x"),
|
|
]
|
|
playlist = Playlist(id="p1", name="Order Test", songCount=3, songs=songs)
|
|
|
|
temp_cache.setPlaylist(playlist)
|
|
stored = temp_cache.getPlaylist("p1")
|
|
assert stored is not None
|
|
assert [s.id for s in stored.songs] == ["s1", "s2", "s3"]
|
|
|
|
|
|
def test_sync_state_stores_timestamps(tmp_path: Path):
|
|
db = CacheDatabase("https://fake.example", tmp_path)
|
|
cache = CacheManager(db)
|
|
before = datetime.now()
|
|
cache.setSyncState("incremental", 5)
|
|
recorded = cache.getSyncState("incremental")
|
|
assert recorded is not None
|
|
assert recorded["itemCount"] == 5
|
|
assert recorded["lastSync"] >= before
|
|
db.close()
|
|
|
|
|
|
def test_cached_client_exposes_last_sync(tmp_path: Path):
|
|
client = CachedClient(FakeClient(), tmp_path)
|
|
# First incremental run should set sync state
|
|
stats = client.syncIncremental()
|
|
assert stats["artists"] == 2
|
|
assert client.getLastSyncTime("incremental") is not None
|
|
# Future check should consider recency
|
|
client.setSyncState("incremental", 0)
|
|
assert client.getLastSyncTime("incremental") is not None
|