#!/usr/bin/env python3 """Reproducible test suite for the Deemix MCP server. Run: python test_deemix_mcp.py (from the package directory) The suite spins up a local HTTP server that emulates the Deemix webui API (connect / search / getQueue / addToQueue / loginArl / removeFromQueue / retryDownload / removeFinishedDownloads / cancelAllDownloads) and points the MCP at it. No real Deezer access, no real downloads, no queue residue: * read + auth flow (fresh first call, one-shot re-login on NotLoggedIn, 401, 403, re-login failure abort) * bitrate validation (valid TrackFormats ids, invalid rejected, not kbit/s) * url/term/result limits * invalid url/id handling * compact queue output * secret redaction (ARL, cookie headers, long hex) Write-path tools that would enqueue real downloads are verified here ONLY against the emulated backend; against production Deemix they are considered verified against the source code, not practically executed. """ from __future__ import annotations import json import os import sys import threading import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Optional HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) # -------------------------------------------------------------------------- # Emulated Deemix backend # -------------------------------------------------------------------------- FAKE_ARL = "a1" * 96 # 192 hex chars, like a real Deezer ARL class DeemixState: def __init__(self) -> None: self.logged_in = False self.connect_calls = 0 self.login_calls = 0 self.requests: list[str] = [] self.fail_next_relogin = False self.force_401_once = False self.force_403_once = False self.queue = { "album_1_1": { "uuid": "album_1_1", "type": "album", "id": "1", "title": "Folge 240", "artist": "Die drei ???", "progress": 37, "downloaded": 52428800, "size": 80530636, "failed": False, "errors": [], }, "album_2_3": { "uuid": "album_2_3", "type": "album", "id": "2", "title": "Folge 239", "artist": "Die drei ???", "progress": 100, "downloaded": 76000000, "size": 76000000, "failed": True, "errors": ["Tagging failed: unknown picture MIME", "second error", "third error", "fourth error"], }, } self.added: list[dict] = [] self.cancelled_all = False STATE = DeemixState() class EmulatedDeemix(BaseHTTPRequestHandler): def log_message(self, *a): # quiet pass def _send(self, obj: dict, status: int = 200) -> None: body = json.dumps(obj).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _auth_required(self) -> bool: """True when the session cookie is missing (not logged in).""" cookie = self.headers.get("Cookie", "") return "deemix_session" not in cookie def do_GET(self): STATE.requests.append("GET " + self.path) if self.path.startswith("/deemix/api/connect"): STATE.connect_calls += 1 # In real Deemix the session IS the cookie: without it the # server reports singleUser + the ARL again, so a client can # re-login after any session loss. authed = not self._auth_required() obj: dict = { "update": {"deemixVersion": "3.13.7", "webuiVersion": "4.6.0"}, "autologin": not authed, "currentUser": None, "deezerAvailable": "yes", "spotifyEnabled": True, "settingsData": {"settings": { "downloadLocation": "/downloads", "maxBitrate": 9, }}, } if not authed: obj["singleUser"] = {"arl": FAKE_ARL} else: obj["currentUser"] = {"id": 12345, "name": "Test", "lastName": "User", "country": "DE", "subscription": "premium"} self._send(obj) return if self._force_auth_error(): return if self._auth_required(): self._send({"result": False, "errid": "NotLoggedIn"}) return if self.path.startswith("/deemix/api/search"): q = urllib.parse.parse_qs(self.path.split("?", 1)[1]) if "?" in self.path else {} nb = int(q.get("nb", ["10"])[0]) total = 31 data = [{"id": 985501051, "title": "Die drei ??? Folge 240", "release_date": "2026-08-18", "nb_tracks": 48, "related": {"artists": [{"name": "Die drei ???"}, {"name": "Mystery-Club"}]}}] data += [{"id": i, "title": f"Treffer {i}", "release_date": "2020-01-01", "nb_tracks": 10, "related": {"artists": [{"name": "K"}]}} for i in range(1, total)] self._send({"data": data[:nb], "total": total, "type": "album", "error": ""}) return if self.path.startswith("/deemix/api/getQueue"): self._send({"queue": dict(STATE.queue), "queueOrder": list(STATE.queue.keys())}) return self._send({"result": False, "errid": "NotFound"}, 404) def _force_auth_error(self) -> bool: """Emulate a session that the server rejects with HTTP 401/403.""" if STATE.force_401_once: STATE.force_401_once = False self._send({"result": False, "errid": "http401"}, 401) return True if STATE.force_403_once: STATE.force_403_once = False self._send({"result": False, "errid": "http403"}, 403) return True return False def do_POST(self): length = int(self.headers.get("Content-Length", 0)) raw = self.rfile.read(length).decode() if length else "" try: body = json.loads(raw) if raw else {} except json.JSONDecodeError: body = {} STATE.requests.append("POST " + self.path) if self.path.startswith("/deemix/api/loginArl"): STATE.login_calls += 1 if STATE.fail_next_relogin: STATE.fail_next_relogin = False self._send({"status": 0, "error": "invalidArl"}) return if body.get("arl") != FAKE_ARL: self._send({"status": 0, "error": "invalidArl"}) return STATE.logged_in = True # Emulate setting the session cookie (real Deemix sets a cookie). self.send_response(200) self.send_header("Set-Cookie", "deemix_session=abc123; Path=/") self.send_header("Content-Type", "application/json") payload = json.dumps({"status": 1, "user": {"id": 12345, "name": "Test", "lastName": "User", "country": "DE"}}) self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload.encode()) return if self._auth_required(): self._send({"result": False, "errid": "NotLoggedIn"}) return if self._force_auth_error(): return if self.path.startswith("/deemix/api/addToQueue"): url = body.get("url", "") bitrate = body.get("bitrate") # Mirror the real route: empty/"null" falls back to settings. if bitrate == "null" or not bitrate: bitrate = 9 bitrate = int(bitrate) if "badid" in url or "9999999999" in url: # Real Deemix behaviour: an unknown ID is accepted (result # true) but resolves to zero items, so nothing is queued. self._send({"result": True, "data": { "url": [u for u in url.split()], "bitrate": bitrate, "obj": []}}) return STATE.added.append({"url": url, "bitrate": bitrate}) ids = [u.rsplit("/", 1)[-1] for u in url.split()] self._send({"result": True, "data": { "url": ids, "bitrate": bitrate, "obj": [{"id": i, "type": "album"} for i in ids]}}) return if self.path.startswith("/deemix/api/removeFromQueue"): uuid = urllib.parse.parse_qs(self.path.split("?", 1)[1]).get("uuid", [""])[0] if uuid not in STATE.queue: self._send({"result": False, "errid": "UnknownUuid"}) return del STATE.queue[uuid] self._send({"result": True}) return if self.path.startswith("/deemix/api/retryDownload"): uuid = body.get("uuid", "") if uuid not in STATE.queue: self._send({"result": False, "errid": "UnknownUuid"}) return STATE.queue[uuid]["failed"] = False self._send({"result": True}) return if self.path.startswith("/deemix/api/removeFinishedDownloads"): for k in list(STATE.queue): if not STATE.queue[k]["failed"]: del STATE.queue[k] self._send({"result": True}) return if self.path.startswith("/deemix/api/cancelAllDownloads"): STATE.cancelled_all = True self._send({"result": True}) return self._send({"result": False, "errid": "NotFound"}, 404) # -------------------------------------------------------------------------- # Tests # -------------------------------------------------------------------------- import urllib.parse # noqa: E402 (used by the emulated handler above) PASS: list[str] = [] FAIL: list[str] = [] def check(name: str, cond: bool, detail: str = "") -> None: (PASS if cond else FAIL).append(name) print(("PASS " if cond else "FAIL ") + name + (f" [{detail}]" if detail and not cond else "")) def main() -> int: server = ThreadingHTTPServer(("127.0.0.1", 0), EmulatedDeemix) port = server.server_address[1] threading.Thread(target=server.serve_forever, daemon=True).start() os.environ["DEEMIX_URL"] = f"http://127.0.0.1:{port}" os.environ["DEEMIX_ARL"] = "" # force single-user path # fresh import so module-level env reads pick up the emulated URL if "deemix_mcp" in sys.modules: del sys.modules["deemix_mcp"] import importlib import deemix_mcp importlib.reload(deemix_mcp) from deemix_mcp import (Session, DeemixError, redact, _clean_error, deemix_status, deemix_search, deemix_add_to_queue, deemix_queue, deemix_retry_download, deemix_remove_from_queue, TRACKFORMATS, MAX_URLS_PER_CALL, MAX_SEARCH_RESULTS, MAX_SEARCH_TERM) S = deemix_mcp.SESSION S._logged_in = False S.jar.clear() # 1) fresh first call right after "container start" -------------------- out = json.loads(deemix_status()) check("erstaufruf: status nach frischem Start", out["logged_in_user"].get("name") == "Test", str(out.get("logged_in_user"))) check("erstaufruf: ARL nicht in Status-Output", FAKE_ARL not in json.dumps(out)) check("erstaufruf: genau ein Login-Versuch", STATE.login_calls == 1, f"login_calls={STATE.login_calls}") # 2) direct first call of search (session already warm) ----------------- out = json.loads(deemix_search("Die drei ??? Folge 240")) check("erstaufruf: search funktioniert", out["total"] == 31 and len(out["results"]) == 10) check("search: Deezer-Link praeSENT", out["results"][0]["link"] == "https://www.deezer.com/album/985501051") out = json.loads(deemix_queue()) check("erstaufruf: queue funktioniert", out["total"] == 2) check("queue: kompakt (keine groben Rohfelder)", all("downloaded_mb" in r for r in out["rows"]) and all("size" not in r for r in out["rows"])) check("queue: errors auf 3 begrenzt", [r for r in out["rows"] if r["uuid"] == "album_2_3"][0]["errors"] and len([r for r in out["rows"] if r["uuid"] == "album_2_3"][0]["errors"]) == 3) # 3) automatic ONE re-login on NotExpired session ------------------------ S._logged_in = True S.jar.clear() # server now sees no session cookie -> NotLoggedIn out = json.loads(deemix_queue()) check("re-login: NotLoggedIn wird automatisch einxmal behoben", out["total"] == 2) check("re-login: genau EIN zusaeztlicher Login", STATE.login_calls == 2, f"login_calls={STATE.login_calls}") # 4) HTTP 401 triggers the same bounded re-login ------------------------- STATE.force_401_once = True S._logged_in = True out = json.loads(deemix_queue()) check("re-login: HTTP 401 wird automatisch einxmal behoben", out["total"] == 2 and STATE.login_calls == 3) # 5) HTTP 403 likewise ---------------------------------------------------- STATE.force_403_once = True out = json.loads(deemix_queue()) check("re-login: HTTP 403 wird automatisch einxmal behoben", out["total"] == 2 and STATE.login_calls == 4) # 6) failed re-login aborts (no loop) ------------------------------------ S._logged_in = False S.jar.clear() STATE.fail_next_relogin = True try: deemix_queue() check("re-login: fehlgeschlagener Re-Login bricht ab", False, "kein Fehler geworfen") except DeemixError as e: ok = "re-login" in str(e).lower() or "session" in str(e).lower() check("re-login: fehlgeschlagener Re-Login bricht ab", ok, str(e)) check("re-login: kein Endlosschleifen-Verhalten (max. 1 Versuch)", STATE.login_calls == 5, f"login_calls={STATE.login_calls}") # restore a working session for the remaining tests S._logged_in = False S.jar.clear() deemix_status() # 7) bitrate validation ---------------------------------------------------- out = json.loads(deemix_add_to_queue( "https://www.deezer.com/album/985501051", bitrate=3)) check("bitrate: 3 (MP3 320) akzeptiert und uebergeben", out["queued"] and out["bitrate_used"] == 3) out = json.loads(deemix_add_to_queue( "https://www.deezer.com/album/985501051", bitrate=9)) check("bitrate: 9 (FLAC) akzeptiert", out["queued"] and out["bitrate_used"] == 9) out = json.loads(deemix_add_to_queue( "https://www.deezer.com/album/985501051", bitrate=1)) check("bitrate: 1 (MP3 128) akzeptiert", out["queued"] and out["bitrate_used"] == 1) out = json.loads(deemix_add_to_queue("https://www.deezer.com/album/985501051")) check("bitrate: weggelassen -> Deemix-Setting (9) greift", out["bitrate_requested"] is None and out["bitrate_used"] == 9) try: deemix_add_to_queue("https://www.deezer.com/album/985501051", bitrate=320) check("bitrate: 320 (kbit/s-Irrtum) wird abgelehnt", False, "kein Fehler") except DeemixError as e: check("bitrate: 320 (kbit/s-Irrtum) wird abgelehnt", "320" in str(e) and "not a valid" in str(e), str(e)) try: deemix_add_to_queue("https://www.deezer.com/album/985501051", bitrate=0) check("bitrate: 0 (LOCAL) wird abgelehnt", False, "kein Fehler") except DeemixError: check("bitrate: 0 (LOCAL) wird abgelehnt", True) check("bitrate: nur die WebUI-Werte 1/3/9 sind gueltig", set(TRACKFORMATS) == {1, 3, 9}, str(TRACKFORMATS)) # 8) url limits ----------------------------------------------------------- many = " ".join(f"https://www.deezer.com/album/{i}" for i in range(MAX_URLS_PER_CALL + 1)) try: deemix_add_to_queue(many) check("limit: >10 URLs abgelehnt", False, "kein Fehler") except DeemixError as e: check("limit: >10 URLs abgelehnt", "11" in str(e), str(e)) ok10 = " ".join(f"https://www.deezer.com/album/{i}" for i in range(10)) out = json.loads(deemix_add_to_queue(ok10)) check("limit: genau 10 URLs akzeptiert", out["queued"]) # 9) search limits ---------------------------------------------------------- out = json.loads(deemix_search("x" * (MAX_SEARCH_TERM + 50), nb=999)) check("limit: Suchtext wird auf 100 Zeichen gekuerzt", out["total"] == 31) # emulated server answers regardless; no crash out = json.loads(deemix_search("test", nb=999)) check("limit: nb wird auf 20 begrenzt", len(out["results"]) <= MAX_SEARCH_RESULTS) # 10) invalid url / id -------------------------------------------------------- # Real Deemix behaviour: unknown IDs are accepted but resolve to zero # items -> the tool reports queued=false with a warning and nothing is # queued (verified against production Deemix 3.13.7 / webui 4.6.0). out = json.loads(deemix_add_to_queue("https://www.deeeeezer.com/badid/9999999999")) check("ungueltige URL/ID: wird ehrlich gemeldet (queued=false + Warnung)", out["queued"] is False and "NO items" in (out.get("warning") or ""), str(out)) check("ungueltige URL/ID: nichts in der Queue (kein Download)", STATE.added == [] or all("9999999999" not in a["url"] for a in STATE.added), str(STATE.added)) try: deemix_add_to_queue("") check("leere urls abgelehnt", False, "kein Fehler") except DeemixError: check("leere urls abgelehnt", True) try: deemix_retry_download("album_1_1") check("retry: vorhandene UUID ok", True) except DeemixError as e: check("retry: vorhandene UUID ok", False, str(e)) try: deemix_retry_download("album_kein_kein") check("retry: unbekannte UUID sauberer Fehler", False, "kein Fehler") except DeemixError as e: check("retry: unbekannte UUID sauberer Fehler", "UnknownUuid" in str(e), str(e)) # 11) remove / finished (against emulated backend only) ---------------------- out = json.loads(deemix_remove_from_queue("album_2_3")) check("remove_from_queue: UUID entfernt", out["result"] and "album_2_3" not in STATE.queue) out = json.loads(deemix_queue()) check("queue: nach Entfernen nur 1 Eintrag", out["total"] == 1) try: deemix_remove_from_queue("album_unbekannt") check("remove_from_queue: unbekannte UUID sauberer Fehler", False, "kein Fehler") except DeemixError: check("remove_from_queue: unbekannte UUID sauberer Fehler", True) # 12) secret redaction --------------------------------------------------------- blob = { "arl": FAKE_ARL, "cookie": "deemix_session=abc123", "note": f"token {FAKE_ARL[:100]} embedded", "nested": {"set-cookie": "a=1", "ok": "bleibt"}, "list": [FAKE_ARL], } red = redact(blob) s = json.dumps(red) check("redaction: ARL verschwaengt", FAKE_ARL not in s) check("redaction: Cookie-Feld verschwaengt", red["cookie"] == "[REDACTED]" and red["nested"]["set-cookie"] == "[REDACTED]") check("redaction: unfaellige Felder bleiben", red["nested"]["ok"] == "bleibt") check("redaction: lange Hexfolgen werden maskiert", "[REDACTED]" in s) msg = _clean_error(f"boom arl={FAKE_ARL}") check("redaction: Fehlermeldungen werden bereinigt", FAKE_ARL not in msg and "[REDACTED]" in msg) # 13) compact queue output ------------------------------------------------------ big = [{"uuid": f"u{i}", "type": "album", "id": i, "title": "t", "artist": "a", "progress": i, "downloaded": 1048576 * i, "size": 2097152 * i, "failed": i % 7 == 0} for i in range(60)] q = json.loads(deemix_queue()) check("queue: Ausgabe kompakt und begrenzt", q["total"] <= 50 or q.get("truncated")) server.shutdown() print(f"\n{len(PASS)} bestanden, {len(FAIL)} fehlgeschlagen") if FAIL: print("Fehlschlaege:", *FAIL, sep="\n - ") return 1 print("ALLE TESTS BESTANDEN (emuliertes Deemix-Backend, keine echten Downloads)") return 0 if __name__ == "__main__": sys.exit(main())