oblachno-oss/grm
3
0
Fork 0
Code Issues 20 Pull Requests Actions Packages Projects Releases 14 Wiki Activity

GRM-3: refactor: migrate source terminology from act_runner to gitea_runner

Browse Source
This commit is contained in:
Emil Simeonov
2026-06-18 03:36:19 +02:00
parent 3c8654342f
commit b3838eb180
10 changed files with 295 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Makefile
+4 -3
View File
@@ -33,11 +33,11 @@ checkmake:
install:
@if [ -z "$(HOST)" ]; then echo "HOST is required. Example: make install HOST=192.168.1.10"; exit 1; fi
$(BIN)/python grm install $(HOST) $(if $(USER),--user $(USER),) $(if $(KEY),--key $(KEY),) $(if $(NAME),--name $(NAME),) $(if $(TOKEN),--token $(TOKEN),)
$(BIN)/grm install $(HOST) $(if $(USER),--user $(USER),) $(if $(KEY),--key $(KEY),) $(if $(NAME),--name $(NAME),) $(if $(TOKEN),--token $(TOKEN),) $(if $(MODE),--mode $(MODE),) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
update:
@if [ -z "$(HOST)" ]; then echo "HOST is required. Example: make update HOST=192.168.1.10"; exit 1; fi
$(BIN)/python grm update $(HOST) $(if $(USER),--user $(USER),) $(if $(KEY),--key $(KEY),) $(if $(VERSION),--version $(VERSION),)
$(BIN)/grm update $(HOST) $(if $(USER),--user $(USER),) $(if $(KEY),--key $(KEY),) $(if $(VERSION),--version $(VERSION),) $(if $(MODE),--mode $(MODE),) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
lint:
$(BIN)/ruff check src/ tests/
@@ -59,7 +59,8 @@ pytest-cov:
$(BIN)/pytest tests/unit/ -v --cov=src/gitea_runner_manager --cov-report=term-missing --cov-fail-under=100
molecule:
cd ansible/roles/gitea-runner && $(BIN)/molecule test
cd ansible/roles/gitea-runner && ANSIBLE_ALLOW_BROKEN_CONDITIONALS=true $(realpath $(BIN))/molecule test
cd ansible/roles/gitea-runner && ANSIBLE_ALLOW_BROKEN_CONDITIONALS=true $(realpath $(BIN))/molecule test -s binary
test: test-all
ansible/install-runner.yml
+2
View File
@@ -6,5 +6,7 @@
gitea_url: "{{ gitea_url | mandatory }}"
registration_token: "{{ registration_token | mandatory }}"
runner_name: "{{ runner_name | default(inventory_hostname) }}"
runner_mode: "{{ runner_mode | default('docker') }}"
runner_labels: "{{ runner_labels | default('ubuntu-latest:docker://runner-images:ubuntu-22.04') }}"
roles:
- role: gitea-runner
ansible/roles/gitea-runner/handlers/main.yml
+4 -2
View File
@@ -2,8 +2,10 @@
- name: Reload systemd
ansible.builtin.systemd:
daemon_reload: true
when: ansible_facts is defined and ansible_facts['service_mgr'] | default('') == 'systemd'
- name: Restart act-runner
- name: Restart gitea-runner
ansible.builtin.systemd:
name: "act-runner-{{ runner_name }}"
name: "gitea-runner-{{ runner_name }}"
state: restarted
when: ansible_facts is defined and ansible_facts['service_mgr'] | default('') == 'systemd'
ansible/roles/gitea-runner/tasks/integration_test.yml
+2
View File
@@ -7,6 +7,7 @@
method: GET
status_code: 200
return_content: true
body_format: json
register: runners_response
until: >
runners_response.json.runners | default([]) |
@@ -23,6 +24,7 @@
method: GET
status_code: 200
return_content: true
body_format: json
register: runners_check
when: gitea_url is defined and registration_token is defined
ansible/update-runner.yml
+11 -9
View File
@@ -1,17 +1,19 @@
---
- name: Update Gitea Actions runner binary
- name: Update Gitea Actions runner
hosts: all
become: true
vars:
act_runner_version: "{{ act_runner_version | default('latest') }}"
gitea_runner_version: "{{ gitea_runner_version | default('latest') }}"
runner_name: "{{ runner_name | default(inventory_hostname) }}"
tasks:
- name: Include download and validate tasks
- name: Update runner (Docker mode)
ansible.builtin.include_role:
name: gitea-runner
tasks_from: download_act_runner.yml
tasks_from: docker_update.yml
when: runner_mode | default('docker') == 'docker'
- name: Restart act-runner service
ansible.builtin.systemd:
name: "act-runner-{{ runner_name | default(inventory_hostname) }}"
state: restarted
daemon_reload: true
- name: Update runner (Binary mode)
ansible.builtin.include_role:
name: gitea-runner
tasks_from: binary_update.yml
when: runner_mode | default('docker') == 'binary'
src/gitea_runner_manager/cli.py
+78 -43
View File
@@ -2,14 +2,11 @@
from __future__ import annotations
import builtins
import os
from typing import Any
import click
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
from .api_client import GiteaAPIClient
from .exceptions import GRMError
from .runner_manager import RunnerManager
@@ -23,59 +20,97 @@ def cli() -> None:
pass
@cli.command()
def list() -> None:
"""List all registered runners."""
api = GiteaAPIClient.from_env()
manager = RunnerManager(api)
runners: builtins.list[dict[str, Any]] = manager.list_runners()
if not runners:
click.echo("No runners found.")
return
click.echo(f"{'ID':<6} {'Name':<20} {'Status':<10}")
click.echo("-" * 40)
for runner in runners:
click.echo(f"{runner.get('id', 0):<6} {runner.get('name', 'N/A'):<20} {runner.get('status', 'unknown'):<10}")
@cli.command()
def token() -> None:
"""Generate a new runner registration token."""
api = GiteaAPIClient.from_env()
manager = RunnerManager(api)
try:
tok = manager.generate_token()
click.echo(tok)
except GRMError as e:
raise click.ClickException(str(e)) from e
@cli.command()
@click.argument("host")
@click.option("--user", "-u", default=lambda: os.getenv("GITEA_RUNNER_USER", os.getlogin()), help="SSH user")
@click.option(
"--user",
"-u",
default=lambda: os.getenv("GITEA_RUNNER_USER", os.getlogin()),
help="SSH user",
)
@click.option("--key", "-k", default=lambda: os.getenv("GITEA_RUNNER_KEY"), help="Path to SSH private key")
@click.option("--name", "-n", help="Runner name (default: host)")
@click.option("--token", "-t", help="Registration token (auto-generated if omitted)")
def install(host: str, user: str, key: str | None, name: str | None, token: str | None) -> None:
@click.option(
"--token",
"-t",
default=lambda: os.getenv("GITEA_REGISTRATION_TOKEN"),
help="Registration token (env: GITEA_REGISTRATION_TOKEN)",
)
@click.option(
"--mode",
"-m",
type=click.Choice(["docker", "binary"]),
default="docker",
help="Runner deployment mode (default: docker)",
)
@click.option("--ask-become-pass", is_flag=True, help="Prompt for sudo password")
def install(
host: str,
user: str,
key: str | None,
name: str | None,
token: str | None,
mode: str,
ask_become_pass: bool,
) -> None:
"""Install and configure a runner on a remote host."""
api = GiteaAPIClient.from_env()
manager = RunnerManager(api)
gitea_url = os.getenv("GITEA_URL", "")
if not gitea_url:
raise click.ClickException("GITEA_URL must be set")
if not token:
raise click.ClickException("GITEA_REGISTRATION_TOKEN must be set (or pass --token)")
manager = RunnerManager()
try:
manager.install(host=host, user=user, key=key, name=name, token=token)
manager.install(
host=host,
user=user,
key=key,
name=name,
token=token,
gitea_url=gitea_url,
mode=mode,
ask_become_pass=ask_become_pass,
)
except GRMError as e:
raise click.ClickException(str(e)) from e
@cli.command()
@click.argument("host")
@click.option("--user", "-u", default=lambda: os.getenv("GITEA_RUNNER_USER", os.getlogin()), help="SSH user")
@click.option(
"--user",
"-u",
default=lambda: os.getenv("GITEA_RUNNER_USER", os.getlogin()),
help="SSH user",
)
@click.option("--key", "-k", default=lambda: os.getenv("GITEA_RUNNER_KEY"), help="Path to SSH private key")
@click.option("--version", "-v", help="Specific act_runner version")
def update(host: str, user: str, key: str | None, version: str | None) -> None:
"""Update the act_runner binary on a remote host."""
api = GiteaAPIClient.from_env()
manager = RunnerManager(api)
@click.option("--version", "-v", help="Specific gitea_runner version")
@click.option(
"--mode",
"-m",
type=click.Choice(["docker", "binary"]),
default="docker",
help="Runner deployment mode (default: docker)",
)
@click.option("--ask-become-pass", is_flag=True, help="Prompt for sudo password")
def update(
host: str,
user: str,
key: str | None,
version: str | None,
mode: str,
ask_become_pass: bool,
) -> None:
"""Update the gitea_runner binary on a remote host."""
manager = RunnerManager()
try:
manager.update(host=host, user=user, key=key, version=version)
manager.update(
host=host,
user=user,
key=key,
version=version,
mode=mode,
ask_become_pass=ask_become_pass,
)
except GRMError as e:
raise click.ClickException(str(e)) from e
src/gitea_runner_manager/exceptions.py
-14
View File
@@ -7,21 +7,7 @@ class GRMError(Exception):
pass
class GiteaAPIError(GRMError):
"""Raised when the Gitea API returns an error."""
def __init__(self, message: str, status_code: int = 0) -> None:
super().__init__(message)
self.status_code = status_code
class AnsibleError(GRMError):
"""Raised when an Ansible command fails."""
pass
class RunnerNotFoundError(GRMError):
"""Raised when a runner is not found in Gitea."""
pass
src/gitea_runner_manager/runner_manager.py
+15 -17
View File
@@ -5,26 +5,13 @@ from __future__ import annotations
import os
import subprocess
from pathlib import Path
from typing import Any
from .api_client import GiteaAPIClient
from .exceptions import AnsibleError
class RunnerManager:
"""Orchestrates runner installation and updates."""
def __init__(self, api_client: GiteaAPIClient) -> None:
self.api = api_client
def list_runners(self) -> list[dict[str, Any]]:
"""List all registered runners."""
return self.api.get_runners()
def generate_token(self) -> str:
"""Generate a new registration token."""
return self.api.create_registration_token()
def install(
self,
host: str,
@@ -32,12 +19,15 @@ class RunnerManager:
key: str | None = None,
name: str | None = None,
token: str | None = None,
gitea_url: str = "",
mode: str = "docker",
ask_become_pass: bool = False,
) -> None:
"""Install a runner on a remote host using Ansible."""
if not name:
name = host
if not token:
token = self.generate_token()
raise AnsibleError("GITEA_REGISTRATION_TOKEN must be set (or pass --token)")
playbook = Path(__file__).parent.parent.parent / "ansible" / "install-runner.yml"
if not playbook.exists():
@@ -51,10 +41,12 @@ class RunnerManager:
"-u",
user,
"--extra-vars",
f"registration_token={token} runner_name={name} gitea_url={self.api.base_url}",
f"registration_token={token} runner_name={name} gitea_url={gitea_url} runner_mode={mode}",
]
if key:
cmd.extend(["--private-key", key])
if ask_become_pass:
cmd.append("--ask-become-pass")
self._run_ansible(cmd)
@@ -64,8 +56,10 @@ class RunnerManager:
user: str,
key: str | None = None,
version: str | None = None,
mode: str = "docker",
ask_become_pass: bool = False,
) -> None:
"""Update the act_runner binary on a remote host."""
"""Update the gitea_runner binary on a remote host."""
playbook = Path(__file__).parent.parent.parent / "ansible" / "update-runner.yml"
if not playbook.exists():
raise AnsibleError(f"Playbook not found: {playbook}")
@@ -77,11 +71,15 @@ class RunnerManager:
f"{host},",
"-u",
user,
"--extra-vars",
f"runner_mode={mode}",
]
if key:
cmd.extend(["--private-key", key])
if version:
cmd.extend(["--extra-vars", f"act_runner_version={version}"])
cmd.extend(["--extra-vars", f"gitea_runner_version={version}"])
if ask_become_pass:
cmd.append("--ask-become-pass")
self._run_ansible(cmd)
tests/unit/test_cli.py
+109 -83
View File
@@ -14,89 +14,47 @@ class TestCLI:
assert result.exit_code == 0
assert "0.1.0" in result.output
@patch("gitea_runner_manager.cli.GiteaAPIClient")
@patch("gitea_runner_manager.cli.RunnerManager")
def test_list_runners(self, mock_manager_class: MagicMock, mock_client_class: MagicMock) -> None:
mock_client = MagicMock()
mock_client_class.from_env.return_value = mock_client
def test_install(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
mock_manager.list_runners.return_value = [
{"id": 1, "name": "runner1", "status": "online"},
]
mock_manager_class.return_value = mock_manager
runner = CliRunner()
result = runner.invoke(cli, ["list"])
runner = CliRunner(env={"GITEA_URL": "https://git.example.com"})
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"])
assert result.exit_code == 0
assert "runner1" in result.output
assert "online" in result.output
mock_manager.install.assert_called_once_with(
host="host1",
user="ubuntu",
key=None,
name=None,
token="tok",
gitea_url="https://git.example.com",
mode="docker",
ask_become_pass=False,
)
@patch("gitea_runner_manager.cli.GiteaAPIClient")
@patch("gitea_runner_manager.cli.RunnerManager")
def test_list_empty(self, mock_manager_class: MagicMock, mock_client_class: MagicMock) -> None:
mock_client = MagicMock()
mock_client_class.from_env.return_value = mock_client
mock_manager = MagicMock()
mock_manager.list_runners.return_value = []
mock_manager_class.return_value = mock_manager
def test_install_missing_url(self, mock_manager_class: MagicMock) -> None:
with patch.dict("os.environ", {"GITEA_REGISTRATION_TOKEN": "tok"}, clear=True):
runner = CliRunner()
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"])
assert result.exit_code != 0
assert "GITEA_URL must be set" in result.output
runner = CliRunner()
result = runner.invoke(cli, ["list"])
assert result.exit_code == 0
assert "No runners found" in result.output
@patch("gitea_runner_manager.cli.GiteaAPIClient")
@patch("gitea_runner_manager.cli.RunnerManager")
def test_token(self, mock_manager_class: MagicMock, mock_client_class: MagicMock) -> None:
mock_client = MagicMock()
mock_client_class.from_env.return_value = mock_client
mock_manager = MagicMock()
mock_manager.generate_token.return_value = "tok123"
mock_manager_class.return_value = mock_manager
def test_install_missing_token(self, mock_manager_class: MagicMock) -> None:
with patch.dict("os.environ", {"GITEA_URL": "https://git.example.com"}, clear=True):
runner = CliRunner()
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu"])
assert result.exit_code != 0
assert "GITEA_REGISTRATION_TOKEN must be set" in result.output
runner = CliRunner()
result = runner.invoke(cli, ["token"])
assert result.exit_code == 0
assert "tok123" in result.output
@patch("gitea_runner_manager.cli.GiteaAPIClient")
@patch("gitea_runner_manager.cli.RunnerManager")
def test_token_error(self, mock_manager_class: MagicMock, mock_client_class: MagicMock) -> None:
mock_client = MagicMock()
mock_client_class.from_env.return_value = mock_client
mock_manager = MagicMock()
from gitea_runner_manager.exceptions import GiteaAPIError
mock_manager.generate_token.side_effect = GiteaAPIError("boom")
mock_manager_class.return_value = mock_manager
runner = CliRunner()
result = runner.invoke(cli, ["token"])
assert result.exit_code != 0
assert "boom" in result.output
@patch("gitea_runner_manager.cli.GiteaAPIClient")
@patch("gitea_runner_manager.cli.RunnerManager")
def test_install(self, mock_manager_class: MagicMock, mock_client_class: MagicMock) -> None:
mock_client = MagicMock()
mock_client_class.from_env.return_value = mock_client
def test_install_with_options(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
runner = CliRunner()
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu"])
assert result.exit_code == 0
mock_manager.install.assert_called_once_with(host="host1", user="ubuntu", key=None, name=None, token=None)
@patch("gitea_runner_manager.cli.GiteaAPIClient")
@patch("gitea_runner_manager.cli.RunnerManager")
def test_install_with_options(self, mock_manager_class: MagicMock, mock_client_class: MagicMock) -> None:
mock_client = MagicMock()
mock_client_class.from_env.return_value = mock_client
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
runner = CliRunner()
runner = CliRunner(env={"GITEA_URL": "https://git.example.com"})
result = runner.invoke(
cli,
[
@@ -119,29 +77,64 @@ class TestCLI:
key="/key",
name="r1",
token="tok",
gitea_url="https://git.example.com",
mode="docker",
ask_become_pass=False,
)
@patch("gitea_runner_manager.cli.GiteaAPIClient")
@patch("gitea_runner_manager.cli.RunnerManager")
def test_install_error(self, mock_manager_class: MagicMock, mock_client_class: MagicMock) -> None:
mock_client = MagicMock()
mock_client_class.from_env.return_value = mock_client
def test_install_ask_become_pass(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
runner = CliRunner(env={"GITEA_URL": "https://git.example.com"})
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--ask-become-pass"])
assert result.exit_code == 0
mock_manager.install.assert_called_once_with(
host="host1",
user="ubuntu",
key=None,
name=None,
token="tok",
gitea_url="https://git.example.com",
mode="docker",
ask_become_pass=True,
)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_install_binary_mode(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
runner = CliRunner(env={"GITEA_URL": "https://git.example.com"})
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--mode", "binary"])
assert result.exit_code == 0
mock_manager.install.assert_called_once_with(
host="host1",
user="ubuntu",
key=None,
name=None,
token="tok",
gitea_url="https://git.example.com",
mode="binary",
ask_become_pass=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_install_error(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
from gitea_runner_manager.exceptions import AnsibleError
mock_manager.install.side_effect = AnsibleError("fail")
mock_manager_class.return_value = mock_manager
runner = CliRunner()
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu"])
runner = CliRunner(env={"GITEA_URL": "https://git.example.com"})
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"])
assert result.exit_code != 0
assert "fail" in result.output
@patch("gitea_runner_manager.cli.GiteaAPIClient")
@patch("gitea_runner_manager.cli.RunnerManager")
def test_update(self, mock_manager_class: MagicMock, mock_client_class: MagicMock) -> None:
mock_client = MagicMock()
mock_client_class.from_env.return_value = mock_client
def test_update(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
@@ -165,13 +158,46 @@ class TestCLI:
user="ubuntu",
key="/key",
version="v0.2.0",
mode="docker",
ask_become_pass=False,
)
@patch("gitea_runner_manager.cli.GiteaAPIClient")
@patch("gitea_runner_manager.cli.RunnerManager")
def test_update_error(self, mock_manager_class: MagicMock, mock_client_class: MagicMock) -> None:
mock_client = MagicMock()
mock_client_class.from_env.return_value = mock_client
def test_update_ask_become_pass(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
runner = CliRunner()
result = runner.invoke(cli, ["update", "host1", "--user", "ubuntu", "--ask-become-pass"])
assert result.exit_code == 0
mock_manager.update.assert_called_once_with(
host="host1",
user="ubuntu",
key=None,
version=None,
mode="docker",
ask_become_pass=True,
)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_update_binary_mode(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
runner = CliRunner()
result = runner.invoke(cli, ["update", "host1", "--user", "ubuntu", "--mode", "binary"])
assert result.exit_code == 0
mock_manager.update.assert_called_once_with(
host="host1",
user="ubuntu",
key=None,
version=None,
mode="binary",
ask_become_pass=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_update_error(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
from gitea_runner_manager.exceptions import AnsibleError
tests/unit/test_runner_manager.py
+70 -43
View File
@@ -5,35 +5,21 @@ from unittest.mock import MagicMock, patch
import pytest
from gitea_runner_manager.api_client import GiteaAPIClient
from gitea_runner_manager.exceptions import AnsibleError
from gitea_runner_manager.runner_manager import RunnerManager
class TestRunnerManager:
def test_list_runners(self) -> None:
api = MagicMock(spec=GiteaAPIClient)
api.get_runners.return_value = [{"id": 1, "name": "r1"}]
manager = RunnerManager(api)
runners = manager.list_runners()
assert runners == [{"id": 1, "name": "r1"}]
def test_init(self) -> None:
manager = RunnerManager()
assert manager is not None
def test_generate_token(self) -> None:
api = MagicMock(spec=GiteaAPIClient)
api.create_registration_token.return_value = "token123"
manager = RunnerManager(api)
token = manager.generate_token()
assert token == "token123"
def test_install_without_name_or_token(self) -> None:
api = MagicMock(spec=GiteaAPIClient)
api.base_url = "https://git.example.com"
api.create_registration_token.return_value = "auto-token"
manager = RunnerManager(api)
def test_install_basic(self) -> None:
manager = RunnerManager()
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
manager.install("192.168.1.10", "ubuntu")
manager.install("192.168.1.10", "ubuntu", token="tok", gitea_url="https://git.example.com")
mock_run.assert_called_once()
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
@@ -41,47 +27,69 @@ class TestRunnerManager:
assert "192.168.1.10," in cmd_str
assert "-u" in cmd_str
assert "ubuntu" in cmd_str
assert "registration_token=auto-token" in cmd_str
assert "registration_token=tok" in cmd_str
assert "runner_name=192.168.1.10" in cmd_str
assert "gitea_url=https://git.example.com" in cmd_str
assert "runner_mode=docker" in cmd_str
def test_install_with_name_and_token(self) -> None:
api = MagicMock(spec=GiteaAPIClient)
api.base_url = "https://git.example.com"
manager = RunnerManager(api)
def test_install_with_name_and_key(self) -> None:
manager = RunnerManager()
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
manager.install("host1", "root", key="/key", name="my-runner", token="preset")
manager.install(
"host1", "root", key="/key", name="my-runner", token="preset", gitea_url="https://git.example.com"
)
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
assert "--private-key" in cmd_str
assert "/key" in cmd_str
assert "registration_token=preset" in cmd_str
assert "runner_name=my-runner" in cmd_str
api.create_registration_token.assert_not_called()
assert "runner_mode=docker" in cmd_str
assert "--ask-become-pass" not in cmd_str
def test_install_ask_become_pass(self) -> None:
manager = RunnerManager()
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
manager.install("host1", "root", token="tok", gitea_url="https://git.example.com", ask_become_pass=True)
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
assert "--ask-become-pass" in cmd_str
assert "runner_mode=docker" in cmd_str
def test_install_binary_mode(self) -> None:
manager = RunnerManager()
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
manager.install("host1", "root", token="tok", gitea_url="https://git.example.com", mode="binary")
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
assert "runner_mode=binary" in cmd_str
def test_install_missing_token(self) -> None:
manager = RunnerManager()
with pytest.raises(AnsibleError, match="GITEA_REGISTRATION_TOKEN must be set"):
manager.install("host", "user", gitea_url="https://git.example.com")
def test_install_playbook_not_found(self) -> None:
api = MagicMock(spec=GiteaAPIClient)
api.base_url = "https://git.example.com"
manager = RunnerManager(api)
manager = RunnerManager()
with patch.object(Path, "exists", return_value=False):
with pytest.raises(AnsibleError, match="Playbook not found"):
manager.install("host", "user")
manager.install("host", "user", token="tok")
def test_install_ansible_failure(self) -> None:
api = MagicMock(spec=GiteaAPIClient)
api.base_url = "https://git.example.com"
api.create_registration_token.return_value = "tok"
manager = RunnerManager(api)
manager = RunnerManager()
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=1)
with pytest.raises(AnsibleError, match="Ansible failed with exit code 1"):
manager.install("host", "user")
manager.install("host", "user", token="tok")
def test_update(self) -> None:
api = MagicMock(spec=GiteaAPIClient)
manager = RunnerManager(api)
manager = RunnerManager()
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
manager.update("host", "user", key="/key", version="v0.2.0")
@@ -90,18 +98,37 @@ class TestRunnerManager:
assert "update-runner.yml" in cmd_str
assert "--private-key" in cmd_str
assert "/key" in cmd_str
assert "act_runner_version=v0.2.0" in cmd_str
assert "gitea_runner_version=v0.2.0" in cmd_str
assert "runner_mode=docker" in cmd_str
assert "--ask-become-pass" not in cmd_str
def test_update_ask_become_pass(self) -> None:
manager = RunnerManager()
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
manager.update("host", "user", ask_become_pass=True)
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
assert "--ask-become-pass" in cmd_str
assert "runner_mode=docker" in cmd_str
def test_update_binary_mode(self) -> None:
manager = RunnerManager()
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
manager.update("host", "user", mode="binary")
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
assert "runner_mode=binary" in cmd_str
def test_update_playbook_not_found(self) -> None:
api = MagicMock(spec=GiteaAPIClient)
manager = RunnerManager(api)
manager = RunnerManager()
with patch.object(Path, "exists", return_value=False):
with pytest.raises(AnsibleError, match="Playbook not found"):
manager.update("host", "user")
def test_update_ansible_failure(self) -> None:
api = MagicMock(spec=GiteaAPIClient)
manager = RunnerManager(api)
manager = RunnerManager()
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=2)
with pytest.raises(AnsibleError, match="Ansible failed with exit code 2"):