548 lines
20 KiB
Python
548 lines
20 KiB
Python
"""Tests for MCP server endpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from aegis_gitea_mcp.config import reset_settings
|
|
from aegis_gitea_mcp.oauth import reset_oauth_validator
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_state() -> None:
|
|
"""Reset global state between tests."""
|
|
reset_settings()
|
|
reset_oauth_validator()
|
|
yield
|
|
reset_settings()
|
|
reset_oauth_validator()
|
|
|
|
|
|
@pytest.fixture
|
|
def oauth_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Set OAuth-first environment for server tests."""
|
|
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
|
|
monkeypatch.setenv("OAUTH_MODE", "true")
|
|
monkeypatch.setenv("GITEA_OAUTH_CLIENT_ID", "test-client-id")
|
|
monkeypatch.setenv("GITEA_OAUTH_CLIENT_SECRET", "test-client-secret")
|
|
monkeypatch.setenv("ENVIRONMENT", "test")
|
|
monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false")
|
|
monkeypatch.setenv("WRITE_MODE", "false")
|
|
monkeypatch.setenv("PUBLIC_BASE_URL", "")
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_oauth_validation(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Mock OAuth validator outcomes by token value."""
|
|
|
|
async def _validate(_self, token: str | None, _ip: str, _ua: str):
|
|
if token == "valid-read":
|
|
return True, None, {"login": "alice", "scopes": ["read:repository"]}
|
|
if token == "valid-write":
|
|
return (
|
|
True,
|
|
None,
|
|
{
|
|
"login": "alice",
|
|
"scopes": ["read:repository", "write:repository"],
|
|
},
|
|
)
|
|
return False, "Invalid or expired OAuth token.", None
|
|
|
|
monkeypatch.setattr(
|
|
"aegis_gitea_mcp.oauth.GiteaOAuthValidator.validate_oauth_token",
|
|
_validate,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(oauth_env: None, mock_oauth_validation: None) -> TestClient:
|
|
"""Create FastAPI test client."""
|
|
from aegis_gitea_mcp.server import app
|
|
|
|
return TestClient(app)
|
|
|
|
|
|
def test_root_endpoint(client: TestClient) -> None:
|
|
"""Root endpoint returns server metadata."""
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["name"] == "AegisGitea MCP Server"
|
|
assert data["status"] == "running"
|
|
|
|
|
|
def test_health_endpoint(client: TestClient) -> None:
|
|
"""Health endpoint does not require auth."""
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "healthy"
|
|
|
|
|
|
def test_oauth_protected_resource_metadata(client: TestClient) -> None:
|
|
"""OAuth protected-resource metadata contains required OpenAI-compatible fields."""
|
|
response = client.get("/.well-known/oauth-protected-resource")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["resource"] == "https://gitea.example.com"
|
|
assert data["authorization_servers"] == ["https://gitea.example.com"]
|
|
assert data["bearer_methods_supported"] == ["header"]
|
|
assert data["scopes_supported"] == ["read:repository", "write:repository"]
|
|
assert "resource_documentation" in data
|
|
|
|
|
|
def test_oauth_authorization_server_metadata(client: TestClient) -> None:
|
|
"""Auth server metadata includes expected OAuth endpoints and scopes."""
|
|
response = client.get("/.well-known/oauth-authorization-server")
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["authorization_endpoint"].endswith("/login/oauth/authorize")
|
|
assert payload["token_endpoint"].endswith("/oauth/token")
|
|
assert payload["scopes_supported"] == ["read:repository", "write:repository"]
|
|
|
|
|
|
def test_oauth_metadata_uses_public_base_url(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Public base URL is used for externally advertised OAuth metadata links."""
|
|
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
|
|
monkeypatch.setenv("OAUTH_MODE", "true")
|
|
monkeypatch.setenv("GITEA_OAUTH_CLIENT_ID", "test-client-id")
|
|
monkeypatch.setenv("GITEA_OAUTH_CLIENT_SECRET", "test-client-secret")
|
|
monkeypatch.setenv("PUBLIC_BASE_URL", "https://mcp.example.com")
|
|
monkeypatch.setenv("ENVIRONMENT", "test")
|
|
monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false")
|
|
|
|
from aegis_gitea_mcp.server import app
|
|
|
|
client = TestClient(app)
|
|
metadata_response = client.get("/.well-known/oauth-authorization-server")
|
|
assert metadata_response.status_code == 200
|
|
payload = metadata_response.json()
|
|
assert payload["token_endpoint"] == "https://mcp.example.com/oauth/token"
|
|
|
|
challenge_response = client.post(
|
|
"/mcp/tool/call",
|
|
json={"tool": "list_repositories", "arguments": {}},
|
|
)
|
|
assert challenge_response.status_code == 401
|
|
challenge = challenge_response.headers["WWW-Authenticate"]
|
|
assert (
|
|
'resource_metadata="https://mcp.example.com/.well-known/oauth-protected-resource"'
|
|
in challenge
|
|
)
|
|
|
|
|
|
def test_scope_compatibility_write_implies_read() -> None:
|
|
"""write:repository grants read-level access for read tools."""
|
|
from aegis_gitea_mcp.server import READ_SCOPE, _has_required_scope
|
|
|
|
assert _has_required_scope(READ_SCOPE, {"write:repository"})
|
|
|
|
|
|
def test_scope_compatibility_repository_aliases() -> None:
|
|
"""Legacy/broad repository scopes satisfy MCP read/write requirements."""
|
|
from aegis_gitea_mcp.server import READ_SCOPE, WRITE_SCOPE, _has_required_scope
|
|
|
|
assert _has_required_scope(READ_SCOPE, {"repository"})
|
|
assert _has_required_scope(WRITE_SCOPE, {"repository"})
|
|
assert _has_required_scope(WRITE_SCOPE, {"repo"})
|
|
|
|
|
|
def test_list_tools_without_auth(client: TestClient) -> None:
|
|
"""Tool listing remains discoverable without auth."""
|
|
response = client.get("/mcp/tools")
|
|
assert response.status_code == 200
|
|
assert "tools" in response.json()
|
|
|
|
|
|
def test_call_tool_without_auth_returns_challenge(client: TestClient) -> None:
|
|
"""Tool calls without bearer token return 401 + WWW-Authenticate challenge."""
|
|
response = client.post("/mcp/tool/call", json={"tool": "list_repositories", "arguments": {}})
|
|
|
|
assert response.status_code == 401
|
|
assert "WWW-Authenticate" in response.headers
|
|
challenge = response.headers["WWW-Authenticate"]
|
|
assert 'resource_metadata="http://testserver/.well-known/oauth-protected-resource"' in challenge
|
|
assert 'scope="read:repository"' in challenge
|
|
|
|
|
|
def test_call_tool_invalid_token_returns_challenge(client: TestClient) -> None:
|
|
"""Invalid bearer token returns 401 + WWW-Authenticate challenge."""
|
|
response = client.post(
|
|
"/mcp/tool/call",
|
|
headers={"Authorization": "Bearer invalid-token"},
|
|
json={"tool": "list_repositories", "arguments": {}},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert "WWW-Authenticate" in response.headers
|
|
|
|
|
|
def test_sse_tools_list_returns_camel_case_schema(client: TestClient) -> None:
|
|
"""SSE tools/list returns MCP-compatible camelCase inputSchema."""
|
|
response = client.post(
|
|
"/mcp/sse",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={"jsonrpc": "2.0", "id": "1", "method": "tools/list"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "result" in data
|
|
assert "tools" in data["result"]
|
|
assert "inputSchema" in data["result"]["tools"][0]
|
|
|
|
|
|
def test_sse_initialize_message(client: TestClient) -> None:
|
|
"""SSE initialize message returns protocol and server metadata."""
|
|
response = client.post(
|
|
"/mcp/sse",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={"jsonrpc": "2.0", "id": "init-1", "method": "initialize"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["result"]["protocolVersion"] == "2024-11-05"
|
|
assert payload["result"]["serverInfo"]["name"] == "AegisGitea MCP"
|
|
|
|
|
|
def test_sse_tools_call_success_response(
|
|
client: TestClient, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""SSE tools/call wraps successful tool output in text content."""
|
|
|
|
async def _fake_execute(tool_name: str, arguments: dict, correlation_id: str) -> dict:
|
|
assert tool_name == "list_repositories"
|
|
assert isinstance(arguments, dict)
|
|
assert correlation_id
|
|
return {"ok": True}
|
|
|
|
monkeypatch.setattr("aegis_gitea_mcp.server._execute_tool_call", _fake_execute)
|
|
|
|
response = client.post(
|
|
"/mcp/sse",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": "call-1",
|
|
"method": "tools/call",
|
|
"params": {"name": "list_repositories", "arguments": {}},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert '"ok": true' in response.json()["result"]["content"][0]["text"].lower()
|
|
|
|
|
|
def test_sse_tools_call_http_exception(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""SSE tools/call maps HTTPException to JSON-RPC error envelope."""
|
|
|
|
async def _fake_execute(_tool_name: str, _arguments: dict, _correlation_id: str) -> dict:
|
|
from fastapi import HTTPException
|
|
|
|
raise HTTPException(status_code=403, detail="Insufficient scope")
|
|
|
|
monkeypatch.setattr("aegis_gitea_mcp.server._execute_tool_call", _fake_execute)
|
|
|
|
response = client.post(
|
|
"/mcp/sse",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": "call-2",
|
|
"method": "tools/call",
|
|
"params": {"name": "create_issue", "arguments": {}},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["error"]["code"] == -32000
|
|
assert "insufficient scope" in body["error"]["message"].lower()
|
|
|
|
|
|
def test_call_nonexistent_tool(client: TestClient) -> None:
|
|
"""Unknown tools return 404 after successful auth."""
|
|
response = client.post(
|
|
"/mcp/tool/call",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={"tool": "nonexistent_tool", "arguments": {}},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
assert "not found" in response.json()["detail"].lower()
|
|
|
|
|
|
def test_write_scope_enforced_before_policy(client: TestClient) -> None:
|
|
"""Write tools require write:repository scope."""
|
|
response = client.post(
|
|
"/mcp/tool/call",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={
|
|
"tool": "create_issue",
|
|
"arguments": {"owner": "acme", "repo": "demo", "title": "test"},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert "required scope: write:repository" in response.json()["detail"].lower()
|
|
|
|
|
|
def test_write_tool_denied_by_default_policy(client: TestClient) -> None:
|
|
"""Even with write scope, write mode stays denied by default policy."""
|
|
response = client.post(
|
|
"/mcp/tool/call",
|
|
headers={"Authorization": "Bearer valid-write"},
|
|
json={
|
|
"tool": "create_issue",
|
|
"arguments": {"owner": "acme", "repo": "demo", "title": "test"},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert "write mode is disabled" in response.json()["detail"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_startup_event_fails_when_discovery_unreachable(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Startup validation fails with clear guidance if OIDC discovery is unreachable."""
|
|
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
|
|
monkeypatch.setenv("ENVIRONMENT", "production")
|
|
monkeypatch.setenv("OAUTH_MODE", "true")
|
|
monkeypatch.setenv("GITEA_OAUTH_CLIENT_ID", "test-client-id")
|
|
monkeypatch.setenv("GITEA_OAUTH_CLIENT_SECRET", "test-client-secret")
|
|
monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "true")
|
|
|
|
from aegis_gitea_mcp import server
|
|
|
|
with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(
|
|
side_effect=httpx.RequestError("connect failed", request=MagicMock())
|
|
)
|
|
mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with pytest.raises(
|
|
RuntimeError,
|
|
match="unable to reach Gitea OIDC discovery endpoint",
|
|
):
|
|
await server.startup_event()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_startup_event_succeeds_when_discovery_ready(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Startup validation succeeds when OIDC discovery returns HTTP 200."""
|
|
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
|
|
monkeypatch.setenv("ENVIRONMENT", "production")
|
|
monkeypatch.setenv("OAUTH_MODE", "true")
|
|
monkeypatch.setenv("GITEA_OAUTH_CLIENT_ID", "test-client-id")
|
|
monkeypatch.setenv("GITEA_OAUTH_CLIENT_SECRET", "test-client-secret")
|
|
monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "true")
|
|
|
|
from aegis_gitea_mcp import server
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
|
|
with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(return_value=mock_response)
|
|
mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
await server.startup_event()
|
|
|
|
|
|
def test_probe_scopeless_token_returns_401(
|
|
oauth_env: None, mock_oauth_validation: None, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Middleware returns 401 with re-auth guidance when Gitea probe returns 403."""
|
|
from aegis_gitea_mcp import server
|
|
|
|
server._api_scope_cache.clear()
|
|
|
|
mock_probe_response = MagicMock()
|
|
mock_probe_response.status_code = 403
|
|
|
|
with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(return_value=mock_probe_response)
|
|
mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
from aegis_gitea_mcp.server import app
|
|
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
response = client.post(
|
|
"/mcp/tool/call",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={"tool": "list_repositories", "arguments": {}},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert "WWW-Authenticate" in response.headers
|
|
|
|
|
|
def test_probe_valid_token_proceeds(
|
|
oauth_env: None, mock_oauth_validation: None, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Middleware allows request through when Gitea probe returns 200."""
|
|
from aegis_gitea_mcp import server
|
|
|
|
server._api_scope_cache.clear()
|
|
|
|
mock_probe_response = MagicMock()
|
|
mock_probe_response.status_code = 200
|
|
|
|
async def _fake_execute(tool_name: str, arguments: dict, correlation_id: str) -> dict:
|
|
return {"ok": True}
|
|
|
|
monkeypatch.setattr("aegis_gitea_mcp.server._execute_tool_call", _fake_execute)
|
|
|
|
with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(return_value=mock_probe_response)
|
|
mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
from aegis_gitea_mcp.server import app
|
|
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
response = client.post(
|
|
"/mcp/tool/call",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={"tool": "list_repositories", "arguments": {}},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["success"] is True
|
|
|
|
|
|
def test_gitea_authorization_error_returns_403(
|
|
oauth_env: None, mock_oauth_validation: None, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""GiteaAuthorizationError from tool handler returns 403 with re-auth guidance."""
|
|
from aegis_gitea_mcp import server
|
|
from aegis_gitea_mcp.gitea_client import GiteaAuthorizationError
|
|
|
|
server._api_scope_cache.clear()
|
|
|
|
async def _fake_execute(tool_name: str, arguments: dict, correlation_id: str) -> dict:
|
|
raise GiteaAuthorizationError("403 Forbidden")
|
|
|
|
monkeypatch.setattr("aegis_gitea_mcp.server._execute_tool_call", _fake_execute)
|
|
|
|
mock_probe_response = MagicMock()
|
|
mock_probe_response.status_code = 200
|
|
|
|
with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(return_value=mock_probe_response)
|
|
mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
from aegis_gitea_mcp.server import app
|
|
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
response = client.post(
|
|
"/mcp/tool/call",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={"tool": "list_repositories", "arguments": {}},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
body = response.json()
|
|
assert body["success"] is False
|
|
assert "re-authorize" in body["error"].lower()
|
|
|
|
|
|
def test_gitea_authentication_error_returns_401(
|
|
oauth_env: None, mock_oauth_validation: None, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""GiteaAuthenticationError from tool handler returns 401."""
|
|
from aegis_gitea_mcp import server
|
|
from aegis_gitea_mcp.gitea_client import GiteaAuthenticationError
|
|
|
|
server._api_scope_cache.clear()
|
|
|
|
async def _fake_execute(tool_name: str, arguments: dict, correlation_id: str) -> dict:
|
|
raise GiteaAuthenticationError("401 Unauthorized")
|
|
|
|
monkeypatch.setattr("aegis_gitea_mcp.server._execute_tool_call", _fake_execute)
|
|
|
|
mock_probe_response = MagicMock()
|
|
mock_probe_response.status_code = 200
|
|
|
|
with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(return_value=mock_probe_response)
|
|
mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
from aegis_gitea_mcp.server import app
|
|
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
response = client.post(
|
|
"/mcp/tool/call",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={"tool": "list_repositories", "arguments": {}},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
body = response.json()
|
|
assert body["success"] is False
|
|
assert "re-authenticate" in body["error"].lower()
|
|
|
|
|
|
def test_sse_gitea_authorization_error_returns_jsonrpc_error(
|
|
oauth_env: None, mock_oauth_validation: None, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""GiteaAuthorizationError in SSE handler returns JSON-RPC -32000 with guidance."""
|
|
from aegis_gitea_mcp import server
|
|
from aegis_gitea_mcp.gitea_client import GiteaAuthorizationError
|
|
|
|
server._api_scope_cache.clear()
|
|
|
|
async def _fake_execute(tool_name: str, arguments: dict, correlation_id: str) -> dict:
|
|
raise GiteaAuthorizationError("403 Forbidden")
|
|
|
|
monkeypatch.setattr("aegis_gitea_mcp.server._execute_tool_call", _fake_execute)
|
|
|
|
mock_probe_response = MagicMock()
|
|
mock_probe_response.status_code = 200
|
|
|
|
with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.get = AsyncMock(return_value=mock_probe_response)
|
|
mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
from aegis_gitea_mcp.server import app
|
|
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
response = client.post(
|
|
"/mcp/sse",
|
|
headers={"Authorization": "Bearer valid-read"},
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": "err-1",
|
|
"method": "tools/call",
|
|
"params": {"name": "list_repositories", "arguments": {}},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["error"]["code"] == -32000
|
|
assert "re-authorize" in body["error"]["message"].lower()
|