music-assistant-server

1"""Tests for the DatabaseConnection helper."""
2
3import asyncio
4import os
5import pathlib
6from collections.abc import AsyncGenerator
7from sqlite3 import OperationalError
8from typing import Any
9
10import pytest
11
12from music_assistant.helpers.database import (
13    DatabaseConnection,
14    get_sqlite_memory_settings,
15    query_params,
16)
17from music_assistant.mass import MusicAssistant
18
19GIB = 1024**3
20
21# PRAGMA temp_store integer values (sqlite docs)
22TEMP_STORE_FILE = 1
23TEMP_STORE_MEMORY = 2
24
25
26@pytest.fixture
27async def db_connection(tmp_path: pathlib.Path) -> AsyncGenerator[DatabaseConnection]:
28    """Return an initialized DatabaseConnection backed by a temp file."""
29    db = DatabaseConnection(str(tmp_path / "test.db"))
30    await db.setup()
31    yield db
32    await db.close()
33
34
35@pytest.fixture
36async def db_with_table(db_connection: DatabaseConnection) -> DatabaseConnection:
37    """Return an initialized DatabaseConnection with a simple test table."""
38    await db_connection.execute(
39        "CREATE TABLE items(id INTEGER PRIMARY KEY AUTOINCREMENT, "
40        "name TEXT NOT NULL UNIQUE, url TEXT, plays INTEGER)"
41    )
42    await db_connection.commit()
43    return db_connection
44
45
46def _count_commits(db: DatabaseConnection) -> list[None]:
47    """Wrap the raw connection commit so every commit appends to the returned list."""
48    calls: list[None] = []
49    original_commit = db._db.commit
50
51    async def counting_commit() -> None:
52        calls.append(None)
53        await original_commit()
54
55    db._db.commit = counting_commit  # type: ignore[method-assign]
56    return calls
57
58
59async def _get_temp_store(db: DatabaseConnection) -> int:
60    async with db._db.execute("PRAGMA temp_store") as cursor:
61        row = await cursor.fetchone()
62    assert row is not None
63    return int(row[0])
64
65
66async def _read_pragma_int(db: DatabaseConnection, pragma: str) -> int:
67    async with db._db.execute(f"PRAGMA {pragma}") as cursor:
68        row = await cursor.fetchone()
69    assert row is not None
70    return int(row[0])
71
72
73async def test_vacuum_spills_temp_storage_to_disk(db_connection: DatabaseConnection) -> None:
74    """Test that vacuum runs with temp_store=FILE and restores temp_store=memory after."""
75    executed: list[str] = []
76    original_execute = db_connection._db.execute
77
78    def record(sql: str, *args: Any, **kwargs: Any) -> Any:
79        executed.append(sql)
80        return original_execute(sql, *args, **kwargs)
81
82    db_connection._db.execute = record  # type: ignore[method-assign]
83    await db_connection.vacuum()
84    db_connection._db.execute = original_execute  # type: ignore[method-assign]
85
86    vacuum_idx = executed.index("VACUUM")
87    assert any("temp_store=FILE" in sql for sql in executed[:vacuum_idx])
88    assert any("temp_store=memory" in sql for sql in executed[vacuum_idx:])
89    assert await _get_temp_store(db_connection) == TEMP_STORE_MEMORY
90
91
92async def test_vacuum_restores_temp_store_on_failure(
93    db_connection: DatabaseConnection,
94) -> None:
95    """Test that temp_store is restored to memory even when the vacuum itself fails."""
96    original_execute = db_connection._db.execute
97
98    def explode(sql: str, *args: Any, **kwargs: Any) -> Any:
99        if sql == "VACUUM":
100            raise OperationalError("database or disk is full")
101        return original_execute(sql, *args, **kwargs)
102
103    db_connection._db.execute = explode  # type: ignore[method-assign]
104    with pytest.raises(OperationalError):
105        await db_connection.vacuum()
106    db_connection._db.execute = original_execute  # type: ignore[method-assign]
107
108    assert await _get_temp_store(db_connection) == TEMP_STORE_MEMORY
109
110
111def test_sqlite_tmpdir_defaults_to_storage_path(tmp_path: pathlib.Path) -> None:
112    """Test that SQLITE_TMPDIR is pointed at the storage path on server init."""
113    original = os.environ.pop("SQLITE_TMPDIR", None)
114    try:
115        MusicAssistant(str(tmp_path / "data"), str(tmp_path / "cache"))
116        assert os.environ.get("SQLITE_TMPDIR") == str(tmp_path / "data")
117    finally:
118        if original is None:
119            os.environ.pop("SQLITE_TMPDIR", None)
120        else:
121            os.environ["SQLITE_TMPDIR"] = original
122
123
124def test_sqlite_tmpdir_respects_existing_value(
125    tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
126) -> None:
127    """Test that a user-provided SQLITE_TMPDIR is not overwritten."""
128    monkeypatch.setenv("SQLITE_TMPDIR", "/custom/tmp")
129    MusicAssistant(str(tmp_path / "data"), str(tmp_path / "cache"))
130    assert os.environ["SQLITE_TMPDIR"] == "/custom/tmp"
131
132
133@pytest.mark.parametrize(
134    ("total_ram_gb", "expected_cache_kib", "expected_mmap_bytes"),
135    [
136        (0.0, 64000, 2 * GIB),  # unknown -> fail open to the generous "capable" tier
137        (1.0, 16000, 256 * 1024 * 1024),
138        (2.0, 32000, GIB),
139        (4.0, 64000, 2 * GIB),  # capable: unchanged from the previous 64MB cache
140        (8.0, 128000, 2 * GIB),  # plenty of RAM: larger cache for performance
141        (12.0, 512000, 2 * GIB),  # large host: keep a big library hot
142        (16.0, 1024000, 2 * GIB),  # very large host
143        (32.0, 1024000, 2 * GIB),
144    ],
145)
146def test_sqlite_memory_settings_scale_with_ram(
147    monkeypatch: pytest.MonkeyPatch,
148    total_ram_gb: float,
149    expected_cache_kib: int,
150    expected_mmap_bytes: int,
151) -> None:
152    """Test that the SQLite cache/mmap ceilings scale with available system memory."""
153    monkeypatch.setattr(
154        "music_assistant.helpers.util.get_total_system_memory", lambda: total_ram_gb
155    )
156    assert get_sqlite_memory_settings() == (expected_cache_kib, expected_mmap_bytes)
157
158
159async def test_setup_applies_ram_scaled_pragmas(
160    tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
161) -> None:
162    """Test that setup() applies the RAM-scaled cache_size/mmap_size pragmas by default."""
163    # use the 2-4GB tier so the 1GiB mmap stays below SQLite's ~2GiB build cap and reads back
164    monkeypatch.setattr("music_assistant.helpers.util.get_total_system_memory", lambda: 2.0)
165    db = DatabaseConnection(str(tmp_path / "scaled.db"))
166    await db.setup()
167    try:
168        assert await _read_pragma_int(db, "cache_size") == -32000
169        assert await _read_pragma_int(db, "mmap_size") == GIB
170    finally:
171        await db.close()
172
173
174async def test_setup_clamps_pragma_values(tmp_path: pathlib.Path) -> None:
175    """Test that setup() clamps explicit cache/mmap values to non-negative integers."""
176    db = DatabaseConnection(str(tmp_path / "clamped.db"))
177    await db.setup(cache_size_kib=-50, mmap_size_bytes=-1)
178    try:
179        assert await _read_pragma_int(db, "cache_size") == 0
180        assert await _read_pragma_int(db, "mmap_size") == 0
181    finally:
182        await db.close()
183
184
185async def test_deferred_commit_batches_writes_into_single_commit(
186    db_with_table: DatabaseConnection,
187) -> None:
188    """Test that writes within a deferred_commit scope result in one commit at scope exit."""
189    commits = _count_commits(db_with_table)
190    async with db_with_table.deferred_commit():
191        for i in range(10):
192            await db_with_table.insert("items", {"name": f"item{i}"})
193        await db_with_table.update("items", {"name": "item0"}, {"url": "http://test"})
194        await db_with_table.delete("items", {"name": "item9"})
195        assert len(commits) == 0
196    assert len(commits) == 1
197    assert len(await db_with_table.get_rows("items")) == 9
198
199
200async def test_deferred_commit_nested_scopes_commit_once(
201    db_with_table: DatabaseConnection,
202) -> None:
203    """Test that nested deferred_commit scopes only commit when the outermost scope exits."""
204    commits = _count_commits(db_with_table)
205    async with db_with_table.deferred_commit():
206        await db_with_table.insert("items", {"name": "outer"})
207        async with db_with_table.deferred_commit():
208            await db_with_table.insert("items", {"name": "inner"})
209        assert len(commits) == 0
210    assert len(commits) == 1
211    assert len(await db_with_table.get_rows("items")) == 2
212
213
214async def test_deferred_commit_commits_pending_writes_on_error(
215    db_with_table: DatabaseConnection,
216) -> None:
217    """Test that a deferred_commit scope commits (not rolls back) already executed writes."""
218
219    async def write_and_fail() -> None:
220        async with db_with_table.deferred_commit():
221            await db_with_table.insert("items", {"name": "kept"})
222            raise RuntimeError("boom")
223
224    with pytest.raises(RuntimeError):
225        await write_and_fail()
226    assert await db_with_table.get_row("items", {"name": "kept"}) is not None
227
228
229async def test_deferred_commit_commits_on_cancellation(
230    db_with_table: DatabaseConnection,
231) -> None:
232    """Test that cancelling a task inside a scope commits its writes and keeps the db usable."""
233    started = asyncio.Event()
234    release = asyncio.Event()
235
236    async def writer() -> None:
237        async with db_with_table.deferred_commit():
238            await db_with_table.insert("items", {"name": "written-before-cancel"})
239            started.set()
240            await release.wait()
241
242    task = asyncio.get_running_loop().create_task(writer())
243    await started.wait()
244    task.cancel()
245    with pytest.raises(asyncio.CancelledError):
246        await task
247    assert not db_with_table._db.in_transaction
248    assert await db_with_table.get_row("items", {"name": "written-before-cancel"}) is not None
249    # the connection remains fully usable for subsequent writes
250    await db_with_table.insert("items", {"name": "after-cancel"})
251    assert await db_with_table.get_row("items", {"name": "after-cancel"}) is not None
252
253
254async def test_deferred_commit_scope_only_defers_own_task(
255    db_with_table: DatabaseConnection,
256) -> None:
257    """Test that writes from tasks outside a scope still commit immediately."""
258    commits = _count_commits(db_with_table)
259    in_scope = asyncio.Event()
260    release = asyncio.Event()
261
262    async def scoped_writer() -> None:
263        async with db_with_table.deferred_commit():
264            await db_with_table.insert("items", {"name": "scoped"})
265            in_scope.set()
266            await release.wait()
267
268    task = asyncio.get_running_loop().create_task(scoped_writer())
269    await in_scope.wait()
270    # this write happens on the test task (no scope) while the writer's scope is open
271    commits_before = len(commits)
272    await db_with_table.insert("items", {"name": "plain"})
273    assert len(commits) == commits_before + 1
274    release.set()
275    await task
276    assert len(await db_with_table.get_rows("items")) == 2
277
278
279async def test_deferred_commit_concurrent_scopes_do_not_interfere(
280    db_with_table: DatabaseConnection,
281) -> None:
282    """Test that concurrent tasks (e.g. parallel syncs) each batch their own writes."""
283    commits = _count_commits(db_with_table)
284    barrier = asyncio.Barrier(2)
285
286    async def sync_task(name: str) -> None:
287        async with db_with_table.deferred_commit():
288            await db_with_table.insert("items", {"name": f"{name}-1"})
289            # force both scopes to be open at the same time
290            await barrier.wait()
291            await db_with_table.insert("items", {"name": f"{name}-2"})
292
293    async with asyncio.TaskGroup() as tg:
294        tg.create_task(sync_task("a"))
295        tg.create_task(sync_task("b"))
296
297    # all writes from both tasks landed, with one commit per scope exit
298    assert len(await db_with_table.get_rows("items")) == 4
299    assert len(commits) == 2
300
301
302async def test_update_returns_none(db_with_table: DatabaseConnection) -> None:
303    """Test that update() no longer fetches and returns the updated row."""
304    row_id = await db_with_table.insert("items", {"name": "old"})
305    result = await db_with_table.update("items", {"id": row_id}, {"name": "new"})  # type: ignore[func-returns-value]
306    assert result is None
307    row = await db_with_table.get_row("items", {"id": row_id})
308    assert row is not None
309    assert row["name"] == "new"
310
311
312async def test_upsert_many(db_with_table: DatabaseConnection) -> None:
313    """Test that upsert_many inserts/updates multiple rows with a single commit."""
314    commits = _count_commits(db_with_table)
315    # rows with different column sets are handled in a single call
316    await db_with_table.upsert_many(
317        "items",
318        [
319            {"name": "a", "url": "http://a", "plays": 1},
320            {"name": "b", "plays": 2},
321            {"name": "c", "plays": 3},
322        ],
323    )
324    assert len(commits) == 1
325    rows = {row["name"]: row for row in await db_with_table.get_rows("items")}
326    assert len(rows) == 3
327    assert rows["a"]["url"] == "http://a"
328    # upserting again updates on conflict; omitted columns keep their existing value
329    await db_with_table.upsert_many("items", [{"name": "a", "plays": 10}])
330    row = await db_with_table.get_row("items", {"name": "a"})
331    assert row is not None
332    assert row["plays"] == 10
333    assert row["url"] == "http://a"
334
335
336async def test_upsert_many_empty_is_noop(db_with_table: DatabaseConnection) -> None:
337    """Test that upsert_many with no rows does nothing."""
338    commits = _count_commits(db_with_table)
339    await db_with_table.upsert_many("items", [])
340    assert len(commits) == 0
341
342
343def test_query_params_expands_list_values() -> None:
344    """Test that list params are expanded into placeholders in all placeholder notations."""
345    query, params = query_params(
346        "SELECT * FROM items WHERE id IN :ids AND name = :name",
347        {"ids": [1, 2], "name": "foo"},
348    )
349    assert query == "SELECT * FROM items WHERE id IN (:_param_0,:_param_1) AND name = :name"
350    assert params == {"_param_0": 1, "_param_1": 2, "name": "foo"}
351    # placeholder already wrapped in parens must not end up double-wrapped
352    query, params = query_params("SELECT * FROM items WHERE id IN(:ids)", {"ids": [1, 2]})
353    assert query == "SELECT * FROM items WHERE id IN(:_param_0,:_param_1)"
354    assert params == {"_param_0": 1, "_param_1": 2}
355
356
357def test_query_params_leaves_prefixed_placeholders_untouched() -> None:
358    """Test that expanding a list param does not corrupt placeholders sharing its prefix."""
359    query, params = query_params(
360        "SELECT * FROM items WHERE id IN :ids AND other = :ids_extra",
361        {"ids": [1], "ids_extra": 2},
362    )
363    assert query == "SELECT * FROM items WHERE id IN (:_param_0) AND other = :ids_extra"
364    assert params == {"_param_0": 1, "ids_extra": 2}
365