emil
f67dff8458
Post-merge / detect-type (push) Successful in 51s
Post-merge / release (push) Successful in 1m14s
Post-merge / validate-commit-msg (push) Successful in 1m14s
Post-merge / vikunja (push) Successful in 1m14s
Post-merge / badges (push) Successful in 1m25s
Post-merge / sync-wiki (push) Successful in 1m53s
Post-merge / configure-repo (push) Successful in 1m20s
Post-merge / publish (push) Successful in 1m21s
719 lines
29 KiB
Python
719 lines
29 KiB
Python
"""Unit tests for runner_manager module."""
|
|
|
|
import json
|
|
import os
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from gitea_runner_manager.exceptions import AnsibleError
|
|
from gitea_runner_manager.runner_manager import RunnerManager
|
|
|
|
|
|
@contextmanager
|
|
def _capture_extra_vars(self, extra_vars: dict[str, str | int] | None):
|
|
"""Capture extra_vars dict for inspection instead of writing to file."""
|
|
self._captured_extra_vars = extra_vars
|
|
yield "/tmp/fake-vars.json" if extra_vars else None
|
|
|
|
|
|
class TestRunnerManager:
|
|
@pytest.fixture(autouse=True)
|
|
def _patch_extra_vars_file(self) -> None:
|
|
"""Patch ``_extra_vars_file`` so no real temp files are created."""
|
|
with patch.object(RunnerManager, "_extra_vars_file", _capture_extra_vars):
|
|
yield
|
|
|
|
def test_init(self) -> None:
|
|
manager = RunnerManager()
|
|
assert manager is not None
|
|
assert manager._executor is not None
|
|
assert manager._registry is not None
|
|
|
|
def test_install_basic(self) -> None:
|
|
mock_registry = MagicMock()
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.install("192.168.1.10", "ubuntu", token="tok", gitea_url="https://git.example.com")
|
|
mock_executor.run.assert_called_once()
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "ansible-playbook" in cmd_str
|
|
assert "192.168.1.10," in cmd_str
|
|
assert "-u" in cmd_str
|
|
assert "ubuntu" in cmd_str
|
|
assert "--extra-vars" in cmd_str
|
|
assert "@/tmp/fake-vars.json" in cmd_str
|
|
assert manager._captured_extra_vars["registration_token"] == "tok"
|
|
assert manager._captured_extra_vars["runner_name"] == "192.168.1.10"
|
|
assert manager._captured_extra_vars["gitea_url"] == "https://git.example.com"
|
|
assert "Installing Gitea Runner on 192.168.1.10" in mock_executor.run.call_args.kwargs["description"]
|
|
mock_registry.add.assert_called_once_with(
|
|
name="192.168.1.10",
|
|
host="192.168.1.10",
|
|
user="ubuntu",
|
|
key=None,
|
|
gitea_url="https://git.example.com",
|
|
labels=None,
|
|
)
|
|
|
|
def test_install_with_name_and_key(self) -> None:
|
|
mock_registry = MagicMock()
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.install(
|
|
"host1", "root", key="/key", name="my-runner", token="preset", gitea_url="https://git.example.com"
|
|
)
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "--private-key" in cmd_str
|
|
assert "/key" in cmd_str
|
|
assert manager._captured_extra_vars["registration_token"] == "preset"
|
|
assert manager._captured_extra_vars["runner_name"] == "my-runner"
|
|
assert "--ask-become-pass" not in cmd_str
|
|
mock_registry.add.assert_called_once_with(
|
|
name="my-runner",
|
|
host="host1",
|
|
user="root",
|
|
key="/key",
|
|
gitea_url="https://git.example.com",
|
|
labels=None,
|
|
)
|
|
|
|
def test_install_ask_become_pass(self) -> None:
|
|
mock_registry = MagicMock()
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.install("host1", "root", token="tok", gitea_url="https://git.example.com", ask_become_pass=True)
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "--ask-become-pass" in cmd_str
|
|
|
|
def test_install_missing_gitea_url(self) -> None:
|
|
manager = RunnerManager()
|
|
with pytest.raises(AnsibleError, match="GITEA_URL must be set"):
|
|
manager.install("host", "user", token="tok")
|
|
|
|
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:
|
|
manager = RunnerManager()
|
|
with patch.object(Path, "exists", return_value=False):
|
|
with pytest.raises(AnsibleError, match="Playbook not found"):
|
|
manager.install("host", "user", token="tok", gitea_url="https://git.example.com")
|
|
|
|
def test_install_with_labels(self) -> None:
|
|
mock_registry = MagicMock()
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.install(
|
|
"host1",
|
|
"root",
|
|
token="tok",
|
|
gitea_url="https://git.example.com",
|
|
labels="docker:docker://alpine:latest",
|
|
)
|
|
assert manager._captured_extra_vars["runner_labels"] == "docker:docker://alpine:latest"
|
|
|
|
def test_install_no_labels(self) -> None:
|
|
mock_registry = MagicMock()
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.install("host1", "root", token="tok", gitea_url="https://git.example.com")
|
|
assert "runner_labels" not in manager._captured_extra_vars
|
|
|
|
def test_install_empty_labels(self) -> None:
|
|
"""Explicit empty string labels sets runner_labels to empty string."""
|
|
mock_registry = MagicMock()
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.install("host1", "root", token="tok", gitea_url="https://git.example.com", labels="")
|
|
assert manager._captured_extra_vars["runner_labels"] == ""
|
|
|
|
def test_install_with_admin_token(self) -> None:
|
|
mock_registry = MagicMock()
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.install("host1", "root", token="tok", gitea_url="https://git.example.com", admin_token="admin-tok")
|
|
assert manager._captured_extra_vars["gitea_admin_token"] == "admin-tok"
|
|
|
|
def test_install_ansible_failure(self) -> None:
|
|
mock_registry = MagicMock()
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
mock_executor.run.side_effect = AnsibleError("Ansible failed with exit code 1. See full log: /tmp/test.log")
|
|
with pytest.raises(AnsibleError, match="Ansible failed with exit code 1"):
|
|
manager.install("host", "user", token="tok", gitea_url="https://git.example.com")
|
|
mock_registry.add.assert_not_called()
|
|
|
|
def test_update(self) -> None:
|
|
manager = RunnerManager()
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.update("host", "user", key="/key", version="v0.2.0")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "update-runner.yml" in cmd_str
|
|
assert "--private-key" in cmd_str
|
|
assert "/key" in cmd_str
|
|
assert manager._captured_extra_vars["gitea_runner_version"] == "v0.2.0"
|
|
assert "--ask-become-pass" not in cmd_str
|
|
assert "Updating Gitea Runner on host" in mock_executor.run.call_args.kwargs["description"]
|
|
|
|
def test_update_ask_become_pass(self) -> None:
|
|
manager = RunnerManager()
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.update("host", "user", ask_become_pass=True)
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "--ask-become-pass" in cmd_str
|
|
assert "--extra-vars" not in cmd_str
|
|
|
|
def test_update_playbook_not_found(self) -> None:
|
|
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:
|
|
manager = RunnerManager()
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
mock_executor.run.side_effect = AnsibleError("Ansible failed with exit code 2. See full log: /tmp/test.log")
|
|
with pytest.raises(AnsibleError, match="Ansible failed with exit code 2"):
|
|
manager.update("host", "user")
|
|
|
|
def test_resolve_runner_from_registry(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {
|
|
"host": "10.0.0.1",
|
|
"user": "ubuntu",
|
|
"key": "/key",
|
|
"gitea_url": "https://git.example.com",
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
host, user, key, gitea_url = manager._resolve_runner("r1")
|
|
assert host == "10.0.0.1"
|
|
assert user == "ubuntu"
|
|
assert key == "/key"
|
|
assert gitea_url == "https://git.example.com"
|
|
mock_registry.get.assert_called_once_with("r1")
|
|
|
|
def test_resolve_runner_explicit_host_user(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
manager = RunnerManager(registry=mock_registry)
|
|
host, user, key, gitea_url = manager._resolve_runner("r1", host="10.0.0.2", user="root")
|
|
assert host == "10.0.0.2"
|
|
assert user == "root"
|
|
assert key is None
|
|
assert gitea_url == ""
|
|
|
|
def test_resolve_runner_override(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {
|
|
"host": "10.0.0.1",
|
|
"user": "ubuntu",
|
|
"key": "/key",
|
|
"gitea_url": "https://git.example.com",
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
host, user, key, gitea_url = manager._resolve_runner("r1", host="10.0.0.2", user="root", key="/new")
|
|
assert host == "10.0.0.2"
|
|
assert user == "root"
|
|
assert key == "/new"
|
|
assert gitea_url == "https://git.example.com"
|
|
|
|
def test_resolve_runner_not_found(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
manager = RunnerManager(registry=mock_registry)
|
|
with pytest.raises(AnsibleError, match="not found in registry"):
|
|
manager._resolve_runner("missing")
|
|
|
|
def test_start(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user", "key": None}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.start("r1")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "start-runner.yml" in cmd_str
|
|
assert manager._captured_extra_vars["runner_name"] == "r1"
|
|
assert "Starting Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
|
|
|
|
def test_start_with_override(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user", "key": None}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.start("r1", host="newhost", user="newuser")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "newhost," in cmd_str
|
|
assert "newuser" in cmd_str
|
|
|
|
def test_start_no_registry_with_explicit(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = None
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.start("r1", host="host", user="user")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "start-runner.yml" in cmd_str
|
|
|
|
def test_stop(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user", "key": None}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.stop("r1")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "stop-runner.yml" in cmd_str
|
|
assert manager._captured_extra_vars["runner_name"] == "r1"
|
|
assert "Stopping Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
|
|
|
|
def test_restart(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user", "key": None}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.restart("r1")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "restart-runner.yml" in cmd_str
|
|
assert manager._captured_extra_vars["runner_name"] == "r1"
|
|
assert "Restarting Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
|
|
|
|
def test_enable(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user", "key": None}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.enable("r1")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "enable-runner.yml" in cmd_str
|
|
assert manager._captured_extra_vars["runner_name"] == "r1"
|
|
assert "Enabling Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
|
|
|
|
def test_disable(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {
|
|
"host": "host",
|
|
"user": "user",
|
|
"key": None,
|
|
"gitea_url": "",
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.disable("r1", token="tok", gitea_url="https://git.example.com")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "disable-runner.yml" in cmd_str
|
|
assert manager._captured_extra_vars["runner_name"] == "r1"
|
|
assert manager._captured_extra_vars["registration_token"] == "tok"
|
|
assert manager._captured_extra_vars["gitea_url"] == "https://git.example.com"
|
|
assert "Disabling Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
|
|
|
|
def test_disable_uses_registry_gitea_url(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {
|
|
"host": "host",
|
|
"user": "user",
|
|
"key": None,
|
|
"gitea_url": "https://registry.example.com",
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.disable("r1", token="tok")
|
|
assert manager._captured_extra_vars["gitea_url"] == "https://registry.example.com"
|
|
|
|
def test_disable_missing_token(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user"}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
with pytest.raises(AnsibleError, match="GITEA_REGISTRATION_TOKEN must be set"):
|
|
manager.disable("r1", gitea_url="https://git.example.com")
|
|
|
|
def test_disable_missing_gitea_url(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user"}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
with pytest.raises(AnsibleError, match="GITEA_URL must be set"):
|
|
manager.disable("r1", token="tok")
|
|
|
|
def test_status(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user", "key": None}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.status("r1")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "status-runner.yml" in cmd_str
|
|
assert manager._captured_extra_vars["runner_name"] == "r1"
|
|
assert "Checking status of Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
|
|
|
|
def test_remove(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {
|
|
"host": "host",
|
|
"user": "user",
|
|
"key": None,
|
|
"gitea_url": "",
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.remove("r1", token="tok", gitea_url="https://git.example.com")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "remove-runner.yml" in cmd_str
|
|
assert manager._captured_extra_vars["runner_name"] == "r1"
|
|
assert manager._captured_extra_vars["registration_token"] == "tok"
|
|
assert manager._captured_extra_vars["gitea_url"] == "https://git.example.com"
|
|
assert "Removing Gitea Runner r1 from host" in mock_executor.run.call_args.kwargs["description"]
|
|
mock_registry.remove.assert_called_once_with("r1")
|
|
|
|
def test_remove_missing_token(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user"}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
with pytest.raises(AnsibleError, match="GITEA_REGISTRATION_TOKEN must be set"):
|
|
manager.remove("r1", gitea_url="https://git.example.com")
|
|
|
|
def test_remove_missing_gitea_url(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user"}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
with pytest.raises(AnsibleError, match="GITEA_URL must be set"):
|
|
manager.remove("r1", token="tok")
|
|
|
|
def test_remove_registry_deleted_on_failure(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user"}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
mock_executor.run.side_effect = AnsibleError("fail")
|
|
|
|
with pytest.raises(AnsibleError):
|
|
manager.remove("r1", token="tok")
|
|
mock_registry.remove.assert_not_called()
|
|
|
|
def test_remove_force_skips_playbook(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {
|
|
"host": "host",
|
|
"user": "user",
|
|
"key": None,
|
|
"gitea_url": "",
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.remove("r1", token="tok", gitea_url="https://git.example.com", force=True)
|
|
mock_executor.run.assert_not_called()
|
|
mock_registry.remove.assert_called_once_with("r1")
|
|
|
|
def test_remove_force_without_token_or_url(self) -> None:
|
|
"""Force removal should not require token or gitea_url."""
|
|
mock_registry = MagicMock()
|
|
mock_registry.get.return_value = {"host": "host", "user": "user"}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.remove("r1", force=True)
|
|
mock_executor.run.assert_not_called()
|
|
mock_registry.remove.assert_called_once_with("r1")
|
|
|
|
def test_list_runners(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": "/key"},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
mock_executor.run_ad_hoc.return_value = "active"
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners()
|
|
assert len(runners) == 1
|
|
assert runners[0]["name"] == "r1"
|
|
assert runners[0]["host"] == "10.0.0.1"
|
|
assert runners[0]["status"] == "active"
|
|
mock_executor.run_ad_hoc.assert_called_once_with(
|
|
"10.0.0.1",
|
|
"ubuntu",
|
|
"/key",
|
|
"shell",
|
|
"sudo -u grm-r1 "
|
|
"XDG_RUNTIME_DIR=/run/user/$(id -u grm-r1) "
|
|
"systemctl --user is-active gitea-runner 2>/dev/null",
|
|
become=True,
|
|
ask_become_pass=False,
|
|
check=False,
|
|
become_pass=None,
|
|
)
|
|
|
|
def test_list_runners_exception(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
mock_executor.run_ad_hoc.side_effect = AnsibleError("ssh fail")
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners()
|
|
assert runners[0]["status"] == "unknown"
|
|
|
|
def test_list_runners_empty(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
assert manager.list_runners() == []
|
|
|
|
def test_list_runners_without_labels(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
mock_executor.run_ad_hoc.return_value = "running"
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners()
|
|
assert runners[0]["labels"] == ""
|
|
|
|
def test_list_runners_inactive(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
mock_executor.run_ad_hoc.return_value = "inactive"
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners()
|
|
assert runners[0]["status"] == "inactive"
|
|
|
|
def test_list_runners_unknown_status(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
mock_executor.run_ad_hoc.return_value = ""
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners()
|
|
assert runners[0]["status"] == "unknown"
|
|
|
|
def test_list_runners_ad_hoc_error(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
mock_executor.run_ad_hoc.side_effect = AnsibleError("ssh fail")
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners()
|
|
assert runners[0]["status"] == "unknown"
|
|
|
|
def test_list_runners_with_become_pass(self) -> None:
|
|
"""When become_pass is provided, ask_become_pass=True in ad-hoc call."""
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
mock_executor.run_ad_hoc.return_value = "active"
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners(become_pass="secret")
|
|
assert runners[0]["status"] == "active"
|
|
mock_executor.run_ad_hoc.assert_called_once_with(
|
|
"10.0.0.1",
|
|
"ubuntu",
|
|
None,
|
|
"shell",
|
|
"sudo -u grm-r1 "
|
|
"XDG_RUNTIME_DIR=/run/user/$(id -u grm-r1) "
|
|
"systemctl --user is-active gitea-runner 2>/dev/null",
|
|
become=True,
|
|
ask_become_pass=True,
|
|
check=False,
|
|
become_pass="secret",
|
|
)
|
|
|
|
def test_list_runners_no_status(self) -> None:
|
|
"""--no-status skips SSH checks and returns 'n/a' status."""
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None, "labels": "docker:docker://alpine:latest"},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners(no_status=True)
|
|
assert len(runners) == 1
|
|
assert runners[0]["status"] == "n/a"
|
|
assert runners[0]["labels"] == "docker:docker://alpine:latest"
|
|
mock_executor.run_ad_hoc.assert_not_called()
|
|
|
|
def test_list_runners_no_status_empty(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
assert manager.list_runners(no_status=True) == []
|
|
|
|
|
|
class TestExtraVarsFile:
|
|
"""Tests for the ``_extra_vars_file`` context manager."""
|
|
|
|
def test_writes_temp_file_with_content(self) -> None:
|
|
manager = RunnerManager()
|
|
with manager._extra_vars_file({"foo": "bar", "count": 3}) as path:
|
|
assert path is not None
|
|
assert path.endswith(".json")
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
assert data == {"foo": "bar", "count": 3}
|
|
mode = os.stat(path).st_mode & 0o777
|
|
assert mode == 0o600
|
|
assert not os.path.exists(path)
|
|
|
|
def test_none_yields_none(self) -> None:
|
|
manager = RunnerManager()
|
|
with manager._extra_vars_file(None) as path:
|
|
assert path is None
|
|
|
|
def test_empty_dict_yields_none(self) -> None:
|
|
manager = RunnerManager()
|
|
with manager._extra_vars_file({}) as path:
|
|
assert path is None
|
|
|
|
def test_cleans_up_even_if_file_deleted(self) -> None:
|
|
manager = RunnerManager()
|
|
with manager._extra_vars_file({"foo": "bar"}) as path:
|
|
os.unlink(path)
|
|
# Should not raise despite missing file on cleanup.
|
|
|
|
|
|
class TestBuildCmd:
|
|
def test_build_cmd_basic(self) -> None:
|
|
manager = RunnerManager()
|
|
with patch.object(Path, "exists", return_value=True):
|
|
cmd = manager._build_cmd("test.yml", "host1", "user1", "/tmp/vars.json")
|
|
cmd_str = " ".join(cmd)
|
|
assert "ansible-playbook" in cmd_str
|
|
assert "test.yml" in cmd_str
|
|
assert "host1," in cmd_str
|
|
assert "user1" in cmd_str
|
|
assert "--extra-vars" in cmd_str
|
|
assert "@/tmp/vars.json" in cmd_str
|
|
|
|
def test_build_cmd_with_key(self) -> None:
|
|
manager = RunnerManager()
|
|
with patch.object(Path, "exists", return_value=True):
|
|
cmd = manager._build_cmd("test.yml", "host1", "user1", "/tmp/vars.json", key="/key")
|
|
assert "--private-key" in cmd
|
|
assert "/key" in cmd
|
|
|
|
def test_build_cmd_ask_become_pass(self) -> None:
|
|
manager = RunnerManager()
|
|
with patch.object(Path, "exists", return_value=True):
|
|
cmd = manager._build_cmd("test.yml", "host1", "user1", "/tmp/vars.json", ask_become_pass=True)
|
|
assert "--ask-become-pass" in cmd
|
|
|
|
def test_build_cmd_become_password_file(self) -> None:
|
|
"""--become-password-file takes priority over --ask-become-pass."""
|
|
manager = RunnerManager()
|
|
with patch.object(Path, "exists", return_value=True):
|
|
cmd = manager._build_cmd(
|
|
"test.yml",
|
|
"host1",
|
|
"user1",
|
|
"/tmp/vars.json",
|
|
ask_become_pass=True,
|
|
become_password_file="/tmp/pw.txt",
|
|
)
|
|
assert "--become-password-file" in cmd
|
|
assert "/tmp/pw.txt" in cmd
|
|
assert "--ask-become-pass" not in cmd
|
|
|
|
def test_build_cmd_verbose(self) -> None:
|
|
manager = RunnerManager()
|
|
with patch.object(Path, "exists", return_value=True):
|
|
cmd = manager._build_cmd("test.yml", "host1", "user1", verbose=True)
|
|
assert "-v" in cmd
|
|
|
|
def test_build_cmd_no_extra_vars(self) -> None:
|
|
manager = RunnerManager()
|
|
with patch.object(Path, "exists", return_value=True):
|
|
cmd = manager._build_cmd("test.yml", "host1", "user1")
|
|
assert "--extra-vars" not in cmd
|
|
|
|
def test_build_cmd_playbook_not_found(self) -> None:
|
|
manager = RunnerManager()
|
|
with patch.object(Path, "exists", return_value=False):
|
|
with pytest.raises(AnsibleError, match="Playbook not found"):
|
|
manager._build_cmd("missing.yml", "host1", "user1", "/tmp/vars.json")
|