From 22e8b03e6e2dd9822ac73ce1cabaa9f3c24c770b Mon Sep 17 00:00:00 2001 From: Uyanide Date: Wed, 1 Apr 2026 20:38:30 +0200 Subject: [PATCH] test: add test for cache db --- tests/test_cache.py | 347 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 347 insertions(+) create mode 100644 tests/test_cache.py diff --git a/tests/test_cache.py b/tests/test_cache.py new file mode 100644 index 0000000..7a4047e --- /dev/null +++ b/tests/test_cache.py @@ -0,0 +1,347 @@ +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from lrx_cli.cache import ( + CacheEngine, + _generate_key, + _normalize_artist, + _normalize_for_match, +) +from lrx_cli.config import DURATION_TOLERANCE_MS +from lrx_cli.models import CacheStatus, LyricResult, TrackMeta + + +def _track( + *, + artist: str | None = "Artist", + title: str | None = "Song", + album: str | None = "Album", + length: int | None = 180000, + trackid: str | None = None, + url: str | None = None, +) -> TrackMeta: + return TrackMeta( + artist=artist, + title=title, + album=album, + length=length, + trackid=trackid, + url=url, + ) + + +def _result( + status: CacheStatus, + lyrics: str | None, + source: str, +) -> LyricResult: + return LyricResult(status=status, lyrics=lyrics, source=source) + + +@pytest.fixture +def cache_db(tmp_path: Path) -> CacheEngine: + db_path = tmp_path / "cache.db" + return CacheEngine(str(db_path)) + + +def test_normalize_for_match_covers_nfkc_punct_feat_and_whitespace() -> None: + text = " Ｔｅｓｔ！ feat. SOMEONE " + + normalized = _normalize_for_match(text) + + assert normalized == "test" + + +def test_normalize_artist_splits_separators_and_sorts_parts() -> None: + artist = "B / A feat. C; D vs. E × F 、 G" + + normalized = _normalize_artist(artist) + + assert normalized == "a\0b\0d\0e\0f\0g" + + +def test_generate_key_uses_spotify_trackid_and_url_fallback() -> None: + spotify_track = _track( + trackid="abc123", artist=None, title=None, album=None, length=None + ) + local_track = _track( + artist=None, title=None, album=None, length=None, url="file:///x.flac" + ) + + assert _generate_key(spotify_track, "spotify") == "spotify:abc123" + assert _generate_key(local_track, "local") == "local:url:file:///x.flac" + + +def test_generate_key_raises_when_metadata_missing() -> None: + with pytest.raises(ValueError): + _generate_key( + _track(artist=None, title=None, album=None, length=None, url=None), "lrclib" + ) + + +def test_set_and_get_roundtrip_with_ttl( + monkeypatch: pytest.MonkeyPatch, cache_db: CacheEngine +) -> None: + monkeypatch.setattr("lrx_cli.cache.time.time", lambda: 1_000_000) + + track = _track() + cache_db.set( + track, + "lrclib", + _result(CacheStatus.SUCCESS_SYNCED, "[00:01.00]line", "lrclib"), + ttl_seconds=120, + ) + + cached = cache_db.get(track, "lrclib") + + assert cached is not None + assert cached.status is CacheStatus.SUCCESS_SYNCED + assert cached.lyrics == "[00:01.00]line" + assert cached.source == "lrclib" + assert cached.ttl == 120 + + +def test_get_expired_entry_returns_none_and_removes_row( + monkeypatch: pytest.MonkeyPatch, cache_db: CacheEngine +) -> None: + track = _track() + monkeypatch.setattr("lrx_cli.cache.time.time", lambda: 2_000_000) + cache_db.set( + track, + "netease", + _result(CacheStatus.SUCCESS_UNSYNCED, "line", "netease"), + ttl_seconds=10, + ) + + monkeypatch.setattr("lrx_cli.cache.time.time", lambda: 2_000_020) + cached = cache_db.get(track, "netease") + + assert cached is None + assert cache_db.query_all() == [] + + +def test_get_backfills_missing_length_when_track_provides_it( + cache_db: CacheEngine, +) -> None: + track_without_length = _track( + trackid="spotify-track-1", + artist=None, + title=None, + album=None, + length=None, + ) + cache_db.set( + track_without_length, + "spotify", + _result(CacheStatus.SUCCESS_SYNCED, "line", "spotify"), + ) + + track_with_length = _track( + trackid="spotify-track-1", + artist=None, + title=None, + album=None, + length=200000, + ) + cached = cache_db.get(track_with_length, "spotify") + + assert cached is not None + + with sqlite3.connect(cache_db.db_path) as conn: + row = conn.execute("SELECT length FROM cache LIMIT 1").fetchone() + assert row is not None + assert row[0] == 200000 + + +def test_get_best_prefers_synced_over_unsynced_and_negative( + cache_db: CacheEngine, +) -> None: + track = _track() + cache_db.set( + track, + "source-a", + _result(CacheStatus.NOT_FOUND, None, "source-a"), + ) + cache_db.set( + track, + "source-b", + _result(CacheStatus.SUCCESS_UNSYNCED, "unsynced", "source-b"), + ) + cache_db.set( + track, + "source-c", + _result(CacheStatus.SUCCESS_SYNCED, "synced", "source-c"), + ) + + best = cache_db.get_best(track, ["source-a", "source-b", "source-c"]) + + assert best is not None + assert best.status is CacheStatus.SUCCESS_SYNCED + assert best.lyrics == "synced" + + +def test_clear_track_and_clear_all_affect_expected_rows(cache_db: CacheEngine) -> None: + track_a = _track(artist="A", title="T", album="X") + track_b = _track(artist="B", title="T", album="X") + + cache_db.set(track_a, "s1", _result(CacheStatus.SUCCESS_SYNCED, "a1", "s1")) + cache_db.set(track_a, "s2", _result(CacheStatus.SUCCESS_UNSYNCED, "a2", "s2")) + cache_db.set(track_b, "s1", _result(CacheStatus.SUCCESS_SYNCED, "b1", "s1")) + + cache_db.clear_track(track_a) + rows_after_track_clear = cache_db.query_all() + assert len(rows_after_track_clear) == 1 + assert rows_after_track_clear[0]["artist"] == "B" + + cache_db.clear_all() + assert cache_db.query_all() == [] + + +def test_prune_removes_only_expired_rows( + monkeypatch: pytest.MonkeyPatch, cache_db: CacheEngine +) -> None: + track = _track() + monkeypatch.setattr("lrx_cli.cache.time.time", lambda: 3_000_000) + cache_db.set( + track, + "s-expired", + _result(CacheStatus.SUCCESS_SYNCED, "x", "s-expired"), + ttl_seconds=1, + ) + cache_db.set( + track, + "s-active", + _result(CacheStatus.SUCCESS_SYNCED, "y", "s-active"), + ttl_seconds=100, + ) + + monkeypatch.setattr("lrx_cli.cache.time.time", lambda: 3_000_010) + deleted = cache_db.prune() + + assert deleted == 1 + rows = cache_db.query_all() + assert len(rows) == 1 + assert rows[0]["source"] == "s-active" + + +def test_find_best_positive_uses_exact_match_and_prefers_synced( + cache_db: CacheEngine, +) -> None: + track = _track(artist="Artist", title="Song", album="Album") + cache_db.set(track, "s1", _result(CacheStatus.SUCCESS_UNSYNCED, "u", "s1")) + cache_db.set(track, "s2", _result(CacheStatus.SUCCESS_SYNCED, "s", "s2")) + + best = cache_db.find_best_positive(track) + + assert best is not None + assert best.status is CacheStatus.SUCCESS_SYNCED + assert best.lyrics == "s" + # find_best_positive always reports cache-search source + assert best.source == "cache-search" + + +def test_search_by_meta_fuzzy_rules_and_duration_sorting(cache_db: CacheEngine) -> None: + # Same logical title/artist after normalization, different length quality. + base = _track( + artist="A / B", + title="Hello，World!", + album="Album", + length=200000, + ) + close_synced = _track( + artist="B vs. A", + title="hello world", + album="Else", + length=200500, + ) + close_unsynced = _track( + artist="A feat. C / B", + title="HELLO WORLD", + album="Else2", + length=201000, + ) + unknown_len = _track( + artist="A & B", + title="Hello World", + album="Else3", + length=None, + ) + far_len = _track( + artist="A / B", + title="Hello World", + album="Else4", + length=200000 + DURATION_TOLERANCE_MS + 1, + ) + + cache_db.set(base, "seed", _result(CacheStatus.SUCCESS_SYNCED, "seed", "seed")) + cache_db.set( + close_synced, + "close-synced", + _result(CacheStatus.SUCCESS_SYNCED, "cs", "close-synced"), + ) + cache_db.set( + close_unsynced, + "close-unsynced", + _result(CacheStatus.SUCCESS_UNSYNCED, "cu", "close-unsynced"), + ) + cache_db.set( + unknown_len, + "unknown-len", + _result(CacheStatus.SUCCESS_SYNCED, "ul", "unknown-len"), + ) + cache_db.set( + far_len, + "far-len", + _result(CacheStatus.SUCCESS_SYNCED, "fl", "far-len"), + ) + # Negative status should never appear in search results. + cache_db.set( + _track(artist="A / B", title="Hello World", album="Else5", length=200000), + "negative", + _result(CacheStatus.NOT_FOUND, None, "negative"), + ) + + rows = cache_db.search_by_meta( + artist="Ｂ ; A", + title=" hello world ", + length=200000, + ) + + sources = [r["source"] for r in rows] + assert "negative" not in sources + assert "far-len" not in sources + # Sorted by duration diff, then synced before unsynced for equal diff. + assert sources[0] == "seed" + assert sources[1] == "close-synced" + assert sources[2] == "close-unsynced" + # Unknown length remains candidate with fallback distance priority. + assert sources[-1] == "unknown-len" + + +def test_query_track_and_stats_return_expected_aggregates( + cache_db: CacheEngine, +) -> None: + cache_db.set( + _track(artist="A", title="T", album="AL"), + "s1", + _result(CacheStatus.SUCCESS_SYNCED, "x", "s1"), + ) + cache_db.set( + _track(artist="A", title="T", album="AL"), + "s2", + _result(CacheStatus.SUCCESS_UNSYNCED, "y", "s2"), + ) + + rows = cache_db.query_track(_track(artist="A", title="T", album="AL")) + stats = cache_db.stats() + + assert len(rows) == 2 + assert stats["total"] == 2 + assert stats["active"] == 2 + assert stats["expired"] == 0 + assert stats["by_status"][CacheStatus.SUCCESS_SYNCED.value] == 1 + assert stats["by_status"][CacheStatus.SUCCESS_UNSYNCED.value] == 1