refactor: extract transport-agnostic core and shared tool registry
Introduce aegis_gitea_mcp.registry as the single name->handler source of truth consumed by every transport adapter, moving TOOL_HANDLERS out of the FastAPI server module. Add aegis_gitea_mcp.errors.ToolError so core handlers no longer import fastapi.HTTPException; raw_tools now raises ToolError and the HTTP adapter maps it back to HTTPException, preserving status codes and audit behavior. Add a subprocess boundary test asserting the core imports without pulling in fastapi/uvicorn/starlette. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,64 @@
|
||||
"""Lock the transport-agnostic core boundary.
|
||||
|
||||
The core (tool registry, Gitea client, policy, audit, config, request context,
|
||||
tools) must import cleanly without dragging in the web stack. If a stray
|
||||
``import fastapi`` creeps back into a core module, the local stdio package would
|
||||
gain a needless heavy dependency and the ``[server]`` extra split would leak.
|
||||
|
||||
The check runs in a subprocess because, within the pytest process, FastAPI is
|
||||
already imported by the server tests — so ``'fastapi' in sys.modules`` would be
|
||||
true regardless. A clean interpreter is the only reliable probe.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_SRC = Path(__file__).resolve().parents[1] / "src"
|
||||
|
||||
# Core modules that must stay free of the web stack.
|
||||
_CORE_MODULES = [
|
||||
"aegis_gitea_mcp.registry",
|
||||
"aegis_gitea_mcp.gitea_client",
|
||||
"aegis_gitea_mcp.policy",
|
||||
"aegis_gitea_mcp.audit",
|
||||
"aegis_gitea_mcp.config",
|
||||
"aegis_gitea_mcp.request_context",
|
||||
"aegis_gitea_mcp.response_limits",
|
||||
"aegis_gitea_mcp.security",
|
||||
"aegis_gitea_mcp.cache",
|
||||
"aegis_gitea_mcp.logging_utils",
|
||||
"aegis_gitea_mcp.mcp_protocol",
|
||||
"aegis_gitea_mcp.errors",
|
||||
"aegis_gitea_mcp.tools.raw_tools",
|
||||
"aegis_gitea_mcp.tools.read_tools",
|
||||
"aegis_gitea_mcp.tools.write_tools",
|
||||
"aegis_gitea_mcp.tools.repository",
|
||||
"aegis_gitea_mcp.tools.arguments",
|
||||
]
|
||||
|
||||
|
||||
def test_core_does_not_import_fastapi() -> None:
|
||||
"""Importing the core in a clean interpreter must not import FastAPI."""
|
||||
imports = "\n".join(f"import {module}" for module in _CORE_MODULES)
|
||||
program = (
|
||||
f"import sys\n{imports}\n"
|
||||
"leaked = [m for m in ('fastapi', 'uvicorn', 'starlette') if m in sys.modules]\n"
|
||||
"assert not leaked, f'core leaked web stack: {leaked}'\n"
|
||||
"print('ok')\n"
|
||||
)
|
||||
env = dict(os.environ)
|
||||
env["PYTHONPATH"] = str(_SRC)
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", program],
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
assert result.returncode == 0, (
|
||||
f"core import boundary violated.\nstdout: {result.stdout}\nstderr: {result.stderr}"
|
||||
)
|
||||
assert "ok" in result.stdout
|
||||
@@ -6,10 +6,10 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from pydantic import ValidationError
|
||||
|
||||
from aegis_gitea_mcp.config import reset_settings
|
||||
from aegis_gitea_mcp.errors import ToolError
|
||||
from aegis_gitea_mcp.tools.arguments import (
|
||||
extract_repository,
|
||||
extract_target_path,
|
||||
@@ -83,7 +83,7 @@ async def test_lowercase_method_is_normalized(raw_env: None) -> None:
|
||||
async def test_delete_denied_when_write_mode_off(raw_env: None) -> None:
|
||||
"""A write method is denied (no network call) while write-mode is disabled."""
|
||||
stub = StubRawGitea()
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
with pytest.raises(ToolError) as exc_info:
|
||||
await raw_api_request_tool(stub, {"method": "DELETE", "path": "/repos/acme/app/issues/1"})
|
||||
|
||||
assert exc_info.value.status_code == 403
|
||||
@@ -135,7 +135,7 @@ async def test_write_denied_for_repo_outside_whitelist(
|
||||
monkeypatch.setenv("WRITE_REPOSITORY_WHITELIST", "acme/other")
|
||||
|
||||
stub = StubRawGitea()
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
with pytest.raises(ToolError) as exc_info:
|
||||
await raw_api_request_tool(stub, {"method": "POST", "path": "/repos/acme/app/issues"})
|
||||
|
||||
assert exc_info.value.status_code == 403
|
||||
@@ -158,7 +158,7 @@ async def test_non_repository_write_denied(monkeypatch: pytest.MonkeyPatch, tmp_
|
||||
monkeypatch.setenv("WRITE_REPOSITORY_WHITELIST", "acme/app")
|
||||
|
||||
stub = StubRawGitea()
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
with pytest.raises(ToolError) as exc_info:
|
||||
await raw_api_request_tool(stub, {"method": "POST", "path": "/user/repos"})
|
||||
|
||||
assert exc_info.value.status_code == 403
|
||||
@@ -173,7 +173,7 @@ async def test_non_repository_write_denied(monkeypatch: pytest.MonkeyPatch, tmp_
|
||||
async def test_sensitive_paths_denied_on_get(raw_env: None, path: str) -> None:
|
||||
"""Admin/credential surfaces are denied for every method, including GET."""
|
||||
stub = StubRawGitea()
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
with pytest.raises(ToolError) as exc_info:
|
||||
await raw_api_request_tool(stub, {"method": "GET", "path": path})
|
||||
|
||||
assert exc_info.value.status_code == 403
|
||||
@@ -253,7 +253,7 @@ async def test_raw_api_disabled(monkeypatch: pytest.MonkeyPatch, tmp_path: Path)
|
||||
monkeypatch.setenv("RAW_API_ENABLED", "false")
|
||||
|
||||
stub = StubRawGitea()
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
with pytest.raises(ToolError) as exc_info:
|
||||
await raw_api_request_tool(stub, {"method": "GET", "path": "/repos/acme/app"})
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "disabled" in str(exc_info.value.detail)
|
||||
|
||||
Reference in New Issue
Block a user