tag-albums/tests/test_app.py

import re
import sqlite3

from conftest import AUTH

from tag_albums.cli import fetch_covers, import_yaml
from tag_albums.store import Store


def test_api_requires_token(client):
    assert client.post("/api/scans", json={"tag_id": "AA"}).status_code == 401
    assert client.get("/api/tags/AA", headers={"Authorization": "Bearer wrong"}).status_code == 401


def test_first_scan_creates_unassigned_tag(client, store):
    r = client.post("/api/scans", json={"tag_id": "53-13-0b-2a"}, headers=AUTH)
    assert r.status_code == 200
    assert r.json() == {"tag_id": "53-13-0B-2A", "assigned": False}
    tag = store.get("53-13-0B-2A")
    assert tag.scan_count == 1 and not tag.assigned


def test_scan_of_assigned_tag_returns_album(client, store):
    store.assign("86-2C-1D-BD", "Tool", "Lateralus")
    r = client.post("/api/scans", json={"tag_id": "86-2C-1D-BD"}, headers=AUTH)
    assert r.json() == {"tag_id": "86-2C-1D-BD", "assigned": True, "artist": "Tool", "album": "Lateralus"}
    assert store.get("86-2C-1D-BD").scan_count == 1


def test_empty_tag_id_rejected(client):
    assert client.post("/api/scans", json={"tag_id": "  "}, headers=AUTH).status_code == 422


def test_lookup_is_read_only(client, store):
    assert client.get("/api/tags/AA-BB", headers=AUTH).status_code == 404
    assert store.get("AA-BB") is None

    store.record_scan("AA-BB")
    r = client.get("/api/tags/aa-bb", headers=AUTH)
    assert r.json() == {"tag_id": "AA-BB", "assigned": False}
    assert store.get("AA-BB").scan_count == 1


def test_assign_unassign_delete_from_web_page(client, store):
    store.record_scan("AA-BB")
    r = client.post("/tags/AA-BB/assign", data={"artist": "Tool", "album": "Lateralus", "image": ""},
                    follow_redirects=False)
    assert r.status_code == 303
    tag = store.get("AA-BB")
    assert (tag.artist, tag.album, tag.image) == ("Tool", "Lateralus", None)

    client.post("/tags/AA-BB/unassign")
    assert not store.get("AA-BB").assigned

    client.post("/tags/AA-BB/delete")
    assert store.get("AA-BB") is None


def test_pages_render(client, store):
    store.record_scan("AA-BB")
    store.assign("CC-DD", "Tool", "Lateralus")
    home = client.get("/")
    assert "AA-BB" in home.text and "Lateralus" in home.text
    assert "AA-BB" in client.get("/unassigned").text
    assert client.get("/tags/AA-BB").status_code == 200
    assert client.get("/tags/ZZ").status_code == 404


def test_htmx_is_served_locally(client):
    assert 'src="/static/htmx-2.0.4.min.js"' in client.get("/").text
    r = client.get("/static/htmx-2.0.4.min.js")
    assert r.status_code == 200 and r.text.startswith("var htmx=")


def test_search_results(client, library):
    r = client.get("/tags/AA-BB/search", params={"q": " lateralus "})
    assert library.queries == ["lateralus"]
    assert "Lateralus" in r.text and 'action="/tags/AA-BB/assign"' in r.text
    # Covers go through the service, never straight to Music Assistant's http URL.
    assert 'src="http://img/1"' not in r.text and 'src="/covers?src=' in r.text


def _search_form(client):
    """The hidden fields of the first search result's Assign form."""
    html = client.get("/tags/AA-BB/search", params={"q": "lateralus"}).text
    return dict(re.findall(r'name="(\w+)" value="([^"]*)"', html))


def test_assign_from_search_stores_cover(client, store, library):
    store.record_scan("AA-BB")
    form = _search_form(client)
    assert form["image"] == "http://img/1" and form["sig"]
    client.post("/tags/AA-BB/assign", data=form)

    tag = store.get("AA-BB")
    assert tag.has_cover and tag.image == "http://img/1"
    r = client.get("/tags/AA-BB/cover")
    assert (r.status_code, r.content, r.headers["content-type"]) == (200, b"\x89PNG-lateralus", "image/png")
    assert f'src="/tags/AA-BB/cover?v=' in client.get("/").text


def test_assign_ignores_unsigned_image(client, store, library):
    client.post("/tags/AA-BB/assign",
                data={"artist": "Tool", "album": "Lateralus", "image": "http://evil/x", "sig": "nope"})
    tag = store.get("AA-BB")
    assert tag.image is None and not tag.has_cover and library.fetched == []


def test_reassigning_drops_the_old_cover(client, store):
    client.post("/tags/AA-BB/assign", data=_search_form(client))
    assert store.get("AA-BB").has_cover
    client.post("/tags/AA-BB/assign", data={"artist": "Mogwai", "album": "Come On Die Young"})
    assert not store.get("AA-BB").has_cover
    assert client.get("/tags/AA-BB/cover").status_code == 404


def test_cover_preview_only_serves_signed_urls(client):
    form = _search_form(client)
    ok = client.get("/covers", params={"src": form["image"], "sig": form["sig"]})
    assert ok.status_code == 200 and ok.content == b"\x89PNG-lateralus"
    assert client.get("/covers", params={"src": "http://evil/x", "sig": form["sig"]}).status_code == 403
    assert client.get("/covers", params={"src": "http://evil/x"}).status_code == 403


def test_assignments_sort(client, store):
    store.assign("A", "Tool", "Lateralus")
    store.assign("B", "Amenra", "De Doorn")
    store._conn.execute("UPDATE tags SET assigned_at = '2026-01-01T00:00:00+00:00' WHERE tag_id = 'B'")
    store.record_scan("B")

    def order(sort):
        return [t.tag_id for t in store.assigned(sort)]

    assert order("artist") == ["B", "A"]
    assert order("assigned") == ["A", "B"]
    assert order("scanned") == ["B", "A"]          # never-scanned tags last
    page = client.get("/", params={"sort": "assigned"}).text
    assert page.index("Lateralus") < page.index("De Doorn")
    assert '<strong aria-current="true">Recently assigned</strong>' in page
    assert "not scanned yet" in page and "last scanned" in page
    assert client.get("/", params={"sort": "bogus"}).status_code == 200


def test_old_database_is_migrated(tmp_path):
    db = str(tmp_path / "old.sqlite3")
    conn = sqlite3.connect(db)
    conn.execute("CREATE TABLE tags (tag_id TEXT PRIMARY KEY, first_seen TEXT NOT NULL, last_seen TEXT,"
                 " scan_count INTEGER NOT NULL DEFAULT 0, artist TEXT, album TEXT, image TEXT, assigned_at TEXT)")
    conn.execute("INSERT INTO tags (tag_id, first_seen, artist, album, assigned_at)"
                 " VALUES ('AA', '2026-01-01', 'Tool', 'Lateralus', '2026-01-01')")
    conn.commit()
    conn.close()
    store = Store(db)
    tag = store.get("AA")
    assert tag.album == "Lateralus" and not tag.has_cover
    store.set_cover("AA", b"img", "image/jpeg")
    assert store.get_cover("AA") == (b"img", "image/jpeg")
    store.close()


def test_fetch_covers_for_imported_assignments(store, library):
    store.assign("A", "Tool", "Lateralus")                   # imported: no image URL
    store.assign("B", "Amenra", "De Doorn")                  # not in the library
    found, missing = fetch_covers(store, library)
    assert found == ["Tool – Lateralus"] and missing == ["Amenra – De Doorn"]
    tag = store.get("A")
    assert tag.has_cover and tag.image == "http://img/1"
    assert fetch_covers(store, library) == ([], ["Amenra – De Doorn"])   # already stored: skipped


def test_import_yaml(tmp_path):
    src = tmp_path / "tag_albums.yaml"
    src.write_text('"29-92-68-C5":\n  artist: ONRUST\n  album: Van Woede Tot Wanhoop\n')
    db = str(tmp_path / "db.sqlite3")
    assert import_yaml(str(src), db) == 1
    store = Store(db)
    assert store.get("29-92-68-C5").album == "Van Woede Tot Wanhoop"
    store.close()


def test_fetch_covers_stops_when_music_assistant_is_unreachable(store, library):
    import httpx
    import pytest

    def unreachable(url):
        raise httpx.ConnectTimeout("timed out")

    library.fetch_image = unreachable
    library.image_download_url = lambda url: url
    store.assign("A", "Tool", "Lateralus")
    store.assign("B", "Kiasmos", "Kiasmos")
    with pytest.raises(SystemExit, match="Set MA_URL"):
        fetch_covers(store, library)


def test_settings_read_ma_url(monkeypatch):
    from tag_albums.config import Settings

    for k, v in {"API_TOKEN": "t", "HA_URL": "https://ha/", "HA_TOKEN": "h", "MA_URL": "https://ma.example/"}.items():
        monkeypatch.setenv(k, v)
    assert Settings.from_env().ma_url == "https://ma.example"
    monkeypatch.delenv("MA_URL")
    assert Settings.from_env().ma_url is None