Public Access
emil
55c530eb00
Post-merge / detect-type (push) Successful in 8s
Post-merge / validate-commit-msg (push) Successful in 9s
Post-merge / vikunja (push) Successful in 11s
Post-merge / configure-repo (push) Successful in 9s
Post-merge / sync-wiki (push) Successful in 18s
Post-merge / release (push) Successful in 25s
Post-merge / badges (push) Successful in 28s
Build Images / detect-type (push) Successful in 41s
Post-merge / publish (push) Successful in 15s
Build Images / build-and-push (push) Successful in 3m1s
Build Images / cleanup (push) Successful in 2m25s
900 lines
41 KiB
Python
900 lines
41 KiB
Python
"""Unit tests for api_clients module."""
|
|
|
|
import http
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from devx.api_clients import GiteaClient, VikunjaClient, _parse_error
|
|
from devx.config import (
|
|
DEFAULT_PER_PAGE,
|
|
DEFAULT_TIMEOUT,
|
|
VIKUNJA_PROJECT_ID,
|
|
)
|
|
from devx.exceptions import APIError
|
|
|
|
# Local test config — TEST_BP_CONFIG moved to configure_repo.py defaults in devx
|
|
TEST_BP_CONFIG: dict[str, object] = {
|
|
"branch_name": "master",
|
|
"enable_push": True,
|
|
"enable_push_whitelist": True,
|
|
"push_whitelist_usernames": [],
|
|
"enable_status_check": True,
|
|
"status_check_contexts": ["CI / quality (pull_request)"],
|
|
"required_approvals": 0,
|
|
}
|
|
|
|
|
|
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
|
|
|
|
|
|
def _mock_http_error(status_code: int, message: str = "") -> requests.HTTPError:
|
|
"""Create an HTTPError with a proper response attached (for _parse_error)."""
|
|
resp = MagicMock()
|
|
resp.status_code = status_code
|
|
resp.json.return_value = {"message": message or str(status_code)}
|
|
err = requests.HTTPError(f"{status_code} {message}", response=resp)
|
|
return err
|
|
|
|
|
|
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_ensure_label_creates_when_others_exist(self) -> None:
|
|
"""When labels exist but none match the target name, create a new one."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client.list_labels = MagicMock(
|
|
return_value=[{"name": "bug", "color": "ff0000"}, {"name": "docs", "color": "007ec6"}]
|
|
)
|
|
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_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(TEST_BP_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=TEST_BP_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", TEST_BP_CONFIG)
|
|
assert result["id"] == 1
|
|
client.create_branch_protection.assert_called_once_with(TEST_BP_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", TEST_BP_CONFIG)
|
|
assert result["required_approvals"] == 1
|
|
expected_update = {k: v for k, v in TEST_BP_CONFIG.items() if k != "branch_name"}
|
|
client.update_branch_protection.assert_called_once_with("master", expected_update)
|
|
|
|
def test_ensure_branch_protection_creates_when_none_match(self) -> None:
|
|
"""When existing protections exist but none match the target branch, create a new one."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client.list_branch_protections = MagicMock(
|
|
return_value=[{"branch_name": "develop"}, {"branch_name": "staging"}]
|
|
)
|
|
client.create_branch_protection = MagicMock(return_value={"id": 5, "branch_name": "master"})
|
|
|
|
result = client.ensure_branch_protection("master", TEST_BP_CONFIG)
|
|
assert result["id"] == 5
|
|
client.create_branch_protection.assert_called_once_with(TEST_BP_CONFIG)
|
|
|
|
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_get_pr_labels(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response([{"name": "ready-to-merge"}]))
|
|
|
|
result = client.get_pr_labels(5)
|
|
assert result == [{"name": "ready-to-merge"}]
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/issues/5/labels",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_get_commit_status(self) -> None:
|
|
"""Uses combined status endpoint (/status, not /statuses)."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response({"statuses": [{"context": "CI / quality", "status": "success"}]})
|
|
)
|
|
|
|
result = client.get_commit_status("abc123")
|
|
assert result == [{"context": "CI / quality", "status": "success"}]
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/commits/abc123/status",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_get_pr(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"number": 7, "head": {"sha": "abc123"}}))
|
|
|
|
result = client.get_pr(7)
|
|
assert result["number"] == 7
|
|
assert result["head"]["sha"] == "abc123"
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/pulls/7",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_create_pr(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response({"number": 15, "html_url": "https://git.example.com/pr/15"})
|
|
)
|
|
result = client.create_pr(title="DEVX-42: Add feature", head="DEVX-42-fix", body="desc")
|
|
assert result["number"] == 15
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/pulls",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"title": "DEVX-42: Add feature", "head": "DEVX-42-fix", "base": "master", "body": "desc"},
|
|
)
|
|
|
|
def test_create_pr_no_body(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response({"number": 16, "html_url": "https://git.example.com/pr/16"})
|
|
)
|
|
result = client.create_pr(title="DEVX-43: Fix bug", head="DEVX-43-fix")
|
|
assert result["number"] == 16
|
|
call_kwargs = client._session.request.call_args.kwargs
|
|
assert "body" not in call_kwargs["json"]
|
|
|
|
def test_create_pr_custom_base(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"number": 17}))
|
|
client.create_pr(title="Test", head="branch", base="develop")
|
|
call_kwargs = client._session.request.call_args.kwargs
|
|
assert call_kwargs["json"]["base"] == "develop"
|
|
|
|
def test_get_pr_files(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response([{"filename": "src/main.py", "status": "modified"}])
|
|
)
|
|
|
|
result = client.get_pr_files(7)
|
|
assert len(result) == 1
|
|
assert result[0]["filename"] == "src/main.py"
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/pulls/7/files",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_get_pr_commits(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response([{"sha": "abc123", "commit": {"message": "fix: bug"}}])
|
|
)
|
|
|
|
result = client.get_pr_commits(7)
|
|
assert len(result) == 1
|
|
assert result[0]["commit"]["message"] == "fix: bug"
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/pulls/7/commits",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_list_prs(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response([{"number": 1, "title": "feat: add"}, {"number": 2, "title": "fix: bug"}])
|
|
)
|
|
|
|
result = client.list_prs()
|
|
assert len(result) == 2
|
|
assert result[0]["number"] == 1
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/pulls",
|
|
params={"state": "all"},
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_list_prs_with_params(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response([{"number": 3, "title": "docs: update"}]))
|
|
|
|
result = client.list_prs(state="closed", q="docs")
|
|
assert len(result) == 1
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/pulls",
|
|
params={"state": "closed", "q": "docs"},
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_get_pr_reviews(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response([{"id": 1, "state": "APPROVED"}]))
|
|
|
|
result = client.get_pr_reviews(7)
|
|
assert len(result) == 1
|
|
assert result[0]["state"] == "APPROVED"
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/pulls/7/reviews",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_create_issue(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"id": 42, "title": "bug"}))
|
|
|
|
result = client.create_issue(title="bug", body="description", labels=[1])
|
|
assert result["id"] == 42
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/issues",
|
|
json={"title": "bug", "body": "description", "labels": [1]},
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_create_issue_no_labels(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"id": 43, "title": "bug"}))
|
|
|
|
result = client.create_issue(title="bug", body="description")
|
|
assert result["id"] == 43
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/issues",
|
|
json={"title": "bug", "body": "description"},
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_create_review_comment(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"id": 42}))
|
|
|
|
result = client.create_review(7, event="COMMENT", body="Looks good")
|
|
assert result["id"] == 42
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/pulls/7/reviews",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"event": "COMMENT", "body": "Looks good"},
|
|
)
|
|
|
|
def test_create_review_approve_maps_to_approved(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"id": 44, "state": "APPROVED"}))
|
|
|
|
result = client.create_review(7, event="APPROVE", body="Good work")
|
|
assert result["id"] == 44
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/pulls/7/reviews",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"event": "APPROVED", "body": "Good work"},
|
|
)
|
|
|
|
def test_create_review_with_inline_comments(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"id": 43}))
|
|
|
|
comments = [{"path": "src/main.py", "body": "Fix this", "new_position": 10}]
|
|
result = client.create_review(7, event="REQUEST_CHANGES", body="Please fix", comments=comments)
|
|
assert result["id"] == 43
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/pulls/7/reviews",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"event": "REQUEST_CHANGES", "body": "Please fix", "comments": comments},
|
|
)
|
|
|
|
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},
|
|
)
|
|
|
|
def test_get_release_by_tag_found(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({"id": 1, "tag_name": "v1.0.0"}))
|
|
result = client.get_release_by_tag("v1.0.0")
|
|
assert result is not None
|
|
assert result["id"] == 1
|
|
|
|
def test_get_release_by_tag_not_found(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status.side_effect = requests.HTTPError("404")
|
|
mock_resp.status_code = 404
|
|
client._session.request = MagicMock(return_value=mock_resp)
|
|
result = client.get_release_by_tag("v9.9.9")
|
|
assert result is None
|
|
|
|
def test_create_release_idempotent_existing(self) -> None:
|
|
"""If release already exists, should return it without creating a new one."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
existing_response = _mock_response({"id": 42, "tag_name": "v1.0.0"})
|
|
client._session.request = MagicMock(return_value=existing_response)
|
|
result = client.create_release_idempotent("v1.0.0")
|
|
assert result["id"] == 42
|
|
# Should only call GET (check), not POST (create)
|
|
assert client._session.request.call_count == 1
|
|
assert client._session.request.call_args[0][0] == "GET"
|
|
|
|
def test_create_release_idempotent_new(self) -> None:
|
|
"""If release doesn't exist, should create it."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
not_found_resp = MagicMock()
|
|
not_found_resp.raise_for_status.side_effect = requests.HTTPError("404")
|
|
not_found_resp.status_code = 404
|
|
create_resp = _mock_response({"id": 1, "tag_name": "v1.0.0"})
|
|
client._session.request = MagicMock(side_effect=[not_found_resp, create_resp])
|
|
result = client.create_release_idempotent("v1.0.0")
|
|
assert result["id"] == 1
|
|
assert client._session.request.call_count == 2
|
|
|
|
@patch("time.sleep")
|
|
def test_request_retries_on_429(self, mock_sleep: MagicMock) -> None:
|
|
"""Should retry on 429 rate limit with exponential backoff."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
rate_limited = MagicMock()
|
|
rate_limited.raise_for_status.side_effect = _mock_http_error(429, "rate limited")
|
|
success = _mock_response({"ok": True})
|
|
client._session.request = MagicMock(side_effect=[rate_limited, rate_limited, success])
|
|
result = client._request("GET", "/test")
|
|
assert result.json() == {"ok": True}
|
|
assert client._session.request.call_count == 3
|
|
assert mock_sleep.call_count == 2
|
|
|
|
@patch("time.sleep")
|
|
def test_request_retries_on_503(self, mock_sleep: MagicMock) -> None:
|
|
"""Should retry on 503 service unavailable."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
unavailable = MagicMock()
|
|
unavailable.raise_for_status.side_effect = _mock_http_error(503, "unavailable")
|
|
success = _mock_response({"ok": True})
|
|
client._session.request = MagicMock(side_effect=[unavailable, success])
|
|
result = client._request("GET", "/test")
|
|
assert result.json() == {"ok": True}
|
|
assert client._session.request.call_count == 2
|
|
|
|
@patch("time.sleep")
|
|
def test_request_no_retry_on_404(self, mock_sleep: MagicMock) -> None:
|
|
"""Should NOT retry on 404 — it's not a transient error."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
not_found = MagicMock()
|
|
not_found.raise_for_status.side_effect = _mock_http_error(404, "not found")
|
|
client._session.request = MagicMock(return_value=not_found)
|
|
with pytest.raises(APIError) as exc_info:
|
|
client._request("GET", "/test")
|
|
assert exc_info.value.status == 404
|
|
assert client._session.request.call_count == 1
|
|
mock_sleep.assert_not_called()
|
|
|
|
@patch("time.sleep")
|
|
def test_request_retries_on_connection_error(self, mock_sleep: MagicMock) -> None:
|
|
"""Should retry on connection errors."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
success = _mock_response({"ok": True})
|
|
client._session.request = MagicMock(side_effect=[requests.ConnectionError("refused"), success])
|
|
result = client._request("GET", "/test")
|
|
assert result.json() == {"ok": True}
|
|
assert client._session.request.call_count == 2
|
|
|
|
@patch("time.sleep")
|
|
def test_request_max_retries_exhausted(self, mock_sleep: MagicMock) -> None:
|
|
"""Should raise APIError after max retries on persistent 503."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
unavailable = MagicMock()
|
|
unavailable.raise_for_status.side_effect = _mock_http_error(503, "unavailable")
|
|
client._session.request = MagicMock(return_value=unavailable)
|
|
with pytest.raises(APIError) as exc_info:
|
|
client._request("GET", "/test")
|
|
assert exc_info.value.status == 503
|
|
assert client._session.request.call_count == 3 # MAX_RETRIES
|
|
|
|
@patch("time.sleep")
|
|
def test_request_connection_error_exhausted(self, mock_sleep: MagicMock) -> None:
|
|
"""Should raise APIError after max retries on persistent connection errors."""
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(side_effect=requests.ConnectionError("refused"))
|
|
with pytest.raises(APIError) as exc_info:
|
|
client._request("GET", "/test")
|
|
assert exc_info.value.status == 0
|
|
assert client._session.request.call_count == 3 # MAX_RETRIES
|
|
|
|
|
|
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": "DEVX-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_list_project_tasks(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(return_value=_mock_response([{"id": 1, "identifier": "DEVX-19"}]))
|
|
|
|
result = client.list_project_tasks(VIKUNJA_PROJECT_ID, page=1, per_page=DEFAULT_PER_PAGE)
|
|
assert len(result) == 1
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
f"https://work.example.com/projects/{VIKUNJA_PROJECT_ID}/tasks",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
params={"page": 1, "per_page": DEFAULT_PER_PAGE},
|
|
)
|
|
|
|
def test_get_task(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response({"id": 292, "identifier": "DEVX-32", "title": "Some task"})
|
|
)
|
|
|
|
result = client.get_task(292)
|
|
assert result["identifier"] == "DEVX-32"
|
|
assert result["title"] == "Some task"
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://work.example.com/tasks/292",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
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(
|
|
"PUT",
|
|
"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_list_comments(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response([{"id": 1, "comment": "first"}, {"id": 2, "comment": "second"}])
|
|
)
|
|
|
|
result = client.list_comments(42)
|
|
assert len(result) == 2
|
|
assert result[0]["comment"] == "first"
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://work.example.com/tasks/42/comments",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_update_task_safe(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(
|
|
side_effect=[
|
|
_mock_response({"id": 42, "title": "My task", "done": False}),
|
|
_mock_response({"id": 42, "title": "My task", "done": True}),
|
|
]
|
|
)
|
|
|
|
result = client.update_task_safe(42, done=True)
|
|
assert result["done"] is True
|
|
assert result["title"] == "My task"
|
|
assert client._session.request.call_count == 2
|
|
client._session.request.assert_any_call(
|
|
"GET",
|
|
"https://work.example.com/tasks/42",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
client._session.request.assert_any_call(
|
|
"POST",
|
|
"https://work.example.com/tasks/42",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"id": 42, "title": "My task", "done": True},
|
|
)
|
|
|
|
@patch("time.sleep")
|
|
def test_http_error_raises_api_error(self, mock_sleep: MagicMock) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status.side_effect = _mock_http_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error")
|
|
client._session.request = MagicMock(return_value=mock_resp)
|
|
|
|
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]
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status.side_effect = err
|
|
client._session.request = MagicMock(return_value=mock_resp)
|
|
|
|
with pytest.raises(APIError) as exc_info:
|
|
client.list_tasks()
|
|
assert "connection failed" in str(exc_info.value)
|
|
|
|
@patch("time.sleep")
|
|
def test_vikunja_retries_on_503(self, mock_sleep: MagicMock) -> None:
|
|
"""VikunjaClient should also retry on 503."""
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
unavailable = MagicMock()
|
|
unavailable.raise_for_status.side_effect = _mock_http_error(503, "unavailable")
|
|
success = _mock_response([{"id": 1}])
|
|
client._session.request = MagicMock(side_effect=[unavailable, success])
|
|
result = client.list_tasks()
|
|
assert len(result) == 1
|
|
assert client._session.request.call_count == 2
|
|
|
|
@patch("time.sleep")
|
|
def test_vikunja_retries_on_connection_error(self, mock_sleep: MagicMock) -> None:
|
|
"""VikunjaClient should retry on connection errors."""
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
success = _mock_response([{"id": 1}])
|
|
client._session.request = MagicMock(side_effect=[requests.ConnectionError("refused"), success])
|
|
result = client.list_tasks()
|
|
assert len(result) == 1
|
|
assert client._session.request.call_count == 2
|
|
|
|
@patch("time.sleep")
|
|
def test_vikunja_max_retries_exhausted(self, mock_sleep: MagicMock) -> None:
|
|
"""VikunjaClient should raise APIError after max retries on persistent 503."""
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
unavailable = MagicMock()
|
|
unavailable.raise_for_status.side_effect = _mock_http_error(503, "unavailable")
|
|
client._session.request = MagicMock(return_value=unavailable)
|
|
with pytest.raises(APIError) as exc_info:
|
|
client.list_tasks()
|
|
assert exc_info.value.status == 503
|
|
assert client._session.request.call_count == 3 # MAX_RETRIES
|
|
|
|
@patch("time.sleep")
|
|
def test_vikunja_connection_error_exhausted(self, mock_sleep: MagicMock) -> None:
|
|
"""VikunjaClient should raise APIError after max retries on persistent connection errors."""
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(side_effect=requests.ConnectionError("refused"))
|
|
with pytest.raises(APIError) as exc_info:
|
|
client.list_tasks()
|
|
assert exc_info.value.status == 0
|
|
assert client._session.request.call_count == 3 # MAX_RETRIES
|
|
|
|
def test_vikunja_create_task(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response({"id": 1, "identifier": "DEVX-1", "title": "Test"})
|
|
)
|
|
result = client.create_task(6, "Test", "<p>desc</p>")
|
|
assert result["identifier"] == "DEVX-1"
|
|
client._session.request.assert_called_once_with(
|
|
"PUT",
|
|
"https://work.example.com/projects/6/tasks",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"title": "Test", "description": "<p>desc</p>"},
|
|
)
|
|
|
|
def test_vikunja_create_task_no_description(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response({"id": 2, "identifier": "DEVX-2", "title": "No desc"})
|
|
)
|
|
result = client.create_task(6, "No desc")
|
|
assert result["id"] == 2
|
|
call_kwargs = client._session.request.call_args.kwargs
|
|
assert call_kwargs["json"]["description"] == ""
|
|
|
|
def test_find_task_by_identifier_found(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response([{"identifier": "DEVX-1"}, {"identifier": "DEVX-42", "title": "Found"}])
|
|
)
|
|
result = client.find_task_by_identifier(6, "DEVX-42", per_page=50)
|
|
assert result is not None
|
|
assert result["title"] == "Found"
|
|
|
|
def test_find_task_by_identifier_not_found(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response([{"identifier": "DEVX-1"}, {"identifier": "DEVX-2"}])
|
|
)
|
|
result = client.find_task_by_identifier(6, "DEVX-99", per_page=50)
|
|
assert result is None
|
|
|
|
def test_find_task_by_identifier_empty_project(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
client._session.request = MagicMock(return_value=_mock_response([]))
|
|
result = client.find_task_by_identifier(6, "DEVX-1", per_page=50)
|
|
assert result is None
|
|
|
|
def test_find_task_by_identifier_paginates(self) -> None:
|
|
client = VikunjaClient("https://work.example.com", "tok")
|
|
full_page = [{"identifier": f"DEVX-{i}"} for i in range(50)]
|
|
client._session.request = MagicMock(
|
|
side_effect=[
|
|
_mock_response(full_page),
|
|
_mock_response([{"identifier": "DEVX-50", "title": "Found on page 2"}]),
|
|
]
|
|
)
|
|
result = client.find_task_by_identifier(6, "DEVX-50", per_page=50)
|
|
assert result is not None
|
|
assert result["title"] == "Found on page 2"
|
|
|
|
|
|
class TestGiteaClientPrLabels:
|
|
def test_add_pr_label(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({}))
|
|
client.add_pr_label(42, ["ready-to-merge"])
|
|
client._session.request.assert_called_once_with(
|
|
"POST",
|
|
"https://git.example.com/repos/owner/repo/issues/42/labels",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
json={"labels": ["ready-to-merge"]},
|
|
)
|
|
|
|
def test_add_pr_label_multiple(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({}))
|
|
client.add_pr_label(42, ["ready-to-merge", "reviewed"])
|
|
call_kwargs = client._session.request.call_args.kwargs
|
|
assert call_kwargs["json"]["labels"] == ["ready-to-merge", "reviewed"]
|
|
|
|
def test_get_pr_label_names(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response([{"name": "bug"}, {"name": "ready-to-merge"}]))
|
|
result = client.get_pr_label_names(42)
|
|
assert result == ["bug", "ready-to-merge"]
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/issues/42/labels",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
|
|
class TestGiteaClientActions:
|
|
def test_list_action_runs(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response({"workflow_runs": [{"id": 1, "status": "completed"}], "total_count": 1})
|
|
)
|
|
result = client.list_action_runs(branch="feature-branch", limit=1)
|
|
assert result["total_count"] == 1
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/actions/runs",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
params={"branch": "feature-branch", "limit": 1},
|
|
)
|
|
|
|
def test_get_action_run_jobs(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(
|
|
return_value=_mock_response({"jobs": [{"id": 100, "name": "quality", "conclusion": "failure"}]})
|
|
)
|
|
result = client.get_action_run_jobs(1410)
|
|
assert len(result) == 1
|
|
assert result[0]["name"] == "quality"
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/actions/runs/1410/jobs",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
def test_get_action_run_jobs_empty(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
client._session.request = MagicMock(return_value=_mock_response({}))
|
|
result = client.get_action_run_jobs(1410)
|
|
assert result == []
|
|
|
|
def test_get_action_job_logs(self) -> None:
|
|
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
|
mock_resp = MagicMock()
|
|
mock_resp.text = "log line 1\nlog line 2"
|
|
mock_resp.raise_for_status = MagicMock()
|
|
client._session.request = MagicMock(return_value=mock_resp)
|
|
result = client.get_action_job_logs(10026)
|
|
assert "log line 1" in result
|
|
client._session.request.assert_called_once_with(
|
|
"GET",
|
|
"https://git.example.com/repos/owner/repo/actions/jobs/10026/logs",
|
|
timeout=DEFAULT_TIMEOUT,
|
|
)
|