"""Tests für die Gastseite: Einladungszugang, Suche, Wünschen mit Schutzregeln, Status.""" import pytest from fastapi.testclient import TestClient from app import config, db, gaeste from app.spotify import SpotifyFehler, spotify URI = {k: "spotify:track:" + k * 22 for k in "ABCDE"} KUENSTLER_GESPERRT = "spotify:artist:" + "Z" * 22 def titel(k, explicit=False, kuenstler="spotify:artist:" + "X" * 22): return {"track_uri": URI[k], "title": f"Lied {k}", "artist": f"Band {k}", "artist_uris": [kuenstler], "album": "Album", "cover_url": "https://i.scdn.co/image/x", "duration_ms": 200_000, "explicit": explicit} class FakeSpotify: def __init__(self): self.titel_db = {URI["A"]: titel("A"), URI["B"]: titel("B"), URI["C"]: titel("C", explicit=True), URI["D"]: titel("D", kuenstler=KUENSTLER_GESPERRT), URI["E"]: titel("E")} def verfuegbar(self): return True def suche(self, q): return list(self.titel_db.values()) def titel(self, uri): if uri not in self.titel_db: raise SpotifyFehler(404) return dict(self.titel_db[uri]) @pytest.fixture def client(tmp_path, monkeypatch): env = tmp_path / ".env" env.write_text(f"DB_PATH={tmp_path / 'audiofy.db'}\nSESSION_SECRET=geheim\nCOOKIE_SECURE=0\n") monkeypatch.setenv("AUDIOFY_ENV", str(env)) config.konfig.cache_clear() for name in ("suche_drossel", "wunsch_drossel", "einladung_drossel"): monkeypatch.setattr(gaeste, name, gaeste.Drossel(1000, 60)) from app.main import app app.dependency_overrides[spotify] = FakeSpotify with TestClient(app) as c: yield c app.dependency_overrides.clear() config.konfig.cache_clear() def setzen(**werte): conn = db.verbinden() for k, v in werte.items(): db.einstellung_setzen(conn, k, v) conn.close() def token(): conn = db.verbinden() t = db.einstellungen(conn)["access_token"] conn.close() return t def eintreten(client): """Zugang einschalten und Einladungslink öffnen.""" setzen(access_enabled=True) r = client.get(f"/z/{token()}", follow_redirects=False) assert r.status_code == 303 and r.headers["location"] == "/" return client def sql(abfrage, *params): conn = db.verbinden() with conn: zeilen = conn.execute(abfrage, params).fetchall() conn.close() return zeilen # --- Einladungszugang ------------------------------------------------------ def test_ohne_einladung_kein_zugang(client): r = client.get("/") assert r.status_code == 403 and "Nur für Gäste" in r.text assert client.get("/api/status").status_code == 401 assert client.get("/api/search?q=ab").status_code == 401 assert client.post("/api/wish", json={"track_uri": URI["A"]}).status_code == 401 def test_token_vorhanden_und_zugang_anfangs_aus(client): assert len(token()) >= 12 assert client.get(f"/z/{token()}", follow_redirects=False).status_code == 403 # access_enabled = False def test_einladung_gibt_zugang(client): eintreten(client) r = client.get("/") assert r.status_code == 200 and "Song wünschen" in r.text assert client.get("/api/status").status_code == 200 def test_falscher_link(client): setzen(access_enabled=True) assert client.get("/z/falsch", follow_redirects=False).status_code == 403 assert client.get("/api/status").status_code == 401 def test_zugang_aus_sperrt_alle(client): eintreten(client) setzen(access_enabled=False) assert client.get("/api/status").status_code == 401 assert client.get("/").status_code == 403 def test_neuer_link_macht_alte_zugaenge_ungueltig(client): eintreten(client) alt = token() setzen(access_token=db.neuer_zugangs_token()) assert client.get("/api/status").status_code == 401 assert client.get(f"/z/{alt}", follow_redirects=False).status_code == 403 # --- Suche ------------------------------------------------------------------- def test_suche_markiert_gesperrte(client): eintreten(client) setzen(block_explicit=True) sql("INSERT INTO blocklist (kind, value, created_at) VALUES ('artist', ?, 'x')", KUENSTLER_GESPERRT) client.post("/api/wish", json={"track_uri": URI["B"]}) treffer = {t["track_uri"][-1]: t["blocked"] for t in client.get("/api/search?q=lied").json()["results"]} assert treffer["A"] is None assert "Wunschliste" in treffer["B"] assert "explizit" in treffer["C"] assert "nicht gewünscht" in treffer["D"] assert "artist_uris" not in client.get("/api/search?q=lied").json()["results"][0] def test_suche_mindestlaenge(client): eintreten(client) assert client.get("/api/search?q=a").status_code == 422 # --- Wünschen ---------------------------------------------------------------- def test_wunsch_landet_in_warteschlange(client): eintreten(client) r = client.post("/api/wish", json={"track_uri": URI["A"], "name": " Tante Erna "}) assert r.status_code == 200 and r.json()["status"] == "queued" w = sql("SELECT title, guest_name, guest_ip, guest_id FROM wishes")[0] assert (w["title"], w["guest_name"]) == ("Lied A", "Tante Erna") assert w["guest_ip"] != "testclient" and len(w["guest_ip"]) == 24 # nur Hash s = client.get("/api/status").json() assert [q["title"] for q in s["queue"]] == ["Lied A"] and s["queue"][0]["mine"] is True assert s["mine"][0]["status"] == "queued" and s["limit"]["remaining"] == 1 def test_titelinfos_kommen_vom_server(client): eintreten(client) client.post("/api/wish", json={"track_uri": URI["A"], "title": "Gefälscht"}) assert sql("SELECT title FROM wishes")[0]["title"] == "Lied A" def test_moderation(client): eintreten(client) setzen(moderation=True) r = client.post("/api/wish", json={"track_uri": URI["A"]}) assert r.json()["status"] == "pending" s = client.get("/api/status").json() assert s["queue"] == [] and s["pending_count"] == 1 def test_pausiert(client): eintreten(client) setzen(wishes_paused=True) r = client.post("/api/wish", json={"track_uri": URI["A"]}) assert r.status_code == 403 and "pausiert" in r.json()["detail"] def test_limit_pro_gast(client): eintreten(client) assert client.post("/api/wish", json={"track_uri": URI["A"]}).status_code == 200 assert client.post("/api/wish", json={"track_uri": URI["B"]}).status_code == 200 r = client.post("/api/wish", json={"track_uri": URI["E"]}) assert r.status_code == 429 and "Limit" in r.json()["detail"] def test_keine_duplikate(client): eintreten(client) setzen(wish_limit_count=10) client.post("/api/wish", json={"track_uri": URI["A"]}) r = client.post("/api/wish", json={"track_uri": URI["A"]}) assert r.status_code == 409 and "Wunschliste" in r.json()["detail"] def test_kuerzlich_gespielt_und_laeuft_gerade(client): eintreten(client) setzen(wish_limit_count=10) sql("INSERT INTO history (track_uri, title, artist, started_at) VALUES (?, 'x', 'y', ?)", URI["A"], db.jetzt()) sql("UPDATE pi_state SET track_uri = ? WHERE id = 1", URI["B"]) assert client.post("/api/wish", json={"track_uri": URI["A"]}).status_code == 409 assert client.post("/api/wish", json={"track_uri": URI["B"]}).status_code == 409 alt = "2020-01-01T00:00:00+00:00" # außerhalb der Sperrzeit → wieder erlaubt sql("UPDATE history SET started_at = ?", alt) assert client.post("/api/wish", json={"track_uri": URI["A"]}).status_code == 200 def test_explizit_und_sperrliste(client): eintreten(client) setzen(block_explicit=True, wish_limit_count=10) assert client.post("/api/wish", json={"track_uri": URI["C"]}).status_code == 409 sql("INSERT INTO blocklist (kind, value, created_at) VALUES ('artist', ?, 'x')", KUENSTLER_GESPERRT) assert client.post("/api/wish", json={"track_uri": URI["D"]}).status_code == 409 sql("INSERT INTO blocklist (kind, value, created_at) VALUES ('track', ?, 'x')", URI["E"]) assert client.post("/api/wish", json={"track_uri": URI["E"]}).status_code == 409 def test_gesperrter_gast(client): eintreten(client) gid = client.cookies.get(gaeste.GAST_COOKIE) sql("INSERT INTO blocklist (kind, value, created_at) VALUES ('guest', ?, 'x')", gid) r = client.post("/api/wish", json={"track_uri": URI["A"]}) assert r.status_code == 403 and "Gerät" in r.json()["detail"] def test_ungueltige_eingaben(client): eintreten(client) assert client.post("/api/wish", json={"track_uri": "spotify:album:x"}).status_code == 422 assert client.post("/api/wish", json={"track_uri": URI["A"], "name": "x" * 31}).status_code == 422 unbekannt = "spotify:track:" + "Q" * 22 assert client.post("/api/wish", json={"track_uri": unbekannt}).status_code == 404 def test_fremde_herkunft(client): eintreten(client) r = client.post("/api/wish", json={"track_uri": URI["A"]}, headers={"Origin": "https://boese.example"}) assert r.status_code == 403 # --- Status ------------------------------------------------------------------ def test_laeuft_gerade_und_offline(client): eintreten(client) assert client.get("/api/status").json()["now_playing"] is None sql("UPDATE pi_state SET payload = ?, updated_at = ?, last_seen = ? WHERE id = 1", '{"track": {"track_uri": "%s", "title": "Lied A", "artist": "Band"}, "is_playing": true, ' '"progress_ms": 5000, "source": "wish"}' % URI["A"], db.jetzt(), db.jetzt()) jetzt = client.get("/api/status").json()["now_playing"] assert jetzt["title"] == "Lied A" and jetzt["is_wish"] is True sql("UPDATE pi_state SET last_seen = '2020-01-01T00:00:00+00:00'") # Player seit Langem still → offline s = client.get("/api/status").json() assert s["online"] is False and s["now_playing"] is None def test_sicherheits_header(client): r = client.get("/") assert "default-src 'self'" in r.headers["content-security-policy"]