latte
59e1ea53a8
Introduce a GiteaOAuthValidator for JWT and userinfo validation and fallbacks, add /oauth/token proxy, and thread per-user tokens through the request context and automation paths. Update config and .env.example for OAuth-first mode, add OpenAPI, extensive unit/integration tests, GitHub/Gitea CI workflows, docs, and lint/test enforcement (>=80% cov).
169 lines
7.2 KiB
Python
169 lines
7.2 KiB
Python
"""Unit tests for Gitea client request behavior."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from httpx import Request, Response
|
|
|
|
from aegis_gitea_mcp.config import reset_settings
|
|
from aegis_gitea_mcp.gitea_client import (
|
|
GiteaAuthenticationError,
|
|
GiteaAuthorizationError,
|
|
GiteaClient,
|
|
GiteaError,
|
|
GiteaNotFoundError,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def gitea_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Provide minimal environment for client initialization."""
|
|
reset_settings()
|
|
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
|
|
monkeypatch.setenv("GITEA_TOKEN", "legacy-token")
|
|
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
|
|
monkeypatch.setenv("ENVIRONMENT", "test")
|
|
yield
|
|
reset_settings()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_context_uses_bearer_header() -> None:
|
|
"""HTTP client is created with bearer token and closed on exit."""
|
|
with patch("aegis_gitea_mcp.gitea_client.AsyncClient") as mock_async_client:
|
|
mock_instance = AsyncMock()
|
|
mock_async_client.return_value = mock_instance
|
|
|
|
async with GiteaClient(token="user-oauth-token"):
|
|
pass
|
|
|
|
_, kwargs = mock_async_client.call_args
|
|
assert kwargs["headers"]["Authorization"] == "Bearer user-oauth-token"
|
|
mock_instance.aclose.assert_awaited_once()
|
|
|
|
|
|
def test_client_requires_non_empty_token() -> None:
|
|
"""Client construction fails when token is missing."""
|
|
with pytest.raises(ValueError, match="non-empty"):
|
|
GiteaClient(token=" ")
|
|
|
|
|
|
def test_handle_response_maps_error_codes() -> None:
|
|
"""HTTP status codes map to explicit domain exceptions."""
|
|
client = GiteaClient(token="user-token")
|
|
request = Request("GET", "https://gitea.example.com/api/v1/user")
|
|
|
|
with pytest.raises(GiteaAuthenticationError):
|
|
client._handle_response(Response(401, request=request), correlation_id="c1")
|
|
|
|
with pytest.raises(GiteaAuthorizationError):
|
|
client._handle_response(Response(403, request=request), correlation_id="c2")
|
|
|
|
with pytest.raises(GiteaNotFoundError):
|
|
client._handle_response(Response(404, request=request), correlation_id="c3")
|
|
|
|
with pytest.raises(GiteaError, match="boom"):
|
|
client._handle_response(
|
|
Response(500, request=request, json={"message": "boom"}),
|
|
correlation_id="c4",
|
|
)
|
|
|
|
assert client._handle_response(Response(200, request=request, json={"ok": True}), "c5") == {
|
|
"ok": True
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_public_methods_delegate_to_request_and_normalize() -> None:
|
|
"""Wrapper methods call shared request logic and normalize return types."""
|
|
client = GiteaClient(token="user-token")
|
|
|
|
async def fake_request(method: str, endpoint: str, **kwargs):
|
|
if endpoint == "/api/v1/user":
|
|
return {"login": "alice"}
|
|
if endpoint == "/api/v1/user/repos":
|
|
return [{"name": "repo"}]
|
|
if endpoint == "/api/v1/repos/acme/demo":
|
|
return {"name": "demo"}
|
|
if endpoint == "/api/v1/repos/acme/demo/contents/README.md":
|
|
return {"size": 8, "content": "aGVsbG8=", "encoding": "base64"}
|
|
if endpoint == "/api/v1/repos/acme/demo/git/trees/main":
|
|
return {"tree": [{"path": "README.md"}]}
|
|
if endpoint == "/api/v1/repos/acme/demo/search":
|
|
return {"hits": []}
|
|
if endpoint == "/api/v1/repos/acme/demo/commits":
|
|
return [{"sha": "abc"}]
|
|
if endpoint == "/api/v1/repos/acme/demo/git/commits/abc":
|
|
return {"sha": "abc"}
|
|
if endpoint == "/api/v1/repos/acme/demo/compare/main...feature":
|
|
return {"total_commits": 1}
|
|
if endpoint == "/api/v1/repos/acme/demo/issues":
|
|
if method == "GET":
|
|
return [{"number": 1}]
|
|
return {"number": 12}
|
|
if endpoint == "/api/v1/repos/acme/demo/issues/1":
|
|
if method == "GET":
|
|
return {"number": 1}
|
|
return {"number": 1, "state": "closed"}
|
|
if endpoint == "/api/v1/repos/acme/demo/pulls":
|
|
return [{"number": 2}]
|
|
if endpoint == "/api/v1/repos/acme/demo/pulls/2":
|
|
return {"number": 2}
|
|
if endpoint == "/api/v1/repos/acme/demo/labels":
|
|
return [{"name": "bug"}]
|
|
if endpoint == "/api/v1/repos/acme/demo/tags":
|
|
return [{"name": "v1"}]
|
|
if endpoint == "/api/v1/repos/acme/demo/releases":
|
|
return [{"id": 1}]
|
|
if endpoint == "/api/v1/repos/acme/demo/issues/1/comments":
|
|
return {"id": 9}
|
|
if endpoint == "/api/v1/repos/acme/demo/issues/1/labels":
|
|
return {"labels": [{"name": "bug"}]}
|
|
if endpoint == "/api/v1/repos/acme/demo/issues/1/assignees":
|
|
return {"assignees": [{"login": "alice"}]}
|
|
return {}
|
|
|
|
client._request = AsyncMock(side_effect=fake_request) # type: ignore[method-assign]
|
|
|
|
assert (await client.get_current_user())["login"] == "alice"
|
|
assert len(await client.list_repositories()) == 1
|
|
assert (await client.get_repository("acme", "demo"))["name"] == "demo"
|
|
assert (await client.get_file_contents("acme", "demo", "README.md"))["size"] == 8
|
|
assert len((await client.get_tree("acme", "demo"))["tree"]) == 1
|
|
assert isinstance(
|
|
await client.search_code("acme", "demo", "needle", ref="main", page=1, limit=5), dict
|
|
)
|
|
assert len(await client.list_commits("acme", "demo", ref="main", page=1, limit=5)) == 1
|
|
assert (await client.get_commit_diff("acme", "demo", "abc"))["sha"] == "abc"
|
|
assert isinstance(await client.compare_refs("acme", "demo", "main", "feature"), dict)
|
|
assert len(await client.list_issues("acme", "demo", state="open", page=1, limit=10)) == 1
|
|
assert (await client.get_issue("acme", "demo", 1))["number"] == 1
|
|
assert len(await client.list_pull_requests("acme", "demo", state="open", page=1, limit=10)) == 1
|
|
assert (await client.get_pull_request("acme", "demo", 2))["number"] == 2
|
|
assert len(await client.list_labels("acme", "demo", page=1, limit=10)) == 1
|
|
assert len(await client.list_tags("acme", "demo", page=1, limit=10)) == 1
|
|
assert len(await client.list_releases("acme", "demo", page=1, limit=10)) == 1
|
|
assert (await client.create_issue("acme", "demo", title="Hi", body="Body"))["number"] == 12
|
|
assert (await client.update_issue("acme", "demo", 1, state="closed"))["state"] == "closed"
|
|
assert (await client.create_issue_comment("acme", "demo", 1, "comment"))["id"] == 9
|
|
assert (await client.create_pr_comment("acme", "demo", 1, "comment"))["id"] == 9
|
|
assert isinstance(await client.add_labels("acme", "demo", 1, ["bug"]), dict)
|
|
assert isinstance(await client.assign_issue("acme", "demo", 1, ["alice"]), dict)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_file_contents_blocks_oversized_payload(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""File size limits are enforced before returning content."""
|
|
monkeypatch.setenv("MAX_FILE_SIZE_BYTES", "5")
|
|
reset_settings()
|
|
client = GiteaClient(token="user-token")
|
|
|
|
client._request = AsyncMock( # type: ignore[method-assign]
|
|
return_value={"size": 50, "content": "x", "encoding": "base64"}
|
|
)
|
|
|
|
with pytest.raises(GiteaError, match="exceeds limit"):
|
|
await client.get_file_contents("acme", "demo", "big.bin")
|