/
/
/
1"""Tests for cache controller oversized cache detection and reset."""
2
3import os
4from collections.abc import Callable
5from typing import Any
6from unittest.mock import AsyncMock, patch
7
8import aiofiles
9import pytest
10
11from music_assistant.controllers.cache import MAX_CACHE_DB_SIZE_MB
12from music_assistant.mass import MusicAssistant
13
14
15async def _create_db_files(cache_path: str) -> list[str]:
16 """Create small cache.db, cache.db-wal, and cache.db-shm files.
17
18 :param cache_path: Path to the cache directory.
19 """
20 db_path = os.path.join(cache_path, "cache.db")
21 paths = [db_path + suffix for suffix in ("", "-wal", "-shm")]
22 for path in paths:
23 async with aiofiles.open(path, "wb") as f:
24 await f.write(b"\0")
25 return paths
26
27
28async def test_cache_reset_when_exceeding_limit(mass_minimal: MusicAssistant) -> None:
29 """Test that the cache database is removed when it exceeds MAX_CACHE_DB_SIZE_MB.
30
31 :param mass_minimal: Minimal MusicAssistant instance.
32 """
33 cache = mass_minimal.cache
34 db_files = await _create_db_files(mass_minimal.cache_path)
35
36 with patch("asyncio.to_thread", new_callable=AsyncMock) as mock_to_thread:
37
38 async def _side_effect(func: Callable[..., Any], *args: Any) -> Any:
39 if getattr(func, "__name__", "") == "_get_db_size":
40 return float(MAX_CACHE_DB_SIZE_MB + 100)
41 return func(*args)
42
43 mock_to_thread.side_effect = _side_effect
44 result = await cache._check_and_reset_oversized_cache()
45
46 assert result is True
47 for path in db_files:
48 assert not os.path.exists(path)
49
50
51async def test_cache_not_reset_when_under_limit(mass_minimal: MusicAssistant) -> None:
52 """Test that the cache database is kept when it is under MAX_CACHE_DB_SIZE_MB.
53
54 :param mass_minimal: Minimal MusicAssistant instance.
55 """
56 cache = mass_minimal.cache
57 db_files = await _create_db_files(mass_minimal.cache_path)
58
59 with patch("asyncio.to_thread", new_callable=AsyncMock) as mock_to_thread:
60
61 async def _side_effect(func: Callable[..., Any], *args: Any) -> Any:
62 if getattr(func, "__name__", "") == "_get_db_size":
63 return 1.0
64 return func(*args)
65
66 mock_to_thread.side_effect = _side_effect
67 result = await cache._check_and_reset_oversized_cache()
68
69 assert result is False
70 for path in db_files:
71 assert os.path.exists(path)
72
73
74async def test_all_three_db_files_included_in_size(mass_minimal: MusicAssistant) -> None:
75 """Test that cache.db, cache.db-wal, and cache.db-shm are all summed for size check.
76
77 :param mass_minimal: Minimal MusicAssistant instance.
78 """
79 cache = mass_minimal.cache
80 db_path = os.path.join(mass_minimal.cache_path, "cache.db")
81
82 # Create 3 files of 100 bytes each (300 bytes total)
83 for suffix in ("", "-wal", "-shm"):
84 async with aiofiles.open(db_path + suffix, "wb") as f:
85 await f.write(b"\0" * 100)
86
87 # Set threshold to ~200 bytes so 2 files pass but 3 files exceed it
88 size_threshold_mb = 0.0002
89 with patch("music_assistant.controllers.cache.MAX_CACHE_DB_SIZE_MB", size_threshold_mb):
90 result = await cache._check_and_reset_oversized_cache()
91
92 # 300 bytes exceeds the ~200 byte threshold, proving all 3 files are summed
93 assert result is True
94 assert not os.path.exists(db_path)
95 assert not os.path.exists(db_path + "-wal")
96 assert not os.path.exists(db_path + "-shm")
97
98
99async def test_skip_migration_when_cache_reset(
100 mass_minimal: MusicAssistant,
101 caplog: pytest.LogCaptureFixture,
102) -> None:
103 """Test that database migration is skipped when the cache was reset.
104
105 :param mass_minimal: Minimal MusicAssistant instance.
106 :param caplog: Log capture fixture.
107 """
108 cache = mass_minimal.cache
109
110 with patch.object(cache, "_check_and_reset_oversized_cache", return_value=True):
111 await cache._setup_database()
112
113 assert "Performing database migration" not in caplog.text
114
115
116async def test_skip_vacuum_when_cache_reset(
117 mass_minimal: MusicAssistant,
118 caplog: pytest.LogCaptureFixture,
119) -> None:
120 """Test that database vacuum is skipped when the cache was reset.
121
122 :param mass_minimal: Minimal MusicAssistant instance.
123 :param caplog: Log capture fixture.
124 """
125 cache = mass_minimal.cache
126
127 with patch.object(cache, "_check_and_reset_oversized_cache", return_value=True):
128 await cache._setup_database()
129
130 assert "Compacting database" not in caplog.text
131