GRM-17: fix grm list still showing unknown status for active runners
- run_ad_hoc() now accepts ask_become_pass and check parameters - list_runners() passes ask_become_pass=True so --ask-become-pass is added when running in a TTY, matching playbook behavior - list_runners() passes check=False so systemctl is-active non-zero exit codes (inactive=3, unknown=4) don't raise exceptions; the actual status string is parsed from stdout instead - TTY guard prevents --ask-become-pass from hanging in non-interactive environments (CI, scripts) - 123 tests, 100% coverage, pyright clean, ruff clean
This commit is contained in:
@@ -6,6 +6,7 @@ import logging
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
@@ -85,6 +86,8 @@ class AnsibleExecutor:
|
||||
module: str,
|
||||
args: str,
|
||||
become: bool = False,
|
||||
ask_become_pass: bool = False,
|
||||
check: bool = True,
|
||||
) -> str:
|
||||
"""Run an Ansible ad-hoc command and return stdout."""
|
||||
cmd = [
|
||||
@@ -101,6 +104,8 @@ class AnsibleExecutor:
|
||||
cmd.extend(["--private-key", key])
|
||||
if become:
|
||||
cmd.append("--become")
|
||||
if become and ask_become_pass and sys.stdin.isatty():
|
||||
cmd.append("--ask-become-pass")
|
||||
|
||||
env = os.environ.copy()
|
||||
proc = subprocess.run(
|
||||
@@ -109,7 +114,7 @@ class AnsibleExecutor:
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
if proc.returncode != 0:
|
||||
if check and proc.returncode != 0:
|
||||
stderr = proc.stderr.strip() if proc.stderr else ""
|
||||
raise AnsibleError(
|
||||
_(
|
||||
|
||||
@@ -282,7 +282,14 @@ class RunnerManager:
|
||||
mode = info.get("mode", "docker")
|
||||
try:
|
||||
stdout = self._executor.run_ad_hoc(
|
||||
host, user, key, "shell", f"systemctl is-active gitea-runner@{name}", become=True
|
||||
host,
|
||||
user,
|
||||
key,
|
||||
"shell",
|
||||
f"systemctl is-active gitea-runner@{name}",
|
||||
become=True,
|
||||
ask_become_pass=True,
|
||||
check=False,
|
||||
)
|
||||
# Filter out ansible noise (SSH warnings etc)
|
||||
lines = [ln for ln in stdout.splitlines() if ln.strip() and not ln.startswith(("[WARNING]", "ssh:"))]
|
||||
|
||||
@@ -207,6 +207,54 @@ class TestAnsibleExecutorAdHoc:
|
||||
cmd = mock_run.call_args.args[0]
|
||||
assert "--become" in cmd
|
||||
|
||||
def test_run_ad_hoc_with_ask_become_pass(self, tmp_path: Path) -> None:
|
||||
executor = AnsibleExecutor(log_dir=tmp_path)
|
||||
result_mock = MagicMock()
|
||||
result_mock.stdout = "ok\n"
|
||||
result_mock.returncode = 0
|
||||
result_mock.stderr = ""
|
||||
|
||||
with patch("subprocess.run", return_value=result_mock) as mock_run:
|
||||
with patch("sys.stdin.isatty", return_value=True):
|
||||
result = executor.run_ad_hoc(
|
||||
"10.0.0.1", "ubuntu", None, "shell", "cmd", become=True, ask_become_pass=True
|
||||
)
|
||||
|
||||
assert result == "ok"
|
||||
cmd = mock_run.call_args.args[0]
|
||||
assert "--become" in cmd
|
||||
assert "--ask-become-pass" in cmd
|
||||
|
||||
def test_run_ad_hoc_ask_become_pass_no_tty(self, tmp_path: Path) -> None:
|
||||
executor = AnsibleExecutor(log_dir=tmp_path)
|
||||
result_mock = MagicMock()
|
||||
result_mock.stdout = "ok\n"
|
||||
result_mock.returncode = 0
|
||||
result_mock.stderr = ""
|
||||
|
||||
with patch("subprocess.run", return_value=result_mock) as mock_run:
|
||||
with patch("sys.stdin.isatty", return_value=False):
|
||||
result = executor.run_ad_hoc(
|
||||
"10.0.0.1", "ubuntu", None, "shell", "cmd", become=True, ask_become_pass=True
|
||||
)
|
||||
|
||||
assert result == "ok"
|
||||
cmd = mock_run.call_args.args[0]
|
||||
assert "--become" in cmd
|
||||
assert "--ask-become-pass" not in cmd
|
||||
|
||||
def test_run_ad_hoc_check_false(self, tmp_path: Path) -> None:
|
||||
executor = AnsibleExecutor(log_dir=tmp_path)
|
||||
result_mock = MagicMock()
|
||||
result_mock.stdout = "inactive\n"
|
||||
result_mock.returncode = 3
|
||||
result_mock.stderr = ""
|
||||
|
||||
with patch("subprocess.run", return_value=result_mock):
|
||||
result = executor.run_ad_hoc("10.0.0.1", "ubuntu", None, "shell", "systemctl is-active svc", check=False)
|
||||
|
||||
assert result == "inactive"
|
||||
|
||||
def test_run_ad_hoc_failure_raises(self, tmp_path: Path) -> None:
|
||||
executor = AnsibleExecutor(log_dir=tmp_path)
|
||||
result_mock = MagicMock()
|
||||
|
||||
@@ -408,7 +408,14 @@ class TestRunnerManager:
|
||||
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", "systemctl is-active gitea-runner@r1", become=True
|
||||
"10.0.0.1",
|
||||
"ubuntu",
|
||||
"/key",
|
||||
"shell",
|
||||
"systemctl is-active gitea-runner@r1",
|
||||
become=True,
|
||||
ask_become_pass=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
def test_list_runners_empty(self) -> None:
|
||||
|
||||
Reference in New Issue
Block a user