test: stdio adapter dispatch, owner resolution and local env bootstrap

Cover the stdio adapter: local-mode env bootstrap (OAuth off, API-key gate off,
per-user audit path), missing-env failure, PAT owner resolution, and dispatch
(unknown tool, write-mode policy denial, and the happy path pinning request
context to the PAT owner via the shared registry). Tidy the boundary-test
assertion so ruff and black agree.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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, (
+    detail = f"core import boundary violated.\nstdout: {result.stdout}\nstderr: {result.stderr}"
-        f"core import boundary violated.\nstdout: {result.stdout}\nstderr: {result.stderr}"
+    assert result.returncode == 0, detail
-    )
     assert "ok" in result.stdout

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"