Public Access
twislaandClaude Opus 5.5
5878adb756
An assignment is now an ordered list of albums (e.g. one vinyl compiling two EPs that are separate releases in the library). Albums move to their own table, with a migration for existing databases. The scan and lookup responses return an albums list instead of the single artist/album/uri fields. The tag page lists its albums with up, down and remove, and search results get an Add button. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
343 lines
15 KiB
Python
343 lines
15 KiB
Python
import re
|
||
import sqlite3
|
||
|
||
from conftest import AUTH
|
||
|
||
from tag_albums.cli import import_yaml, sync_library
|
||
from tag_albums.library import Album
|
||
from tag_albums.store import Store
|
||
|
||
|
||
def _scan(client, tag_id):
|
||
return client.post("/api/scans", json={"tag_id": tag_id}, headers=AUTH).json()
|
||
|
||
|
||
def _names(tag):
|
||
return [(a.artist, a.album) for a in tag.albums]
|
||
|
||
|
||
def _search_form(client, tag_id="AA-BB"):
|
||
"""The hidden fields of the first search result's form."""
|
||
html = client.get(f"/tags/{tag_id}/search", params={"q": "lateralus"}).text
|
||
return dict(re.findall(r'name="(\w+)" value="([^"]*)"', html))
|
||
|
||
|
||
# --- API for Home Assistant -------------------------------------------------
|
||
|
||
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, "albums": []}
|
||
tag = store.get("53-13-0B-2A")
|
||
assert tag.scan_count == 1 and not tag.assigned
|
||
|
||
|
||
def test_scan_returns_albums_in_play_order(client, store):
|
||
store.assign("REC", "Tool", "Lateralus", uri="library://album/1868")
|
||
store.add_album("REC", "Amenra", "De Doorn")
|
||
assert _scan(client, "REC") == {
|
||
"tag_id": "REC",
|
||
"assigned": True,
|
||
"albums": [
|
||
{"artist": "Tool", "album": "Lateralus", "uri": "library://album/1868"},
|
||
{"artist": "Amenra", "album": "De Doorn", "uri": None}, # not in the library: by name
|
||
],
|
||
}
|
||
assert store.get("REC").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, library):
|
||
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, "albums": []}
|
||
assert store.get("AA-BB").scan_count == 1
|
||
|
||
store.assign("AA-BB", "Tool", "Lateralus", uri="library://album/9999")
|
||
albums = client.get("/api/tags/AA-BB", headers=AUTH).json()["albums"]
|
||
assert albums == [{"artist": "Tool", "album": "Lateralus", "uri": "library://album/9999"}]
|
||
assert library.queries == [] # stored URI, unchecked
|
||
|
||
|
||
# --- Playing by URI -----------------------------------------------------------
|
||
|
||
def test_scan_links_names_only_album_to_exact_spelling(client, store, library):
|
||
# The case that broke: two library entries differing only in capitals,
|
||
# and only the one spelled like the assignment is playable.
|
||
library.library = [
|
||
Album("The Mob", "Let The Tribe Increase", None, "library://album/1183"),
|
||
Album("The Mob", "Let the Tribe Increase", None, "library://album/2846"),
|
||
]
|
||
store.assign("MOB", "The Mob", "Let the Tribe Increase")
|
||
assert _scan(client, "MOB")["albums"][0]["uri"] == "library://album/2846"
|
||
assert store.get("MOB").albums[0].uri == "library://album/2846"
|
||
|
||
|
||
def test_scan_replaces_a_uri_the_library_no_longer_has(client, store):
|
||
store.assign("A", "Tool", "Lateralus", uri="library://album/9999") # before a rebuild
|
||
assert _scan(client, "A")["albums"][0]["uri"] == "library://album/1868"
|
||
assert store.get("A").albums[0].uri == "library://album/1868"
|
||
|
||
|
||
def test_scan_falls_back_to_names_when_nothing_matches(client, store):
|
||
store.assign("A", "Amenra", "De Doorn", uri="library://album/9999")
|
||
album = _scan(client, "A")["albums"][0]
|
||
assert album["uri"] is None and album["album"] == "De Doorn"
|
||
assert store.get("A").albums[0].uri == "library://album/9999" # kept, in case it comes back
|
||
|
||
|
||
def test_scan_trusts_stored_uri_when_library_is_unreachable(client, store, library):
|
||
store.assign("A", "Tool", "Lateralus", uri="library://album/1868")
|
||
library.fail_search = True
|
||
assert _scan(client, "A")["albums"][0]["uri"] == "library://album/1868"
|
||
|
||
|
||
# --- Web page -----------------------------------------------------------------
|
||
|
||
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
|
||
(album,) = store.get("AA-BB").albums
|
||
assert (album.artist, album.album, album.image) == ("Tool", "Lateralus", None)
|
||
|
||
client.post("/tags/AA-BB/unassign")
|
||
tag = store.get("AA-BB")
|
||
assert not tag.assigned and tag.assigned_at is None
|
||
|
||
client.post("/tags/AA-BB/delete")
|
||
assert store.get("AA-BB") is None
|
||
|
||
|
||
def test_add_puts_albums_after_the_others(client, store):
|
||
client.post("/tags/REC/assign", data={"artist": "Amenra", "album": "EP 1"})
|
||
r = client.post("/tags/REC/add", data={"artist": "Amenra", "album": "EP 2"}, follow_redirects=False)
|
||
assert r.headers["location"] == "/tags/REC"
|
||
client.post("/tags/REC/add", data={"artist": "Amenra", "album": "EP 3"})
|
||
assert [a.album for a in store.get("REC").albums] == ["EP 1", "EP 2", "EP 3"]
|
||
|
||
# Assign replaces the whole list.
|
||
client.post("/tags/REC/assign", data={"artist": "Tool", "album": "Lateralus"})
|
||
assert _names(store.get("REC")) == [("Tool", "Lateralus")]
|
||
|
||
|
||
def test_reorder_and_remove(client, store):
|
||
for name in ("EP 1", "EP 2", "EP 3"):
|
||
store.add_album("REC", "Amenra", name)
|
||
|
||
client.post("/tags/REC/albums/2/up")
|
||
assert [a.album for a in store.get("REC").albums] == ["EP 1", "EP 3", "EP 2"]
|
||
client.post("/tags/REC/albums/0/down")
|
||
assert [a.album for a in store.get("REC").albums] == ["EP 3", "EP 1", "EP 2"]
|
||
client.post("/tags/REC/albums/0/up") # already first: no-op
|
||
client.post("/tags/REC/albums/2/down") # already last: no-op
|
||
assert [a.album for a in store.get("REC").albums] == ["EP 3", "EP 1", "EP 2"]
|
||
|
||
client.post("/tags/REC/albums/0/remove")
|
||
tag = store.get("REC")
|
||
assert [(a.position, a.album) for a in tag.albums] == [(0, "EP 1"), (1, "EP 2")]
|
||
|
||
client.post("/tags/REC/albums/1/remove")
|
||
client.post("/tags/REC/albums/0/remove")
|
||
tag = store.get("REC")
|
||
assert not tag.assigned and tag.assigned_at is None # back to unassigned
|
||
assert "REC" in client.get("/unassigned").text
|
||
|
||
|
||
def test_pages_render(client, store):
|
||
store.record_scan("AA-BB")
|
||
store.assign("CC-DD", "Tool", "Lateralus")
|
||
store.add_album("CC-DD", "Amenra", "De Doorn")
|
||
home = client.get("/").text
|
||
assert "AA-BB" in home
|
||
assert re.search(r"Lateralus</strong> · Tool\s*<span class=\"muted\">\+</span>\s*<strong>De Doorn", home)
|
||
assert "AA-BB" in client.get("/unassigned").text
|
||
page = client.get("/tags/CC-DD").text
|
||
assert "Plays, in this order" in page and page.index("Lateralus") < page.index("De Doorn")
|
||
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, store, 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
|
||
# Add only makes sense once the tag has an album.
|
||
assert 'formaction="/tags/AA-BB/add"' not in r.text
|
||
store.assign("AA-BB", "Amenra", "De Doorn")
|
||
assert 'formaction="/tags/AA-BB/add"' in client.get("/tags/AA-BB/search", params={"q": "x"}).text
|
||
|
||
|
||
def test_assign_from_search_stores_cover_and_uri(client, store):
|
||
store.record_scan("AA-BB")
|
||
form = _search_form(client)
|
||
assert form["image"] == "http://img/1" and form["sig"] and form["uri"] == "library://album/1868"
|
||
client.post("/tags/AA-BB/assign", data=form)
|
||
|
||
(album,) = store.get("AA-BB").albums
|
||
assert album.has_cover and album.image == "http://img/1" and album.uri == "library://album/1868"
|
||
r = client.get("/tags/AA-BB/albums/0/cover")
|
||
assert (r.status_code, r.content, r.headers["content-type"]) == (200, b"\x89PNG-lateralus", "image/png")
|
||
assert 'src="/tags/AA-BB/albums/0/cover?v=' in client.get("/").text
|
||
|
||
|
||
def test_added_album_gets_its_own_cover(client, store):
|
||
client.post("/tags/REC/assign", data={"artist": "Amenra", "album": "De Doorn"})
|
||
client.post("/tags/REC/add", data=_search_form(client, "REC"))
|
||
first, second = store.get("REC").albums
|
||
assert not first.has_cover and second.has_cover
|
||
assert client.get("/tags/REC/albums/1/cover").status_code == 200
|
||
assert client.get("/tags/REC/albums/0/cover").status_code == 404
|
||
|
||
|
||
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"})
|
||
(album,) = store.get("AA-BB").albums
|
||
assert album.image is None and not album.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").albums[0].has_cover
|
||
client.post("/tags/AA-BB/assign", data={"artist": "Mogwai", "album": "Come On Die Young"})
|
||
assert not store.get("AA-BB").albums[0].has_cover
|
||
assert client.get("/tags/AA-BB/albums/0/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_by_first_album(client, store):
|
||
store.assign("A", "Tool", "Lateralus")
|
||
store.assign("B", "Amenra", "De Doorn")
|
||
store.add_album("B", "Zzz", "Last") # only the first album counts
|
||
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
|
||
|
||
|
||
# --- Storage, import, sync ----------------------------------------------------
|
||
|
||
def test_database_from_before_multiple_albums_is_migrated(tmp_path):
|
||
db = str(tmp_path / "v1.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, uri TEXT,"
|
||
" assigned_at TEXT, cover BLOB, cover_type TEXT)")
|
||
conn.execute("INSERT INTO tags VALUES ('AA', '2026-01-01', '2026-02-01', 3, 'Tool', 'Lateralus',"
|
||
" 'http://img/1', 'library://album/1868', '2026-01-02', X'89', 'image/png')")
|
||
conn.execute("INSERT INTO tags (tag_id, first_seen) VALUES ('BB', '2026-01-01')")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
store = Store(db)
|
||
tag = store.get("AA")
|
||
assert (tag.scan_count, tag.last_seen, tag.assigned_at) == (3, "2026-02-01", "2026-01-02")
|
||
(album,) = tag.albums
|
||
assert (album.position, album.artist, album.album, album.image, album.uri, album.has_cover) == (
|
||
0, "Tool", "Lateralus", "http://img/1", "library://album/1868", True)
|
||
assert store.get_cover("AA", 0) == (b"\x89", "image/png")
|
||
assert not store.get("BB").assigned
|
||
store.close()
|
||
|
||
store = Store(db) # reopening doesn't migrate again
|
||
assert len(store.get("AA").albums) == 1
|
||
store.close()
|
||
|
||
|
||
def test_even_older_database_without_uri_or_cover_is_migrated(tmp_path):
|
||
db = str(tmp_path / "v0.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)
|
||
(album,) = store.get("AA").albums
|
||
assert album.album == "Lateralus" and album.uri is None and not album.has_cover
|
||
store.close()
|
||
|
||
|
||
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 _names(store.get("29-92-68-C5")) == [("ONRUST", "Van Woede Tot Wanhoop")]
|
||
store.close()
|
||
|
||
|
||
def test_sync_library_links_and_fetches_covers(store, library):
|
||
store.assign("A", "Tool", "Lateralus") # imported: names only
|
||
store.add_album("A", "Amenra", "De Doorn") # not in the library
|
||
report = sync_library(store, library)
|
||
assert report.linked == ["Tool – Lateralus"] and report.unmatched == ["Amenra – De Doorn"]
|
||
assert report.covers == ["Tool – Lateralus"] and report.coverless == ["Amenra – De Doorn"]
|
||
first = store.get("A").albums[0]
|
||
assert first.uri == "library://album/1868" and first.has_cover and first.image == "http://img/1"
|
||
again = sync_library(store, library) # done ones are skipped
|
||
assert (again.linked, again.covers) == ([], [])
|
||
|
||
|
||
def test_sync_library_stops_when_music_assistant_is_unreachable(store, library):
|
||
import httpx
|
||
import pytest
|
||
|
||
def unreachable(url):
|
||
raise httpx.ConnectTimeout("timed out")
|
||
|
||
library.fetch_image = unreachable
|
||
store.assign("A", "Tool", "Lateralus")
|
||
with pytest.raises(SystemExit, match="Set MA_URL"):
|
||
sync_library(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
|