AegisGitea-MCP/tests/test_automation.py

"""Tests for automation endpoints and controls."""

from pathlib import Path

import pytest
from fastapi.testclient import TestClient


@pytest.fixture
def allow_oauth(monkeypatch: pytest.MonkeyPatch) -> None:
    """Mock OAuth validation to return a deterministic authenticated principal."""

    async def _validate(_self, token, _ip, _ua):
        if token == "a" * 64:
            return True, None, {"login": "automation-user", "scopes": ["read:repository"]}
        return False, "Invalid or expired OAuth token.", None

    monkeypatch.setattr(
        "aegis_gitea_mcp.oauth.GiteaOAuthValidator.validate_oauth_token",
        _validate,
    )


def _set_base_env(
    monkeypatch: pytest.MonkeyPatch, automation_enabled: bool, policy_path: Path
) -> None:
    monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
    monkeypatch.setenv("GITEA_TOKEN", "test-token")
    monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
    monkeypatch.setenv("AUTH_ENABLED", "true")
    monkeypatch.setenv("ENVIRONMENT", "test")
    monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false")
    monkeypatch.setenv("AUTOMATION_ENABLED", "true" if automation_enabled else "false")
    monkeypatch.setenv("POLICY_FILE_PATH", str(policy_path))


def test_automation_job_denied_when_disabled(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, allow_oauth: None
) -> None:
    """Automation endpoints should deny requests when automation mode is disabled."""
    policy_path = tmp_path / "policy.yaml"
    policy_path.write_text("defaults:\n  read: allow\n  write: deny\n", encoding="utf-8")
    _set_base_env(monkeypatch, automation_enabled=False, policy_path=policy_path)

    from aegis_gitea_mcp.server import app

    client = TestClient(app)
    response = client.post(
        "/automation/jobs/run",
        headers={"Authorization": f"Bearer {'a' * 64}"},
        json={"job_name": "dependency_hygiene_scan", "owner": "acme", "repo": "app"},
    )

    assert response.status_code == 403
    assert "disabled" in response.json()["detail"]


def test_automation_job_executes_when_enabled(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, allow_oauth: None
) -> None:
    """Dependency scan job should execute when automation is enabled and policy allows it."""
    policy_path = tmp_path / "policy.yaml"
    policy_path.write_text(
        """
defaults:
  read: allow
  write: deny
tools:
  allow:
    - automation_dependency_hygiene_scan
    - automation_webhook_ingest
""".strip() + "\n",
        encoding="utf-8",
    )
    _set_base_env(monkeypatch, automation_enabled=True, policy_path=policy_path)

    from aegis_gitea_mcp.server import app

    client = TestClient(app)
    response = client.post(
        "/automation/jobs/run",
        headers={"Authorization": f"Bearer {'a' * 64}"},
        json={"job_name": "dependency_hygiene_scan", "owner": "acme", "repo": "app"},
    )

    assert response.status_code == 200
    payload = response.json()
    assert payload["success"] is True
    assert payload["result"]["job"] == "dependency_hygiene_scan"


def test_automation_webhook_policy_denied(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, allow_oauth: None
) -> None:
    """Webhook ingestion must respect policy deny rules."""
    policy_path = tmp_path / "policy.yaml"
    policy_path.write_text(
        """
defaults:
  read: allow
  write: deny
tools:
  deny:
    - automation_webhook_ingest
""".strip() + "\n",
        encoding="utf-8",
    )
    _set_base_env(monkeypatch, automation_enabled=True, policy_path=policy_path)

    from aegis_gitea_mcp.server import app

    client = TestClient(app)
    response = client.post(
        "/automation/webhook",
        headers={"Authorization": f"Bearer {'a' * 64}"},
        json={"event_type": "scan.completed", "payload": {"status": "ok"}},
    )

    assert response.status_code == 403
    assert "policy denied" in response.json()["detail"].lower()


def test_auto_issue_creation_denied_without_write_mode(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, allow_oauth: None
) -> None:
    """Auto issue creation job should be denied unless write mode is enabled."""
    policy_path = tmp_path / "policy.yaml"
    policy_path.write_text(
        """
defaults:
  read: allow
  write: allow
tools:
  allow:
    - automation_auto_issue_creation
""".strip() + "\n",
        encoding="utf-8",
    )
    _set_base_env(monkeypatch, automation_enabled=True, policy_path=policy_path)

    from aegis_gitea_mcp.server import app

    client = TestClient(app)
    response = client.post(
        "/automation/jobs/run",
        headers={"Authorization": f"Bearer {'a' * 64}"},
        json={"job_name": "auto_issue_creation", "owner": "acme", "repo": "app"},
    )

    assert response.status_code == 403
    assert "write mode is disabled" in response.json()["detail"].lower()