Public Access
Let a tag play several albums in order
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>
This commit is contained in:
Claude Opus 5.5
+190
-92
@@ -8,6 +8,22 @@ 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
|
||||
@@ -16,54 +32,140 @@ def test_api_requires_token(client):
|
||||
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}
|
||||
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_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",
|
||||
"uri": "library://album/1868"}
|
||||
assert store.get("86-2C-1D-BD").scan_count == 1
|
||||
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):
|
||||
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}
|
||||
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
|
||||
tag = store.get("AA-BB")
|
||||
assert (tag.artist, tag.album, tag.image) == ("Tool", "Lateralus", None)
|
||||
(album,) = store.get("AA-BB").albums
|
||||
assert (album.artist, album.album, album.image) == ("Tool", "Lateralus", None)
|
||||
|
||||
client.post("/tags/AA-BB/unassign")
|
||||
assert not store.get("AA-BB").assigned
|
||||
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")
|
||||
home = client.get("/")
|
||||
assert "AA-BB" in home.text and "Lateralus" in home.text
|
||||
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
|
||||
|
||||
@@ -74,46 +176,53 @@ def test_htmx_is_served_locally(client):
|
||||
assert r.status_code == 200 and r.text.startswith("var htmx=")
|
||||
|
||||
|
||||
def test_search_results(client, library):
|
||||
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 _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):
|
||||
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"]
|
||||
assert form["image"] == "http://img/1" and form["sig"] and form["uri"] == "library://album/1868"
|
||||
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")
|
||||
(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 f'src="/tags/AA-BB/cover?v=' in client.get("/").text
|
||||
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"})
|
||||
tag = store.get("AA-BB")
|
||||
assert tag.image is None and not tag.has_cover and library.fetched == []
|
||||
(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").has_cover
|
||||
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").has_cover
|
||||
assert client.get("/tags/AA-BB/cover").status_code == 404
|
||||
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):
|
||||
@@ -124,9 +233,10 @@ def test_cover_preview_only_serves_signed_urls(client):
|
||||
assert client.get("/covers", params={"src": "http://evil/x"}).status_code == 403
|
||||
|
||||
|
||||
def test_assignments_sort(client, store):
|
||||
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")
|
||||
|
||||
@@ -135,7 +245,7 @@ def test_assignments_sort(client, store):
|
||||
|
||||
assert order("artist") == ["B", "A"]
|
||||
assert order("assigned") == ["A", "B"]
|
||||
assert order("scanned") == ["B", "A"] # never-scanned tags last
|
||||
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
|
||||
@@ -143,8 +253,37 @@ def test_assignments_sort(client, store):
|
||||
assert client.get("/", params={"sort": "bogus"}).status_code == 200
|
||||
|
||||
|
||||
def test_old_database_is_migrated(tmp_path):
|
||||
db = str(tmp_path / "old.sqlite3")
|
||||
# --- 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)")
|
||||
@@ -153,21 +292,29 @@ def test_old_database_is_migrated(tmp_path):
|
||||
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")
|
||||
(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.assign("B", "Amenra", "De Doorn") # not in the library
|
||||
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"]
|
||||
tag = store.get("A")
|
||||
assert tag.uri == "library://album/1868" and tag.has_cover and tag.image == "http://img/1"
|
||||
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) == ([], [])
|
||||
|
||||
@@ -193,52 +340,3 @@ def test_settings_read_ma_url(monkeypatch):
|
||||
assert Settings.from_env().ma_url == "https://ma.example"
|
||||
monkeypatch.delenv("MA_URL")
|
||||
assert Settings.from_env().ma_url is None
|
||||
|
||||
|
||||
# --- Playing by URI -----------------------------------------------------------
|
||||
|
||||
def _scan(client, tag_id):
|
||||
return client.post("/api/scans", json={"tag_id": tag_id}, headers=AUTH).json()
|
||||
|
||||
|
||||
def test_assign_from_search_stores_the_uri(client, store):
|
||||
client.post("/tags/AA-BB/assign", data=_search_form(client))
|
||||
assert store.get("AA-BB").uri == "library://album/1868"
|
||||
assert _scan(client, "AA-BB")["uri"] == "library://album/1868"
|
||||
|
||||
|
||||
def test_scan_links_names_only_assignment_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")["uri"] == "library://album/2846"
|
||||
assert store.get("MOB").uri == "library://album/2846"
|
||||
|
||||
|
||||
def test_scan_replaces_a_uri_the_library_no_longer_has(client, store, library):
|
||||
store.assign("A", "Tool", "Lateralus", uri="library://album/9999") # before a rebuild
|
||||
assert _scan(client, "A")["uri"] == "library://album/1868"
|
||||
assert store.get("A").uri == "library://album/1868"
|
||||
|
||||
|
||||
def test_scan_falls_back_to_names_when_nothing_matches(client, store, library):
|
||||
store.assign("A", "Amenra", "De Doorn", uri="library://album/9999")
|
||||
body = _scan(client, "A")
|
||||
assert body["uri"] is None and body["album"] == "De Doorn"
|
||||
assert store.get("A").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")["uri"] == "library://album/1868"
|
||||
|
||||
|
||||
def test_lookup_returns_stored_uri_without_searching(client, store, library):
|
||||
store.assign("A", "Tool", "Lateralus", uri="library://album/1868")
|
||||
r = client.get("/api/tags/A", headers=AUTH).json()
|
||||
assert r["uri"] == "library://album/1868" and library.queries == []
|
||||
|
||||
Reference in New Issue
Block a user