200 lines
8.0 KiB
Python
200 lines
8.0 KiB
Python
"""Unit tests for AnsibleExecutor."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, mock_open, patch
|
|
|
|
import pytest
|
|
|
|
from gitea_runner_manager.exceptions import AnsibleError
|
|
from gitea_runner_manager.executor import AnsibleExecutor
|
|
|
|
|
|
def _mock_popen_process(returncode: int = 0) -> MagicMock:
|
|
"""Create a mock Popen process that yields no stdout and exits with given code."""
|
|
proc = MagicMock()
|
|
proc.stdout = iter([])
|
|
proc.wait.return_value = returncode
|
|
proc.returncode = returncode
|
|
return proc
|
|
|
|
|
|
class TestAnsibleExecutorRun:
|
|
def test_run_success(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
|
executor = AnsibleExecutor(log_dir=tmp_path)
|
|
with patch("subprocess.Popen") as mock_popen, patch("builtins.open", mock_open()):
|
|
mock_popen.return_value = _mock_popen_process(returncode=0)
|
|
executor.run(["echo", "hello"], description="Test run")
|
|
|
|
captured = capsys.readouterr()
|
|
assert "[GRM] Test run" in captured.out
|
|
assert "[GRM] Done" in captured.out
|
|
|
|
def test_run_failure(self, tmp_path: Path) -> None:
|
|
executor = AnsibleExecutor(log_dir=tmp_path)
|
|
with patch("subprocess.Popen") as mock_popen, patch("builtins.open", mock_open()):
|
|
mock_popen.return_value = _mock_popen_process(returncode=1)
|
|
with pytest.raises(AnsibleError, match="exit code 1"):
|
|
executor.run(["false"], description="Test run")
|
|
|
|
def test_run_default_description(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
|
executor = AnsibleExecutor(log_dir=tmp_path)
|
|
with patch("subprocess.Popen") as mock_popen, patch("builtins.open", mock_open()):
|
|
mock_popen.return_value = _mock_popen_process(returncode=0)
|
|
executor.run(["echo", "hello"])
|
|
|
|
captured = capsys.readouterr()
|
|
assert "[GRM] Running Ansible playbook" in captured.out
|
|
|
|
def test_run_with_stdout_lines(self, tmp_path: Path) -> None:
|
|
executor = AnsibleExecutor(log_dir=tmp_path)
|
|
proc = MagicMock()
|
|
proc.stdout = iter(["line1\n", "line2\n"])
|
|
proc.wait.return_value = 0
|
|
proc.returncode = 0
|
|
mock_file = mock_open()
|
|
|
|
with patch("subprocess.Popen", return_value=proc), patch("builtins.open", mock_file):
|
|
executor.run(["echo", "hello"], description="Test run")
|
|
handle = mock_file()
|
|
handle.write.assert_any_call("line1\n")
|
|
handle.write.assert_any_call("line2\n")
|
|
|
|
def test_run_extracts_status(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
|
executor = AnsibleExecutor(log_dir=tmp_path)
|
|
log_content = (
|
|
"TASK [gitea-runner : Report runner status]\n"
|
|
"ok: [127.0.0.1] => {\n"
|
|
' "msg": "Runner \'127.0.0.1\' is installed and running."\n'
|
|
"}\n"
|
|
)
|
|
with patch("subprocess.Popen") as mock_popen, patch("builtins.open", mock_open(read_data=log_content)):
|
|
mock_popen.return_value = _mock_popen_process(returncode=0)
|
|
executor.run(["echo", "hello"], description="Test run")
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Runner '127.0.0.1' is installed and running" in captured.out
|
|
|
|
|
|
class TestAnsibleExecutorExtractStatus:
|
|
def test_extract_status_found(self, tmp_path: Path) -> None:
|
|
executor = AnsibleExecutor()
|
|
log_content = (
|
|
"TASK [gitea-runner : Report runner status]\n"
|
|
"ok: [127.0.0.1] => {\n"
|
|
' "msg": "Runner \'127.0.0.1\' is installed and running."\n'
|
|
"}\n"
|
|
)
|
|
log_file = tmp_path / "test.log"
|
|
log_file.write_text(log_content)
|
|
status = executor._extract_status(log_file)
|
|
assert status is not None
|
|
assert "Runner '127.0.0.1' is installed and running" in status
|
|
|
|
def test_extract_status_not_found(self, tmp_path: Path) -> None:
|
|
executor = AnsibleExecutor()
|
|
log_file = tmp_path / "test.log"
|
|
log_file.write_text('TASK [other]\nok: [host] => {\n "msg": "something else"\n}\n')
|
|
status = executor._extract_status(log_file)
|
|
assert status is None
|
|
|
|
def test_extract_status_exception(self, tmp_path: Path) -> None:
|
|
executor = AnsibleExecutor()
|
|
log_file = tmp_path / "test.log"
|
|
log_file.write_text("incomplete")
|
|
log_file.chmod(0o000)
|
|
try:
|
|
status = executor._extract_status(log_file)
|
|
assert status is None
|
|
finally:
|
|
log_file.chmod(0o644)
|
|
|
|
|
|
class TestAnsibleExecutorPrepareLog:
|
|
def test_prepare_log(self, tmp_path: Path) -> None:
|
|
executor = AnsibleExecutor(log_dir=tmp_path)
|
|
log_file = executor._prepare_log(["ansible-playbook", "test.yml"])
|
|
assert log_file.parent == tmp_path
|
|
assert log_file.name.startswith("ansible-")
|
|
content = log_file.read_text()
|
|
assert "=== GRM Ansible Run:" in content
|
|
assert "Command: ansible-playbook test.yml" in content
|
|
|
|
def test_prepare_log_creates_directories(self, tmp_path: Path) -> None:
|
|
nested = tmp_path / "a" / "b" / "c"
|
|
executor = AnsibleExecutor(log_dir=nested)
|
|
log_file = executor._prepare_log(["echo"])
|
|
assert nested.exists()
|
|
assert log_file.exists()
|
|
|
|
|
|
class TestAnsibleExecutorTranslation:
|
|
def test_run_translated(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
|
executor = AnsibleExecutor(log_dir=tmp_path)
|
|
with (
|
|
patch.dict(os.environ, {"GRM_LANG": "de"}),
|
|
patch("subprocess.Popen") as mock_popen,
|
|
patch("builtins.open", mock_open()),
|
|
):
|
|
mock_popen.return_value = _mock_popen_process(returncode=0)
|
|
executor.run(["echo", "hello"]) # use default description to test translation
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Ansible-Playbook wird ausgeführt" in captured.out
|
|
assert "Fertig" in captured.out
|
|
|
|
def test_unsupported_lang_fallback(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
|
executor = AnsibleExecutor(log_dir=tmp_path)
|
|
with (
|
|
patch.dict(os.environ, {"GRM_LANG": "xx"}),
|
|
patch("subprocess.Popen") as mock_popen,
|
|
patch("builtins.open", mock_open()),
|
|
):
|
|
mock_popen.return_value = _mock_popen_process(returncode=0)
|
|
executor.run(["echo", "hello"])
|
|
|
|
captured = capsys.readouterr()
|
|
assert "[GRM] Running Ansible playbook" in captured.out # English fallback
|
|
|
|
|
|
class TestAnsibleExecutorAdHoc:
|
|
def test_run_ad_hoc_success(self, tmp_path: Path) -> None:
|
|
executor = AnsibleExecutor(log_dir=tmp_path)
|
|
result_mock = MagicMock()
|
|
result_mock.stdout = "active\n"
|
|
result_mock.returncode = 0
|
|
result_mock.stderr = ""
|
|
|
|
with patch("subprocess.run", return_value=result_mock):
|
|
result = executor.run_ad_hoc("10.0.0.1", "ubuntu", "/key", "shell", "systemctl is-active svc")
|
|
|
|
assert result == "active"
|
|
|
|
def test_run_ad_hoc_without_key(self, tmp_path: Path) -> None:
|
|
executor = AnsibleExecutor(log_dir=tmp_path)
|
|
result_mock = MagicMock()
|
|
result_mock.stdout = "inactive\n"
|
|
result_mock.returncode = 0
|
|
result_mock.stderr = ""
|
|
|
|
with patch("subprocess.run", return_value=result_mock) as mock_run:
|
|
result = executor.run_ad_hoc("10.0.0.1", "ubuntu", None, "shell", "cmd")
|
|
|
|
assert result == "inactive"
|
|
cmd = mock_run.call_args.args[0]
|
|
assert "--private-key" not in cmd
|
|
|
|
def test_run_ad_hoc_with_become(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:
|
|
result = executor.run_ad_hoc("10.0.0.1", "ubuntu", None, "shell", "cmd", become=True)
|
|
|
|
assert result == "ok"
|
|
cmd = mock_run.call_args.args[0]
|
|
assert "--become" in cmd
|