feedBack/tests/test_library_providers.py

import importlib
import sys

import pytest
from fastapi.responses import Response
from fastapi.testclient import TestClient


@pytest.fixture()
def server_mod(tmp_path, monkeypatch):
    monkeypatch.setenv("CONFIG_DIR", str(tmp_path))
    monkeypatch.setenv("FEEDBACK_SKIP_STARTUP_TASKS", "1")
    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 _put(server_mod, filename="local.archive", title="Local Song", artist="Local Artist"):
    server_mod.meta_db.put(filename, 1.0, 1, {
        "title": title,
        "artist": artist,
        "album": "Local Album",
        "year": "2026",
        "duration": 180.0,
        "tuning": "E Standard",
        "arrangements": [{"index": 0, "name": "Lead", "notes": 1}],
        "has_lyrics": True,
        "format": "archive",
        "stem_count": 0,
        "stem_ids": [],
        "tuning_name": "E Standard",
        "tuning_sort_key": 0,
    })


class FakeLibraryProvider:
    id = "remote:frodo"
    label = "Frodo's Library"
    kind = "remote"
    capabilities = ("library.read", "art.read", "song.sync")

    def __init__(self):
        self.page_kwargs = None
        self.artist_kwargs = None
        self.stats_kwargs = None
        self.art_song_id = None
        self.sync_song_id = None

    def query_page(self, **kwargs):
        self.page_kwargs = kwargs
        return ([{
            "filename": "remote-song-id",
            "title": "Remote Song",
            "artist": "Remote Artist",
            "album": "Remote Album",
            "arrangements": [],
            "format": "archive",
        }], 1)

    def query_artists(self, **kwargs):
        self.artist_kwargs = kwargs
        return ([{
            "name": "Remote Artist",
            "album_count": 1,
            "song_count": 1,
            "albums": [{"name": "Remote Album", "songs": []}],
        }], 1)

    def query_stats(self, **kwargs):
        self.stats_kwargs = kwargs
        return {"total_songs": 1, "total_artists": 1, "letters": {"R": 1}}

    def tuning_names(self):
        return {"tunings": [{"name": "E Standard", "sort_key": 0, "count": 1}]}

    def get_art(self, song_id: str):
        self.art_song_id = song_id
        return Response(content=b"fake-art", media_type="image/png")

    def sync_song(self, song_id: str):
        self.sync_song_id = song_id
        return {"ok": True, "filename": "synced.archive", "song_id": song_id}


class ReadOnlyLibraryProvider:
    id = "remote:readonly"
    label = "Readonly Library"
    kind = "remote"
    capabilities = ("library.read",)

    def query_page(self, **kwargs):
        return ([], 0)

    def query_artists(self, **kwargs):
        return ([], 0)

    def query_stats(self, **kwargs):
        return {"total_songs": 0, "total_artists": 0, "letters": {}}

    def tuning_names(self):
        return {"tunings": []}

    def get_art(self, song_id: str):
        raise AssertionError("provider without art.read capability should not be dispatched")

    def sync_song(self, song_id: str):
        raise AssertionError("provider without song.sync capability should not be dispatched")


class NonBrowsableLibraryProvider:
    id = "remote:actions-only"
    label = "Actions Only"
    kind = "remote"
    capabilities = ("art.read", "song.sync")

    def query_page(self, **kwargs):
        raise AssertionError("provider without library.read capability should not be dispatched")

    def query_artists(self, **kwargs):
        raise AssertionError("provider without library.read capability should not be dispatched")

    def query_stats(self, **kwargs):
        raise AssertionError("provider without library.read capability should not be dispatched")

    def tuning_names(self):
        raise AssertionError("provider without library.read capability should not be dispatched")

    def get_art(self, song_id: str):
        raise AssertionError("get_art should not be dispatched in this test")

    def sync_song(self, song_id: str):
        raise AssertionError("sync_song should not be dispatched in this test")


def test_local_provider_is_default_library_provider(server_mod, client):
    _put(server_mod)

    providers = client.get("/api/library/providers").json()["providers"]
    local = next(provider for provider in providers if provider["id"] == "local")
    assert local["label"] == "My Library"
    assert local["kind"] == "local"
    assert local["default"] is True
    assert "library.read" in local["capabilities"]

    default_payload = client.get("/api/library").json()
    explicit_payload = client.get("/api/library", params={"provider": "local"}).json()
    assert default_payload == explicit_payload
    assert default_payload["songs"][0]["filename"] == "local.archive"


def test_registered_provider_handles_library_endpoints(server_mod, client):
    provider = FakeLibraryProvider()
    server_mod.register_library_provider(provider)

    providers = client.get("/api/library/providers").json()["providers"]
    remote = next(item for item in providers if item["id"] == "remote:frodo")
    assert remote["label"] == "Frodo's Library"
    assert remote["kind"] == "remote"
    assert remote["default"] is False
    assert remote["capabilities"] == ["art.read", "library.read", "song.sync"]

    songs = client.get("/api/library", params={
        "provider": "remote:frodo",
        "q": "needle",
        "page": "2",
        "size": "12",
        "sort": "title-desc",
        "dir": "desc",
        "format": "archive",
        "favorites": "1",
        "arrangements_has": "Lead,Rhythm",
        "has_lyrics": "1",
        "tunings": "E Standard,Drop D",
    }).json()
    assert songs["total"] == 1
    assert songs["songs"][0]["title"] == "Remote Song"
    assert provider.page_kwargs["q"] == "needle"
    assert provider.page_kwargs["page"] == 2
    assert provider.page_kwargs["size"] == 12
    assert provider.page_kwargs["sort"] == "title-desc"
    assert provider.page_kwargs["direction"] == "desc"
    assert provider.page_kwargs["favorites_only"] is True
    assert provider.page_kwargs["format_filter"] == "archive"
    assert provider.page_kwargs["arrangements_has"] == ["Lead", "Rhythm"]
    assert provider.page_kwargs["has_lyrics"] == 1
    assert provider.page_kwargs["tunings"] == ["E Standard", "Drop D"]

    artists = client.get("/api/library/artists", params={
        "provider": "remote:frodo",
        "letter": "R",
        "page": "1",
        "size": "5",
    }).json()
    assert artists["total_artists"] == 1
    assert provider.artist_kwargs["letter"] == "R"
    assert provider.artist_kwargs["page"] == 1
    assert provider.artist_kwargs["size"] == 5
    assert "sort" not in provider.artist_kwargs

    stats = client.get("/api/library/stats", params={"provider": "remote:frodo"}).json()
    assert stats["letters"] == {"R": 1}
    assert "page" not in provider.stats_kwargs
    assert "size" not in provider.stats_kwargs
    # `sort` is forwarded to query_stats now (the v3 jump rail keys its
    # present-letter breakdown on the active sort column); defaults to "artist".
    assert provider.stats_kwargs.get("sort") == "artist"

    tunings = client.get("/api/library/tuning-names", params={"provider": "remote:frodo"}).json()
    assert tunings["tunings"][0]["name"] == "E Standard"

    art = client.get("/api/library/providers/remote:frodo/songs/remote-song-id/art")
    assert art.status_code == 200
    assert art.content == b"fake-art"
    assert art.headers["content-type"].startswith("image/png")
    assert provider.art_song_id == "remote-song-id"

    synced = client.post("/api/library/providers/remote:frodo/songs/remote-song-id/sync").json()
    assert synced == {"ok": True, "filename": "synced.archive", "song_id": "remote-song-id"}
    assert provider.sync_song_id == "remote-song-id"


def test_unknown_library_provider_returns_404(server_mod, client):
    response = client.get("/api/library", params={"provider": "missing"})
    assert response.status_code == 404
    assert "Unknown library provider" in response.json()["detail"]


def test_provider_art_and_sync_report_unsupported_when_missing(server_mod, client):
    server_mod.register_library_provider(ReadOnlyLibraryProvider())

    art = client.get("/api/library/providers/remote:readonly/songs/remote-song-id/art")
    assert art.status_code == 501
    assert "does not declare capability 'art.read'" in art.json()["detail"]

    synced = client.post("/api/library/providers/remote:readonly/songs/remote-song-id/sync")
    assert synced.status_code == 501
    assert "does not declare capability 'song.sync'" in synced.json()["detail"]


def test_library_read_endpoints_require_library_read_capability(server_mod, client):
    server_mod.register_library_provider(NonBrowsableLibraryProvider())

    responses = [
        client.get("/api/library", params={"provider": "remote:actions-only"}),
        client.get("/api/library/artists", params={"provider": "remote:actions-only"}),
        client.get("/api/library/stats", params={"provider": "remote:actions-only"}),
        client.get("/api/library/tuning-names", params={"provider": "remote:actions-only"}),
    ]
    for response in responses:
        assert response.status_code == 501
        assert "does not declare capability 'library.read'" in response.json()["detail"]


def test_local_library_provider_cannot_be_replaced(server_mod):
    provider = FakeLibraryProvider()
    provider.id = "local"
    provider.label = "Not Actually Local"

    try:
        server_mod.register_library_provider(provider, replace=True)
    except ValueError as exc:
        assert "local library provider cannot be replaced" in str(exc)
    else:
        raise AssertionError("expected replacing the local provider to fail")


def test_library_provider_registration_is_available_to_plugins(tmp_path, monkeypatch):
    monkeypatch.setenv("CONFIG_DIR", str(tmp_path))
    monkeypatch.setenv("FEEDBACK_SYNC_STARTUP", "1")
    sys.modules.pop("server", None)
    server = importlib.import_module("server")

    captured = {}

    def _capturing_load_plugins(app, context, **kwargs):
        captured.update(context)

    monkeypatch.setattr(server, "load_plugins", _capturing_load_plugins)
    monkeypatch.setattr(server, "startup_scan", lambda: None)

    conn = getattr(getattr(server, "meta_db", None), "conn", None)
    try:
        with TestClient(server.app):
            assert captured["library_providers"] is server.library_providers
            assert captured["register_library_provider"] is server.register_library_provider
            assert captured["unregister_library_provider"] is server.unregister_library_provider
    finally:
        if conn is not None:
            getattr(__import__("sys").modules.get("server"), "_join_background_db_threads", lambda: None)()
            conn.close()