Serve album covers from the service, show and sort by dates

Covers are stored with the assignment and served by the service, since Music Assistant's plain-http image URLs are blocked on an https page. Search results load covers through a signed /covers proxy. Adds a fetch-covers command for imported assignments, a migration for existing databases, and sorting assignments by artist, recently assigned or recently scanned, with both dates shown. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-25 21:21:05 +00:00
parent 8b8a3f6184
commit 1abba34cfe
11 changed files with 388 additions and 28 deletions
@@ -13,11 +13,23 @@ AUTH = {"Authorization": f"Bearer {TOKEN}"}
 class FakeLibrary:
    def __init__(self):
        self.queries = []
+        self.fetched = []
+        self.images = {"http://img/1": (b"\x89PNG-lateralus", "image/png")}

    def search_albums(self, query, limit=12):
        self.queries.append(query)
        return [Album(artist="Tool", album="Lateralus", image="http://img/1")]

+    def find_album(self, artist, album):
+        found = self.search_albums(album)[0]
+        return found if (found.artist, found.album) == (artist, album) else None
+
+    def fetch_image(self, url):
+        self.fetched.append(url)
+        if url not in self.images:
+            raise ValueError("no such image")
+        return self.images[url]
+

@pytest.fixture
 def store(tmp_path):
@@ -1,6 +1,9 @@
+import re
+import sqlite3
+
 from conftest import AUTH

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


@@ -73,6 +76,96 @@ 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):
@@ -1,8 +1,9 @@
 import json

 import httpx
+import pytest

-from tag_albums.library import MusicLibrary
+from tag_albums.library import Album, MusicLibrary


 def test_search_albums_parses_music_assistant_response():
@@ -30,3 +31,35 @@ def test_search_albums_parses_music_assistant_response():
    library.search_albums("lateralus")
    # The Music Assistant config entry is looked up once, then cached.
    assert sum(c.url.path == "/api/config/config_entries/entry" for c in calls) == 1
+
+
+def _library_with_images(responses):
+    seen = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        return responses[str(request.url)]
+
+    images = httpx.Client(transport=httpx.MockTransport(handler))
+    return MusicLibrary("http://ha", "ha-token", client=httpx.Client(), image_client=images), seen
+
+
+def test_fetch_image_checks_type_and_never_sends_ha_token():
+    library, seen = _library_with_images({
+        "http://ma/cover": httpx.Response(200, content=b"jpg", headers={"content-type": "image/jpeg"}),
+        "http://ma/page": httpx.Response(200, content=b"<html>", headers={"content-type": "text/html"}),
+    })
+    assert library.fetch_image("http://ma/cover") == (b"jpg", "image/jpeg")
+    assert "authorization" not in seen[0].headers
+    with pytest.raises(ValueError):
+        library.fetch_image("http://ma/page")
+
+
+def test_find_album_needs_exact_names():
+    library = MusicLibrary("http://ha", "t", client=httpx.Client())
+    library.search_albums = lambda query, limit=12: [
+        Album("Tool", "Lateralus (Live)", "http://img/live"),
+        Album("TOOL", "lateralus", "http://img/1"),
+    ]
+    assert library.find_album("Tool", "Lateralus").image == "http://img/1"
+    assert library.find_album("Mogwai", "Lateralus") is None