"""Тесты per-user MCP-токенов (ТЗ 3.10): генерация, отзыв, авторизация /mcp."""
from typing import Any
from fastapi.testclient import TestClient
from app import mcp_server
from app.api.mcp_tokens import hash_token
from app.main import app
from app.models import Task
from tests.conftest import _test_session_factory # type: ignore[attr-defined]
from tests.test_multiuser import switch_user
def _create_token(client: TestClient, label: str = "") -> dict[str, Any]:
resp = client.post("/api/mcp-tokens", json={"label": label})
assert resp.status_code == 200, resp.text
return resp.json()
def test_create_token_returns_plaintext_once(client: TestClient) -> None:
created = _create_token(client, "мой агент")
assert created["token"].startswith("gnt_")
assert created["label"] == "мой агент"
# в БД — только хэш, plaintext не хранится
session = _test_session_factory()
try:
from sqlalchemy import select
from app.models import McpToken
rows = session.scalars(select(McpToken)).all()
assert len(rows) == 1
assert rows[0].token_hash == hash_token(created["token"])
assert rows[0].user_id == "1"
assert created["token"] not in str(rows[0].__dict__)
finally:
session.close()
# список не содержит plaintext
listed = client.get("/api/mcp-tokens").json()
assert len(listed) == 1
assert "token" not in listed[0]
assert "token_hash" not in listed[0]
def test_delete_token(client: TestClient) -> None:
created = _create_token(client)
resp = client.delete(f"/api/mcp-tokens/{created['id']}")
assert resp.status_code == 200
assert client.get("/api/mcp-tokens").json() == []
# чужой/несуществующий — 404
assert client.delete("/api/mcp-tokens/999").status_code == 404
def test_token_grants_mcp_access_and_scopes_to_owner(client: TestClient) -> None:
"""Токен из API авторизует /mcp; тул действует от имени владельца токена."""
created = _create_token(client, "агент")
# middleware валидирует токен по хэшу из БД: без lifespan дальше mounted-app
# не идёт (307 redirect на /mcp/ — это уже после авторизации), 401 нет
tc = TestClient(app, follow_redirects=False)
resp = tc.post(
"/mcp",
json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}},
headers={"Authorization": f"Bearer {created['token']}"},
)
assert resp.status_code == 307 # 401 был бы до редиректа
# тул от имени владельца токена создаёт задачу его пользователя
from tests.conftest import mcp_ctx # type: ignore[attr-defined]
result = mcp_server.create_task("Через токен", ctx=mcp_ctx("1"))
session = _test_session_factory()
try:
task = session.get(Task, result["id"])
assert task is not None and task.user_id == "1"
finally:
session.close()
def test_revoked_token_gets_401(client: TestClient) -> None:
created = _create_token(client)
headers = {"Authorization": f"Bearer {created['token']}"}
assert client.delete(f"/api/mcp-tokens/{created['id']}").status_code == 200
# 401 отдаёт middleware до монтирования MCP — lifespan не нужен
tc = TestClient(app)
assert tc.post("/mcp", json={}, headers=headers).status_code == 401
def test_foreign_token_isolated_per_user(client: TestClient) -> None:
"""Токен другого пользователя не виден в списке и не удаляется по id."""
mine = _create_token(client, "мой")
with switch_user("other-user"):
theirs = _create_token(client, "чужой")
listed = {t["id"] for t in client.get("/api/mcp-tokens").json()}
assert mine["id"] in listed and theirs["id"] not in listed
assert client.delete(f"/api/mcp-tokens/{theirs['id']}").status_code == 404
# неверный токен — 401 (middleware до монтирования, lifespan не нужен)
fake = TestClient(app).post("/mcp", json={}, headers={"Authorization": "Bearer gnt_fake"})
assert fake.status_code == 401