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 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", "uri": "library://album/1868"} 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 'Recently assigned' 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_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 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" 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 # --- 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 == []