oblachno-oss/grm
1
0
Fork 0
Code Issues Pull Requests Actions 14 Packages Projects Releases Wiki Activity

GRM-17: fix docker inspect Jinja2 conflict and add per-iteration exception handling

Browse Source 
- docker inspect -f "{{.State.Status}}" used Go template braces that
  conflicted with Ansible Jinja2 templating in the shell module.
  Ansible tried to parse {{.State.Status}} as a Jinja2 variable (which
  starts with a dot, making it invalid), causing a local template error.
  The outer except Exception caught this immediately, so the fallback
  loop never reached the legacy container name or systemctl checks.

- Replaced with: docker inspect <name> | python3 -c JSON parsing,
  which avoids any brace syntax and uses python3 (already required by
  Ansible on managed nodes).

- Added per-iteration try/except inside the fallback loop so a failure
  on one container name continues to the next fallback instead of
  aborting the entire check.

- Added tests for fallback behavior and binary mode exception path.

128 tests, 100% coverage, ruff + pyright clean
This commit is contained in:
Emil Simeonov
2026-06-19 02:53:26 +02:00
parent 1bbe3b4f43
commit a9adb71a08
2 changed files with 75 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gitea_runner_manager/runner_manager.py
+27 -15
View File
@@ -295,27 +295,36 @@ class RunnerManager:
lines = [ln for ln in stdout.splitlines() if ln.strip() and not any(p in ln for p in ansible_noise)]
return lines[-1].strip() if lines else "unknown"
try:
if mode == "docker":
# Try expected container name first, then host-based fallback for legacy installs
for container_name in (f"gitea-runner-{name}", f"gitea-runner-{host}"):
service_status = "unknown"
if mode == "docker":
# Try expected container name first, then host-based fallback for legacy installs.
# Use python3 JSON parsing to avoid Jinja2 brace conflicts in Ansible shell args.
for container_name in (f"gitea-runner-{name}", f"gitea-runner-{host}"):
try:
stdout = self._executor.run_ad_hoc(
host,
user,
key,
"shell",
f"docker inspect -f '{{{{.State.Status}}}}' {container_name} 2>/dev/null",
(
f"docker inspect {container_name} 2>/dev/null | "
f"python3 -c \"import sys,json; print(json.load(sys.stdin)[0]['State']['Status'])\" "
f"2>/dev/null || echo unknown"
),
become=True,
ask_become_pass=True,
check=False,
)
service_status = _parse_status(stdout)
if service_status != "unknown":
docker_map = {"running": "active", "exited": "inactive", "dead": "failed"}
service_status = docker_map.get(service_status, "unknown")
break
else:
# Final fallback: check systemd service (for installs using the fixed template)
except Exception:
continue
status = _parse_status(stdout)
if status != "unknown":
docker_map = {"running": "active", "exited": "inactive", "dead": "failed"}
service_status = docker_map.get(status, "unknown")
break
else:
# Final fallback: check systemd service (for installs using the fixed template)
try:
stdout = self._executor.run_ad_hoc(
host,
user,
@@ -327,7 +336,10 @@ class RunnerManager:
check=False,
)
service_status = _parse_status(stdout)
else:
except Exception:
service_status = "unknown"
else:
try:
stdout = self._executor.run_ad_hoc(
host,
user,
@@ -339,8 +351,8 @@ class RunnerManager:
check=False,
)
service_status = _parse_status(stdout)
except Exception:
service_status = "unknown"
except Exception:
service_status = "unknown"
# Translate known status values
translated_status = _(
service_status if service_status in {"active", "inactive", "failed", "unknown"} else "unknown"
tests/unit/test_runner_manager.py
+48 -1
View File
@@ -412,7 +412,11 @@ class TestRunnerManager:
"ubuntu",
"/key",
"shell",
"docker inspect -f '{{.State.Status}}' gitea-runner-r1 2>/dev/null",
(
"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,
@@ -442,6 +446,19 @@ class TestRunnerManager:
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 = {}
@@ -461,6 +478,21 @@ class TestRunnerManager:
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 = {
@@ -474,6 +506,21 @@ class TestRunnerManager:
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 = {