"""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 TestHealth: """Tests for the ``health`` method.""" def test_health_single_runner_healthy(self) -> None: mock_registry = MagicMock() mock_registry.get.return_value = {"host": "10.0.0.1", "user": "ubuntu", "key": "/key"} manager = RunnerManager(registry=mock_registry) mock_executor = MagicMock() mock_executor.run_ad_hoc.return_value = "OK: runner healthy, disk at 42%" manager._executor = mock_executor results = manager.health(name="r1") assert len(results) == 1 assert results[0]["name"] == "r1" assert results[0]["healthy"] == "yes" assert "runner healthy" in results[0]["message"] mock_executor.run_ad_hoc.assert_called_once() def test_health_single_runner_unhealthy(self) -> None: mock_registry = MagicMock() mock_registry.get.return_value = {"host": "10.0.0.1", "user": "ubuntu", "key": None} manager = RunnerManager(registry=mock_registry) mock_executor = MagicMock() mock_executor.run_ad_hoc.return_value = "CRITICAL: Docker daemon still down after restart" manager._executor = mock_executor results = manager.health(name="r1") assert len(results) == 1 assert results[0]["healthy"] == "no" assert "Docker daemon still down" in results[0]["message"] def test_health_all_runners(self) -> None: mock_registry = MagicMock() mock_registry.list.return_value = { "r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None}, "r2": {"host": "10.0.0.2", "user": "ubuntu", "key": None}, } manager = RunnerManager(registry=mock_registry) mock_executor = MagicMock() mock_executor.run_ad_hoc.side_effect = [ "OK: runner healthy, disk at 42%", "ERROR: gitea-runner service is inactive, restarting", ] manager._executor = mock_executor results = manager.health() assert len(results) == 2 assert results[0]["healthy"] == "yes" assert results[1]["healthy"] == "no" def test_health_empty_registry(self) -> None: mock_registry = MagicMock() mock_registry.list.return_value = {} manager = RunnerManager(registry=mock_registry) results = manager.health() assert results == [] def test_health_ansible_error(self) -> None: mock_registry = MagicMock() mock_registry.get.return_value = {"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 unreachable") manager._executor = mock_executor results = manager.health(name="r1") assert len(results) == 1 assert results[0]["healthy"] == "no" assert "ssh unreachable" in results[0]["message"] def test_health_with_host_override(self) -> None: mock_registry = MagicMock() mock_registry.get.return_value = {"host": "10.0.0.1", "user": "ubuntu", "key": None} manager = RunnerManager(registry=mock_registry) mock_executor = MagicMock() mock_executor.run_ad_hoc.return_value = "OK: runner healthy, disk at 50%" manager._executor = mock_executor results = manager.health(name="r1", host="10.0.0.99", user="root") assert len(results) == 1 assert results[0]["host"] == "10.0.0.99" call_args = mock_executor.run_ad_hoc.call_args.args assert call_args[0] == "10.0.0.99" assert call_args[1] == "root" def test_health_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.health(name="nonexistent") def test_health_passes_become_pass(self) -> None: """become_pass is forwarded to run_ad_hoc for sudo authentication.""" mock_registry = MagicMock() mock_registry.get.return_value = {"host": "10.0.0.1", "user": "ubuntu", "key": None} manager = RunnerManager(registry=mock_registry) mock_executor = MagicMock() mock_executor.run_ad_hoc.return_value = "OK: runner healthy, disk at 42%" manager._executor = mock_executor manager.health(name="r1", become_pass="s3cr3t") call_kwargs = mock_executor.run_ad_hoc.call_args.kwargs assert call_kwargs["become_pass"] == "s3cr3t" 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")