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