feedBack/tests/test_settings_export_library_db.py

"""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
from routers import settings as settings_router
import sqlite3
import sys
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

# The startup DB swap lives with the DB layer it guards, not with server.py.
# metadata_db reads no environment at import, so a plain module import is safe
# alongside the env-patched `server_mod` fixture below.
from metadata_db import _apply_pending_db_restore


@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:
        getattr(__import__("sys").modules.get("server"), "_join_background_db_threads", lambda: 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:
        getattr(__import__("sys").modules.get("server"), "_join_background_db_threads", lambda: None)()
        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": settings_router.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": settings_router.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": settings_router.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": settings_router.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": settings_router.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": settings_router.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)

    _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)

    _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")
    _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()
    _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:
        getattr(__import__("sys").modules.get("server"), "_join_background_db_threads", lambda: None)()
        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(settings_router, "_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 = settings_router._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(settings_router, "_atomic_write_file", boom)
    r = client.post("/api/settings/import", json={
        "schema": settings_router.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()