michael/AI-Profile-Router
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add managed Deemix MCP integration

Browse Source
This commit is contained in:
Athena Operator
2026-08-25 10:26:37 +02:00
parent 4a260ae7d7
commit 1056784642
11 changed files with 1093 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dev/test_deemix_mcp.py
+469
View File
@@ -0,0 +1,469 @@
#!/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())