music-assistant-server

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