b803811701
Implements F-14: thin FastMCP wrapper over existing codex domain modules.
Seven read-only tools: search, ask, wiki_read, wiki_list, discover_leads,
provenance_verify, synthesis_browse. All optional-feature tools degrade
gracefully to {"error": "feature not available"} instead of crashing.
Adds mcp[cli]>=1.0 dependency and codex-mcp console_script entry-point.
28 new tests across test_server, test_search, test_readonly, test_graceful,
test_http_auth — all green; 0 regressions in existing 172 tests.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
"""test_http_auth.py — HTTP transport token-auth enforcement.
|
|
|
|
Tests:
|
|
1. _require_http_token() raises RuntimeError when MCP_AUTH_TOKEN is absent.
|
|
2. _require_http_token() returns the token when it is set.
|
|
3. main() with transport='http' raises without token (refuses to start).
|
|
4. main() with transport='stdio' runs mcp.run (happy path, no token needed).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _require_http_token
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_require_http_token_raises_without_token() -> None:
|
|
"""_require_http_token() must raise RuntimeError when mcp_auth_token is None."""
|
|
from codex.mcp_server import _require_http_token
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.mcp_auth_token = None
|
|
|
|
with (
|
|
patch("codex.config.get_settings", return_value=mock_settings),
|
|
pytest.raises(RuntimeError, match="MCP_AUTH_TOKEN"),
|
|
):
|
|
_require_http_token()
|
|
|
|
|
|
def test_require_http_token_returns_token_when_set() -> None:
|
|
"""_require_http_token() must return the token string when it is configured."""
|
|
from codex.mcp_server import _require_http_token
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.mcp_auth_token = "super-secret-token"
|
|
|
|
with patch("codex.config.get_settings", return_value=mock_settings):
|
|
result = _require_http_token()
|
|
|
|
assert result == "super-secret-token"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# main() — HTTP transport without token → RuntimeError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_main_http_without_token_raises() -> None:
|
|
"""main() with MCP_TRANSPORT=http and no token must raise RuntimeError."""
|
|
from codex.mcp_server import main
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.mcp_transport = "http"
|
|
mock_settings.mcp_auth_token = None
|
|
|
|
with (
|
|
patch("codex.config.get_settings", return_value=mock_settings),
|
|
pytest.raises(RuntimeError, match="MCP_AUTH_TOKEN"),
|
|
):
|
|
main()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# main() — stdio transport runs mcp.run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_main_stdio_calls_mcp_run() -> None:
|
|
"""main() with stdio transport must call mcp.run(transport='stdio')."""
|
|
from codex import mcp_server
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.mcp_transport = "stdio"
|
|
|
|
with (
|
|
patch("codex.config.get_settings", return_value=mock_settings),
|
|
patch.object(mcp_server.mcp, "run") as mock_run,
|
|
):
|
|
mcp_server.main()
|
|
|
|
mock_run.assert_called_once_with(transport="stdio")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# main() — HTTP transport with valid token attempts uvicorn.run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_main_http_with_token_calls_uvicorn() -> None:
|
|
"""main() with http transport + valid token must invoke uvicorn.run."""
|
|
from codex import mcp_server
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.mcp_transport = "http"
|
|
mock_settings.mcp_auth_token = "test-token"
|
|
mock_settings.mcp_host = "127.0.0.1"
|
|
mock_settings.mcp_port = 8765
|
|
|
|
# Stub the ASGI app returned by streamable_http_app
|
|
mock_app = MagicMock()
|
|
mock_app.add_middleware = MagicMock()
|
|
|
|
with (
|
|
patch("codex.config.get_settings", return_value=mock_settings),
|
|
patch.object(mcp_server.mcp, "streamable_http_app", return_value=mock_app),
|
|
patch("uvicorn.run") as mock_uvicorn,
|
|
):
|
|
mcp_server.main()
|
|
|
|
mock_uvicorn.assert_called_once()
|
|
call_kwargs = mock_uvicorn.call_args
|
|
assert call_kwargs is not None
|
|
# host and port must match settings
|
|
_, kwargs = call_kwargs
|
|
assert kwargs.get("host") == "127.0.0.1"
|
|
assert kwargs.get("port") == 8765
|