GRM-7: feat: integrate AnsibleExecutor and i18n into CLI and RunnerManager
This commit is contained in:
Emil Simeonov
@@ -8,55 +8,55 @@ import click
|
||||
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
|
||||
|
||||
from .exceptions import GRMError
|
||||
from .i18n import _
|
||||
from .runner_manager import RunnerManager
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
|
||||
@click.group()
|
||||
@click.group(help=_("Gitea Runner Manager — manage Gitea Actions runners."))
|
||||
@click.version_option(version="0.1.0")
|
||||
def cli() -> None:
|
||||
"""Gitea Runner Manager — manage Gitea Actions runners."""
|
||||
pass
|
||||
|
||||
|
||||
@cli.command()
|
||||
@cli.command(help=_("Install and configure a Gitea Runner on a remote host."))
|
||||
@click.argument("host")
|
||||
@click.option(
|
||||
"--user",
|
||||
"-u",
|
||||
default=lambda: os.getenv("GITEA_RUNNER_USER", os.getlogin()),
|
||||
help="SSH user",
|
||||
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("--key", "-k", default=lambda: os.getenv("GITEA_RUNNER_KEY"), help=_("Path to SSH private key"))
|
||||
@click.option("--name", "-n", help=_("Gitea Runner name (default: host)"))
|
||||
@click.option(
|
||||
"--token",
|
||||
"-t",
|
||||
default=lambda: os.getenv("GITEA_REGISTRATION_TOKEN"),
|
||||
help="Registration token (env: 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)",
|
||||
help=_("Gitea Runner deployment mode (default: docker)"),
|
||||
)
|
||||
@click.option(
|
||||
"--admin-token",
|
||||
"-a",
|
||||
default=lambda: os.getenv("GITEA_ADMIN_TOKEN"),
|
||||
help="Gitea admin API token for integration test (env: GITEA_ADMIN_TOKEN)",
|
||||
help=_("Gitea admin API token for integration test (env: GITEA_ADMIN_TOKEN)"),
|
||||
)
|
||||
@click.option(
|
||||
"--integration-retries",
|
||||
"-r",
|
||||
type=int,
|
||||
default=lambda: int(os.getenv("GITEA_INTEGRATION_RETRIES", "3")),
|
||||
help="Integration test API retries (default: 3, env: GITEA_INTEGRATION_RETRIES)",
|
||||
help=_("Integration test API retries (default: 3, env: GITEA_INTEGRATION_RETRIES)"),
|
||||
)
|
||||
@click.option("--ask-become-pass", is_flag=True, help="Prompt for sudo password")
|
||||
@click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password"))
|
||||
def install(
|
||||
host: str,
|
||||
user: str,
|
||||
@@ -68,12 +68,9 @@ def install(
|
||||
integration_retries: int,
|
||||
ask_become_pass: bool,
|
||||
) -> None:
|
||||
"""Install and configure a runner on a remote host."""
|
||||
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)")
|
||||
raise click.ClickException(_("GITEA_URL must be set (or pass --url)"))
|
||||
manager = RunnerManager()
|
||||
try:
|
||||
manager.install(
|
||||
@@ -89,27 +86,27 @@ def install(
|
||||
ask_become_pass=ask_become_pass,
|
||||
)
|
||||
except GRMError as e:
|
||||
raise click.ClickException(str(e)) from e
|
||||
raise click.ClickException(_("Installation failed: {error}", error=e)) from e
|
||||
|
||||
|
||||
@cli.command()
|
||||
@cli.command(help=_("Update the Gitea Runner binary on a remote host."))
|
||||
@click.argument("host")
|
||||
@click.option(
|
||||
"--user",
|
||||
"-u",
|
||||
default=lambda: os.getenv("GITEA_RUNNER_USER", os.getlogin()),
|
||||
help="SSH user",
|
||||
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 gitea_runner version")
|
||||
@click.option("--key", "-k", default=lambda: os.getenv("GITEA_RUNNER_KEY"), help=_("Path to SSH private key"))
|
||||
@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)",
|
||||
help=_("Gitea Runner deployment mode (default: docker)"),
|
||||
)
|
||||
@click.option("--ask-become-pass", is_flag=True, help="Prompt for sudo password")
|
||||
@click.option("--ask-become-pass", is_flag=True, help=_("Prompt for sudo password"))
|
||||
def update(
|
||||
host: str,
|
||||
user: str,
|
||||
@@ -118,7 +115,6 @@ def update(
|
||||
mode: str,
|
||||
ask_become_pass: bool,
|
||||
) -> None:
|
||||
"""Update the gitea_runner binary on a remote host."""
|
||||
manager = RunnerManager()
|
||||
try:
|
||||
manager.update(
|
||||
@@ -130,4 +126,4 @@ def update(
|
||||
ask_become_pass=ask_become_pass,
|
||||
)
|
||||
except GRMError as e:
|
||||
raise click.ClickException(str(e)) from e
|
||||
raise click.ClickException(_("Update failed: {error}", error=e)) from e
|
||||
|
||||
@@ -2,16 +2,19 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from .exceptions import AnsibleError
|
||||
from .executor import AnsibleExecutor
|
||||
from .i18n import _
|
||||
|
||||
|
||||
class RunnerManager:
|
||||
"""Orchestrates runner installation and updates."""
|
||||
|
||||
def __init__(self, executor: AnsibleExecutor | None = None) -> None:
|
||||
self._executor = executor or AnsibleExecutor()
|
||||
|
||||
def install(
|
||||
self,
|
||||
host: str,
|
||||
@@ -29,11 +32,7 @@ class RunnerManager:
|
||||
if not name:
|
||||
name = host
|
||||
if not 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():
|
||||
raise AnsibleError(f"Playbook not found: {playbook}")
|
||||
raise AnsibleError(_("GITEA_REGISTRATION_TOKEN must be set (or pass --token)"))
|
||||
|
||||
extra_vars = (
|
||||
f"registration_token={token} runner_name={name} gitea_url={gitea_url}"
|
||||
@@ -43,6 +42,40 @@ class RunnerManager:
|
||||
if admin_token:
|
||||
extra_vars += f" gitea_admin_token={admin_token}"
|
||||
|
||||
cmd = self._build_cmd("install-runner.yml", host, user, extra_vars, key, ask_become_pass)
|
||||
self._executor.run(cmd, description=_("Installing Gitea Runner on {host}", host=host))
|
||||
|
||||
def update(
|
||||
self,
|
||||
host: str,
|
||||
user: str,
|
||||
key: str | None = None,
|
||||
version: str | None = None,
|
||||
mode: str = "docker",
|
||||
ask_become_pass: bool = False,
|
||||
) -> None:
|
||||
"""Update the gitea_runner binary on a remote host."""
|
||||
extra_vars = f"runner_mode={mode}"
|
||||
if version:
|
||||
extra_vars += f" gitea_runner_version={version}"
|
||||
|
||||
cmd = self._build_cmd("update-runner.yml", host, user, extra_vars, key, ask_become_pass)
|
||||
self._executor.run(cmd, description=_("Updating Gitea Runner on {host}", host=host))
|
||||
|
||||
def _build_cmd(
|
||||
self,
|
||||
playbook_name: str,
|
||||
host: str,
|
||||
user: str,
|
||||
extra_vars: str,
|
||||
key: str | None = None,
|
||||
ask_become_pass: bool = False,
|
||||
) -> list[str]:
|
||||
"""Build the ansible-playbook command."""
|
||||
playbook = Path(__file__).parent.parent.parent / "ansible" / playbook_name
|
||||
if not playbook.exists():
|
||||
raise AnsibleError(_("Playbook not found: {playbook}", playbook=playbook))
|
||||
|
||||
cmd = [
|
||||
"ansible-playbook",
|
||||
str(playbook),
|
||||
@@ -57,45 +90,4 @@ class RunnerManager:
|
||||
cmd.extend(["--private-key", key])
|
||||
if ask_become_pass:
|
||||
cmd.append("--ask-become-pass")
|
||||
|
||||
self._run_ansible(cmd)
|
||||
|
||||
def update(
|
||||
self,
|
||||
host: str,
|
||||
user: str,
|
||||
key: str | None = None,
|
||||
version: str | None = None,
|
||||
mode: str = "docker",
|
||||
ask_become_pass: bool = False,
|
||||
) -> None:
|
||||
"""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}")
|
||||
|
||||
cmd = [
|
||||
"ansible-playbook",
|
||||
str(playbook),
|
||||
"-i",
|
||||
f"{host},",
|
||||
"-u",
|
||||
user,
|
||||
"--extra-vars",
|
||||
f"runner_mode={mode}",
|
||||
]
|
||||
if key:
|
||||
cmd.extend(["--private-key", key])
|
||||
if version:
|
||||
cmd.extend(["--extra-vars", f"gitea_runner_version={version}"])
|
||||
if ask_become_pass:
|
||||
cmd.append("--ask-become-pass")
|
||||
|
||||
self._run_ansible(cmd)
|
||||
|
||||
def _run_ansible(self, cmd: list[str]) -> None:
|
||||
"""Execute an Ansible command, streaming output."""
|
||||
env = os.environ.copy()
|
||||
result = subprocess.run(cmd, env=env, check=False) # noqa: S603
|
||||
if result.returncode != 0:
|
||||
raise AnsibleError(f"Ansible failed with exit code {result.returncode}")
|
||||
return cmd
|
||||
|
||||
+18
-4
@@ -19,7 +19,7 @@ class TestCLI:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager_class.return_value = mock_manager
|
||||
|
||||
runner = CliRunner(env={"GITEA_URL": "https://git.example.com"})
|
||||
runner = CliRunner(env={"GITEA_URL": "https://git.example.com", "GITEA_ADMIN_TOKEN": ""})
|
||||
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"])
|
||||
assert result.exit_code == 0
|
||||
mock_manager.install.assert_called_once_with(
|
||||
@@ -30,6 +30,8 @@ class TestCLI:
|
||||
token="tok",
|
||||
gitea_url="https://git.example.com",
|
||||
mode="docker",
|
||||
admin_token="",
|
||||
integration_retries=3,
|
||||
ask_become_pass=False,
|
||||
)
|
||||
|
||||
@@ -43,6 +45,12 @@ class TestCLI:
|
||||
|
||||
@patch("gitea_runner_manager.cli.RunnerManager")
|
||||
def test_install_missing_token(self, mock_manager_class: MagicMock) -> None:
|
||||
mock_manager = MagicMock()
|
||||
from gitea_runner_manager.exceptions import AnsibleError
|
||||
|
||||
mock_manager.install.side_effect = AnsibleError("GITEA_REGISTRATION_TOKEN must be set (or pass --token)")
|
||||
mock_manager_class.return_value = mock_manager
|
||||
|
||||
with patch.dict("os.environ", {"GITEA_URL": "https://git.example.com"}, clear=True):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu"])
|
||||
@@ -54,7 +62,7 @@ class TestCLI:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager_class.return_value = mock_manager
|
||||
|
||||
runner = CliRunner(env={"GITEA_URL": "https://git.example.com"})
|
||||
runner = CliRunner(env={"GITEA_URL": "https://git.example.com", "GITEA_ADMIN_TOKEN": ""})
|
||||
result = runner.invoke(
|
||||
cli,
|
||||
[
|
||||
@@ -79,6 +87,8 @@ class TestCLI:
|
||||
token="tok",
|
||||
gitea_url="https://git.example.com",
|
||||
mode="docker",
|
||||
admin_token="",
|
||||
integration_retries=3,
|
||||
ask_become_pass=False,
|
||||
)
|
||||
|
||||
@@ -87,7 +97,7 @@ class TestCLI:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager_class.return_value = mock_manager
|
||||
|
||||
runner = CliRunner(env={"GITEA_URL": "https://git.example.com"})
|
||||
runner = CliRunner(env={"GITEA_URL": "https://git.example.com", "GITEA_ADMIN_TOKEN": ""})
|
||||
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(
|
||||
@@ -98,6 +108,8 @@ class TestCLI:
|
||||
token="tok",
|
||||
gitea_url="https://git.example.com",
|
||||
mode="docker",
|
||||
admin_token="",
|
||||
integration_retries=3,
|
||||
ask_become_pass=True,
|
||||
)
|
||||
|
||||
@@ -106,7 +118,7 @@ class TestCLI:
|
||||
mock_manager = MagicMock()
|
||||
mock_manager_class.return_value = mock_manager
|
||||
|
||||
runner = CliRunner(env={"GITEA_URL": "https://git.example.com"})
|
||||
runner = CliRunner(env={"GITEA_URL": "https://git.example.com", "GITEA_ADMIN_TOKEN": ""})
|
||||
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--mode", "binary"])
|
||||
assert result.exit_code == 0
|
||||
mock_manager.install.assert_called_once_with(
|
||||
@@ -117,6 +129,8 @@ class TestCLI:
|
||||
token="tok",
|
||||
gitea_url="https://git.example.com",
|
||||
mode="binary",
|
||||
admin_token="",
|
||||
integration_retries=3,
|
||||
ask_become_pass=False,
|
||||
)
|
||||
|
||||
|
||||
@@ -13,62 +13,64 @@ class TestRunnerManager:
|
||||
def test_init(self) -> None:
|
||||
manager = RunnerManager()
|
||||
assert manager is not None
|
||||
assert manager._executor is not None
|
||||
|
||||
def test_install_basic(self) -> None:
|
||||
manager = RunnerManager()
|
||||
mock_executor = MagicMock()
|
||||
manager._executor = mock_executor
|
||||
|
||||
with patch("subprocess.run") as mock_run:
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
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)
|
||||
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
|
||||
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"]
|
||||
|
||||
def test_install_with_name_and_key(self) -> None:
|
||||
manager = RunnerManager()
|
||||
mock_executor = MagicMock()
|
||||
manager._executor = mock_executor
|
||||
|
||||
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", 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
|
||||
assert "runner_mode=docker" in cmd_str
|
||||
assert "--ask-become-pass" not in cmd_str
|
||||
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
|
||||
|
||||
def test_install_ask_become_pass(self) -> None:
|
||||
manager = RunnerManager()
|
||||
mock_executor = MagicMock()
|
||||
manager._executor = mock_executor
|
||||
|
||||
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
|
||||
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:
|
||||
manager = RunnerManager()
|
||||
mock_executor = MagicMock()
|
||||
manager._executor = mock_executor
|
||||
|
||||
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
|
||||
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
|
||||
|
||||
def test_install_missing_token(self) -> None:
|
||||
manager = RunnerManager()
|
||||
@@ -81,45 +83,60 @@ class TestRunnerManager:
|
||||
with pytest.raises(AnsibleError, match="Playbook not found"):
|
||||
manager.install("host", "user", token="tok")
|
||||
|
||||
def test_install_with_admin_token(self) -> None:
|
||||
manager = RunnerManager()
|
||||
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:
|
||||
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", token="tok")
|
||||
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")
|
||||
|
||||
def test_update(self) -> None:
|
||||
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")
|
||||
cmd = mock_run.call_args[0][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
|
||||
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()
|
||||
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
|
||||
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()
|
||||
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
|
||||
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()
|
||||
@@ -129,7 +146,40 @@ class TestRunnerManager:
|
||||
|
||||
def test_update_ansible_failure(self) -> None:
|
||||
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"):
|
||||
manager.update("host", "user")
|
||||
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")
|
||||
|
||||
|
||||
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")
|
||||
|
||||
Reference in New Issue
Block a user