GRM-113: feat: upgrade all dependencies, add trigger-workflow command
This commit is contained in:
emil
@@ -918,3 +918,91 @@ class TestCLI:
|
||||
|
||||
with patch.dict("os.environ", {"ANSIBLE_BECOME_PASSWORD_FILE": "/tmp/ansible.txt"}, clear=True):
|
||||
assert _get_become_password_file() == "/tmp/ansible.txt"
|
||||
|
||||
|
||||
class TestTriggerWorkflow:
|
||||
"""Tests for the trigger-workflow CLI command."""
|
||||
|
||||
def test_trigger_workflow_success(self) -> None:
|
||||
runner = CliRunner(env=_TEST_ENV)
|
||||
with patch("gitea_runner_manager.cli.GiteaWorkflowClient") as mock_client_cls:
|
||||
mock_client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
mock_client.dispatch_workflow.return_value = {
|
||||
"id": 42,
|
||||
"html_url": "https://git.example.com/oblachno-oss/grm/actions/runs/42",
|
||||
}
|
||||
result = runner.invoke(cli, ["trigger-workflow", "ci.yml", "--token", "tok"])
|
||||
assert result.exit_code == 0
|
||||
assert "42" in result.output
|
||||
mock_client.dispatch_workflow.assert_called_once_with("oblachno-oss", "grm", "ci.yml", "master")
|
||||
|
||||
def test_trigger_workflow_no_url(self) -> None:
|
||||
runner = CliRunner()
|
||||
with patch.dict("os.environ", {}, clear=True):
|
||||
result = runner.invoke(cli, ["trigger-workflow", "ci.yml", "--token", "tok"])
|
||||
assert result.exit_code != 0
|
||||
assert "GITEA_URL" in result.output
|
||||
|
||||
def test_trigger_workflow_no_token(self) -> None:
|
||||
runner = CliRunner()
|
||||
with patch.dict("os.environ", {"GITEA_URL": "https://git.example.com"}, clear=True):
|
||||
result = runner.invoke(cli, ["trigger-workflow", "ci.yml"])
|
||||
assert result.exit_code != 0
|
||||
assert "CI_GITEA_TOKEN" in result.output
|
||||
|
||||
def test_trigger_workflow_list(self) -> None:
|
||||
runner = CliRunner(env=_TEST_ENV)
|
||||
with patch("gitea_runner_manager.cli.GiteaWorkflowClient") as mock_client_cls:
|
||||
mock_client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
mock_client.list_workflows.return_value = [
|
||||
{"id": 1, "name": "CI", "path": "ci.yml", "state": "active"},
|
||||
{"id": 2, "name": "Post-merge", "path": "post-merge.yml", "state": "active"},
|
||||
]
|
||||
result = runner.invoke(cli, ["trigger-workflow", "--list", "--token", "tok"])
|
||||
assert result.exit_code == 0
|
||||
assert "CI" in result.output
|
||||
assert "Post-merge" in result.output
|
||||
mock_client.list_workflows.assert_called_once_with("oblachno-oss", "grm")
|
||||
|
||||
def test_trigger_workflow_list_empty(self) -> None:
|
||||
runner = CliRunner(env=_TEST_ENV)
|
||||
with patch("gitea_runner_manager.cli.GiteaWorkflowClient") as mock_client_cls:
|
||||
mock_client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
mock_client.list_workflows.return_value = []
|
||||
result = runner.invoke(cli, ["trigger-workflow", "--list", "--token", "tok"])
|
||||
assert result.exit_code == 0
|
||||
assert "No workflows" in result.output
|
||||
|
||||
def test_trigger_workflow_no_workflow_id(self) -> None:
|
||||
runner = CliRunner(env=_TEST_ENV)
|
||||
result = runner.invoke(cli, ["trigger-workflow", "--token", "tok"])
|
||||
assert result.exit_code != 0
|
||||
assert "WORKFLOW_ID" in result.output
|
||||
|
||||
def test_trigger_workflow_api_error(self) -> None:
|
||||
from gitea_runner_manager.gitea_client import GiteaAPIError
|
||||
|
||||
runner = CliRunner(env=_TEST_ENV)
|
||||
with patch("gitea_runner_manager.cli.GiteaWorkflowClient") as mock_client_cls:
|
||||
mock_client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
mock_client.dispatch_workflow.side_effect = GiteaAPIError(404, "workflow not found")
|
||||
result = runner.invoke(cli, ["trigger-workflow", "nonexistent.yml", "--token", "tok"])
|
||||
assert result.exit_code != 0
|
||||
assert "404" in result.output
|
||||
|
||||
def test_trigger_workflow_custom_repo_and_ref(self) -> None:
|
||||
runner = CliRunner(env=_TEST_ENV)
|
||||
with patch("gitea_runner_manager.cli.GiteaWorkflowClient") as mock_client_cls:
|
||||
mock_client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
mock_client.dispatch_workflow.return_value = None
|
||||
result = runner.invoke(
|
||||
cli,
|
||||
["trigger-workflow", "build.yml", "--repo", "myorg/myrepo", "--ref", "develop", "--token", "tok"],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
mock_client.dispatch_workflow.assert_called_once_with("myorg", "myrepo", "build.yml", "develop")
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
"""Unit tests for gitea_client module."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from gitea_runner_manager.gitea_client import GiteaAPIError, GiteaWorkflowClient
|
||||
|
||||
|
||||
class TestGiteaWorkflowClient:
|
||||
def _client(self) -> GiteaWorkflowClient:
|
||||
return GiteaWorkflowClient("https://git.example.com", "test-token")
|
||||
|
||||
def test_list_workflows(self) -> None:
|
||||
client = self._client()
|
||||
mock_response = {"workflows": [{"id": 1, "name": "CI", "path": "ci.yml", "state": "active"}]}
|
||||
with patch.object(client, "_request", return_value=mock_response) as mock_req:
|
||||
result = client.list_workflows("oblachno-oss", "grm")
|
||||
assert len(result) == 1
|
||||
assert result[0]["name"] == "CI"
|
||||
mock_req.assert_called_once_with("GET", "/repos/oblachno-oss/grm/actions/workflows")
|
||||
|
||||
def test_list_workflows_empty(self) -> None:
|
||||
client = self._client()
|
||||
with patch.object(client, "_request", return_value=None):
|
||||
result = client.list_workflows("oblachno-oss", "grm")
|
||||
assert result == []
|
||||
|
||||
def test_dispatch_workflow(self) -> None:
|
||||
client = self._client()
|
||||
mock_response = {"id": 42, "html_url": "https://git.example.com/oblachno-oss/grm/actions/runs/42"}
|
||||
with patch.object(client, "_request", return_value=mock_response) as mock_req:
|
||||
result = client.dispatch_workflow("oblachno-oss", "grm", "ci.yml", ref="master")
|
||||
assert result is not None
|
||||
assert result["id"] == 42
|
||||
mock_req.assert_called_once_with(
|
||||
"POST",
|
||||
"/repos/oblachno-oss/grm/actions/workflows/ci.yml/dispatches?return_run_details=true",
|
||||
{"ref": "master"},
|
||||
)
|
||||
|
||||
def test_dispatch_workflow_with_inputs(self) -> None:
|
||||
client = self._client()
|
||||
with patch.object(client, "_request", return_value=None) as mock_req:
|
||||
client.dispatch_workflow("oblachno-oss", "grm", "build.yml", ref="master", inputs={"env": "prod"})
|
||||
mock_req.assert_called_once_with(
|
||||
"POST",
|
||||
"/repos/oblachno-oss/grm/actions/workflows/build.yml/dispatches?return_run_details=true",
|
||||
{"ref": "master", "inputs": {"env": "prod"}},
|
||||
)
|
||||
|
||||
def test_dispatch_workflow_api_error(self) -> None:
|
||||
client = self._client()
|
||||
with patch.object(client, "_request", side_effect=GiteaAPIError(404, "workflow not found")):
|
||||
with pytest.raises(GiteaAPIError) as exc_info:
|
||||
client.dispatch_workflow("oblachno-oss", "grm", "nonexistent.yml")
|
||||
assert exc_info.value.status == 404
|
||||
|
||||
|
||||
class TestGiteaWorkflowClientRequest:
|
||||
"""Test the underlying _request method with mocked urllib."""
|
||||
|
||||
def test_request_success(self) -> None:
|
||||
client = GiteaWorkflowClient("https://git.example.com/", "tok")
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status = 200
|
||||
mock_resp.read.return_value = json.dumps({"ok": True}).encode()
|
||||
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
||||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||||
with patch("urllib.request.urlopen", return_value=mock_resp) as mock_urlopen:
|
||||
result = client._request("GET", "/test")
|
||||
assert result == {"ok": True}
|
||||
mock_urlopen.assert_called_once()
|
||||
|
||||
def test_request_204_no_content(self) -> None:
|
||||
client = GiteaWorkflowClient("https://git.example.com", "tok")
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status = 204
|
||||
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
||||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||||
with patch("urllib.request.urlopen", return_value=mock_resp):
|
||||
result = client._request("POST", "/test", {"ref": "master"})
|
||||
assert result is None
|
||||
|
||||
def test_request_http_error(self) -> None:
|
||||
import urllib.error
|
||||
|
||||
client = GiteaWorkflowClient("https://git.example.com", "tok")
|
||||
err = urllib.error.HTTPError(
|
||||
"https://git.example.com/api/v1/test",
|
||||
404,
|
||||
"Not Found",
|
||||
{},
|
||||
__import__("io").BytesIO(b'{"message": "resource not found"}'),
|
||||
)
|
||||
with patch("urllib.request.urlopen", side_effect=err):
|
||||
with pytest.raises(GiteaAPIError) as exc_info:
|
||||
client._request("GET", "/test")
|
||||
assert exc_info.value.status == 404
|
||||
assert "resource not found" in exc_info.value.message
|
||||
|
||||
def test_request_http_error_non_json(self) -> None:
|
||||
import urllib.error
|
||||
|
||||
client = GiteaWorkflowClient("https://git.example.com", "tok")
|
||||
err = urllib.error.HTTPError(
|
||||
"https://git.example.com/api/v1/test",
|
||||
500,
|
||||
"Internal Server Error",
|
||||
{},
|
||||
__import__("io").BytesIO(b"plain text error"),
|
||||
)
|
||||
with patch("urllib.request.urlopen", side_effect=err):
|
||||
with pytest.raises(GiteaAPIError) as exc_info:
|
||||
client._request("GET", "/test")
|
||||
assert exc_info.value.status == 500
|
||||
assert "plain text error" in exc_info.value.message
|
||||
Reference in New Issue
Block a user