GRM-123: fix: cast disk threshold to string in template-content verify assertion
Post-merge / detect-type (push) Successful in 49s
Post-merge / validate-commit-msg (push) Successful in 1m6s
Post-merge / vikunja (push) Successful in 1m11s
Post-merge / release (push) Successful in 1m21s
Post-merge / badges (push) Successful in 1m37s
Post-merge / configure-repo (push) Successful in 1m20s
Post-merge / publish (push) Successful in 57s
Post-merge / sync-wiki (push) Successful in 2m35s

This commit was merged in pull request #189.
This commit is contained in:
2026-06-30 23:28:54 +00:00
parent 8fbe2d3f51
commit 63ef5cdbcf
16 changed files with 598 additions and 0 deletions
+106
View File
@@ -625,6 +625,112 @@ class TestRunnerManager:
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."""