99 lines
4.5 KiB
Python
99 lines
4.5 KiB
Python
"""Minimal MCP client and ADT deletion for the EPOD acceptance tests (standard library only, Python 3.9+)."""
|
|
import base64
|
|
import http.cookiejar
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
import urllib.request
|
|
from xml.sax.saxutils import quoteattr
|
|
|
|
URL = os.environ.get("MCP_URL", "http://127.0.0.1:3000/mcp")
|
|
TOKEN = os.environ.get("MCP_TOKEN", "")
|
|
SYSTEM = os.environ.get("MCP_SYSTEM_ID") # optional: Eclipse project name when several systems are connected
|
|
|
|
|
|
class Mcp:
|
|
def __init__(self, timeout=300):
|
|
self.sid, self.n, self.timeout = None, 0, timeout
|
|
|
|
def _post(self, body, method="POST"):
|
|
h = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"}
|
|
if TOKEN:
|
|
h["Authorization"] = "Bearer " + TOKEN
|
|
if self.sid:
|
|
h["Mcp-Session-Id"] = self.sid
|
|
req = urllib.request.Request(URL, json.dumps(body).encode() if body is not None else None, h, method=method)
|
|
with urllib.request.urlopen(req, timeout=self.timeout) as r:
|
|
sid = r.headers.get("Mcp-Session-Id")
|
|
raw = r.read().decode()
|
|
self.sid = sid or self.sid
|
|
if "data:" in raw[:40]:
|
|
raw = "".join(l[5:].strip() for l in raw.splitlines() if l.startswith("data:"))
|
|
return json.loads(raw) if raw.strip() else None
|
|
|
|
def open(self):
|
|
self.n += 1
|
|
self._post({"jsonrpc": "2.0", "id": self.n, "method": "initialize", "params": {
|
|
"protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "epod-acceptance", "version": "1"}}})
|
|
self._post({"jsonrpc": "2.0", "method": "notifications/initialized"})
|
|
return self
|
|
|
|
def close(self):
|
|
try:
|
|
self._post(None, "DELETE")
|
|
except Exception:
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self.open()
|
|
|
|
def __exit__(self, *a):
|
|
self.close()
|
|
|
|
def call(self, tool, args):
|
|
"""Returns (is_error, text). 'Concurrent call detected' is returned as it is (the tests count it)."""
|
|
self.n += 1
|
|
if SYSTEM:
|
|
args = dict(args, systemId=SYSTEM)
|
|
res = self._post({"jsonrpc": "2.0", "id": self.n, "method": "tools/call", "params": {"name": tool, "arguments": args}})
|
|
if "error" in res:
|
|
return True, json.dumps(res["error"])
|
|
r = res["result"]
|
|
return bool(r.get("isError")), "\n".join(c.get("text", "") for c in r.get("content", []))
|
|
|
|
|
|
class Adt:
|
|
"""Deletion through the ADT deletion API (needs A4H_URL, A4H_USER, A4H_PASSWORD; A4H_CLIENT default 001)."""
|
|
|
|
def __init__(self):
|
|
self.base = os.environ.get("A4H_URL", "").rstrip("/")
|
|
self.client = os.environ.get("A4H_CLIENT", "001")
|
|
self.auth = "Basic " + base64.b64encode(("%s:%s" % (os.environ.get("A4H_USER", ""), os.environ.get("A4H_PASSWORD", ""))).encode()).decode()
|
|
self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
|
|
self.csrf = None
|
|
|
|
def available(self):
|
|
return bool(self.base and os.environ.get("A4H_USER"))
|
|
|
|
def _req(self, path, body=None, headers=None):
|
|
h = {"Authorization": self.auth}
|
|
if self.csrf:
|
|
h["x-csrf-token"] = self.csrf
|
|
h.update(headers or {})
|
|
req = urllib.request.Request("%s%s%ssap-client=%s" % (self.base, path, "&" if "?" in path else "?", self.client), body.encode() if body else None, h)
|
|
with self.opener.open(req, timeout=120) as r:
|
|
return dict(r.headers), r.read().decode()
|
|
|
|
def delete(self, uris):
|
|
if not self.available() or not uris:
|
|
return {}
|
|
hdr, _ = self._req("/sap/bc/adt/discovery", headers={"x-csrf-token": "fetch", "Accept": "*/*"})
|
|
self.csrf = hdr.get("x-csrf-token") or hdr.get("X-CSRF-Token")
|
|
objs = "".join("<del:object adtcore:uri=%s><del:transportNumber/></del:object>" % quoteattr(u) for u in uris)
|
|
body = ('<?xml version="1.0" encoding="UTF-8"?><del:deletionRequest xmlns:del="http://www.sap.com/adt/deletion" '
|
|
'xmlns:adtcore="http://www.sap.com/adt/core">' + objs + "</del:deletionRequest>")
|
|
_, text = self._req("/sap/bc/adt/deletion/delete", body, {"Content-Type": "application/vnd.sap.adt.deletion.request.v1+xml",
|
|
"Accept": "application/vnd.sap.adt.deletion.response.v1+xml"})
|
|
return {m.group(1): 'isDeleted="true"' in m.group(0) for m in re.finditer(r'<del:object\b[^>]*adtcore:uri="([^"]+)"[^>]*>', text)}
|