633 lines
26 KiB
Python
633 lines
26 KiB
Python
"""Unit tests for runner_manager module."""
|
|
|
|
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
|
|
|
|
|
|
class TestRunnerManager:
|
|
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 "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
|
|
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,
|
|
mode="docker",
|
|
gitea_url="https://git.example.com",
|
|
)
|
|
|
|
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 "registration_token=preset" in cmd_str
|
|
assert "runner_name=my-runner" in cmd_str
|
|
assert "runner_mode=docker" in cmd_str
|
|
assert "--ask-become-pass" not in cmd_str
|
|
mock_registry.add.assert_called_once_with(
|
|
name="my-runner",
|
|
host="host1",
|
|
user="root",
|
|
key="/key",
|
|
mode="docker",
|
|
gitea_url="https://git.example.com",
|
|
)
|
|
|
|
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
|
|
assert "runner_mode=docker" in cmd_str
|
|
|
|
def test_install_binary_mode(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", mode="binary")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "runner_mode=binary" in cmd_str
|
|
mock_registry.add.assert_called_once_with(
|
|
name="host1", host="host1", user="root", key=None, mode="binary", gitea_url="https://git.example.com"
|
|
)
|
|
|
|
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_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")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "gitea_admin_token=admin-tok" in cmd_str
|
|
|
|
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 "gitea_runner_version=v0.2.0" in cmd_str
|
|
assert "runner_mode=docker" in cmd_str
|
|
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 "runner_mode=docker" in cmd_str
|
|
|
|
def test_update_binary_mode(self) -> None:
|
|
manager = RunnerManager()
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.update("host", "user", mode="binary")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "runner_mode=binary" 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",
|
|
"mode": "docker",
|
|
"gitea_url": "https://git.example.com",
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
host, user, key, mode, gitea_url = manager._resolve_runner("r1")
|
|
assert host == "10.0.0.1"
|
|
assert user == "ubuntu"
|
|
assert key == "/key"
|
|
assert mode == "docker"
|
|
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, mode, 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 mode == "docker"
|
|
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",
|
|
"mode": "docker",
|
|
"gitea_url": "https://git.example.com",
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
host, user, key, mode, gitea_url = manager._resolve_runner(
|
|
"r1", host="10.0.0.2", user="root", key="/new", mode="binary"
|
|
)
|
|
assert host == "10.0.0.2"
|
|
assert user == "root"
|
|
assert key == "/new"
|
|
assert mode == "binary"
|
|
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, "mode": "docker"}
|
|
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 "runner_name=r1" in cmd_str
|
|
assert "runner_mode=docker" in cmd_str
|
|
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, "mode": "docker"}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.start("r1", host="newhost", user="newuser", mode="binary")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "newhost," in cmd_str
|
|
assert "newuser" in cmd_str
|
|
assert "runner_mode=binary" 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", mode="docker")
|
|
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, "mode": "docker"}
|
|
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 "runner_name=r1" in cmd_str
|
|
assert "Stopping 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, "mode": "docker"}
|
|
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 "runner_name=r1" in cmd_str
|
|
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,
|
|
"mode": "docker",
|
|
"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 "runner_name=r1" in cmd_str
|
|
assert "registration_token=tok" in cmd_str
|
|
assert "gitea_url=https://git.example.com" in cmd_str
|
|
assert "runner_mode=docker" in cmd_str
|
|
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,
|
|
"mode": "docker",
|
|
"gitea_url": "https://registry.example.com",
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
manager._executor = mock_executor
|
|
|
|
manager.disable("r1", token="tok")
|
|
cmd = mock_executor.run.call_args.args[0]
|
|
cmd_str = " ".join(cmd)
|
|
assert "gitea_url=https://registry.example.com" in cmd_str
|
|
|
|
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, "mode": "binary"}
|
|
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 "runner_name=r1" in cmd_str
|
|
assert "runner_mode=binary" in cmd_str
|
|
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,
|
|
"mode": "docker",
|
|
"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 "runner_name=r1" in cmd_str
|
|
assert "registration_token=tok" in cmd_str
|
|
assert "gitea_url=https://git.example.com" in cmd_str
|
|
assert "runner_mode=docker" in cmd_str
|
|
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,
|
|
"mode": "docker",
|
|
"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_docker(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": "/key", "mode": "docker"},
|
|
}
|
|
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 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",
|
|
(
|
|
"docker inspect gitea-runner-r1 2>/dev/null | "
|
|
"python3 -c \"import sys,json; print(json.load(sys.stdin)[0]['State']['Status'])\" "
|
|
"2>/dev/null || echo unknown"
|
|
),
|
|
become=True,
|
|
ask_become_pass=True,
|
|
check=False,
|
|
)
|
|
|
|
def test_list_runners_binary(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": "/key", "mode": "binary"},
|
|
}
|
|
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]["status"] == "active"
|
|
mock_executor.run_ad_hoc.assert_called_once_with(
|
|
"10.0.0.1",
|
|
"ubuntu",
|
|
"/key",
|
|
"shell",
|
|
"systemctl is-active gitea-runner@r1 2>/dev/null",
|
|
become=True,
|
|
ask_become_pass=True,
|
|
check=False,
|
|
)
|
|
|
|
def test_list_runners_binary_exception(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None, "mode": "binary"},
|
|
}
|
|
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_docker_exited(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None, "mode": "docker"},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
mock_executor.run_ad_hoc.return_value = "exited"
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners()
|
|
assert runners[0]["status"] == "inactive"
|
|
|
|
def test_list_runners_docker_fallback_to_host_container(self) -> None:
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None, "mode": "docker"},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
# First call (name-based) returns unknown, second (host-based) returns running
|
|
mock_executor.run_ad_hoc.side_effect = ["unknown", "running"]
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners()
|
|
assert runners[0]["status"] == "active"
|
|
assert mock_executor.run_ad_hoc.call_count == 2
|
|
|
|
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, "mode": "docker"},
|
|
}
|
|
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_docker_first_command_falls_back(self) -> None:
|
|
"""When the first docker inspect raises, the loop continues to the next fallback."""
|
|
mock_registry = MagicMock()
|
|
mock_registry.list.return_value = {
|
|
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None, "mode": "docker"},
|
|
}
|
|
manager = RunnerManager(registry=mock_registry)
|
|
mock_executor = MagicMock()
|
|
mock_executor.run_ad_hoc.side_effect = [AnsibleError("ssh fail"), "running"]
|
|
manager._executor = mock_executor
|
|
|
|
runners = manager.list_runners()
|
|
assert runners[0]["status"] == "active"
|
|
assert mock_executor.run_ad_hoc.call_count == 2
|
|
|
|
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, "mode": "docker"},
|
|
}
|
|
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"
|
|
|
|
|
|
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", "foo=bar")
|
|
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 "foo=bar" 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", "foo=bar", 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", "foo=bar", ask_become_pass=True)
|
|
assert "--ask-become-pass" 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", "foo=bar")
|