"""Unit tests for cli module.""" import os from unittest.mock import MagicMock, patch import pytest from click.testing import CliRunner from grm import __version__ from grm.cli import cli _TEST_ENV = { "GITEA_URL": "https://git.example.com", "DEVELOPER_GITEA_API_TOKEN": "", "CI_GITEA_API_TOKEN": "", "CI_GITEA_TOKEN": "", } class TestCLI: @pytest.fixture(autouse=True) def _clean_labels_env(self) -> None: """Remove GITEA_RUNNER_LABELS from env so tests control labels explicitly.""" old = os.environ.pop("GITEA_RUNNER_LABELS", None) yield if old is not None: os.environ["GITEA_RUNNER_LABELS"] = old def test_cli_version(self) -> None: runner = CliRunner() result = runner.invoke(cli, ["--version"]) assert result.exit_code == 0 assert __version__ in result.output @patch("grm.cli.RunnerManager") def test_install(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env=_TEST_ENV) result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"]) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels=None, force_reregister=False, ask_become_pass=True, become_password_file=None, verbose=False, auto_recover_token=None, ) @patch("grm.cli.RunnerManager") def test_install_no_ask_become_pass(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env=_TEST_ENV) result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--no-ask-become-pass"]) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels=None, force_reregister=False, ask_become_pass=False, become_password_file=None, verbose=False, auto_recover_token=None, ) @patch("grm.cli.RunnerManager") def test_install_missing_url(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.install.side_effect = AnsibleError("GITEA_URL must be set (or pass --url)") mock_manager_class.return_value = mock_manager with patch.dict("os.environ", {"GITEA_REGISTRATION_TOKEN": "tok"}, clear=True): runner = CliRunner() result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"]) assert result.exit_code != 0 assert "GITEA_URL must be set" in result.output @patch("grm.cli.RunnerManager") def test_install_with_url_flag(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager with patch.dict("os.environ", {"GITEA_REGISTRATION_TOKEN": "tok"}, clear=True): runner = CliRunner() result = runner.invoke( cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--url", "https://git.example.com"], ) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token=None, integration_retries=3, labels=None, force_reregister=False, ask_become_pass=True, become_password_file=None, verbose=False, auto_recover_token=None, ) @patch("grm.cli.RunnerManager") def test_install_missing_token(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.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"]) assert result.exit_code != 0 assert "GITEA_REGISTRATION_TOKEN must be set" in result.output @patch("grm.cli.RunnerManager") def test_install_with_options(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env=_TEST_ENV) result = runner.invoke( cli, [ "install", "host1", "--user", "ubuntu", "--key", "/key", "--name", "r1", "--token", "tok", ], ) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key="/key", name="r1", token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels=None, force_reregister=False, ask_become_pass=True, become_password_file=None, verbose=False, auto_recover_token=None, ) @patch("grm.cli.RunnerManager") def test_install_ask_become_pass(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env=_TEST_ENV) 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( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels=None, force_reregister=False, ask_become_pass=True, become_password_file=None, verbose=False, auto_recover_token=None, ) @patch("grm.cli.RunnerManager") def test_install_error(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.install.side_effect = AnsibleError("fail") mock_manager_class.return_value = mock_manager runner = CliRunner(env={"GITEA_URL": "https://git.example.com"}) result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"]) assert result.exit_code != 0 assert "fail" in result.output @patch("grm.cli.RunnerManager") def test_install_with_labels(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env=_TEST_ENV) result = runner.invoke( cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--labels", "docker:docker://alpine:latest"] ) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels="docker:docker://alpine:latest", force_reregister=False, ask_become_pass=True, become_password_file=None, verbose=False, auto_recover_token=None, ) @patch("grm.cli.RunnerManager") def test_install_force_reregister(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env=_TEST_ENV) result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--force-reregister"]) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels=None, force_reregister=True, ask_become_pass=True, become_password_file=None, verbose=False, auto_recover_token=None, ) @patch("grm.cli.RunnerManager") def test_install_with_auto_recover_token(self, mock_manager_class: MagicMock) -> None: """--auto-recover-token passes the token to the manager for healthcheck auto-recovery.""" mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env=_TEST_ENV) result = runner.invoke( cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--auto-recover-token", "api-tok"], ) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels=None, force_reregister=False, ask_become_pass=True, become_password_file=None, verbose=False, auto_recover_token="api-tok", ) @patch("grm.cli.RunnerManager") def test_install_auto_recover_token_from_env(self, mock_manager_class: MagicMock) -> None: """GITEA_AUTO_RECOVER_TOKEN env var is used when --auto-recover-token is not passed.""" mock_manager = MagicMock() mock_manager_class.return_value = mock_manager env = {**_TEST_ENV, "GITEA_AUTO_RECOVER_TOKEN": "env-tok"} runner = CliRunner(env=env) result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"]) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels=None, force_reregister=False, ask_become_pass=True, become_password_file=None, verbose=False, auto_recover_token="env-tok", ) @patch("grm.cli.RunnerManager") def test_install_with_empty_labels(self, mock_manager_class: MagicMock) -> None: """Explicit empty string labels means 'no labels' (not 'use default').""" mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env=_TEST_ENV) result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--labels", ""]) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels="", force_reregister=False, ask_become_pass=True, become_password_file=None, verbose=False, auto_recover_token=None, ) @patch("grm.cli.RunnerManager") def test_install_labels_from_env(self, mock_manager_class: MagicMock) -> None: """Labels read from GITEA_RUNNER_LABELS env var when --labels not passed.""" mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env={**_TEST_ENV, "GITEA_RUNNER_LABELS": "docker:docker://alpine:latest"}) result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"]) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels="docker:docker://alpine:latest", force_reregister=False, ask_become_pass=True, become_password_file=None, verbose=False, auto_recover_token=None, ) @patch("grm.cli.RunnerManager") def test_install_with_become_password_file(self, mock_manager_class: MagicMock) -> None: """--become-password-file passes file path to manager.""" import tempfile mock_manager = MagicMock() mock_manager_class.return_value = mock_manager with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: f.write("secret\n") pw_file = f.name try: runner = CliRunner(env=_TEST_ENV) result = runner.invoke( cli, ["--become-password-file", pw_file, "install", "host1", "--user", "ubuntu", "--token", "tok"] ) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels=None, force_reregister=False, ask_become_pass=True, become_password_file=pw_file, verbose=False, auto_recover_token=None, ) finally: import os os.unlink(pw_file) @patch("grm.cli.RunnerManager") def test_install_verbose(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env=_TEST_ENV) result = runner.invoke(cli, ["-v", "install", "host1", "--user", "ubuntu", "--token", "tok"]) assert result.exit_code == 0 mock_manager.install.assert_called_once_with( host="host1", user="ubuntu", key=None, name=None, token="tok", gitea_url="https://git.example.com", admin_token="", integration_retries=3, labels=None, force_reregister=False, ask_become_pass=True, become_password_file=None, verbose=True, auto_recover_token=None, ) @patch("grm.cli.RunnerManager") def test_update(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke( cli, [ "update", "host1", "--user", "ubuntu", "--key", "/key", "--version", "v0.2.0", ], ) assert result.exit_code == 0 mock_manager.update.assert_called_once_with( host="host1", user="ubuntu", key="/key", version="v0.2.0", ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_update_ask_become_pass(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["update", "host1", "--user", "ubuntu", "--ask-become-pass"]) assert result.exit_code == 0 mock_manager.update.assert_called_once_with( host="host1", user="ubuntu", key=None, version=None, ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_update_error(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.update.side_effect = AnsibleError("fail") mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["update", "host1", "--user", "ubuntu"]) assert result.exit_code != 0 assert "fail" in result.output @patch("grm.cli.RunnerManager") def test_start(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["start", "r1"]) assert result.exit_code == 0 mock_manager.start.assert_called_once_with( name="r1", host=None, user=None, key=None, ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_start_with_override(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["start", "r1", "--host", "newhost", "--user", "newuser"]) assert result.exit_code == 0 mock_manager.start.assert_called_once_with( name="r1", host="newhost", user="newuser", key=None, ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_stop(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["stop", "r1"]) assert result.exit_code == 0 mock_manager.stop.assert_called_once_with( name="r1", host=None, user=None, key=None, ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_restart(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["restart", "r1"]) assert result.exit_code == 0 mock_manager.restart.assert_called_once_with( name="r1", host=None, user=None, key=None, ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_enable(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["enable", "r1"]) assert result.exit_code == 0 mock_manager.enable.assert_called_once_with( name="r1", host=None, user=None, key=None, ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_disable(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env={"GITEA_URL": "https://git.example.com"}) result = runner.invoke(cli, ["disable", "r1", "--token", "tok"]) assert result.exit_code == 0 mock_manager.disable.assert_called_once_with( name="r1", host=None, user=None, key=None, token="tok", gitea_url="https://git.example.com", ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_disable_missing_url(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.disable.side_effect = AnsibleError("GITEA_URL must be set (or pass --url)") mock_manager_class.return_value = mock_manager with patch.dict("os.environ", {"GITEA_REGISTRATION_TOKEN": "tok"}, clear=True): runner = CliRunner() result = runner.invoke(cli, ["disable", "r1", "--token", "tok"]) assert result.exit_code != 0 assert "GITEA_URL must be set" in result.output @patch("grm.cli.RunnerManager") def test_disable_with_url_flag(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager with patch.dict("os.environ", {"GITEA_REGISTRATION_TOKEN": "tok"}, clear=True): runner = CliRunner() result = runner.invoke(cli, ["disable", "r1", "--token", "tok", "--url", "https://git.example.com"]) assert result.exit_code == 0 mock_manager.disable.assert_called_once_with( name="r1", host=None, user=None, key=None, token="tok", gitea_url="https://git.example.com", ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_disable_error(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.disable.side_effect = AnsibleError("fail") mock_manager_class.return_value = mock_manager runner = CliRunner(env={"GITEA_URL": "https://git.example.com"}) result = runner.invoke(cli, ["disable", "r1", "--token", "tok"]) assert result.exit_code != 0 assert "fail" in result.output @patch("grm.cli.RunnerManager") def test_status(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["status", "r1"]) assert result.exit_code == 0 mock_manager.status.assert_called_once_with( name="r1", host=None, user=None, key=None, ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_remove(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env={"GITEA_URL": "https://git.example.com"}) result = runner.invoke(cli, ["remove", "r1", "--token", "tok"]) assert result.exit_code == 0 mock_manager.remove.assert_called_once_with( name="r1", host=None, user=None, key=None, token="tok", gitea_url="https://git.example.com", force=False, ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_remove_force_flag(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager runner = CliRunner(env={"GITEA_URL": "https://git.example.com"}) result = runner.invoke(cli, ["remove", "r1", "--token", "tok", "--force"]) assert result.exit_code == 0 mock_manager.remove.assert_called_once_with( name="r1", host=None, user=None, key=None, token="tok", gitea_url="https://git.example.com", force=True, ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_start_error(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.start.side_effect = AnsibleError("fail") mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["start", "r1"]) assert result.exit_code != 0 assert "fail" in result.output @patch("grm.cli.RunnerManager") def test_stop_error(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.stop.side_effect = AnsibleError("fail") mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["stop", "r1"]) assert result.exit_code != 0 assert "fail" in result.output @patch("grm.cli.RunnerManager") def test_enable_error(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.enable.side_effect = AnsibleError("fail") mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["enable", "r1"]) assert result.exit_code != 0 assert "fail" in result.output @patch("grm.cli.RunnerManager") def test_status_error(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.status.side_effect = AnsibleError("fail") mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["status", "r1"]) assert result.exit_code != 0 assert "fail" in result.output @patch("grm.cli.RunnerManager") def test_remove_missing_url(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.remove.side_effect = AnsibleError("GITEA_URL must be set (or pass --url)") mock_manager_class.return_value = mock_manager with patch.dict("os.environ", {"GITEA_REGISTRATION_TOKEN": "tok"}, clear=True): runner = CliRunner() result = runner.invoke(cli, ["remove", "r1", "--token", "tok"]) assert result.exit_code != 0 assert "GITEA_URL must be set" in result.output @patch("grm.cli.RunnerManager") def test_remove_with_url_flag(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager_class.return_value = mock_manager with patch.dict("os.environ", {"GITEA_REGISTRATION_TOKEN": "tok"}, clear=True): runner = CliRunner() result = runner.invoke(cli, ["remove", "r1", "--token", "tok", "--url", "https://git.example.com"]) assert result.exit_code == 0 mock_manager.remove.assert_called_once_with( name="r1", host=None, user=None, key=None, token="tok", gitea_url="https://git.example.com", force=False, ask_become_pass=True, become_password_file=None, verbose=False, ) @patch("grm.cli.RunnerManager") def test_remove_error(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.remove.side_effect = AnsibleError("fail") mock_manager_class.return_value = mock_manager runner = CliRunner(env={"GITEA_URL": "https://git.example.com"}) result = runner.invoke(cli, ["remove", "r1", "--token", "tok"]) assert result.exit_code != 0 assert "fail" in result.output @patch("grm.cli.RunnerManager") def test_list(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager.list_runners.return_value = [ { "name": "r1", "host": "10.0.0.1", "user": "ubuntu", "labels": "docker:docker://alpine:latest", "status": "active", }, ] mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["list"]) assert result.exit_code == 0 assert "r1" in result.output assert "10.0.0.1" in result.output assert "active" in result.output mock_manager.list_runners.assert_called_once_with(become_pass=None, no_status=False) @patch("grm.cli.RunnerManager") def test_list_with_none_labels(self, mock_manager_class: MagicMock) -> None: """Runners with labels=None should not crash the list command.""" mock_manager = MagicMock() mock_manager.list_runners.return_value = [ { "name": "r1", "host": "10.0.0.1", "user": "ubuntu", "labels": None, "status": "active", }, ] mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["list"]) assert result.exit_code == 0 assert "r1" in result.output assert "active" in result.output @patch("grm.cli.RunnerManager") def test_list_no_status(self, mock_manager_class: MagicMock) -> None: """--no-status skips SSH checks and shows registry only.""" mock_manager = MagicMock() mock_manager.list_runners.return_value = [ { "name": "r1", "host": "10.0.0.1", "user": "ubuntu", "labels": "", "status": "n/a", }, ] mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["list", "--no-status"]) assert result.exit_code == 0 assert "r1" in result.output assert "n/a" in result.output mock_manager.list_runners.assert_called_once_with(become_pass=None, no_status=True) @patch("grm.cli.click.prompt", return_value="secret") @patch("grm.cli.sys.stdin") def test_collect_become_pass_tty(self, mock_stdin: MagicMock, mock_prompt: MagicMock) -> None: from grm.cli import _collect_become_pass mock_stdin.isatty.return_value = True assert _collect_become_pass(ask_become_pass=True) == "secret" @patch("grm.cli.sys.stdin") def test_collect_become_pass_no_ask(self, mock_stdin: MagicMock) -> None: from grm.cli import _collect_become_pass mock_stdin.isatty.return_value = True assert _collect_become_pass(ask_become_pass=False) is None @patch("grm.cli.RunnerManager") def test_list_with_piped_become_pass(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager.list_runners.return_value = [] mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["list"], input="secret\n") assert result.exit_code == 0 mock_manager.list_runners.assert_called_once_with(become_pass="secret", no_status=False) @patch("grm.cli.RunnerManager") def test_list_with_become_password_file(self, mock_manager_class: MagicMock) -> None: """--become-password-file reads password from file for grm list.""" import os import tempfile mock_manager = MagicMock() mock_manager.list_runners.return_value = [] mock_manager_class.return_value = mock_manager with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: f.write("secret\n") pw_file = f.name try: runner = CliRunner() result = runner.invoke(cli, ["--become-password-file", pw_file, "list"]) assert result.exit_code == 0 mock_manager.list_runners.assert_called_once_with(become_pass="secret", no_status=False) finally: os.unlink(pw_file) @patch("grm.cli.RunnerManager") def test_list_with_become_password_file_env(self, mock_manager_class: MagicMock) -> None: """GRM_BECOME_PASSWORD_FILE env var works for grm list.""" import os import tempfile mock_manager = MagicMock() mock_manager.list_runners.return_value = [] mock_manager_class.return_value = mock_manager with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: f.write("envpass\n") pw_file = f.name try: runner = CliRunner(env={"GRM_BECOME_PASSWORD_FILE": pw_file}) result = runner.invoke(cli, ["list"]) assert result.exit_code == 0 mock_manager.list_runners.assert_called_once_with(become_pass="envpass", no_status=False) finally: os.unlink(pw_file) @patch("grm.cli.RunnerManager") def test_list_empty(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager.list_runners.return_value = [] mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["list"]) assert result.exit_code == 0 assert "No runners registered" in result.output @patch("grm.cli.RunnerManager") def test_list_error(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.list_runners.side_effect = AnsibleError("fail") mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["list"]) assert result.exit_code != 0 assert "fail" in result.output @patch("grm.cli.RunnerManager") def test_health_all_healthy(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager.health.return_value = [ {"name": "r1", "host": "10.0.0.1", "healthy": "yes", "message": "runner healthy, disk at 42%"}, {"name": "r2", "host": "10.0.0.2", "healthy": "yes", "message": "runner healthy, disk at 50%"}, ] mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["health"], input="secret\n") assert result.exit_code == 0 assert "r1" in result.output assert "r2" in result.output assert "yes" in result.output @patch("grm.cli.RunnerManager") def test_health_with_unhealthy(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager.health.return_value = [ {"name": "r1", "host": "10.0.0.1", "healthy": "yes", "message": "runner healthy, disk at 42%"}, {"name": "r2", "host": "10.0.0.2", "healthy": "no", "message": "Docker daemon down"}, ] mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["health"], input="secret\n") assert result.exit_code != 0 assert "unhealthy" in result.output.lower() assert "r2" in result.output @patch("grm.cli.RunnerManager") def test_health_single_runner(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager.health.return_value = [ {"name": "r1", "host": "10.0.0.1", "healthy": "yes", "message": "runner healthy, disk at 42%"}, ] mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["health", "r1"], input="secret\n") assert result.exit_code == 0 assert "r1" in result.output mock_manager.health.assert_called_once() @patch("grm.cli.RunnerManager") def test_health_empty(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() mock_manager.health.return_value = [] mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["health"], input="secret\n") assert result.exit_code == 0 assert "No runners registered" in result.output @patch("grm.cli.RunnerManager") def test_health_error(self, mock_manager_class: MagicMock) -> None: mock_manager = MagicMock() from grm.exceptions import AnsibleError mock_manager.health.side_effect = AnsibleError("fail") mock_manager_class.return_value = mock_manager runner = CliRunner() result = runner.invoke(cli, ["health"], input="secret\n") assert result.exit_code != 0 assert "fail" in result.output @patch("grm.cli.os.getlogin", side_effect=OSError("no tty")) @patch("grm.cli.RunnerManager") def test_default_user_fallback_on_getlogin_error( self, mock_manager_class: MagicMock, mock_getlogin: MagicMock ) -> None: """os.getlogin() failure falls back to USER env var.""" mock_manager = MagicMock() mock_manager_class.return_value = mock_manager with patch.dict("os.environ", {"USER": "testuser", "GITEA_URL": "https://git.example.com"}, clear=True): runner = CliRunner() result = runner.invoke(cli, ["install", "host1", "--token", "tok"]) assert result.exit_code == 0 call_kwargs = mock_manager.install.call_args.kwargs assert call_kwargs["user"] == "testuser" @patch("grm.cli.os.getlogin", side_effect=OSError("no tty")) @patch("grm.cli.RunnerManager") def test_default_user_fallback_to_root(self, mock_manager_class: MagicMock, mock_getlogin: MagicMock) -> None: """os.getlogin() failure with no USER env falls back to 'root'.""" mock_manager = MagicMock() 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", "--token", "tok"]) assert result.exit_code == 0 call_kwargs = mock_manager.install.call_args.kwargs assert call_kwargs["user"] == "root" @patch("grm.cli.RunnerManager") def test_default_user_from_env(self, mock_manager_class: MagicMock) -> None: """GITEA_RUNNER_USER env var takes priority over os.getlogin().""" mock_manager = MagicMock() mock_manager_class.return_value = mock_manager with patch.dict( "os.environ", {"GITEA_RUNNER_USER": "ciuser", "GITEA_URL": "https://git.example.com"}, clear=True, ): runner = CliRunner() result = runner.invoke(cli, ["install", "host1", "--token", "tok"]) assert result.exit_code == 0 call_kwargs = mock_manager.install.call_args.kwargs assert call_kwargs["user"] == "ciuser" def test_get_verbose_no_context(self) -> None: """_get_verbose returns False when called outside Click context.""" from grm.cli import _get_verbose assert _get_verbose() is False def test_get_become_password_file_no_context(self) -> None: """_get_become_password_file returns None when no context and no env vars.""" from grm.cli import _get_become_password_file with patch.dict("os.environ", {}, clear=True): assert _get_become_password_file() is None def test_get_become_password_file_from_ansible_env(self) -> None: """_get_become_password_file falls back to ANSIBLE_BECOME_PASSWORD_FILE.""" from grm.cli import _get_become_password_file with patch.dict("os.environ", {"ANSIBLE_BECOME_PASSWORD_FILE": "/tmp/ansible.txt"}, clear=True): assert _get_become_password_file() == "/tmp/ansible.txt" class TestTriggerWorkflow: """Tests for the trigger-workflow CLI command.""" def test_trigger_workflow_success(self) -> None: runner = CliRunner(env=_TEST_ENV) with patch("grm.cli.GiteaWorkflowClient") as mock_client_cls: mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.dispatch_workflow.return_value = { "id": 42, "html_url": "https://git.example.com/oblachno-oss/grm/actions/runs/42", } result = runner.invoke(cli, ["trigger-workflow", "ci.yml", "--token", "tok"]) assert result.exit_code == 0 assert "42" in result.output mock_client.dispatch_workflow.assert_called_once_with("oblachno-oss", "grm", "ci.yml", "master") def test_trigger_workflow_no_url(self) -> None: runner = CliRunner() with patch.dict("os.environ", {}, clear=True): result = runner.invoke(cli, ["trigger-workflow", "ci.yml", "--token", "tok"]) assert result.exit_code != 0 assert "GITEA_URL" in result.output def test_trigger_workflow_no_token(self) -> None: runner = CliRunner() with patch.dict("os.environ", {"GITEA_URL": "https://git.example.com"}, clear=True): result = runner.invoke(cli, ["trigger-workflow", "ci.yml"]) assert result.exit_code != 0 assert "CI_GITEA_TOKEN" in result.output def test_trigger_workflow_list(self) -> None: runner = CliRunner(env=_TEST_ENV) with patch("grm.cli.GiteaWorkflowClient") as mock_client_cls: mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.list_workflows.return_value = [ {"id": 1, "name": "CI", "path": "ci.yml", "state": "active"}, {"id": 2, "name": "Post-merge", "path": "post-merge.yml", "state": "active"}, ] result = runner.invoke(cli, ["trigger-workflow", "--list", "--token", "tok"]) assert result.exit_code == 0 assert "CI" in result.output assert "Post-merge" in result.output mock_client.list_workflows.assert_called_once_with("oblachno-oss", "grm") def test_trigger_workflow_list_empty(self) -> None: runner = CliRunner(env=_TEST_ENV) with patch("grm.cli.GiteaWorkflowClient") as mock_client_cls: mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.list_workflows.return_value = [] result = runner.invoke(cli, ["trigger-workflow", "--list", "--token", "tok"]) assert result.exit_code == 0 assert "No workflows" in result.output def test_trigger_workflow_no_workflow_id(self) -> None: runner = CliRunner(env=_TEST_ENV) result = runner.invoke(cli, ["trigger-workflow", "--token", "tok"]) assert result.exit_code != 0 assert "WORKFLOW_ID" in result.output def test_trigger_workflow_api_error(self) -> None: from grm.gitea_client import GiteaAPIError runner = CliRunner(env=_TEST_ENV) with patch("grm.cli.GiteaWorkflowClient") as mock_client_cls: mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.dispatch_workflow.side_effect = GiteaAPIError(404, "workflow not found") result = runner.invoke(cli, ["trigger-workflow", "nonexistent.yml", "--token", "tok"]) assert result.exit_code != 0 assert "404" in result.output def test_trigger_workflow_custom_repo_and_ref(self) -> None: runner = CliRunner(env=_TEST_ENV) with patch("grm.cli.GiteaWorkflowClient") as mock_client_cls: mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.dispatch_workflow.return_value = None result = runner.invoke( cli, ["trigger-workflow", "build.yml", "--repo", "myorg/myrepo", "--ref", "develop", "--token", "tok"], ) assert result.exit_code == 0 mock_client.dispatch_workflow.assert_called_once_with("myorg", "myrepo", "build.yml", "develop")