"""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 = Exception("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 = [Exception("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 = Exception("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")