Newer
Older
gnexus-tasks / backend / tests / test_mcp_tokens.py
"""Тесты 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