"""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 import pytest 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.ci.integration_guard.subprocess.Popen") as mock_popen, patch("devx.ci.integration_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.ci.integration_guard.subprocess.Popen") as mock_popen, patch("devx.ci.integration_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.ci.integration_guard.subprocess.Popen") as mock_popen, patch("devx.ci.integration_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"]) class TestGetRunningJobs: def test_returns_jobs(self) -> None: with patch("devx.ci.integration_guard.requests.get") as mock_get: mock_response = MagicMock() mock_response.json.return_value = { "jobs": [ {"name": "integration-tests (0)", "conclusion": "success"}, {"name": "integration-tests (1)", "conclusion": "failure"}, ] } mock_response.raise_for_status.return_value = None mock_get.return_value = mock_response from devx.ci.integration_guard import get_running_jobs jobs = get_running_jobs("https://gitea.example", "owner", "repo", "token", 123) assert len(jobs) == 2 mock_get.assert_called_once() def test_raises_on_request_error(self) -> None: import requests with patch("devx.ci.integration_guard.requests.get") as mock_get: mock_get.side_effect = requests.RequestException("boom") with pytest.raises(requests.RequestException): from devx.ci.integration_guard import get_running_jobs get_running_jobs("https://gitea.example", "owner", "repo", "token", 123) class TestAnyOtherRunnerFailed: def test_detects_other_failure(self) -> None: from devx.ci.integration_guard import any_other_runner_failed jobs = [ {"name": "integration-tests (0)", "conclusion": "success"}, {"name": "integration-tests (1)", "conclusion": "failure"}, {"name": "integration-tests (2)", "conclusion": "running"}, ] assert any_other_runner_failed(jobs, "integration-tests", 0) is True def test_ignores_current_runner(self) -> None: from devx.ci.integration_guard import any_other_runner_failed jobs = [ {"name": "integration-tests (0)", "conclusion": "failure"}, {"name": "integration-tests (1)", "conclusion": "success"}, ] assert any_other_runner_failed(jobs, "integration-tests", 0) is False def test_ignores_non_matching_jobs(self) -> None: from devx.ci.integration_guard import any_other_runner_failed jobs = [ {"name": "quality", "conclusion": "failure"}, {"name": "integration-tests (1)", "conclusion": "success"}, ] assert any_other_runner_failed(jobs, "integration-tests", 0) is False class TestPollForOtherFailures: def test_sets_failed_event_when_other_runner_fails(self) -> None: from devx.ci.integration_guard import poll_for_other_failures stop_event = MagicMock() failed_event = MagicMock() def side_effect(*args, **kwargs): if stop_event.wait.call_count < 1: return [ {"name": "integration-tests (0)", "conclusion": "success"}, {"name": "integration-tests (1)", "conclusion": "failure"}, ] return [] with patch("devx.ci.integration_guard.get_running_jobs") as mock_get_jobs: mock_get_jobs.side_effect = side_effect stop_event.is_set.side_effect = [False, False] stop_event.wait.return_value = True poll_for_other_failures( "https://gitea.example", "owner", "repo", "token", 123, "integration-tests", 0, stop_event, failed_event, ) failed_event.set.assert_called_once() def test_poll_warns_on_api_error(self) -> None: import requests from devx.ci.integration_guard import poll_for_other_failures stop_event = MagicMock() failed_event = MagicMock() with patch("devx.ci.integration_guard.get_running_jobs") as mock_get_jobs: mock_get_jobs.side_effect = requests.RequestException("boom") stop_event.is_set.side_effect = [False, True] stop_event.wait.return_value = True poll_for_other_failures( "https://gitea.example", "owner", "repo", "token", 123, "integration-tests", 0, stop_event, failed_event, ) failed_event.set.assert_not_called()