feat(settings): back up the library DB + custom art in the export bundle (#639)

Closes the dev-ops lens's #1 finding from the library charrette
(got-feedback/feedBack#636 item 1): scores, favorites, playlists, and play
history — the only library state a rescan can't rebuild — were absent from
the settings backup. Now GET /api/settings/export carries an additive
`core_server_files` section:

- a CONSISTENT snapshot of web_library.db via the SQLite online-backup API
  (a complete single file even while the server runs; taken under the
  MetadataDB write lock), base64-encoded;
- custom playlist covers + avatar (CONFIG_DIR/playlist_covers, /avatars),
  walked with the existing _walk_export_paths machinery.

Restore is DB-safe:
- POST /api/settings/import STAGES the DB to web_library.db.restore (never
  over the live, open file); _apply_pending_db_restore swaps it in at the
  next startup BEFORE the connection opens, clearing stale -wal/-shm so a
  stale WAL can't be replayed onto the restored file. Response sets
  `restart_required` + a warning; custom art applies immediately.
- The staged DB is integrity-checked (open + PRAGMA quick_check) at import
  AND again at startup before the live DB is touched — a corrupt/truncated
  restore is refused/discarded and the live DB is left intact, so a bad
  bundle can never brick startup or lose data.
- Export hard-fails (500) if the snapshot can't be produced (no silent
  DB-less backup); a partial import disarms its own staged restore.

Backward-compatible: older servers ignore the new section; a bundle without
it imports as before. Known gap: custom uploaded *song* art is still
commingled with the rebuildable thumbnail cache in art_cache/, so it isn't
bundled yet (tracked follow-up on #636).

Tests: tests/test_settings_export_library_db.py (snapshot consistency,
staged-not-live restore, sidecar clearing, corrupt-DB refusal at import +
startard, traversal rejection, export hard-fail, disarm-on-failure, full
round-trip).

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Byron Gamatos
2026-06-29 09:34:47 +02:00
committed by GitHub
co-authored by Claude Opus 4.8
parent 6a6efc793a
commit 07ab902604
3 changed files with 531 additions and 0 deletions
+1
View File
@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased] ## [Unreleased]
### Added ### Added
- **The settings backup now includes your library database + custom art — your scores, favorites, playlists, and play history are no longer the one thing a backup can't save.** `GET /api/settings/export` gains an additive `core_server_files` section carrying a **consistent snapshot of `web_library.db`** (taken via the SQLite online-backup API, so it's a complete single file even while the server is running) plus any custom **playlist covers** and **avatar** (`CONFIG_DIR/playlist_covers/`, `CONFIG_DIR/avatars/`). On `POST /api/settings/import` the database is **staged** to `web_library.db.restore` rather than written over the live, open DB; it's swapped in at the next startup (`_apply_pending_db_restore`, before the connection opens), which also clears the old WAL sidecars so a stale `-wal` can't be replayed onto the restored file — the import response sets `restart_required: true` and warns accordingly. Custom art is written immediately. The bundle stays backward-compatible (older servers ignore the new section). Came out of the library design charrette (dev-ops lens's top "protect irreplaceable data" pick, got-feedback/feedBack#636). _Known gap:_ custom uploaded **song** art is still commingled with the rebuildable thumbnail cache in `art_cache/`, so it isn't bundled yet (a tracked follow-up). Tests: `tests/test_settings_export_library_db.py` (snapshot consistency, staged-not-live restore, sidecar clearing, traversal rejection, full round-trip).
- **Practice-aware library home — a "Repertoire" meter + a "Keep practicing" shelf on the v3 Songs page.** The library opened cold into a flat sorted grid; now the unfiltered grid front door leads with two practice-aware surfaces built entirely from data already on hand (no new endpoints or stored state). A **Repertoire meter** shows how much of your library you can actually play — *"Repertoire: 12 of 80 songs · 7 in progress"* with a progress bar — counting songs at or above the same mastery threshold the green accuracy badge uses (≥ 90% best accuracy) over the unfiltered library total. A **"Keep practicing" shelf** is a horizontal row of your recently-played-but-not-yet-mastered songs (newest first, click to play) — the practice-accuracy-driven "continue" rail a media server can't do. Both reuse `/api/stats/best` (already loaded for the card badges) + `/api/stats/recent`; they show **only** on the grid view when you aren't searching/filtering/selecting, refresh after a song is scored, and collapse to nothing on an empty library. Soft-gamification only — descriptive encouragement (goal-gradient / endowed-progress), never content-gating, decay, or nagging. Frontend-only: `static/v3/songs.js` (`renderLibraryHome`/`_repertoireCounts`), `static/v3/v3.css`. Came out of the library design charrette (the UX + gamification lenses' top pick). Tests: `tests/js/v3_keep_practicing.test.js`. - **Practice-aware library home — a "Repertoire" meter + a "Keep practicing" shelf on the v3 Songs page.** The library opened cold into a flat sorted grid; now the unfiltered grid front door leads with two practice-aware surfaces built entirely from data already on hand (no new endpoints or stored state). A **Repertoire meter** shows how much of your library you can actually play — *"Repertoire: 12 of 80 songs · 7 in progress"* with a progress bar — counting songs at or above the same mastery threshold the green accuracy badge uses (≥ 90% best accuracy) over the unfiltered library total. A **"Keep practicing" shelf** is a horizontal row of your recently-played-but-not-yet-mastered songs (newest first, click to play) — the practice-accuracy-driven "continue" rail a media server can't do. Both reuse `/api/stats/best` (already loaded for the card badges) + `/api/stats/recent`; they show **only** on the grid view when you aren't searching/filtering/selecting, refresh after a song is scored, and collapse to nothing on an empty library. Soft-gamification only — descriptive encouragement (goal-gradient / endowed-progress), never content-gating, decay, or nagging. Frontend-only: `static/v3/songs.js` (`renderLibraryHome`/`_repertoireCounts`), `static/v3/v3.css`. Came out of the library design charrette (the UX + gamification lenses' top pick). Tests: `tests/js/v3_keep_practicing.test.js`.
- **AZ fast-scroll rail on the v3 Songs grid.** A vertical letter rail (Plex/Radarr/iOS-contacts pattern) pinned to the right edge next to the scrollbar lets you jump the library to a starting letter — tap a letter, drag to scrub with a live letter bubble, or arrow-key between letters. It shows **only** for the grid view + alphabetical (artist/title) sorts, and only offers letters actually present in the current sort **and filter set**, so a tap always lands on a real card (absent letters are dimmed + non-interactive). Because the grid is forward-only, server-paged infinite scroll, a jump pages through to the target card and scrolls to it (a newer jump supersedes an in-flight one); a keyset-seek + virtualized window is the noted scaling follow-up for very large libraries. Backend: `/api/library/stats` now accepts `sort` and returns an additive `sort_letters` map (songs-per-first-letter of the active sort column — artist or title), filter-synced; the legacy `letters` (distinct-artist) field is unchanged for the dashboard + classic tree. Frontend: `static/v3/songs.js` (`refreshRail`/`jumpToLetter`, cards tagged with `data-letter`), `static/v3/v3.css` (`.v3-azrail`). The classic (v2) tree already had letter selection; this brings the new grid to parity. Tests: `tests/test_library_filters.py` (sort_letters artist/title + song-vs-artist counting), `tests/test_library_providers.py` (sort forwarded to providers), `tests/js/v3_az_rail.test.js`. - **AZ fast-scroll rail on the v3 Songs grid.** A vertical letter rail (Plex/Radarr/iOS-contacts pattern) pinned to the right edge next to the scrollbar lets you jump the library to a starting letter — tap a letter, drag to scrub with a live letter bubble, or arrow-key between letters. It shows **only** for the grid view + alphabetical (artist/title) sorts, and only offers letters actually present in the current sort **and filter set**, so a tap always lands on a real card (absent letters are dimmed + non-interactive). Because the grid is forward-only, server-paged infinite scroll, a jump pages through to the target card and scrolls to it (a newer jump supersedes an in-flight one); a keyset-seek + virtualized window is the noted scaling follow-up for very large libraries. Backend: `/api/library/stats` now accepts `sort` and returns an additive `sort_letters` map (songs-per-first-letter of the active sort column — artist or title), filter-synced; the legacy `letters` (distinct-artist) field is unchanged for the dashboard + classic tree. Frontend: `static/v3/songs.js` (`refreshRail`/`jumpToLetter`, cards tagged with `data-letter`), `static/v3/v3.css` (`.v3-azrail`). The classic (v2) tree already had letter selection; this brings the new grid to parity. Tests: `tests/test_library_filters.py` (sort_letters artist/title + song-vs-artist counting), `tests/test_library_providers.py` (sort forwarded to providers), `tests/js/v3_az_rail.test.js`.
- **Playlists get content-dependent covers + custom art.** Playlist cards were a tiny `🎵` emoji on an empty square. Now a playlist's cover reflects its contents: **empty → the icon**, **a few songs → the first song's album art**, **4+ songs → a 2×2 art mosaic**. You can also **upload a custom cover** (a "Cover" button in the playlist detail view → image picker; "Remove cover" reverts to the content view). `MetadataDB.list_playlists()` now returns each playlist's first few song `art_urls`; `GET /api/playlists` and `GET /api/playlists/{id}` add `cover_url` when a custom cover exists. New routes `POST` / `GET` / `DELETE /api/playlists/{id}/cover` store a small PNG thumbnail under `CONFIG_DIR/playlist_covers/` (PIL-converted, like song-art upload); the cover is removed with the playlist. Frontend: `playlistCoverHtml(p)` in `static/v3/playlists.js`. Tests: `tests/test_playlists_api.py` (art_urls + cover roundtrip / reject-non-image / delete-cleanup), `tests/js/v3_playlist_cover.test.js`. - **Playlists get content-dependent covers + custom art.** Playlist cards were a tiny `🎵` emoji on an empty square. Now a playlist's cover reflects its contents: **empty → the icon**, **a few songs → the first song's album art**, **4+ songs → a 2×2 art mosaic**. You can also **upload a custom cover** (a "Cover" button in the playlist detail view → image picker; "Remove cover" reverts to the content view). `MetadataDB.list_playlists()` now returns each playlist's first few song `art_urls`; `GET /api/playlists` and `GET /api/playlists/{id}` add `cover_url` when a custom cover exists. New routes `POST` / `GET` / `DELETE /api/playlists/{id}/cover` store a small PNG thumbnail under `CONFIG_DIR/playlist_covers/` (PIL-converted, like song-art upload); the cover is removed with the playlist. Frontend: `playlistCoverHtml(p)` in `static/v3/playlists.js`. Tests: `tests/test_playlists_api.py` (art_urls + cover roundtrip / reject-non-image / delete-cleanup), `tests/js/v3_playlist_cover.test.js`.
+219
View File
@@ -364,9 +364,73 @@ def _ensure_smart_names(arrangements: list[dict]) -> list[dict]:
return arrangements return arrangements
def _sqlite_file_integrity_ok(path: Path) -> bool:
"""True if `path` is a SQLite database that opens and passes
`PRAGMA quick_check`. Used to gate a DB restore so a truncated or
corrupt snapshot can never overwrite the live library DB."""
try:
with open(path, "rb") as f:
if f.read(16) != b"SQLite format 3\x00": # cheap header gate, no full read
return False
except OSError:
return False
conn = None
try:
conn = sqlite3.connect(str(path))
row = conn.execute("PRAGMA quick_check").fetchone()
return bool(row) and row[0] == "ok"
except sqlite3.Error:
return False
finally:
if conn is not None:
conn.close()
# quick_check on a non-WAL file makes no sidecars, but a malformed
# file can; sweep them so a probe never litters config_dir.
for suffix in ("-wal", "-shm"):
try:
path.with_name(path.name + suffix).unlink()
except FileNotFoundError:
pass
def _apply_pending_db_restore(config_dir: Path) -> None:
"""Swap in a library DB restored from a settings bundle, if one is
staged. A settings import writes the restored snapshot to
`web_library.db.restore` rather than over the live DB (the running
server holds the old file open, and a stale `-wal`/`-shm` could be
replayed onto a fresh main file corruption). The swap happens here,
at startup, BEFORE the connection opens: delete the old DB and its WAL
sidecars, then rename the staged snapshot into place. The snapshot is a
fully-checkpointed single file (SQLite online-backup API), so it needs
no sidecars of its own. Idempotent and a no-op when nothing is staged.
The staged file is re-validated here before anything is destroyed: a
restore that fails its integrity check is discarded and the live DB is
left untouched, so a bad bundle can never brick startup or lose data."""
pending = config_dir / "web_library.db.restore"
if not pending.exists():
return
if not _sqlite_file_integrity_ok(pending):
log.error("pending library DB restore failed its integrity check; "
"discarding it and keeping the existing database")
try:
pending.unlink()
except FileNotFoundError:
pass
return
for suffix in ("", "-wal", "-shm"):
try:
(config_dir / f"web_library.db{suffix}").unlink()
except FileNotFoundError:
pass
os.replace(pending, config_dir / "web_library.db")
log.info("applied pending library DB restore from settings import")
class MetadataDB: class MetadataDB:
def __init__(self): def __init__(self):
CONFIG_DIR.mkdir(parents=True, exist_ok=True) CONFIG_DIR.mkdir(parents=True, exist_ok=True)
_apply_pending_db_restore(CONFIG_DIR)
self.db_path = str(CONFIG_DIR / "web_library.db") self.db_path = str(CONFIG_DIR / "web_library.db")
self.conn = sqlite3.connect(self.db_path, check_same_thread=False) self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
self.conn.execute("PRAGMA journal_mode=WAL") self.conn.execute("PRAGMA journal_mode=WAL")
@@ -6217,6 +6281,82 @@ def _atomic_write_file(target: Path, payload: bytes):
raise raise
# Core (non-plugin) server-side state that the settings bundle backs up
# alongside config.json. The library DB is the only state a rescan can't
# rebuild (scores, favorites, playlists, play history); the art dirs hold
# custom playlist covers + the user avatar. `web_library.db` is handled
# specially (consistent snapshot on export, staged restore on import) — the
# art dirs are walked like plugin export paths. NOTE: custom uploaded
# *song* art currently lands in `art_cache/` commingled with the derived
# (rebuildable) cache, so it is intentionally NOT bundled here to avoid
# bloating the backup with regenerable thumbnails — splitting custom song
# art into its own dir is a tracked follow-up (got-feedback/feedBack#636).
_CORE_LIBRARY_DB = "web_library.db"
_CORE_EXPORT_ART_DIRS = ("playlist_covers/", "avatars/")
_CORE_IMPORT_ALLOWED = (_CORE_LIBRARY_DB,) + _CORE_EXPORT_ART_DIRS
def _snapshot_library_db() -> dict | None:
"""A consistent, fully-checkpointed single-file copy of the live library
DB, base64-encoded for the bundle. Uses the SQLite online-backup API so
it is safe to call while the server is serving requests; the live write
lock is held for the copy so no write lands mid-snapshot. Returns None if
the DB or backup is unavailable (export proceeds without it)."""
import base64
fd, tmp = tempfile.mkstemp(dir=str(CONFIG_DIR), prefix="._dbsnap.", suffix=".db")
os.close(fd)
try:
dst = sqlite3.connect(tmp)
try:
with meta_db._lock:
meta_db.conn.backup(dst)
finally:
dst.close()
raw = Path(tmp).read_bytes()
except (sqlite3.Error, OSError):
log.warning("library DB snapshot for settings export failed", exc_info=True)
return None
finally:
for suffix in ("", "-wal", "-shm"):
try:
Path(tmp + suffix).unlink()
except FileNotFoundError:
pass
return {"encoding": "base64", "data": base64.b64encode(raw).decode("ascii")}
def _sqlite_payload_integrity_ok(payload: bytes) -> bool:
"""Validate decoded DB bytes by materializing them to a temp file and
running the same integrity probe used at restore time so a corrupt or
truncated snapshot is refused at import, before it's ever staged."""
fd, tmp = tempfile.mkstemp(dir=str(CONFIG_DIR), prefix="._dbcheck.", suffix=".db")
try:
with os.fdopen(fd, "wb") as f:
f.write(payload)
return _sqlite_file_integrity_ok(Path(tmp))
except OSError:
return False
finally:
try:
Path(tmp).unlink()
except FileNotFoundError:
pass
def _core_server_files() -> dict | None:
"""`{relpath: encoded_entry}` for core server-side state in the bundle:
a snapshot of the library DB plus any custom playlist covers / avatar.
Returns None if the DB snapshot could not be produced the caller must
treat that as a hard export failure rather than silently shipping a
backup that's missing the irreplaceable library state."""
snap = _snapshot_library_db()
if snap is None:
return None
out: dict[str, dict] = dict(_walk_export_paths(list(_CORE_EXPORT_ART_DIRS), CONFIG_DIR))
out[_CORE_LIBRARY_DB] = snap
return out
@app.get("/api/settings/export") @app.get("/api/settings/export")
def export_settings(): def export_settings():
"""Build a settings bundle covering server config + opted-in plugin """Build a settings bundle covering server config + opted-in plugin
@@ -6230,6 +6370,17 @@ def export_settings():
if server_config is None: if server_config is None:
server_config = _default_settings() server_config = _default_settings()
# Snapshot the library DB + custom art FIRST: if the irreplaceable state
# can't be captured, abort with an error rather than hand back a bundle
# that looks like a backup but silently omits it.
core_files = _core_server_files()
if core_files is None:
return JSONResponse(
{"ok": False, "error": "could not snapshot the library database; "
"export aborted to avoid an incomplete backup"},
status_code=500,
)
plugin_blocks: dict[str, dict] = {} plugin_blocks: dict[str, dict] = {}
with PLUGINS_LOCK: with PLUGINS_LOCK:
plugins_snapshot = list(LOADED_PLUGINS) plugins_snapshot = list(LOADED_PLUGINS)
@@ -6247,6 +6398,7 @@ def export_settings():
"feedBack_version": _running_version(), "feedBack_version": _running_version(),
"server_config": server_config, "server_config": server_config,
"plugin_server_configs": plugin_blocks, "plugin_server_configs": plugin_blocks,
"core_server_files": core_files,
} }
filename = f"feedBack-settings-{now.strftime('%Y-%m-%d')}.json" filename = f"feedBack-settings-{now.strftime('%Y-%m-%d')}.json"
return JSONResponse( return JSONResponse(
@@ -6386,6 +6538,62 @@ def import_settings(bundle: dict):
if applied_for_plugin: if applied_for_plugin:
applied_plugins.append(plugin_id) applied_plugins.append(plugin_id)
# ── Core server-side files (library DB + custom art) ─────────────
core_blocks = bundle.get("core_server_files") or {}
if not isinstance(core_blocks, dict):
return JSONResponse(
{"ok": False, "error": "core_server_files must be an object"},
status_code=400,
)
db_restore_staged = False
applied_core: list[str] = []
for relpath, file_entry in core_blocks.items():
if not isinstance(relpath, str) or not relpath:
return JSONResponse(
{"ok": False, "error": f"core_server_files: invalid relpath key {relpath!r}"},
status_code=400,
)
if relpath == _CORE_LIBRARY_DB:
# Stage the DB beside the live one; the swap happens at next
# startup (_apply_pending_db_restore), so we never overwrite a DB
# the server holds open or strand a stale WAL against a fresh file.
target = CONFIG_DIR / (_CORE_LIBRARY_DB + ".restore")
db_restore_staged = True
else:
try:
target = _validate_relpath(relpath, list(_CORE_IMPORT_ALLOWED), CONFIG_DIR)
except _UndeclaredFile:
warnings.append(f"core_server_files: skipped undeclared path {relpath!r}")
continue
except ValueError as e:
return JSONResponse(
{"ok": False, "error": f"core_server_files, file {relpath!r}: {e}"},
status_code=400,
)
try:
payload = _decode_entry(file_entry)
except ValueError as e:
return JSONResponse(
{"ok": False, "error": f"core_server_files, file {relpath!r}: {e}"},
status_code=400,
)
# Guard the DB payload: a truncated/corrupt file staged as the restore
# would fail to open at startup and brick the app (after the live DB
# is already gone). Reject anything that doesn't open + pass
# quick_check before it's ever staged.
if relpath == _CORE_LIBRARY_DB and not _sqlite_payload_integrity_ok(payload):
return JSONResponse(
{"ok": False, "error": "core_server_files: web_library.db is not a valid SQLite database"},
status_code=400,
)
staged.append((f"core/{relpath}", target, payload))
applied_core.append(relpath)
if db_restore_staged:
warnings.append(
"library database restored; restart FeedBack to load it "
"(scores, favorites, playlists, and play history)"
)
# ── Phase 2: commit ────────────────────────────────────────────── # ── Phase 2: commit ──────────────────────────────────────────────
written: list[str] = [] written: list[str] = []
try: try:
@@ -6412,6 +6620,15 @@ def import_settings(bundle: dict):
# because we didn't snapshot them — surface what got written # because we didn't snapshot them — surface what got written
# (as relpaths, not absolute server paths) so the user knows # (as relpaths, not absolute server paths) so the user knows
# the state is partial without leaking deployment layout. # the state is partial without leaking deployment layout.
# Disarm a staged DB restore THIS request wrote: a partial import must
# NOT silently swap the library DB on the next restart. Gate on the
# write actually having happened (display key in `written`) so we don't
# delete a valid restore staged by a prior, not-yet-applied import.
if f"core/{_CORE_LIBRARY_DB}" in written:
try:
(CONFIG_DIR / (_CORE_LIBRARY_DB + ".restore")).unlink()
except FileNotFoundError:
pass
return JSONResponse( return JSONResponse(
{ {
"ok": False, "ok": False,
@@ -6427,7 +6644,9 @@ def import_settings(bundle: dict):
"applied": { "applied": {
"server_config": True, "server_config": True,
"plugins": applied_plugins, "plugins": applied_plugins,
"core_files": applied_core,
}, },
"restart_required": db_restore_staged,
} }
+311
View File
@@ -0,0 +1,311 @@
"""Tests for the library-DB + custom-art half of the settings bundle
(got-feedback/feedBack#636 item 1).
The base bundle (config + plugin files) is covered in test_settings_export.py;
this file pins the additive `core_server_files` section:
- the live library DB is exported as a CONSISTENT single-file snapshot
(SQLite online-backup), base64-encoded;
- custom playlist covers / avatar are walked into the bundle;
- on import the DB is STAGED to `web_library.db.restore` (never written
over the live, open DB) and swapped in at next startup, clearing stale
WAL sidecars; custom art is written immediately;
- the whole thing round-trips: export → wipe → import → restart → data back.
"""
import base64
import importlib
import sqlite3
import sys
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
@pytest.fixture()
def server_mod(tmp_path, monkeypatch):
monkeypatch.setenv("CONFIG_DIR", str(tmp_path))
sys.modules.pop("server", None)
mod = importlib.import_module("server")
yield mod
conn = getattr(getattr(mod, "meta_db", None), "conn", None)
if conn is not None:
conn.close()
@pytest.fixture()
def client(server_mod):
c = TestClient(server_mod.app)
try:
yield c
finally:
c.close()
def _valid_db_bytes(tmp_path, name="mk.db", marker="x"):
"""Bytes of a small, valid (quick_check-clean) SQLite database."""
p = tmp_path / name
c = sqlite3.connect(str(p))
try:
c.execute("CREATE TABLE t (x TEXT)")
c.execute("INSERT INTO t VALUES (?)", (marker,))
c.commit()
finally:
c.close()
return p.read_bytes()
def _seed_song(server_mod, filename="marker.archive", title="Marker", artist="Tester"):
server_mod.meta_db.put(filename, 1.0, 1, {
"title": title, "artist": artist, "album": "LP", "year": "",
"duration": 200.0, "tuning": "E Standard", "arrangements": [],
"has_lyrics": False, "format": "archive", "stem_count": 0,
"stem_ids": [], "tuning_name": "E Standard", "tuning_sort_key": 0,
"tuning_offsets": "",
})
# ── Export ──────────────────────────────────────────────────────────────────
def test_export_includes_consistent_library_db_snapshot(client, server_mod, tmp_path):
_seed_song(server_mod, filename="snap.archive", title="SnapSong")
bundle = client.get("/api/settings/export").json()
core = bundle["core_server_files"]
assert "web_library.db" in core
entry = core["web_library.db"]
assert entry["encoding"] == "base64"
# The snapshot must be a complete, openable DB reflecting current data —
# written to its own file (no WAL sidecar needed) and queryable.
snap = tmp_path / "snapshot.db"
snap.write_bytes(base64.b64decode(entry["data"]))
conn = sqlite3.connect(str(snap))
try:
rows = conn.execute(
"SELECT title FROM songs WHERE filename = ?", ("snap.archive",)
).fetchall()
finally:
conn.close()
assert rows == [("SnapSong",)]
def test_export_includes_custom_art_dirs(client, tmp_path):
(tmp_path / "playlist_covers").mkdir()
(tmp_path / "playlist_covers" / "3.png").write_bytes(b"\x89PNG-cover")
(tmp_path / "avatars").mkdir()
(tmp_path / "avatars" / "me.png").write_bytes(b"\x89PNG-avatar")
core = client.get("/api/settings/export").json()["core_server_files"]
assert core["playlist_covers/3.png"]["encoding"] == "base64"
assert base64.b64decode(core["playlist_covers/3.png"]["data"]) == b"\x89PNG-cover"
assert base64.b64decode(core["avatars/me.png"]["data"]) == b"\x89PNG-avatar"
# ── Import: DB is staged, never written over the live file ──────────────────
def test_import_stages_db_restore_without_touching_live_db(client, server_mod, tmp_path):
live = tmp_path / "web_library.db"
live_bytes_before = live.read_bytes()
payload = _valid_db_bytes(tmp_path, name="incoming.db", marker="restored")
r = client.post("/api/settings/import", json={
"schema": server_mod.SETTINGS_BUNDLE_SCHEMA,
"server_config": {},
"core_server_files": {
"web_library.db": {"encoding": "base64",
"data": base64.b64encode(payload).decode()},
},
})
assert r.status_code == 200
body = r.json()
assert body["ok"] is True
assert body["restart_required"] is True
assert any("restart" in w.lower() for w in body["warnings"])
assert "web_library.db" in body["applied"]["core_files"]
# Live DB untouched; the restore is staged beside it for next startup.
assert live.read_bytes() == live_bytes_before
assert (tmp_path / "web_library.db.restore").read_bytes() == payload
def test_import_rejects_corrupt_db_with_valid_magic_header(client, server_mod, tmp_path):
# The dangerous case: SQLite magic header but a corrupt body. It must be
# refused at import — otherwise startup would delete the live DB and then
# fail to open the bad restore.
corrupt = b"SQLite format 3\x00" + b"\xff" * 200
r = client.post("/api/settings/import", json={
"schema": server_mod.SETTINGS_BUNDLE_SCHEMA,
"server_config": {},
"core_server_files": {
"web_library.db": {"encoding": "base64",
"data": base64.b64encode(corrupt).decode()},
},
})
assert r.status_code == 400
assert not (tmp_path / "web_library.db.restore").exists()
def test_import_rejects_non_sqlite_db_payload(client, server_mod, tmp_path):
# A truncated / wrong file staged as the restore would brick startup —
# reject anything lacking the SQLite magic header, before touching disk.
r = client.post("/api/settings/import", json={
"schema": server_mod.SETTINGS_BUNDLE_SCHEMA,
"server_config": {},
"core_server_files": {
"web_library.db": {"encoding": "base64",
"data": base64.b64encode(b"not a database").decode()},
},
})
assert r.status_code == 400
assert not (tmp_path / "web_library.db.restore").exists()
def test_import_writes_custom_art_immediately(client, server_mod, tmp_path):
r = client.post("/api/settings/import", json={
"schema": server_mod.SETTINGS_BUNDLE_SCHEMA,
"server_config": {},
"core_server_files": {
"playlist_covers/7.png": {"encoding": "base64",
"data": base64.b64encode(b"cover7").decode()},
},
})
assert r.status_code == 200
assert r.json()["restart_required"] is False
assert (tmp_path / "playlist_covers" / "7.png").read_bytes() == b"cover7"
def test_import_core_path_traversal_rejected(client, server_mod, tmp_path):
secret = tmp_path.parent / "escape.txt"
r = client.post("/api/settings/import", json={
"schema": server_mod.SETTINGS_BUNDLE_SCHEMA,
"server_config": {},
"core_server_files": {
"../escape.txt": {"encoding": "base64",
"data": base64.b64encode(b"pwned").decode()},
},
})
assert r.status_code == 400
assert not secret.exists()
def test_import_core_undeclared_path_skipped_not_fatal(client, server_mod, tmp_path):
# A relpath outside the core allowlist is a warn-and-skip, not a refusal —
# the rest of the bundle still applies.
r = client.post("/api/settings/import", json={
"schema": server_mod.SETTINGS_BUNDLE_SCHEMA,
"server_config": {},
"core_server_files": {
"audio_cache/x.ogg": {"encoding": "base64",
"data": base64.b64encode(b"nope").decode()},
},
})
assert r.status_code == 200
assert not (tmp_path / "audio_cache" / "x.ogg").exists()
assert any("undeclared" in w.lower() for w in r.json()["warnings"])
# ── Startup swap ────────────────────────────────────────────────────────────
def test_apply_pending_db_restore_swaps_and_clears_sidecars(server_mod, tmp_path):
main = tmp_path / "web_library.db"
new_db = _valid_db_bytes(tmp_path, name="new.db", marker="new")
# Simulate a live DB with stale WAL sidecars + a (valid) staged restore.
main.write_bytes(b"OLD-DB")
(tmp_path / "web_library.db-wal").write_bytes(b"OLD-WAL")
(tmp_path / "web_library.db-shm").write_bytes(b"OLD-SHM")
(tmp_path / "web_library.db.restore").write_bytes(new_db)
server_mod._apply_pending_db_restore(tmp_path)
assert main.read_bytes() == new_db # swapped in
assert not (tmp_path / "web_library.db.restore").exists()
assert not (tmp_path / "web_library.db-wal").exists() # stale sidecars gone
assert not (tmp_path / "web_library.db-shm").exists()
def test_apply_pending_db_restore_discards_corrupt_keeps_live(server_mod, tmp_path):
# A corrupt staged restore must be thrown away WITHOUT destroying the
# live DB — never brick startup or lose data for a bad bundle.
main = tmp_path / "web_library.db"
main.write_bytes(b"LIVE-GOOD-DB")
(tmp_path / "web_library.db.restore").write_bytes(b"SQLite format 3\x00" + b"\xff" * 64)
server_mod._apply_pending_db_restore(tmp_path)
assert main.read_bytes() == b"LIVE-GOOD-DB" # live DB preserved
assert not (tmp_path / "web_library.db.restore").exists() # bad restore dropped
def test_apply_pending_db_restore_noop_without_staging(server_mod, tmp_path):
(tmp_path / "web_library.db").write_bytes(b"LIVE")
server_mod._apply_pending_db_restore(tmp_path) # nothing staged
assert (tmp_path / "web_library.db").read_bytes() == b"LIVE"
# ── Full round-trip ─────────────────────────────────────────────────────────
def test_full_db_backup_restore_round_trip(client, server_mod, tmp_path):
_seed_song(server_mod, filename="keepme.archive", title="KeepMe")
bundle = client.get("/api/settings/export").json()
# Lose the data (a song removed from the live DB after the backup).
server_mod.meta_db.conn.execute("DELETE FROM songs WHERE filename = ?", ("keepme.archive",))
server_mod.meta_db.conn.commit()
assert server_mod.meta_db.conn.execute(
"SELECT COUNT(*) FROM songs WHERE filename = ?", ("keepme.archive",)
).fetchone()[0] == 0
# Re-import the bundle → DB staged, not yet live.
r = client.post("/api/settings/import", json=bundle)
assert r.status_code == 200 and r.json()["restart_required"] is True
# Simulate a restart: close the live conn, apply the staged restore,
# reopen — the song is back.
server_mod.meta_db.conn.close()
server_mod._apply_pending_db_restore(tmp_path)
conn = sqlite3.connect(str(tmp_path / "web_library.db"))
try:
rows = conn.execute(
"SELECT title FROM songs WHERE filename = ?", ("keepme.archive",)
).fetchall()
finally:
conn.close()
assert rows == [("KeepMe",)]
assert not (tmp_path / "web_library.db.restore").exists()
# ── Failure modes ───────────────────────────────────────────────────────────
def test_export_fails_hard_when_db_snapshot_unavailable(client, server_mod, monkeypatch):
# A backup that silently omits the library DB is a data-loss trap — the
# export must error rather than hand back an incomplete-looking bundle.
monkeypatch.setattr(server_mod, "_snapshot_library_db", lambda: None)
r = client.get("/api/settings/export")
assert r.status_code == 500
assert "library database" in r.json()["error"].lower()
def test_failed_import_disarms_staged_db_restore(client, server_mod, tmp_path, monkeypatch):
# If a later write in phase 2 fails, the request 500s — but a staged DB
# restore must NOT survive to swap in on the next restart.
payload = _valid_db_bytes(tmp_path, name="incoming.db")
real_write = server_mod._atomic_write_file
def boom(target, data):
if target.name == "config.json": # last write of the commit
raise OSError("disk full")
return real_write(target, data)
monkeypatch.setattr(server_mod, "_atomic_write_file", boom)
r = client.post("/api/settings/import", json={
"schema": server_mod.SETTINGS_BUNDLE_SCHEMA,
"server_config": {},
"core_server_files": {
"web_library.db": {"encoding": "base64",
"data": base64.b64encode(payload).decode()},
},
})
assert r.status_code == 500
assert not (tmp_path / "web_library.db.restore").exists()