203 lines
7.9 KiB
Python
203 lines
7.9 KiB
Python
"""Écritures MCP minimales (0.26.0) : create_issue + add_comment.
|
|
|
|
Le service d'agents doit pouvoir proposer des cartes et commenter
|
|
via Kanbante, sans passer par l'interface web. Les deux outils
|
|
réutilisent les services existants (création via _do_capture →
|
|
client de forge, commentaire via client.create_comment) et sont
|
|
gardés par la même authentification que les lectures /mcp :
|
|
session, Basic ou jeton d'API kbt_ — jamais l'interface web.
|
|
"""
|
|
|
|
import json
|
|
import uuid
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.main as main
|
|
from app import accounts, models
|
|
|
|
|
|
def _call(client, name, arguments, headers=None):
|
|
payload = {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "tools/call",
|
|
"params": {"name": name, "arguments": arguments},
|
|
}
|
|
return client.post("/mcp", json=payload, headers=headers or {})
|
|
|
|
|
|
def _result_text(response):
|
|
result = response.json()["result"]
|
|
return result, result["content"][0]["text"]
|
|
|
|
|
|
# ── Création de carte ──────────────────────────────────────
|
|
|
|
def test_mcp_create_issue_places_card(client, fake_gitea):
|
|
r = _call(client, "create_issue", {
|
|
"owner": "Projets", "repo": "demo",
|
|
"title": "Carte créée par MCP", "body": "Corps de la carte",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is False
|
|
created = json.loads(text)
|
|
assert created["title"] == "Carte créée par MCP"
|
|
assert created["repo_full_name"] == "Projets/demo"
|
|
assert created["column"] == "backlog"
|
|
assert created["number"] > 0
|
|
# L'écriture est passée par le client de forge existant.
|
|
assert fake_gitea.created == [("Projets", "demo", "Carte créée par MCP")]
|
|
# La carte est placée sur le board (lecture MCP relue).
|
|
r = _call(client, "get_board", {"owner": "Projets", "repo": "demo"})
|
|
board = json.loads(_result_text(r)[1])
|
|
assert any(
|
|
c["title"] == "Carte créée par MCP" for c in board["backlog"]
|
|
)
|
|
# Journal : création tracée avec la source « mcp ».
|
|
events = models.get_events(repo_full_name="Projets/demo", limit=10)
|
|
created_events = [
|
|
e for e in events
|
|
if e["event_type"] == "created"
|
|
and e["issue_number"] == created["number"]
|
|
]
|
|
assert created_events and created_events[0]["source"] == "mcp"
|
|
|
|
|
|
def test_mcp_create_issue_column_choice(client, fake_gitea):
|
|
r = _call(client, "create_issue", {
|
|
"owner": "Projets", "repo": "demo",
|
|
"title": "Directement à faire", "column": "todo",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is False
|
|
assert json.loads(text)["column"] == "todo"
|
|
|
|
|
|
def test_mcp_create_issue_validation_errors(client, fake_gitea):
|
|
# Titre manquant : « Missing argument » (patron d'erreur du serveur).
|
|
r = _call(client, "create_issue", {"owner": "Projets", "repo": "demo"})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is True
|
|
assert "Missing argument" in text
|
|
# Titre vide et colonne inconnue : erreurs métier en isError.
|
|
for arguments in (
|
|
{"owner": "Projets", "repo": "demo", "title": " "},
|
|
{"owner": "Projets", "repo": "demo", "title": "X",
|
|
"column": "nimporte"},
|
|
):
|
|
result, _text = _result_text(_call(client, "create_issue", arguments))
|
|
assert result["isError"] is True
|
|
assert fake_gitea.created == []
|
|
|
|
|
|
# ── Commentaire ────────────────────────────────────────────
|
|
|
|
def test_mcp_add_comment_posts_on_forge(client, fake_gitea):
|
|
r = _call(client, "add_comment", {
|
|
"owner": "Projets", "repo": "demo", "number": 1,
|
|
"body": "Commentaire du service d'agents",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is False
|
|
payload = json.loads(text)
|
|
assert payload["number"] == 1
|
|
assert payload["repo_full_name"] == "Projets/demo"
|
|
assert payload["body"] == "Commentaire du service d'agents"
|
|
assert fake_gitea.posted_comments == [
|
|
("Projets", "demo", 1, "Commentaire du service d'agents")
|
|
]
|
|
events = models.get_events(repo_full_name="Projets/demo", limit=10)
|
|
commented = [e for e in events if e["event_type"] == "commented"]
|
|
assert commented and commented[0]["source"] == "mcp"
|
|
|
|
|
|
def test_mcp_add_comment_validation_errors(client, fake_gitea):
|
|
r = _call(client, "add_comment", {
|
|
"owner": "Projets", "repo": "demo", "number": 1, "body": " ",
|
|
})
|
|
assert _result_text(r)[0]["isError"] is True
|
|
r = _call(client, "add_comment", {
|
|
"owner": "Projets", "repo": "demo", "body": "sans numéro",
|
|
})
|
|
result, text = _result_text(r)
|
|
assert result["isError"] is True
|
|
assert "Missing argument" in text
|
|
assert fake_gitea.posted_comments == []
|
|
|
|
|
|
def test_mcp_write_tools_are_listed(client):
|
|
r = client.post("/mcp", json={
|
|
"jsonrpc": "2.0", "id": 1, "method": "tools/list",
|
|
})
|
|
tools = {t["name"]: t for t in r.json()["result"]["tools"]}
|
|
assert tools["create_issue"]["inputSchema"]["required"] == [
|
|
"owner", "repo", "title",
|
|
]
|
|
assert tools["add_comment"]["inputSchema"]["required"] == [
|
|
"owner", "repo", "number", "body",
|
|
]
|
|
# Aucun outil destructeur n'apparaît.
|
|
assert not any(
|
|
name.startswith(("delete", "remove", "close"))
|
|
for name in tools
|
|
)
|
|
|
|
|
|
# ── Garde par jeton kbt_ (auth active) ─────────────────────
|
|
|
|
@pytest.fixture()
|
|
def utilisateur():
|
|
# Base de test partagée : nom unique, nettoyage complet au
|
|
# teardown (le pont SSO a une sémantique « un seul compte »).
|
|
uid = models.create_user(
|
|
f"mcp-{uuid.uuid4().hex[:8]}",
|
|
accounts.hash_password("motdepasse"), role="member")
|
|
user = models.get_user(uid)
|
|
yield user
|
|
with models.get_db() as conn:
|
|
conn.execute("DELETE FROM api_tokens WHERE user_id = ?", (uid,))
|
|
conn.execute("DELETE FROM users WHERE id = ?", (uid,))
|
|
|
|
|
|
def test_mcp_writes_guarded_by_api_token(monkeypatch, fake_gitea, utilisateur):
|
|
monkeypatch.setenv("KANBANTE_AUTH_USER", "bruno")
|
|
monkeypatch.setenv("KANBANTE_AUTH_PASSWORD", "motdepasse")
|
|
_row, clair = models.create_api_token(utilisateur["id"], "service-agents")
|
|
jeton = {"Authorization": f"Bearer {clair}"}
|
|
create_args = {
|
|
"owner": "Projets", "repo": "demo", "title": "Carte par jeton",
|
|
}
|
|
with TestClient(main.app) as client:
|
|
# Sans jeton : refusé (401), aucune écriture sur la forge.
|
|
assert _call(client, "create_issue", create_args).status_code == 401
|
|
# Jeton inconnu : refusé aussi.
|
|
faux = {"Authorization": "Bearer kbt_inconnu"}
|
|
assert _call(
|
|
client, "create_issue", create_args, headers=faux
|
|
).status_code == 401
|
|
assert fake_gitea.created == []
|
|
# Jeton valide : création et commentaire passent.
|
|
r = _call(client, "create_issue", create_args, headers=jeton)
|
|
assert r.status_code == 200
|
|
assert _result_text(r)[0]["isError"] is False
|
|
assert fake_gitea.created == [("Projets", "demo", "Carte par jeton")]
|
|
r = _call(client, "add_comment", {
|
|
"owner": "Projets", "repo": "demo", "number": 1,
|
|
"body": "Vu par le service d'agents",
|
|
}, headers=jeton)
|
|
assert _result_text(r)[0]["isError"] is False
|
|
# Le même jeton n'ouvre jamais l'interface web.
|
|
assert client.get(
|
|
"/", headers={**jeton, "Accept": "application/json"}
|
|
).status_code == 401
|
|
# Révocation : effet immédiat sur les écritures.
|
|
assert models.revoke_api_token(utilisateur["id"], _row["id"]) is True
|
|
assert _call(
|
|
client, "create_issue", create_args, headers=jeton
|
|
).status_code == 401
|
|
assert _call(client, "add_comment", {
|
|
"owner": "Projets", "repo": "demo", "number": 1, "body": "x",
|
|
}, headers=jeton).status_code == 401
|