Public Access
emil
e796b06a91
Post-merge / detect-type (push) Successful in 9s
Post-merge / validate-commit-msg (push) Successful in 9s
Post-merge / vikunja (push) Successful in 15s
Post-merge / configure-repo (push) Successful in 19s
Post-merge / release (push) Successful in 45s
Post-merge / sync-wiki (push) Successful in 46s
Post-merge / badges (push) Successful in 46s
Post-merge / publish (push) Successful in 17s
292 lines
11 KiB
Python
292 lines
11 KiB
Python
"""Unit tests for devx.ci.integration_guard."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess # nosec B404
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from click.testing import CliRunner
|
|
|
|
from devx.ci.integration_guard import cli
|
|
|
|
|
|
class TestCli:
|
|
def test_all_pass(self) -> None:
|
|
with (
|
|
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
|
|
patch("time.sleep"),
|
|
):
|
|
proc = MagicMock()
|
|
proc.poll.return_value = 0
|
|
proc.returncode = 0
|
|
mock_popen.return_value = proc
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--", "tests/integration/test_foo.py"])
|
|
assert result.exit_code == 0
|
|
assert "Integration tests passed" in result.output
|
|
|
|
def test_failure_exits_nonzero(self) -> None:
|
|
with (
|
|
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
|
|
patch("time.sleep"),
|
|
):
|
|
proc = MagicMock()
|
|
proc.poll.return_value = 1
|
|
proc.returncode = 1
|
|
mock_popen.return_value = proc
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--", "tests/integration/test_foo.py"])
|
|
assert result.exit_code == 1
|
|
assert "failed" in result.output
|
|
|
|
def test_pytest_args_passed_through(self) -> None:
|
|
with (
|
|
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
|
|
patch("time.sleep"),
|
|
):
|
|
proc = MagicMock()
|
|
proc.poll.return_value = 0
|
|
proc.returncode = 0
|
|
mock_popen.return_value = proc
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
cli,
|
|
["--", "-x", "-v", "--tb=short", "test_a.py", "test_b.py"],
|
|
)
|
|
assert result.exit_code == 0
|
|
call_args = mock_popen.call_args[0][0]
|
|
assert "-x" in call_args
|
|
assert "-v" in call_args
|
|
assert "test_a.py" in call_args
|
|
assert "test_b.py" in call_args
|
|
|
|
def test_keyboard_interrupt_kills_process(self) -> None:
|
|
with (
|
|
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
|
|
patch("time.sleep", side_effect=KeyboardInterrupt),
|
|
patch("os.killpg") as mock_killpg,
|
|
patch("os.getpgid") as mock_getpgid,
|
|
):
|
|
mock_getpgid.return_value = 123
|
|
proc = MagicMock()
|
|
proc.poll.return_value = None
|
|
proc.wait.return_value = 0
|
|
mock_popen.return_value = proc
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--", "test_foo.py"])
|
|
assert result.exit_code == 1
|
|
mock_killpg.assert_called()
|
|
|
|
def test_exits_when_other_runner_fails(self) -> None:
|
|
real_sleep = time.sleep
|
|
call_count = [0]
|
|
|
|
def get_jobs_side_effect(*args, **kwargs):
|
|
call_count[0] += 1
|
|
if call_count[0] < 2:
|
|
return [{"name": "integration-tests (1)", "conclusion": "running"}]
|
|
return [
|
|
{"name": "integration-tests (0)", "conclusion": "running"},
|
|
{"name": "integration-tests (1)", "conclusion": "failure"},
|
|
]
|
|
|
|
with (
|
|
patch.dict(
|
|
os.environ,
|
|
{
|
|
"GITEA_URL": "https://gitea.example",
|
|
"CI_GITEA_TOKEN": "token",
|
|
"RUN_ID": "123",
|
|
"JOB_NAME": "integration-tests",
|
|
"MATRIX_INDEX": "0",
|
|
"GITEA_REPOSITORY": "my-org/my-repo",
|
|
"PATH": os.environ.get("PATH", ""),
|
|
},
|
|
clear=True,
|
|
),
|
|
patch("devx.ci.integration_guard.POLL_INTERVAL", 0.01),
|
|
patch("devx.molecule.molecule_ci_guard.POLL_INTERVAL", 0.01),
|
|
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
|
|
patch("devx.molecule.molecule_ci_guard.get_running_jobs", side_effect=get_jobs_side_effect),
|
|
patch("os.killpg") as mock_killpg,
|
|
patch("os.getpgid") as mock_getpgid,
|
|
patch("time.sleep", side_effect=lambda x: real_sleep(0)),
|
|
):
|
|
mock_getpgid.return_value = 123
|
|
proc = MagicMock()
|
|
proc.poll.return_value = None
|
|
proc.wait.return_value = 0
|
|
mock_popen.return_value = proc
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--", "test_foo.py"])
|
|
assert result.exit_code == 1
|
|
mock_killpg.assert_called()
|
|
assert "cancelled" in result.output.lower()
|
|
|
|
def test_process_lookup_error_suppressed(self) -> None:
|
|
real_sleep = time.sleep
|
|
call_count = [0]
|
|
|
|
def get_jobs_side_effect(*args, **kwargs):
|
|
call_count[0] += 1
|
|
if call_count[0] < 2:
|
|
return [{"name": "integration-tests (1)", "conclusion": "running"}]
|
|
return [
|
|
{"name": "integration-tests (0)", "conclusion": "running"},
|
|
{"name": "integration-tests (1)", "conclusion": "failure"},
|
|
]
|
|
|
|
with (
|
|
patch.dict(
|
|
os.environ,
|
|
{
|
|
"GITEA_URL": "https://gitea.example",
|
|
"CI_GITEA_TOKEN": "token",
|
|
"RUN_ID": "123",
|
|
"JOB_NAME": "integration-tests",
|
|
"MATRIX_INDEX": "0",
|
|
"GITEA_REPOSITORY": "my-org/my-repo",
|
|
"PATH": os.environ.get("PATH", ""),
|
|
},
|
|
clear=True,
|
|
),
|
|
patch("devx.ci.integration_guard.POLL_INTERVAL", 0.01),
|
|
patch("devx.molecule.molecule_ci_guard.POLL_INTERVAL", 0.01),
|
|
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
|
|
patch("devx.molecule.molecule_ci_guard.get_running_jobs", side_effect=get_jobs_side_effect),
|
|
patch("os.killpg", side_effect=ProcessLookupError("no such process")),
|
|
patch("os.getpgid") as mock_getpgid,
|
|
patch("time.sleep", side_effect=lambda x: real_sleep(0)),
|
|
):
|
|
mock_getpgid.return_value = 123
|
|
proc = MagicMock()
|
|
proc.poll.return_value = None
|
|
proc.wait.return_value = 0
|
|
mock_popen.return_value = proc
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--", "test_foo.py"])
|
|
assert result.exit_code == 1
|
|
|
|
def test_timeout_expired_kills_with_sigkill(self) -> None:
|
|
real_sleep = time.sleep
|
|
call_count = [0]
|
|
|
|
def get_jobs_side_effect(*args, **kwargs):
|
|
call_count[0] += 1
|
|
if call_count[0] < 2:
|
|
return [{"name": "integration-tests (1)", "conclusion": "running"}]
|
|
return [
|
|
{"name": "integration-tests (0)", "conclusion": "running"},
|
|
{"name": "integration-tests (1)", "conclusion": "failure"},
|
|
]
|
|
|
|
with (
|
|
patch.dict(
|
|
os.environ,
|
|
{
|
|
"GITEA_URL": "https://gitea.example",
|
|
"CI_GITEA_TOKEN": "token",
|
|
"RUN_ID": "123",
|
|
"JOB_NAME": "integration-tests",
|
|
"MATRIX_INDEX": "0",
|
|
"GITEA_REPOSITORY": "my-org/my-repo",
|
|
"PATH": os.environ.get("PATH", ""),
|
|
},
|
|
clear=True,
|
|
),
|
|
patch("devx.ci.integration_guard.POLL_INTERVAL", 0.01),
|
|
patch("devx.molecule.molecule_ci_guard.POLL_INTERVAL", 0.01),
|
|
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
|
|
patch("devx.molecule.molecule_ci_guard.get_running_jobs", side_effect=get_jobs_side_effect),
|
|
patch("os.killpg") as mock_killpg,
|
|
patch("os.getpgid") as mock_getpgid,
|
|
patch("time.sleep", side_effect=lambda x: real_sleep(0)),
|
|
):
|
|
mock_getpgid.return_value = 123
|
|
proc = MagicMock()
|
|
proc.poll.return_value = None
|
|
proc.wait.side_effect = [subprocess.TimeoutExpired("cmd", 10)]
|
|
mock_popen.return_value = proc
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--", "test_foo.py"])
|
|
assert result.exit_code == 1
|
|
# SIGKILL should have been called (second killpg call)
|
|
assert mock_killpg.call_count >= 2
|
|
|
|
def test_no_env_vars_runs_without_polling(self) -> None:
|
|
with (
|
|
patch.dict(os.environ, {"PATH": os.environ.get("PATH", "")}, clear=True),
|
|
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
|
|
patch("time.sleep"),
|
|
):
|
|
proc = MagicMock()
|
|
proc.poll.return_value = 0
|
|
proc.returncode = 0
|
|
mock_popen.return_value = proc
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--", "test_foo.py"])
|
|
assert result.exit_code == 0
|
|
assert "without cross-runner cancellation" in result.output
|
|
|
|
def test_partial_env_vars_runs_without_polling(self) -> None:
|
|
"""Only GITEA_URL set (missing CI_GITEA_TOKEN and RUN_ID) — should skip polling."""
|
|
with (
|
|
patch.dict(
|
|
os.environ,
|
|
{"GITEA_URL": "https://gitea.example", "PATH": os.environ.get("PATH", "")},
|
|
clear=True,
|
|
),
|
|
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
|
|
patch("time.sleep"),
|
|
):
|
|
proc = MagicMock()
|
|
proc.poll.return_value = 0
|
|
proc.returncode = 0
|
|
mock_popen.return_value = proc
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--", "test_foo.py"])
|
|
assert result.exit_code == 0
|
|
assert "without cross-runner cancellation" in result.output
|
|
|
|
def test_invalid_repository_falls_back_to_default(self) -> None:
|
|
"""GITEA_REPOSITORY without '/' falls back to oblachno-oss/devx."""
|
|
with (
|
|
patch.dict(
|
|
os.environ,
|
|
{"GITEA_REPOSITORY": "invalid", "PATH": os.environ.get("PATH", "")},
|
|
clear=True,
|
|
),
|
|
patch("devx.ci.integration_guard.subprocess.Popen") as mock_popen,
|
|
patch("time.sleep"),
|
|
):
|
|
proc = MagicMock()
|
|
proc.poll.return_value = 0
|
|
proc.returncode = 0
|
|
mock_popen.return_value = proc
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--", "test_foo.py"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
def test_main_module_block() -> None:
|
|
import devx.ci.integration_guard as ig
|
|
|
|
with open(ig.__file__) as f:
|
|
source = f.read()
|
|
source = source.replace('if __name__ == "__main__":\n cli()\n', "")
|
|
namespace = dict(ig.__dict__)
|
|
exec(compile(source, ig.__file__, "exec"), namespace)
|
|
assert callable(namespace["cli"])
|