231 lines
8.5 KiB
Python
231 lines
8.5 KiB
Python
"""Unit tests for the connected sources (#92): Gitea & GitHub tools.
|
|
|
|
All HTTP calls are mocked (httpx.request monkeypatched) — the CI never talks
|
|
to a real Gitea/GitHub instance.
|
|
"""
|
|
|
|
import base64
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
import backend.tools.connected as connected
|
|
from backend.tools.context import ToolContext, ToolError, ToolMode
|
|
from backend.tools.registry import get_tool
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, json_data: Any = None, status_code: int = 200):
|
|
self._json = json_data
|
|
self.status_code = status_code
|
|
|
|
def json(self):
|
|
return self._json
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
import httpx
|
|
|
|
raise httpx.HTTPStatusError("boom", request=None, response=self) # type: ignore[arg-type]
|
|
|
|
|
|
def _ctx() -> ToolContext:
|
|
return ToolContext(user={"username": "tester", "vaults": []}, mode=ToolMode.IN_APP)
|
|
|
|
|
|
@pytest.fixture
|
|
def gitea_env(monkeypatch):
|
|
monkeypatch.setenv("OBSIGATE_GITEA_URL", "https://git.example.net")
|
|
monkeypatch.setenv("OBSIGATE_GITEA_TOKEN", "tok-gitea")
|
|
|
|
|
|
@pytest.fixture
|
|
def github_env(monkeypatch):
|
|
monkeypatch.setenv("OBSIGATE_GITHUB_TOKEN", "tok-gh")
|
|
|
|
|
|
def _patch_request(monkeypatch, handler):
|
|
def fake_request(method, url, headers=None, timeout=None, follow_redirects=False, **kw):
|
|
captured = {"method": method, "url": str(url), "headers": headers or {},
|
|
"params": kw.get("params")}
|
|
return handler(captured)
|
|
|
|
monkeypatch.setattr(connected.httpx, "request", fake_request)
|
|
|
|
|
|
class TestRegistration:
|
|
@pytest.mark.parametrize("name", ["git_list_repos", "git_search_issues", "git_get_file"])
|
|
def test_tools_registered_read(self, name):
|
|
from backend.tools.context import ToolRisk
|
|
|
|
spec = get_tool(name)
|
|
assert spec is not None
|
|
assert spec.risk == ToolRisk.READ
|
|
assert "gitea" in spec.description or "github" in spec.description.lower()
|
|
|
|
|
|
class TestConfiguration:
|
|
def test_gitea_requires_base_url(self, monkeypatch):
|
|
monkeypatch.delenv("OBSIGATE_GITEA_URL", raising=False)
|
|
with pytest.raises(ToolError) as ei:
|
|
connected._provider_base("gitea")
|
|
assert ei.value.code == "provider_not_configured"
|
|
|
|
def test_unknown_provider_rejected(self):
|
|
with pytest.raises(ToolError) as ei:
|
|
connected._provider_base("gitlab")
|
|
assert ei.value.code == "invalid_arguments"
|
|
|
|
def test_gitea_token_sent_as_header(self, gitea_env, monkeypatch):
|
|
captured = {}
|
|
|
|
def handler(captured_req):
|
|
captured.update(captured_req)
|
|
return FakeResponse(json_data={"data": []})
|
|
|
|
_patch_request(monkeypatch, handler)
|
|
connected.git_list_repos(_ctx(), connected.GitProviderInput(provider="gitea"))
|
|
assert captured["headers"]["Authorization"] == "token tok-gitea"
|
|
|
|
|
|
class TestListRepos:
|
|
def test_gitea_search_endpoint(self, gitea_env, monkeypatch):
|
|
captured = {}
|
|
|
|
def handler(captured_req):
|
|
captured.update(captured_req)
|
|
return FakeResponse(json_data={"data": [
|
|
{"name": "ObsiGate", "full_name": "bruno/ObsiGate",
|
|
"html_url": "https://git.example.net/bruno/ObsiGate",
|
|
"description": "vault gateway", "updated_at": "2026-09-01",
|
|
"private": False},
|
|
]})
|
|
|
|
_patch_request(monkeypatch, handler)
|
|
out = connected.git_list_repos(_ctx(), connected.GitProviderInput(provider="gitea"))
|
|
assert "/api/v1/repos/search" in captured["url"]
|
|
assert out["repos"][0]["name"] == "ObsiGate"
|
|
assert out["count"] == 1
|
|
|
|
def test_github_single_repo(self, github_env, monkeypatch):
|
|
captured = {}
|
|
|
|
def handler(captured_req):
|
|
captured.update(captured_req)
|
|
return FakeResponse(json_data={
|
|
"name": "ObsiGate", "full_name": "bruno/ObsiGate",
|
|
"html_url": "https://github.com/bruno/ObsiGate",
|
|
"description": "", "updated_at": "2026-09-02", "private": True,
|
|
})
|
|
|
|
_patch_request(monkeypatch, handler)
|
|
out = connected.git_list_repos(_ctx(), connected.GitProviderInput(
|
|
provider="github", repo="bruno/ObsiGate"))
|
|
assert captured["url"].endswith("/repos/bruno/ObsiGate")
|
|
assert out["repos"][0]["full_name"] == "bruno/ObsiGate"
|
|
assert out["repos"][0]["private"] is True
|
|
|
|
|
|
class TestSearchIssues:
|
|
def test_gitea_scoped_to_repo(self, gitea_env, monkeypatch):
|
|
captured = {}
|
|
|
|
def handler(captured_req):
|
|
captured.update(captured_req)
|
|
return FakeResponse(json_data=[
|
|
{"number": 12, "title": "Bug affichage", "html_url": "https://x/12",
|
|
"state": "open"},
|
|
])
|
|
|
|
_patch_request(monkeypatch, handler)
|
|
out = connected.git_search_issues(_ctx(), connected.GitSearchIssuesInput(
|
|
provider="gitea", query="affichage", repo="bruno/ObsiGate"))
|
|
assert "/repos/bruno/ObsiGate/issues" in captured["url"]
|
|
assert out["issues"][0]["id"] == 12
|
|
assert out["issues"][0]["pull_request"] is False
|
|
|
|
def test_github_search_syntax(self, github_env, monkeypatch):
|
|
captured = {}
|
|
|
|
def handler(captured_req):
|
|
captured.update(captured_req)
|
|
return FakeResponse(json_data={"items": [
|
|
{"number": 5, "title": "Crash on save", "html_url": "https://gh/5",
|
|
"state": "open", "pull_request": {"url": "x"}},
|
|
]})
|
|
|
|
_patch_request(monkeypatch, handler)
|
|
out = connected.git_search_issues(_ctx(), connected.GitSearchIssuesInput(
|
|
provider="github", query="crash", repo="bruno/ObsiGate", state="open"))
|
|
assert "/search/issues" in captured["url"]
|
|
assert "repo:bruno/ObsiGate" in captured["params"]["q"]
|
|
assert out["issues"][0]["pull_request"] is True
|
|
|
|
|
|
class TestGetFile:
|
|
def test_gitea_base64_content_decoded(self, gitea_env, monkeypatch):
|
|
payload = base64.b64encode("# Readme\n\nBonjour".encode()).decode()
|
|
|
|
def handler(_captured):
|
|
return FakeResponse(json_data={
|
|
"path": "README.md", "size": 15, "encoding": "base64", "content": payload,
|
|
})
|
|
|
|
_patch_request(monkeypatch, handler)
|
|
out = connected.git_get_file(_ctx(), connected.GitGetFileInput(
|
|
provider="gitea", repo="bruno/ObsiGate", path="README.md"))
|
|
assert "Bonjour" in out["content"]
|
|
assert out["truncated"] is False
|
|
|
|
def test_github_ref_parameter(self, github_env, monkeypatch):
|
|
captured = {}
|
|
|
|
def handler(captured_req):
|
|
captured.update(captured_req)
|
|
return FakeResponse(json_data={
|
|
"path": "a.md", "size": 1, "encoding": "base64",
|
|
"content": base64.b64encode(b"x").decode(),
|
|
})
|
|
|
|
_patch_request(monkeypatch, handler)
|
|
connected.git_get_file(_ctx(), connected.GitGetFileInput(
|
|
provider="github", repo="o/r", path="a.md", ref="v2.9.0"))
|
|
assert captured["url"].endswith("?ref=v2.9.0")
|
|
|
|
def test_missing_repo_or_path_rejected(self, gitea_env):
|
|
with pytest.raises(ToolError) as ei:
|
|
connected.git_get_file(_ctx(), connected.GitGetFileInput(
|
|
provider="gitea", repo="", path="a.md"))
|
|
assert ei.value.code == "invalid_arguments"
|
|
|
|
|
|
class TestErrors:
|
|
def test_404_maps_to_not_found(self, gitea_env, monkeypatch):
|
|
def handler(_captured):
|
|
return FakeResponse(json_data={}, status_code=404)
|
|
|
|
_patch_request(monkeypatch, handler)
|
|
with pytest.raises(ToolError) as ei:
|
|
connected.git_list_repos(_ctx(), connected.GitProviderInput(provider="gitea"))
|
|
assert ei.value.code == "not_found"
|
|
|
|
def test_401_maps_to_permission_denied(self, gitea_env, monkeypatch):
|
|
def handler(_captured):
|
|
return FakeResponse(json_data={}, status_code=401)
|
|
|
|
_patch_request(monkeypatch, handler)
|
|
with pytest.raises(ToolError) as ei:
|
|
connected.git_list_repos(_ctx(), connected.GitProviderInput(provider="gitea"))
|
|
assert ei.value.code == "permission_denied"
|
|
|
|
def test_network_error_maps_to_tool_error(self, gitea_env, monkeypatch):
|
|
import httpx
|
|
|
|
def fake_request(*a, **kw):
|
|
raise httpx.ConnectError("down")
|
|
|
|
monkeypatch.setattr(connected.httpx, "request", fake_request)
|
|
with pytest.raises(ToolError) as ei:
|
|
connected.git_list_repos(_ctx(), connected.GitProviderInput(provider="gitea"))
|
|
assert ei.value.code == "connected_source_unavailable"
|