280 lines
12 KiB
Python
280 lines
12 KiB
Python
"""Unit tests for api_clients module."""
|
|
|
|
import http
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from gitea_runner_manager.api_clients import GiteaClient, VikunjaClient, _parse_error
|
|
from gitea_runner_manager.config import (
|
|
BRANCH_PROTECTION_CONFIG,
|
|
DEFAULT_PER_PAGE,
|
|
DEFAULT_TIMEOUT,
|
|
VIKUNJA_PROJECT_ID,
|
|
)
|
|
from gitea_runner_manager.exceptions import APIError
|
|
|
|
|
|
def _mock_response(json_data: object | None = None, raise_on_status: bool = False) -> MagicMock:
|
|
mock = MagicMock()
|
|
if json_data is not None:
|
|
mock.json.return_value = json_data
|
|
if raise_on_status:
|
|
mock.raise_for_status.side_effect = requests.HTTPError(str(http.HTTPStatus.INTERNAL_SERVER_ERROR))
|
|
return mock
|
|
|
|
|
|
class TestParseError:
|
|
def test_json_parse_fallback(self) -> None:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = http.HTTPStatus.BAD_GATEWAY
|
|
mock_response.json = MagicMock(side_effect=ValueError("not json"))
|
|
err = requests.HTTPError(str(http.HTTPStatus.BAD_GATEWAY), response=mock_response)
|
|
status, message = _parse_error(err)
|
|
assert status == http.HTTPStatus.BAD_GATEWAY
|
|
assert str(http.HTTPStatus.BAD_GATEWAY) in message
|
|
|
|
def test_no_response(self) -> None:
|
|
err = requests.HTTPError("connection failed")
|
|
err.response = None # type: ignore[assignment]
|
|
status, message = _parse_error(err)
|
|
assert status == 0
|
|
assert "connection failed" in message
|
|
|
|
|
|
class TestGiteaClient:
|
|
def test_init_sets_headers(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
assert client._base_url == "https://git.example.com"
|
|
assert client._session.headers["Authorization"] == "token tok"
|
|
assert client._session.headers["Content-Type"] == "application/json"
|
|
|
|
def test_url_constructs_path(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
assert client._url("/labels") == ("https://git.example.com/repos/owner/repo/labels")
|
|
|
|
def test_url_strips_trailing_slash(self) -> None:
|
|
client = GiteaClient("https://git.example.com/", "tok", "owner", "repo")
|
|
assert client._url("/labels") == ("https://git.example.com/repos/owner/repo/labels")
|
|
|
|
def test_list_labels(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response([{"name": "bug", "color": "ff0000"}]))
|
|
result = client.list_labels()
|
|
assert len(result) == 1
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/labels",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_list_labels_raises_api_error(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response(raise_on_status=True))
|
|
with pytest.raises(APIError):
|
|
client.list_labels()
|
|
|
|
def test_http_error_json_parse_fallback(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = http.HTTPStatus.BAD_GATEWAY
|
|
# Make json() itself raise so the except block in _parse_error is hit
|
|
mock_response.json = MagicMock(side_effect=ValueError("not json"))
|
|
mock_response.raise_for_status.side_effect = requests.HTTPError(str(http.HTTPStatus.BAD_GATEWAY))
|
|
client._session.request = MagicMock(return_value=mock_response)
|
|
with pytest.raises(APIError) as exc_info:
|
|
client.list_labels()
|
|
assert str(http.HTTPStatus.BAD_GATEWAY) in str(exc_info.value)
|
|
|
|
def test_create_label(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"name": "ready-to-merge", "color": "2ecc71"}))
|
|
result = client.create_label("ready-to-merge", "2ecc71", "Auto-merge label")
|
|
assert result["name"] == "ready-to-merge"
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/labels",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"name": "ready-to-merge", "color": "2ecc71", "description": "Auto-merge label"},
|
|
)
|
|
|
|
def test_ensure_label_creates_when_not_exists(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client.list_labels = MagicMock(return_value=[])
|
|
client.create_label = MagicMock(return_value={"name": "ready-to-merge", "color": "2ecc71"})
|
|
|
|
result = client.ensure_label("ready-to-merge", "2ecc71", "desc")
|
|
assert result is not None
|
|
assert result["name"] == "ready-to-merge"
|
|
client.create_label.assert_called_once_with("ready-to-merge", "2ecc71", "desc")
|
|
|
|
def test_ensure_label_returns_none_when_exists(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client.list_labels = MagicMock(return_value=[{"name": "ready-to-merge", "color": "2ecc71"}])
|
|
client.create_label = MagicMock()
|
|
|
|
result = client.ensure_label("ready-to-merge", "2ecc71", "desc")
|
|
assert result is None
|
|
client.create_label.assert_not_called()
|
|
|
|
def test_list_branch_protections(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response(
|
|
[
|
|
{"id": 1, "branch_name": "master"},
|
|
{"id": 2, "branch_name": "develop"},
|
|
]
|
|
)
|
|
)
|
|
result = client.list_branch_protections()
|
|
assert len(result) == 2
|
|
assert result[0]["branch_name"] == "master"
|
|
|
|
def test_create_branch_protection(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"id": 3, "branch_name": "master"}))
|
|
result = client.create_branch_protection(BRANCH_PROTECTION_CONFIG)
|
|
assert result["id"] == 3
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/branch_protections",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json=BRANCH_PROTECTION_CONFIG,
|
|
)
|
|
|
|
def test_update_branch_protection(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
resp = {"branch_name": "master", "required_approvals": 2}
|
|
client._session.request = MagicMock(return_value=_mock_response(resp))
|
|
update = {"required_approvals": 2}
|
|
result = client.update_branch_protection("master", update)
|
|
assert result["required_approvals"] == 2
|
|
client._session.request.assert_called_once_with(
|
|
"PATCH",
|
|
"https://git.example.com/repos/owner/repo/branch_protections/master",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json=update,
|
|
)
|
|
|
|
def test_ensure_branch_protection_creates_when_none_exist(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client.list_branch_protections = MagicMock(return_value=[])
|
|
client.create_branch_protection = MagicMock(return_value={"id": 1, "branch_name": "master"})
|
|
|
|
result = client.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
|
|
assert result["id"] == 1
|
|
client.create_branch_protection.assert_called_once_with(BRANCH_PROTECTION_CONFIG)
|
|
|
|
def test_ensure_branch_protection_updates_when_exists(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client.list_branch_protections = MagicMock(return_value=[{"branch_name": "master", "required_approvals": 0}])
|
|
client.update_branch_protection = MagicMock(return_value={"branch_name": "master", "required_approvals": 1})
|
|
|
|
result = client.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
|
|
assert result["required_approvals"] == 1
|
|
expected_update = {k: v for k, v in BRANCH_PROTECTION_CONFIG.items() if k != "branch_name"}
|
|
client.update_branch_protection.assert_called_once_with("master", expected_update)
|
|
|
|
def test_merge_pr(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response())
|
|
|
|
client.merge_pr(1, "fix: bug")
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/pulls/1/merge",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"Do": "squash", "MergeTitleField": "fix: bug"},
|
|
)
|
|
|
|
def test_update_repo_settings(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"default_delete_branch_after_merge": True}))
|
|
|
|
settings = {"default_delete_branch_after_merge": True}
|
|
result = client.update_repo_settings(settings)
|
|
assert result["default_delete_branch_after_merge"] is True
|
|
client._session.request.assert_called_once_with(
|
|
"PATCH",
|
|
"https://git.example.com/repos/owner/repo",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json=settings,
|
|
)
|
|
|
|
def test_create_release(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"id": 1}))
|
|
|
|
client.create_release("v1.0.0")
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/releases",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"tag_name": "v1.0.0", "name": "v1.0.0", "body": "", "draft": False, "prerelease": False},
|
|
)
|
|
|
|
|
|
class TestVikunjaClient:
|
|
def test_init_sets_headers(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
assert client._base_url == "https://work.example.com"
|
|
assert client._session.headers["Authorization"] == "Bearer tok"
|
|
|
|
def test_list_tasks(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response([{"id": 1, "identifier": "GRM-19", "project_id": VIKUNJA_PROJECT_ID}])
|
|
)
|
|
|
|
result = client.list_tasks(per_page=DEFAULT_PER_PAGE)
|
|
assert len(result) == 1
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://work.example.com/tasks",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
params={"per_page": DEFAULT_PER_PAGE},
|
|
)
|
|
|
|
def test_post_comment(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(return_value=_mock_response())
|
|
|
|
client.post_comment(42, "<p>hi</p>")
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://work.example.com/tasks/42/comments",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"comment": "<p>hi</p>"},
|
|
)
|
|
|
|
def test_update_task(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(return_value=_mock_response())
|
|
|
|
client.update_task(42, done=True)
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://work.example.com/tasks/42",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"done": True},
|
|
)
|
|
|
|
def test_http_error_raises_api_error(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(return_value=_mock_response(raise_on_status=True))
|
|
|
|
with pytest.raises(APIError):
|
|
client.list_tasks()
|
|
|
|
def test_http_error_no_response(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
err = requests.HTTPError("connection failed")
|
|
err.response = None # type: ignore[assignment]
|
|
client._session.request = MagicMock(side_effect=err)
|
|
|
|
with pytest.raises(APIError) as exc_info:
|
|
client.list_tasks()
|
|
assert "connection failed" in str(exc_info.value)
|