116 lines
5.9 KiB
Python
116 lines
5.9 KiB
Python
"""MCP-Endpunkt (→ KONZEPT.md 12) und Rückmeldungen (→ 15)."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from test_app import Env, tenant_package
|
|
from hugin_app.app import create_app
|
|
from hugin_app.settings import Settings
|
|
from hugin_app.hugin import Hugin
|
|
from hugin_app.migration import migrate
|
|
|
|
|
|
@pytest.fixture
|
|
def env(database_url):
|
|
url = database_url
|
|
migrate(url)
|
|
hugin = Hugin(Settings(database_url=url, admin_token="admin-geheim"))
|
|
# Mit "with", damit die Lebenszeit der App (MCP-Sitzungsverwalter) läuft
|
|
with TestClient(create_app(hugin)) as client:
|
|
yield Env(client, hugin)
|
|
|
|
|
|
def tool(u: Env, key: str | None, name: str, **arguments) -> tuple[bool, object]:
|
|
header = {"Accept": "application/json, text/event-stream", "MCP-Protocol-Version": "2025-06-18"}
|
|
if key:
|
|
header |= u.h(key)
|
|
response = u.client.post("/mcp", headers=header, json={
|
|
"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": name, "arguments": arguments}})
|
|
assert response.status_code == 200, response.text
|
|
result = response.json()["result"]
|
|
if result["isError"]:
|
|
return False, result["content"][0]["text"]
|
|
return True, result.get("structuredContent", {}).get("result", result.get("structuredContent"))
|
|
|
|
|
|
def test_tools_listed(env):
|
|
response = env.client.post("/mcp/", headers=env.h("shop-lesen") | {
|
|
"Accept": "application/json, text/event-stream"}, json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
|
|
names = {w["name"] for w in response.json()["result"]["tools"]}
|
|
assert names == {"search_packages", "get_package", "report_feedback", "list_subscriptions"}
|
|
|
|
|
|
def test_without_key(env):
|
|
for header in ({}, {"Authorization": "Bearer hk_FALSCH"}, {"Authorization": "Bearer admin-geheim"}):
|
|
response = env.client.post("/mcp", headers=header | {"Accept": "application/json, text/event-stream"},
|
|
json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
|
|
assert response.status_code == 401 and "Key" in response.text
|
|
|
|
|
|
def test_search_and_get(env):
|
|
env.example("pub-munin-nutzen/1.4.0")
|
|
env.example("pub-munin-nutzen/1.5.0")
|
|
tenant_package(env)
|
|
ok, matches = tool(env, "shop-lesen", "search_packages", query="munin")
|
|
assert ok and [t["id"] for t in matches] == ["pub-munin-nutzen"]
|
|
ok, matches = tool(env, "vertrieb-lesen", "search_packages", query="")
|
|
assert ok and "shop-intern" not in json.dumps(matches)
|
|
ok, package = tool(env, "shop-lesen", "get_package", package="pub-munin-nutzen")
|
|
assert ok and package["version"] == "1.5.0" and package["content"].startswith("# Munin nutzen")
|
|
assert "- pub-munin-nutzen" in package["install"] and "hugin sync" in package["install"]
|
|
ok, package = tool(env, "shop-lesen", "get_package", package="pub-munin-nutzen", version="1.4.0")
|
|
assert ok and "referenz/beispiele.md" in package["files"]
|
|
ok, text = tool(env, "vertrieb-lesen", "get_package", package="shop-intern")
|
|
assert not ok and "Nur für den Shop" not in text
|
|
|
|
|
|
def test_mcp_package_explains_installation(env):
|
|
env.example("pub-munin-anbindung/1.0.0")
|
|
ok, package = tool(env, "shop-lesen", "get_package", package="pub-munin-anbindung")
|
|
assert ok and package["install"].startswith("Eine MCP-Anbindung lässt sich nicht einmalig nutzen")
|
|
assert "MUNIN_KEY" in package["install"]
|
|
|
|
|
|
def test_instructions_say_how_to_install(env):
|
|
response = env.client.post("/mcp/", headers=env.h("shop-lesen") | {"Accept": "application/json, text/event-stream"},
|
|
json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
|
|
"protocolVersion": "2025-06-18", "capabilities": {},
|
|
"clientInfo": {"name": "test", "version": "0"}}})
|
|
assert ".hugin.yaml" in response.json()["result"]["instructions"]
|
|
|
|
|
|
def test_feedback_by_agent(env):
|
|
env.example("pub-munin-nutzen/1.4.0")
|
|
ok, r = tool(env, "shop-lesen", "report_feedback", package="pub-munin-nutzen", version="1.4.0",
|
|
kind="unclear", text="recall oder search?", tool="claude")
|
|
assert ok and r["status"] == "open"
|
|
ok, text = tool(env, "shop-lesen", "report_feedback", package="pub-munin-nutzen", version="1.4.0",
|
|
kind="egal", text="x")
|
|
assert not ok and "kind" in text
|
|
# Besitzer sieht die Meldung, aber nicht den Tenant; der Shop sieht seinen eigenen
|
|
owner = env.client.get("/api/v1/packages/pub-munin-nutzen/feedback", headers=env.h("pub-autor")).json()
|
|
assert [(x["source"], x["tenant"]) for x in owner] == [("agent", None)]
|
|
shop = env.client.get("/api/v1/packages/pub-munin-nutzen/feedback", headers=env.h("shop-lesen")).json()
|
|
assert shop[0]["tenant"] == "shop"
|
|
assert env.client.get("/api/v1/packages/pub-munin-nutzen/feedback",
|
|
headers=env.h("vertrieb-lesen")).json() == []
|
|
# Nur der Besitzer erledigt
|
|
rid = owner[0]["id"]
|
|
assert env.client.post(f"/api/v1/packages/pub-munin-nutzen/feedback/{rid}/resolve",
|
|
headers=env.h("shop-lesen")).status_code == 403
|
|
assert env.client.post(f"/api/v1/packages/pub-munin-nutzen/feedback/{rid}/resolve",
|
|
headers=env.h("pub-autor")).status_code == 204
|
|
|
|
|
|
def test_subscriptions(env):
|
|
env.example("pub-munin-nutzen/1.4.0")
|
|
env.example("pub-munin-nutzen/1.5.0")
|
|
env.post("/api/v1/usage", env.h("shop-lesen"),
|
|
{"repo": "git.intern/shop/api", "installed": {"pub-munin-nutzen": "1.4.0"}})
|
|
ok, subscriptions = tool(env, "shop-lesen", "list_subscriptions")
|
|
assert ok and subscriptions == [{"repo": "git.intern/shop/api", "package": "pub-munin-nutzen", "version": "1.4.0",
|
|
"newest": "1.5.0", "newer_available": True}]
|
|
ok, subscriptions = tool(env, "vertrieb-lesen", "list_subscriptions")
|
|
assert ok and subscriptions == []
|