diff --git a/tests/test_core_boundary.py b/tests/test_core_boundary.py index 71803cd..14d73a6 100644 --- a/tests/test_core_boundary.py +++ b/tests/test_core_boundary.py @@ -58,7 +58,6 @@ def test_core_does_not_import_fastapi() -> None: capture_output=True, text=True, ) - assert result.returncode == 0, ( - f"core import boundary violated.\nstdout: {result.stdout}\nstderr: {result.stderr}" - ) + detail = f"core import boundary violated.\nstdout: {result.stdout}\nstderr: {result.stderr}" + assert result.returncode == 0, detail assert "ok" in result.stdout diff --git a/tests/test_stdio_app.py b/tests/test_stdio_app.py new file mode 100644 index 0000000..30bab56 --- /dev/null +++ b/tests/test_stdio_app.py @@ -0,0 +1,140 @@ +"""Tests for the local stdio transport adapter.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import pytest + +from aegis_gitea_mcp import stdio_app +from aegis_gitea_mcp.config import reset_settings +from aegis_gitea_mcp.errors import ToolError +from aegis_gitea_mcp.request_context import get_gitea_user_login + + +@pytest.fixture +def stdio_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Local-mode settings: PAT auth, no OAuth, no API-key requirement.""" + reset_settings() + monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") + monkeypatch.setenv("GITEA_TOKEN", "local-pat-token") + monkeypatch.setenv("AUTH_ENABLED", "false") + monkeypatch.setenv("ENVIRONMENT", "test") + monkeypatch.setenv("POLICY_FILE_PATH", str(tmp_path / "missing-policy.yaml")) + monkeypatch.setattr(stdio_app, "_owner_login", None) + + +def _patch_gitea_client(**methods: object) -> object: + """Patch GiteaClient with an async-context-manager mock exposing methods.""" + patcher = patch("aegis_gitea_mcp.gitea_client.GiteaClient") + cls = patcher.start() + instance = AsyncMock() + for name, value in methods.items(): + setattr(instance, name, AsyncMock(return_value=value)) + cls.return_value.__aenter__ = AsyncMock(return_value=instance) + cls.return_value.__aexit__ = AsyncMock(return_value=False) + return patcher + + +# --- Environment bootstrap -------------------------------------------------- + + +def test_bootstrap_forces_local_defaults(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("OAUTH_MODE", raising=False) + monkeypatch.delenv("AUTH_ENABLED", raising=False) + monkeypatch.delenv("AUDIT_LOG_PATH", raising=False) + stdio_app._bootstrap_env() + import os + + assert os.environ["OAUTH_MODE"] == "false" + assert os.environ["AUTH_ENABLED"] == "false" + assert os.environ["AUDIT_LOG_PATH"].endswith("audit.log") + + +def test_check_required_env_raises_when_missing(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("GITEA_URL", raising=False) + monkeypatch.delenv("GITEA_TOKEN", raising=False) + with pytest.raises(stdio_app.StdioConfigError) as exc_info: + stdio_app._check_required_env() + assert "GITEA_URL" in str(exc_info.value) + assert "GITEA_TOKEN" in str(exc_info.value) + + +def test_check_required_env_passes_when_present(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") + monkeypatch.setenv("GITEA_TOKEN", "tok") + stdio_app._check_required_env() # no raise + + +def test_default_audit_log_path_is_user_scoped() -> None: + path = stdio_app._default_audit_log_path() + assert path.name == "audit.log" + assert "aegis-gitea-mcp" in str(path) + + +def test_main_exits_when_env_missing(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("GITEA_URL", raising=False) + monkeypatch.delenv("GITEA_TOKEN", raising=False) + with pytest.raises(SystemExit) as exc_info: + stdio_app.main() + assert exc_info.value.code == 2 + + +# --- Owner resolution ------------------------------------------------------- + + +async def test_resolve_owner_login(stdio_env: None) -> None: + patcher = _patch_gitea_client(get_current_user={"login": "alice"}) + try: + login = await stdio_app._resolve_owner_login() + finally: + patcher.stop() + assert login == "alice" + + +async def test_resolve_owner_login_empty_raises(stdio_env: None) -> None: + patcher = _patch_gitea_client(get_current_user={"login": ""}) + try: + with pytest.raises(stdio_app.StdioConfigError): + await stdio_app._resolve_owner_login() + finally: + patcher.stop() + + +# --- Dispatch (shared registry + policy + audit) ---------------------------- + + +async def test_dispatch_unknown_tool(stdio_env: None) -> None: + with pytest.raises(ToolError) as exc_info: + await stdio_app._dispatch("nope_not_a_tool", {}) + assert exc_info.value.status_code == 404 + + +async def test_dispatch_policy_denies_write_without_write_mode(stdio_env: None) -> None: + """A write tool is denied by policy/WRITE_MODE before any network call.""" + with pytest.raises(ToolError) as exc_info: + await stdio_app._dispatch("create_issue", {"owner": "acme", "repo": "app", "title": "x"}) + assert exc_info.value.status_code == 403 + assert "write mode is disabled" in str(exc_info.value.detail) + + +async def test_dispatch_pins_owner_login_and_returns( + stdio_env: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """Dispatch pins request context to the PAT owner and runs the shared handler.""" + monkeypatch.setattr(stdio_app, "_owner_login", "alice") + patcher = _patch_gitea_client( + get_repository={ + "owner": {"login": "acme"}, + "name": "app", + "full_name": "acme/app", + } + ) + try: + result = await stdio_app._dispatch("get_repository_info", {"owner": "acme", "repo": "app"}) + finally: + patcher.stop() + assert result["name"] == "app" + # The dispatch pinned the trusted PAT owner onto the request context. + assert get_gitea_user_login() == "alice"