feat: new settings, CLI skeleton and interface skeletons

Adds the pydantic-settings Settings with validated GITEA_URL, OAuth scopes
and toolsets, the argparse CLI skeleton, catalog.model, identity and
GiteaAPIError. Test fixtures now isolate env, cwd and state paths in tmp.
This commit is contained in:
Latte committed 2026-10-14 15:40:37 +02:00
1 parent 4ac9626eb5
commit b82e2866cb
24 files changed
+1512 -605

No files matched your search

+27 -60
View File
@@ -1,4 +1,11 @@
"""Pytest configuration and fixtures."""
"""Pytest configuration and fixtures.
Every test runs with a scrubbed environment, an empty working directory (so a
developer's own ``.env`` is never read) and state paths inside ``tmp_path``.
Nothing may write to ``/var/log``, ``/var/lib`` or the real home directory.
"""
from __future__ import annotations
from collections.abc import Generator
from pathlib import Path
@@ -6,77 +13,37 @@ from pathlib import Path
import pytest
from aegis_gitea_mcp.audit import reset_audit_logger
from aegis_gitea_mcp.auth import reset_validator
from aegis_gitea_mcp.authz import reset_authz_caches
from aegis_gitea_mcp.config import reset_settings
from aegis_gitea_mcp.oauth import reset_oauth_validator
from aegis_gitea_mcp.oauth_flow import reset_oauth_client_registry
from aegis_gitea_mcp.observability import reset_metrics_registry
from aegis_gitea_mcp.policy import reset_policy_engine
from aegis_gitea_mcp.config import Settings, reset_settings
from aegis_gitea_mcp.rate_limit import reset_rate_limiter
from aegis_gitea_mcp.request_context import clear_gitea_auth_context
from aegis_gitea_mcp.server import reset_repo_authz_cache
def _reset_singletons() -> None:
"""Reset every module-level singleton."""
reset_settings()
reset_audit_logger()
reset_rate_limiter()
clear_gitea_auth_context()
@pytest.fixture(autouse=True)
def reset_globals(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[None, None, None]:
"""Reset global singletons between tests and set up temp audit log."""
# Reset singletons before each test to ensure clean state
reset_settings()
reset_audit_logger()
reset_validator()
reset_oauth_validator()
reset_oauth_client_registry()
reset_repo_authz_cache()
reset_authz_caches()
reset_policy_engine()
reset_rate_limiter()
reset_metrics_registry()
clear_gitea_auth_context()
"""Isolate each test: clean env, tmp cwd, tmp state paths, fresh singletons."""
for field_name in Settings.model_fields:
monkeypatch.delenv(field_name.upper(), raising=False)
monkeypatch.chdir(tmp_path)
# Use temporary directory for audit logs in tests
audit_log_path = tmp_path / "audit.log"
monkeypatch.setenv("AUDIT_LOG_PATH", str(audit_log_path))
data_dir = tmp_path / "data"
monkeypatch.setenv("DATA_DIR", str(data_dir))
monkeypatch.setenv("AUDIT_LOG_PATH", str(data_dir / "audit.log"))
_reset_singletons()
yield
# Also reset after test for cleanup
reset_settings()
reset_audit_logger()
reset_validator()
reset_oauth_validator()
reset_oauth_client_registry()
reset_repo_authz_cache()
reset_authz_caches()
reset_policy_engine()
reset_rate_limiter()
reset_metrics_registry()
clear_gitea_auth_context()
_reset_singletons()
@pytest.fixture
def mock_env(monkeypatch: pytest.MonkeyPatch) -> None:
"""Set up mock environment variables for testing (standard API key mode)."""
"""Minimal valid environment: a Gitea URL and a token."""
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("GITEA_TOKEN", "test-token-12345")
monkeypatch.setenv("ENVIRONMENT", "test")
monkeypatch.setenv("MCP_HOST", "127.0.0.1")
monkeypatch.setenv("MCP_PORT", "8080")
monkeypatch.setenv("LOG_LEVEL", "DEBUG")
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false")
@pytest.fixture
def mock_env_oauth(monkeypatch: pytest.MonkeyPatch) -> None:
"""Set up mock environment variables for OAuth mode testing."""
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("ENVIRONMENT", "test")
monkeypatch.setenv("MCP_HOST", "127.0.0.1")
monkeypatch.setenv("MCP_PORT", "8080")
monkeypatch.setenv("LOG_LEVEL", "DEBUG")
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("OAUTH_STATE_SECRET", "test-state-secret-0123456789abcdef")
monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false")
+34 -4
View File
@@ -11,8 +11,6 @@ from aegis_gitea_mcp.audit import AuditLogger, validate_audit_log_integrity
def test_audit_log_integrity_valid(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Fresh audit log should validate with intact hash chain."""
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("GITEA_TOKEN", "token-123")
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
log_path = tmp_path / "audit.log"
logger = AuditLogger(log_path=log_path)
@@ -30,8 +28,6 @@ def test_audit_log_integrity_detects_tamper(
) -> None:
"""Integrity validation should fail when entries are modified."""
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("GITEA_TOKEN", "token-123")
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
log_path = tmp_path / "audit.log"
logger = AuditLogger(log_path=log_path)
@@ -48,3 +44,37 @@ def test_audit_log_integrity_detects_tamper(
valid, errors = validate_audit_log_integrity(log_path)
assert not valid
assert errors
def test_audit_logger_uses_configured_path(mock_env: None, tmp_path: Path) -> None:
"""Without an explicit path the logger writes to AUDIT_LOG_PATH (tmp in tests)."""
from aegis_gitea_mcp.audit import get_audit_logger
logger = get_audit_logger()
logger.log_access_denied("tool", reason="nope")
logger.log_security_event("evt", "description", metadata={"k": "v"})
assert logger.log_path == tmp_path / "data" / "audit.log"
assert len(logger.log_path.read_text(encoding="utf-8").splitlines()) == 2
valid, errors = validate_audit_log_integrity(logger.log_path)
assert valid, errors
def test_audit_masks_secrets_and_continues_chain(mock_env: None, tmp_path: Path) -> None:
"""Secrets are masked before persistence and a reopened log continues the chain."""
log_path = tmp_path / "chain.log"
first = AuditLogger(log_path=log_path)
first.log_tool_invocation("t", params={"note": "key AKIA1234567890ABCDEF"})
first.close()
second = AuditLogger(log_path=log_path)
second.log_tool_invocation("t", error="boom", result_status="error")
second.close()
text = log_path.read_text(encoding="utf-8")
assert "AKIA1234567890ABCDEF" not in text
valid, errors = validate_audit_log_integrity(log_path)
assert valid, errors
def test_validate_missing_log_is_valid(tmp_path: Path) -> None:
assert validate_audit_log_integrity(tmp_path / "absent.log") == (True, [])
+117
View File
@@ -0,0 +1,117 @@
"""Tests for the catalog data model."""
from __future__ import annotations
import json
from typing import Any
import pytest
from aegis_gitea_mcp.catalog import Operation, Parameter
def sample_dict() -> dict[str, Any]:
return {
"tool_name": "gitea_issue_create",
"domain": "issue",
"operation_id": "issueCreateIssue",
"method": "POST",
"path": "/repos/{owner}/{repo}/issues",
"summary": "Create an issue",
"description": "",
"parameters": [
{
"name": "owner",
"arg_name": "owner",
"location": "path",
"required": True,
"schema": {"type": "string"},
"is_file": False,
},
{
"name": "body",
"arg_name": "body",
"location": "body",
"required": False,
"schema": {"type": "object"},
},
],
"input_schema": {"type": "object", "additionalProperties": False},
"consumes": ["application/json"],
"produces": ["application/json"],
"read_only": False,
"destructive": False,
"idempotent": False,
"paginated": False,
}
def test_operation_round_trip_through_json() -> None:
data = sample_dict()
operation = Operation.from_dict(data)
assert operation.method == "POST"
assert isinstance(operation.parameters, tuple)
assert isinstance(operation.consumes, tuple)
assert operation.parameters[0].location == "path"
assert operation.parameters[1].is_file is False
rebuilt = Operation.from_dict(json.loads(json.dumps(operation.to_dict())))
assert rebuilt == operation
def test_to_dict_uses_lists_and_is_json_serializable() -> None:
dumped = Operation.from_dict(sample_dict()).to_dict()
assert isinstance(dumped["parameters"], list)
assert isinstance(dumped["consumes"], list)
json.dumps(dumped)
def test_method_is_upper_cased() -> None:
data = sample_dict()
data["method"] = "post"
assert Operation.from_dict(data).method == "POST"
def test_optional_fields_default() -> None:
data = sample_dict()
del data["summary"], data["description"], data["consumes"], data["produces"]
operation = Operation.from_dict(data)
assert operation.summary == ""
assert operation.consumes == ()
assert operation.produces == ()
def test_invalid_location_rejected() -> None:
with pytest.raises(ValueError, match="location"):
Parameter.from_dict(
{"name": "x", "arg_name": "x", "location": "header", "required": True, "schema": {}}
)
def test_missing_required_key_raises() -> None:
data = sample_dict()
del data["tool_name"]
with pytest.raises(KeyError):
Operation.from_dict(data)
def test_operation_is_frozen() -> None:
operation = Operation.from_dict(sample_dict())
with pytest.raises(AttributeError):
operation.method = "GET" # type: ignore[misc]
def test_file_parameter() -> None:
parameter = Parameter.from_dict(
{
"name": "attachment",
"arg_name": "attachment",
"location": "formData",
"required": True,
"schema": {"type": "string", "contentEncoding": "base64"},
"is_file": True,
}
)
assert parameter.is_file is True
assert parameter.to_dict()["location"] == "formData"
+37
View File
@@ -0,0 +1,37 @@
"""Tests for the command line skeleton."""
from __future__ import annotations
from pathlib import Path
import pytest
from aegis_gitea_mcp import __version__
from aegis_gitea_mcp.cli import build_parser, main
def test_version_prints_package_version(capsys: pytest.CaptureFixture[str]) -> None:
assert main(["version"]) == 0
assert capsys.readouterr().out.strip() == __version__
@pytest.mark.parametrize("argv", [[], ["serve"], ["serve", "--transport", "http"], ["setup"]])
def test_unbuilt_commands_exit_2(argv: list[str], capsys: pytest.CaptureFixture[str]) -> None:
assert main(argv) == 2
captured = capsys.readouterr()
assert "not implemented yet" in captured.err
assert captured.out == ""
def test_serve_transport_choices() -> None:
parser = build_parser()
assert parser.parse_args(["serve", "--transport", "stdio"]).transport == "stdio"
with pytest.raises(SystemExit):
parser.parse_args(["serve", "--transport", "sse"])
def test_package_version_matches_pyproject() -> None:
tomllib = pytest.importorskip("tomllib") # Python 3.11+
data = tomllib.loads((Path(__file__).parent.parent / "pyproject.toml").read_text())
assert data["project"]["version"] == __version__
+358 -105
View File
@@ -1,138 +1,391 @@
"""Tests for configuration management."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from aegis_gitea_mcp.config import get_settings, reset_settings
from aegis_gitea_mcp.config import (
DEFAULT_OAUTH_SCOPES,
GITEA_SCOPES,
Settings,
get_settings,
is_loopback_host,
reset_settings,
)
def test_settings_from_env(mock_env: None) -> None:
"""Test loading settings from environment variables."""
def make(monkeypatch: pytest.MonkeyPatch, **env: str) -> Settings:
"""Build fresh settings from the given environment."""
for key, value in env.items():
monkeypatch.setenv(key, value)
reset_settings()
return get_settings()
def test_defaults(mock_env: None, tmp_path) -> None:
settings = get_settings()
assert settings.gitea_base_url == "https://gitea.example.com"
assert settings.gitea_token == "test-token-12345"
assert settings.gitea_url == "https://gitea.example.com"
assert settings.gitea_api_url == "https://gitea.example.com/api/v1"
assert settings.gitea_token is not None
assert settings.gitea_token.get_secret_value() == "test-token-12345"
assert settings.mcp_transport == "stdio"
assert settings.mcp_host == "127.0.0.1"
assert settings.mcp_port == 8080
assert settings.log_level == "DEBUG"
def test_settings_defaults(monkeypatch: pytest.MonkeyPatch) -> None:
"""Test default values when not specified."""
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("GITEA_TOKEN", "test-token")
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
settings = get_settings()
assert settings.mcp_host == "127.0.0.1"
assert settings.mcp_port == 8080
assert settings.log_level == "INFO"
assert settings.max_file_size_bytes == 1_048_576
assert settings.mcp_path == "/mcp"
assert settings.mcp_bearer_token is None
assert settings.public_base_url is None
assert settings.allowed_hosts == []
assert settings.oauth_enabled is False
assert settings.read_only is False
assert settings.toolsets == ["all"]
assert settings.disabled_tools == []
assert settings.request_timeout_seconds == 30
assert settings.max_tool_response_items == 200
assert settings.max_tool_response_chars == 60000
assert settings.max_binary_bytes == 5_242_880
assert settings.secret_detection_mode == "mask"
assert settings.log_level == "INFO"
assert settings.expose_error_details is False
assert settings.rate_limit_per_minute == 120
assert ",".join(settings.oauth_scopes) == DEFAULT_OAUTH_SCOPES
def test_settings_validation_missing_required(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
"""Test that missing required fields raise validation errors."""
def test_data_dir_defaults_and_audit_path(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("DATA_DIR")
monkeypatch.delenv("AUDIT_LOG_PATH")
settings = make(monkeypatch, GITEA_URL="https://g.example.com")
monkeypatch.delenv("GITEA_URL", raising=False)
monkeypatch.delenv("GITEA_TOKEN", raising=False)
# Change to tmp directory so .env file won't be found
monkeypatch.chdir(tmp_path)
reset_settings()
with pytest.raises(ValidationError):
get_settings()
assert str(settings.data_dir).endswith(".local/state/aegis-gitea-mcp")
assert "~" not in str(settings.data_dir)
assert settings.resolved_audit_log_path == settings.data_dir / "audit.log"
def test_settings_invalid_log_level(monkeypatch: pytest.MonkeyPatch) -> None:
"""Test that invalid log levels are rejected."""
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("GITEA_TOKEN", "test-token")
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
monkeypatch.setenv("LOG_LEVEL", "INVALID")
def test_audit_path_follows_data_dir(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
monkeypatch.delenv("AUDIT_LOG_PATH")
settings = make(monkeypatch, GITEA_URL="https://g.example.com", DATA_DIR=str(tmp_path / "d"))
reset_settings()
with pytest.raises(ValidationError):
get_settings()
assert settings.resolved_audit_log_path == tmp_path / "d" / "audit.log"
def test_settings_empty_token(monkeypatch: pytest.MonkeyPatch) -> None:
"""Test that empty tokens are rejected."""
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("GITEA_TOKEN", " ")
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
def test_blank_values_are_unset(monkeypatch: pytest.MonkeyPatch) -> None:
settings = make(
monkeypatch,
GITEA_URL="https://g.example.com",
GITEA_TOKEN="",
MCP_BEARER_TOKEN=" ",
PUBLIC_BASE_URL="",
AUDIT_LOG_PATH="",
)
reset_settings()
with pytest.raises(ValidationError):
get_settings()
assert settings.gitea_token is None
assert settings.mcp_bearer_token is None
assert settings.public_base_url is None
assert settings.audit_log_path is None
def test_settings_singleton(mock_env: None) -> None:
"""Test that get_settings returns same instance."""
settings1 = get_settings()
settings2 = get_settings()
assert settings1 is settings2
def test_env_names_are_case_insensitive(monkeypatch: pytest.MonkeyPatch) -> None:
settings = make(monkeypatch, gitea_url="https://g.example.com", mcp_port="9000")
assert settings.mcp_port == 9000
def test_write_mode_requires_whitelist_or_allow_all(monkeypatch: pytest.MonkeyPatch) -> None:
"""Write mode without whitelist must be rejected unless allow-all is enabled."""
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("GITEA_TOKEN", "test-token")
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
monkeypatch.setenv("WRITE_MODE", "true")
monkeypatch.delenv("WRITE_REPOSITORY_WHITELIST", raising=False)
monkeypatch.setenv("WRITE_ALLOW_ALL_TOKEN_REPOS", "false")
reset_settings()
with pytest.raises(ValidationError):
get_settings()
def test_write_mode_allows_all_token_repos(monkeypatch: pytest.MonkeyPatch) -> None:
"""Allow-all mode should pass validation without explicit repository whitelist."""
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("GITEA_TOKEN", "test-token")
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
monkeypatch.setenv("WRITE_MODE", "true")
monkeypatch.delenv("WRITE_REPOSITORY_WHITELIST", raising=False)
monkeypatch.setenv("WRITE_ALLOW_ALL_TOKEN_REPOS", "true")
def test_reads_dotenv_file(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
(tmp_path / ".env").write_text("GITEA_URL=https://dotenv.example.com\nREAD_ONLY=true\n")
reset_settings()
settings = get_settings()
assert settings.write_allow_all_token_repos is True
assert settings.gitea_url == "https://dotenv.example.com"
assert settings.read_only is True
def _oauth_env(monkeypatch: pytest.MonkeyPatch) -> None:
"""Apply a minimal, valid OAuth-mode environment."""
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("STARTUP_VALIDATE_GITEA", "false")
def test_oauth_state_secret_too_short_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
"""OAUTH_STATE_SECRET shorter than 32 characters must fail validation."""
_oauth_env(monkeypatch)
monkeypatch.setenv("OAUTH_STATE_SECRET", "short-secret")
reset_settings()
with pytest.raises(ValidationError, match="at least 32 characters"):
def test_missing_gitea_url_is_rejected() -> None:
with pytest.raises(ValidationError):
get_settings()
def test_oauth_state_secret_minimum_length_accepted(monkeypatch: pytest.MonkeyPatch) -> None:
"""A 32+ character OAUTH_STATE_SECRET passes validation."""
_oauth_env(monkeypatch)
monkeypatch.setenv("OAUTH_STATE_SECRET", "x" * 32)
def test_settings_singleton_and_reset(mock_env: None) -> None:
first = get_settings()
assert get_settings() is first
reset_settings()
settings = get_settings()
assert settings.oauth_mode is True
assert len(settings.oauth_state_secret) == 32
assert get_settings() is not first
@pytest.mark.parametrize(
("raw", "expected"),
[
("https://gitea.example.com", "https://gitea.example.com"),
("https://gitea.example.com/", "https://gitea.example.com"),
("http://localhost:3000", "http://localhost:3000"),
("https://gitea.example.com/api/v1", "https://gitea.example.com"),
("https://gitea.example.com/api/v1/", "https://gitea.example.com"),
("https://example.com/gitea", "https://example.com/gitea"),
("https://example.com/gitea/api/v1", "https://example.com/gitea"),
],
)
def test_gitea_url_normalised(monkeypatch: pytest.MonkeyPatch, raw: str, expected: str) -> None:
assert make(monkeypatch, GITEA_URL=raw).gitea_url == expected
@pytest.mark.parametrize(
"raw",
[
"ftp://gitea.example.com",
"gitea.example.com",
"https://",
"https://user:pw@gitea.example.com",
"https://user@gitea.example.com",
"https://gitea.example.com?x=1",
"https://gitea.example.com/#frag",
"https://gitea.example.com:99999",
"file:///etc/passwd",
],
)
def test_gitea_url_rejected(monkeypatch: pytest.MonkeyPatch, raw: str) -> None:
monkeypatch.setenv("GITEA_URL", raw)
with pytest.raises(ValidationError):
get_settings()
def test_bearer_token_minimum_length(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
monkeypatch.setenv("MCP_BEARER_TOKEN", "short")
with pytest.raises(ValidationError, match="at least 32"):
get_settings()
settings = make(monkeypatch, MCP_BEARER_TOKEN="x" * 32)
assert settings.mcp_bearer_token is not None
def test_encryption_key_minimum_length(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
monkeypatch.setenv("TOKEN_ENCRYPTION_KEY", "short")
with pytest.raises(ValidationError, match="TOKEN_ENCRYPTION_KEY"):
get_settings()
@pytest.mark.parametrize(
"raw",
[
"https://mcp.example.com",
"https://mcp.example.com/",
"http://localhost:8080",
"http://127.0.0.1:8080",
"http://[::1]:8080",
],
)
def test_public_base_url_accepted(monkeypatch: pytest.MonkeyPatch, raw: str) -> None:
settings = make(monkeypatch, GITEA_URL="https://g.example.com", PUBLIC_BASE_URL=raw)
assert settings.public_base_url == raw.rstrip("/")
@pytest.mark.parametrize(
"raw",
[
"http://mcp.example.com",
"mcp.example.com",
"https://mcp.example.com?a=b",
"https://u:p@mcp.example.com",
],
)
def test_public_base_url_rejected(monkeypatch: pytest.MonkeyPatch, raw: str) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
monkeypatch.setenv("PUBLIC_BASE_URL", raw)
with pytest.raises(ValidationError):
get_settings()
def test_mcp_path_validation(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
assert make(monkeypatch, MCP_PATH="/api/mcp/").mcp_path == "/api/mcp"
for bad in ("mcp", "//evil", "/a?b", "/", "/a b"):
monkeypatch.setenv("MCP_PATH", bad)
reset_settings()
with pytest.raises(ValidationError):
get_settings()
def test_port_and_transport_validation(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
monkeypatch.setenv("MCP_PORT", "70000")
with pytest.raises(ValidationError):
get_settings()
monkeypatch.setenv("MCP_PORT", "8080")
monkeypatch.setenv("MCP_TRANSPORT", "sse")
with pytest.raises(ValidationError):
get_settings()
assert make(monkeypatch, MCP_TRANSPORT="http").mcp_transport == "http"
def test_comma_separated_lists(monkeypatch: pytest.MonkeyPatch) -> None:
settings = make(
monkeypatch,
GITEA_URL="https://g.example.com",
ALLOWED_HOSTS="mcp.example.com, other.example.com:8443,mcp.example.com",
TOOLSETS="Repo, issue",
DISABLED_TOOLS="Gitea_Repo_Delete, gitea_admin_delete_user",
)
assert settings.allowed_hosts == ["mcp.example.com", "other.example.com:8443"]
assert settings.toolsets == ["repo", "issue"]
assert settings.disabled_tools == ["gitea_repo_delete", "gitea_admin_delete_user"]
def test_unknown_toolset_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
monkeypatch.setenv("TOOLSETS", "repo,bogus")
with pytest.raises(ValidationError, match="bogus"):
get_settings()
def test_empty_toolsets_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
monkeypatch.setenv("TOOLSETS", " , ")
with pytest.raises(ValidationError):
get_settings()
def test_default_oauth_scopes_are_valid() -> None:
assert set(DEFAULT_OAUTH_SCOPES.split(",")) <= GITEA_SCOPES
assert len(GITEA_SCOPES) == 18
def test_oauth_scopes_custom_and_dedup(monkeypatch: pytest.MonkeyPatch) -> None:
settings = make(
monkeypatch,
GITEA_URL="https://g.example.com",
OAUTH_SCOPES="read:repository write:repository,read:repository",
)
assert settings.oauth_scopes == ["read:repository", "write:repository"]
@pytest.mark.parametrize(
"scope", ["all", "repo", "read:bogus", "write:repository,admin", "public-only"]
)
def test_unknown_oauth_scope_rejected(monkeypatch: pytest.MonkeyPatch, scope: str) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
monkeypatch.setenv("OAUTH_SCOPES", scope)
with pytest.raises(ValidationError, match="unknown scope"):
get_settings()
def test_empty_oauth_scopes_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
monkeypatch.setenv("OAUTH_SCOPES", ",")
with pytest.raises(ValidationError, match="must not be empty"):
get_settings()
def test_log_level_normalised_and_validated(monkeypatch: pytest.MonkeyPatch) -> None:
assert (
make(monkeypatch, GITEA_URL="https://g.example.com", LOG_LEVEL="debug").log_level == "DEBUG"
)
monkeypatch.setenv("LOG_LEVEL", "loud")
reset_settings()
with pytest.raises(ValidationError):
get_settings()
def test_secret_detection_mode_validated(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
monkeypatch.setenv("SECRET_DETECTION_MODE", "block")
assert get_settings().secret_detection_mode == "block"
reset_settings()
monkeypatch.setenv("SECRET_DETECTION_MODE", "weird")
with pytest.raises(ValidationError):
get_settings()
@pytest.mark.parametrize(
("host", "expected"),
[
("127.0.0.1", True),
("127.5.6.7", True),
("::1", True),
("[::1]", True),
("localhost", True),
("LOCALHOST", True),
("0.0.0.0", False), # noqa: S104
("::", False),
("192.168.1.5", False),
("gitea.example.com", False),
],
)
def test_is_loopback(monkeypatch: pytest.MonkeyPatch, host: str, expected: bool) -> None:
assert is_loopback_host(host) is expected
settings = make(monkeypatch, GITEA_URL="https://g.example.com", MCP_HOST=host)
assert settings.is_loopback_bind() is expected
def test_oauth_config_errors(monkeypatch: pytest.MonkeyPatch) -> None:
settings = make(monkeypatch, GITEA_URL="https://g.example.com")
assert settings.oauth_config_errors() == []
settings = make(monkeypatch, OAUTH_ENABLED="true")
errors = settings.oauth_config_errors()
assert len(errors) == 4
assert any("GITEA_OAUTH_CLIENT_ID" in e for e in errors)
assert any("GITEA_OAUTH_CLIENT_SECRET" in e for e in errors)
assert any("TOKEN_ENCRYPTION_KEY" in e for e in errors)
assert any("PUBLIC_BASE_URL" in e for e in errors)
settings = make(
monkeypatch,
GITEA_OAUTH_CLIENT_ID="cid",
GITEA_OAUTH_CLIENT_SECRET="csecret",
TOKEN_ENCRYPTION_KEY="k" * 32,
PUBLIC_BASE_URL="https://mcp.example.com",
)
assert settings.oauth_config_errors() == []
def test_http_auth_configured(monkeypatch: pytest.MonkeyPatch) -> None:
assert make(monkeypatch, GITEA_URL="https://g.example.com").http_auth_configured() is False
assert make(monkeypatch, MCP_BEARER_TOKEN="b" * 32).http_auth_configured() is True
monkeypatch.delenv("MCP_BEARER_TOKEN")
assert make(monkeypatch, OAUTH_ENABLED="true").http_auth_configured() is False
settings = make(
monkeypatch,
GITEA_OAUTH_CLIENT_ID="cid",
GITEA_OAUTH_CLIENT_SECRET="csecret",
TOKEN_ENCRYPTION_KEY="k" * 32,
PUBLIC_BASE_URL="https://mcp.example.com",
)
assert settings.http_auth_configured() is True
def test_secrets_never_appear_in_repr_or_dump(monkeypatch: pytest.MonkeyPatch) -> None:
secrets = {
"GITEA_TOKEN": "gitea-token-SECRET-aaaaaaaaaaaa",
"MCP_BEARER_TOKEN": "bearer-SECRET-bbbbbbbbbbbbbbbbbbbbbbbbbb",
"GITEA_OAUTH_CLIENT_SECRET": "client-SECRET-cccccccccccc",
"TOKEN_ENCRYPTION_KEY": "encryption-SECRET-dddddddddddddddddddd",
}
settings = make(monkeypatch, GITEA_URL="https://g.example.com", **secrets)
rendered = [
repr(settings),
str(settings),
str(settings.model_dump()),
settings.model_dump_json(),
str(settings.model_dump(mode="json")),
]
for text in rendered:
for secret in secrets.values():
assert secret not in text
assert settings.gitea_token is not None
assert settings.gitea_token.get_secret_value() == secrets["GITEA_TOKEN"]
def test_env_example_is_valid() -> None:
"""The documented example file must load as-is (blank values mean unset)."""
from pathlib import Path
example = Path(__file__).parent.parent / ".env.example"
settings = Settings(_env_file=example) # type: ignore[call-arg]
assert settings.gitea_url == "https://gitea.example.com"
assert settings.gitea_token is None
assert settings.toolsets == ["all"]
assert ",".join(settings.oauth_scopes) == DEFAULT_OAUTH_SCOPES
+73
View File
@@ -0,0 +1,73 @@
"""Tests for error types and message sanitization."""
from __future__ import annotations
import pytest
from aegis_gitea_mcp.errors import GiteaAPIError, ToolError, kind_for_status, sanitize_message
def test_tool_error_defaults() -> None:
error = ToolError("nope")
assert str(error) == "nope"
assert error.detail == "nope"
assert error.status_code == 400
assert ToolError("x", status_code=403).status_code == 403
@pytest.mark.parametrize(
("status", "kind"),
[
(400, "bad_request"),
(401, "unauthorized"),
(403, "forbidden"),
(404, "not_found"),
(409, "conflict"),
(422, "validation_failed"),
(429, "rate_limited"),
(418, "client_error"),
(500, "server_error"),
(503, "server_error"),
(0, "error"),
],
)
def test_kind_for_status(status: int, kind: str) -> None:
assert kind_for_status(status) == kind
def test_gitea_api_error_fields() -> None:
error = GiteaAPIError(404, "repository does not exist")
assert isinstance(error, ToolError)
assert error.status == 404
assert error.status_code == 404
assert error.kind == "not_found"
assert error.message == "repository does not exist"
assert error.to_dict() == {
"status": 404,
"kind": "not_found",
"message": "repository does not exist",
}
def test_gitea_api_error_explicit_kind() -> None:
assert GiteaAPIError(0, "connection refused", kind="network_error").kind == "network_error"
def test_gitea_api_error_sanitizes_message() -> None:
error = GiteaAPIError(401, "bad header Authorization: Bearer abcdef1234567890abcdef\nretry")
assert "abcdef1234567890abcdef" not in error.message
assert "\n" not in error.message
assert error.detail == error.message
def test_sanitize_message_masks_secret_patterns() -> None:
message = sanitize_message("failed with key AKIA1234567890ABCDEF in body")
assert "AKIA1234567890ABCDEF" not in message
def test_sanitize_message_truncates() -> None:
message = sanitize_message("x" * 2000, max_chars=100)
assert len(message) == 100
assert message.endswith("...")
+31
View File
@@ -0,0 +1,31 @@
"""Tests for the identity type."""
from __future__ import annotations
import dataclasses
import pytest
from aegis_gitea_mcp.identity import Identity
def test_metadata_never_contains_token() -> None:
identity = Identity(login="alice", mode="oauth", token="super-secret-token")
assert identity.metadata() == {"login": "alice", "mode": "oauth", "sudo": None}
assert identity.metadata(sudo="bob") == {"login": "alice", "mode": "oauth", "sudo": "bob"}
assert "super-secret-token" not in str(identity.metadata(sudo="bob"))
def test_token_hidden_from_repr() -> None:
identity = Identity(login=None, mode="admin-token", token="super-secret-token")
assert "super-secret-token" not in repr(identity)
assert "super-secret-token" not in str(identity)
assert "login=None" in repr(identity)
def test_identity_is_frozen() -> None:
identity = Identity(login="a", mode="admin-token", token="t")
with pytest.raises(dataclasses.FrozenInstanceError):
identity.login = "b" # type: ignore[misc]
+1 -51
View File
@@ -1,4 +1,4 @@
"""Tests for structured logging helpers and get_issue instrumentation (#14)."""
"""Tests for structured logging helpers."""
from __future__ import annotations
@@ -7,26 +7,12 @@ import logging
import pytest
from aegis_gitea_mcp.config import reset_settings
from aegis_gitea_mcp.logging_utils import (
JsonLogFormatter,
log_event,
log_nullable_field,
sanitize_context,
)
from aegis_gitea_mcp.tools.read_tools import get_issue_tool
READ_TOOLS_LOGGER = "aegis_gitea_mcp.tools.read_tools"
@pytest.fixture(autouse=True)
def tool_env(monkeypatch: pytest.MonkeyPatch) -> None:
"""Provide minimal settings environment for response limit helpers."""
reset_settings()
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("GITEA_TOKEN", "test-token")
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
monkeypatch.setenv("ENVIRONMENT", "test")
def test_sanitize_context_masks_sensitive_keys() -> None:
@@ -82,39 +68,3 @@ def test_log_nullable_field_characterizes_value(caplog: pytest.LogCaptureFixture
"is_none": False,
"value_type": "list",
}
class _NullIssueGitea:
"""Minimal stub returning an issue with null collection fields."""
async def get_issue(self, owner: str, repo: str, index: int) -> dict[str, object]:
return {
"number": index,
"title": "Issue",
"body": "Body",
"state": "open",
"labels": None,
"assignees": None,
"user": None,
}
@pytest.mark.asyncio
async def test_get_issue_tool_emits_debug_events(caplog: pytest.LogCaptureFixture) -> None:
"""get_issue_tool emits start/shape/field-check debug events and still parses."""
with caplog.at_level(logging.DEBUG, logger=READ_TOOLS_LOGGER):
result = await get_issue_tool(
_NullIssueGitea(), {"owner": "acme", "repo": "app", "issue_number": 7}
)
events = [r.getMessage() for r in caplog.records]
assert "get_issue.start" in events
assert "get_issue.payload_shape" in events
field_checks = [r.context for r in caplog.records if r.getMessage() == "get_issue.field_check"]
assert {"field": "labels", "is_none": True, "value_type": None} in field_checks
assert {"field": "user", "is_none": True, "value_type": None} in field_checks
# Logging must not change behaviour: null collections still parse to empties.
assert result["labels"] == []
assert result["assignees"] == []
+19
View File
@@ -0,0 +1,19 @@
"""Guard: the test environment must never point state at system directories."""
from __future__ import annotations
from pathlib import Path
from aegis_gitea_mcp.audit import get_audit_logger
from aegis_gitea_mcp.config import Settings
def test_state_paths_are_inside_tmp(mock_env: None, tmp_path: Path) -> None:
settings = Settings() # type: ignore[call-arg]
assert settings.data_dir.is_relative_to(tmp_path)
assert settings.resolved_audit_log_path.is_relative_to(tmp_path)
assert get_audit_logger().log_path.is_relative_to(tmp_path)
def test_cwd_is_isolated(tmp_path: Path) -> None:
assert Path.cwd() == tmp_path
+55
View File
@@ -0,0 +1,55 @@
"""Tests for the request rate limiter."""
from __future__ import annotations
import pytest
from aegis_gitea_mcp.audit import validate_audit_log_integrity
from aegis_gitea_mcp.config import get_settings
from aegis_gitea_mcp.rate_limit import (
RequestRateLimiter,
SlidingWindowLimiter,
get_rate_limiter,
reset_rate_limiter,
)
def test_sliding_window_blocks_after_limit(monkeypatch: pytest.MonkeyPatch) -> None:
now = [1000.0]
monkeypatch.setattr("aegis_gitea_mcp.rate_limit.time.time", lambda: now[0])
limiter = SlidingWindowLimiter(max_requests=2, window_seconds=60)
assert limiter.allow("k")
assert limiter.allow("k")
assert not limiter.allow("k")
assert limiter.allow("other")
now[0] += 61
assert limiter.allow("k")
def test_request_limiter_per_ip_and_token(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_URL", "https://g.example.com")
monkeypatch.setenv("RATE_LIMIT_PER_MINUTE", "2")
limiter = RequestRateLimiter()
assert limiter.check("1.1.1.1", "tok").allowed
assert limiter.check("1.1.1.1", None).allowed
assert not limiter.check("1.1.1.1", "tok").allowed
# A different IP with the same token trips the per-token limiter.
other = limiter.check("2.2.2.2", "tok")
assert other.allowed or other.reason == "Per-token rate limit exceeded"
third = limiter.check("3.3.3.3", "tok")
assert not third.allowed
assert third.reason == "Per-token rate limit exceeded"
valid, errors = validate_audit_log_integrity(get_settings().resolved_audit_log_path)
assert valid, errors
def test_global_limiter_singleton(mock_env: None) -> None:
first = get_rate_limiter()
assert get_rate_limiter() is first
reset_rate_limiter()
assert get_rate_limiter() is not first
+35
View File
@@ -0,0 +1,35 @@
"""Tests for request-scoped context variables."""
from __future__ import annotations
from aegis_gitea_mcp.request_context import (
clear_gitea_auth_context,
get_gitea_user_login,
get_gitea_user_scopes,
get_gitea_user_token,
get_request_id,
set_gitea_user_login,
set_gitea_user_scopes,
set_gitea_user_token,
set_request_id,
)
def test_request_id_roundtrip() -> None:
set_request_id("req-1")
assert get_request_id() == "req-1"
def test_auth_context_set_and_clear() -> None:
set_gitea_user_token("tok")
set_gitea_user_login("alice")
set_gitea_user_scopes([" write:repository", "read:user", "read:user", ""])
assert get_gitea_user_token() == "tok"
assert get_gitea_user_login() == "alice"
assert get_gitea_user_scopes() == ("read:user", "write:repository")
clear_gitea_auth_context()
assert get_gitea_user_token() is None
assert get_gitea_user_login() is None
assert get_gitea_user_scopes() == ()
+46
View File
@@ -0,0 +1,46 @@
"""Tests for response size bounds."""
from __future__ import annotations
import pytest
from aegis_gitea_mcp.config import reset_settings
from aegis_gitea_mcp.response_limits import ResponseLimitError, limit_items, limit_text
def test_limit_items_uses_configured_default(
mock_env: None, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("MAX_TOOL_RESPONSE_ITEMS", "3")
reset_settings()
items = [{"i": n} for n in range(5)]
trimmed, omitted = limit_items(items)
assert trimmed == items[:3]
assert omitted == 2
def test_limit_items_within_bound(mock_env: None) -> None:
items = [{"i": 1}]
assert limit_items(items) == (items, 0)
assert limit_items(items, configured_limit=1) == (items, 0)
assert limit_items([{"i": 1}, {"i": 2}], configured_limit=1) == ([{"i": 1}], 1)
def test_limit_text(mock_env: None) -> None:
assert limit_text("abcdef", configured_limit=3) == "abc"
assert limit_text("abc", configured_limit=10) == "abc"
def test_limit_text_default_from_settings(mock_env: None, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("MAX_TOOL_RESPONSE_CHARS", "5")
reset_settings()
assert limit_text("abcdefghij") == "abcde"
def test_invalid_limits_raise(mock_env: None) -> None:
with pytest.raises(ResponseLimitError):
limit_items([], configured_limit=-1)
with pytest.raises(ResponseLimitError):
limit_text("x", configured_limit=-1)
+56
View File
@@ -0,0 +1,56 @@
"""Tests for scripts/validate_audit_log.py."""
from __future__ import annotations
import importlib.util
from pathlib import Path
from types import ModuleType
import pytest
from aegis_gitea_mcp.audit import AuditLogger
SCRIPT = Path(__file__).parent.parent / "scripts" / "validate_audit_log.py"
def load_script() -> ModuleType:
spec = importlib.util.spec_from_file_location("validate_audit_log", SCRIPT)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def test_default_path_from_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
script = load_script()
monkeypatch.setenv("AUDIT_LOG_PATH", str(tmp_path / "x.log"))
assert script.default_audit_path() == tmp_path / "x.log"
monkeypatch.delenv("AUDIT_LOG_PATH")
monkeypatch.setenv("DATA_DIR", str(tmp_path / "d"))
assert script.default_audit_path() == tmp_path / "d" / "audit.log"
monkeypatch.delenv("DATA_DIR")
assert script.default_audit_path().name == "audit.log"
assert "aegis-gitea-mcp" in str(script.default_audit_path())
def test_main_reports_ok_and_failure(
mock_env: None,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
script = load_script()
log_path = tmp_path / "audit.log"
logger = AuditLogger(log_path=log_path)
logger.log_tool_invocation("t")
logger.close()
monkeypatch.setattr("sys.argv", ["x", "--path", str(log_path)])
assert script.main() == 0
assert "integrity OK" in capsys.readouterr().out
log_path.write_text(log_path.read_text().replace('"t"', '"u"'), encoding="utf-8")
assert script.main() == 1
assert "FAILED" in capsys.readouterr().out