"""Unit tests for devx.ci.wait_for_checks.""" from unittest.mock import MagicMock, patch import pytest from click.testing import CliRunner from devx.ci.wait_for_checks import ( main, poll_until_complete, query_job_status, ) def _mock_response(status_code: int = 200, json_data: object | None = None) -> MagicMock: m = MagicMock() m.status_code = status_code if json_data is None: m.json.side_effect = ValueError("no json") else: m.json.return_value = json_data return m class TestQueryJobStatus: @patch("devx.ci.wait_for_checks.requests.get") def test_returns_matching_jobs(self, mock_get: MagicMock) -> None: """Jobs whose name starts with the prefix are returned.""" mock_get.side_effect = [ _mock_response(200, [{"id": 1}, {"id": 2}]), _mock_response(200, [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}]), _mock_response(200, [{"name": "other-job", "status": "completed", "conclusion": "success"}]), ] result = query_job_status("https://api", "tok", "o/r", "molecule-tests") assert len(result) == 1 assert result[0]["name"] == "molecule-tests (1)" assert result[0]["status"] == "completed" assert result[0]["conclusion"] == "success" @patch("devx.ci.wait_for_checks.requests.get") def test_no_matching_jobs(self, mock_get: MagicMock) -> None: """When no job names match the prefix, returns empty list.""" mock_get.side_effect = [ _mock_response(200, [{"id": 1}]), _mock_response(200, [{"name": "other-job", "status": "completed", "conclusion": "success"}]), ] result = query_job_status("https://api", "tok", "o/r", "molecule-tests") assert result == [] @patch("devx.ci.wait_for_checks.requests.get") def test_api_error_returns_empty(self, mock_get: MagicMock) -> None: """Network errors on the runs endpoint return an empty list.""" import requests mock_get.side_effect = requests.ConnectionError("down") result = query_job_status("https://api", "tok", "o/r", "molecule-tests") assert result == [] @patch("devx.ci.wait_for_checks.requests.get") def test_non_200_returns_empty(self, mock_get: MagicMock) -> None: """Non-200 on runs endpoint returns empty list.""" mock_get.side_effect = [_mock_response(500, {"message": "err"})] result = query_job_status("https://api", "tok", "o/r", "molecule-tests") assert result == [] @patch("devx.ci.wait_for_checks.requests.get") def test_jobs_as_dict_with_jobs_key(self, mock_get: MagicMock) -> None: """Jobs endpoint returning {'jobs': [...]} dict is handled.""" mock_get.side_effect = [ _mock_response(200, [{"id": 1}]), _mock_response( 200, {"jobs": [{"name": "molecule-tests (1)", "status": "in_progress", "conclusion": None}]} ), ] result = query_job_status("https://api", "tok", "o/r", "molecule-tests") assert len(result) == 1 assert result[0]["status"] == "in_progress" @patch("devx.ci.wait_for_checks.requests.get") def test_runs_as_dict_with_runs_key(self, mock_get: MagicMock) -> None: """Runs endpoint returning {'runs': [...]} dict is handled.""" mock_get.side_effect = [ _mock_response(200, {"runs": [{"id": 1}]}), _mock_response(200, [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}]), ] result = query_job_status("https://api", "tok", "o/r", "molecule-tests") assert len(result) == 1 @patch("devx.ci.wait_for_checks.requests.get") def test_jobs_endpoint_error_skips_run(self, mock_get: MagicMock) -> None: """A failed jobs query for one run doesn't abort the whole call.""" mock_get.side_effect = [ _mock_response(200, [{"id": 1}, {"id": 2}]), _mock_response(500, {"message": "err"}), _mock_response(200, [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}]), ] result = query_job_status("https://api", "tok", "o/r", "molecule-tests") assert len(result) == 1 @patch("devx.ci.wait_for_checks.requests.get") def test_run_without_id_skipped(self, mock_get: MagicMock) -> None: """Runs missing an 'id' field are skipped.""" mock_get.side_effect = [ _mock_response(200, [{"foo": "bar"}, {"id": 1}]), _mock_response(200, [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}]), ] result = query_job_status("https://api", "tok", "o/r", "molecule-tests") assert len(result) == 1 @patch("devx.ci.wait_for_checks.requests.get") def test_jobs_query_exception_skips_run(self, mock_get: MagicMock) -> None: """A ConnectionError on the jobs endpoint for one run is skipped.""" import requests mock_get.side_effect = [ _mock_response(200, [{"id": 1}, {"id": 2}]), requests.ConnectionError("down"), _mock_response(200, [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}]), ] result = query_job_status("https://api", "tok", "o/r", "molecule-tests") assert len(result) == 1 @patch("devx.ci.wait_for_checks.requests.get") def test_jobs_json_value_error_skips_run(self, mock_get: MagicMock) -> None: """A ValueError (bad JSON) on the jobs endpoint is skipped.""" mock_get.side_effect = [ _mock_response(200, [{"id": 1}]), _mock_response(200), # json raises ValueError by default ] result = query_job_status("https://api", "tok", "o/r", "molecule-tests") assert result == [] class TestPollUntilComplete: @patch("devx.ci.wait_for_checks.time.sleep") @patch("devx.ci.wait_for_checks.time.monotonic") @patch("devx.ci.wait_for_checks.query_job_status") def test_all_jobs_succeed(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: """All jobs completed with success → returns 0.""" mock_query.return_value = [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}] mock_mono.side_effect = [0.0, 0.0] code = poll_until_complete("https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10) assert code == 0 @patch("devx.ci.wait_for_checks.time.sleep") @patch("devx.ci.wait_for_checks.time.monotonic") @patch("devx.ci.wait_for_checks.query_job_status") def test_job_fails(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: """A job with non-success conclusion → returns 1.""" mock_query.return_value = [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "failure"}] mock_mono.side_effect = [0.0, 0.0] code = poll_until_complete("https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10) assert code == 1 @patch("devx.ci.wait_for_checks.time.sleep") @patch("devx.ci.wait_for_checks.time.monotonic") @patch("devx.ci.wait_for_checks.query_job_status") def test_no_require_success(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: """With require_success=False, a failed job returns 0.""" mock_query.return_value = [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "failure"}] mock_mono.side_effect = [0.0, 0.0] code = poll_until_complete( "https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10, require_success=False ) assert code == 0 @patch("devx.ci.wait_for_checks.time.sleep") @patch("devx.ci.wait_for_checks.time.monotonic") @patch("devx.ci.wait_for_checks.query_job_status") def test_timeout(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: """Jobs never complete → returns 2 after timeout.""" mock_query.return_value = [{"name": "molecule-tests (1)", "status": "in_progress", "conclusion": None}] # monotonic calls: deadline=0, while-check=0 (enter), sleep-calc=0, while-check=200 (exit) mock_mono.side_effect = [0.0, 0.0, 0.0, 200.0] code = poll_until_complete("https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10) assert code == 2 @patch("devx.ci.wait_for_checks.time.sleep") @patch("devx.ci.wait_for_checks.time.monotonic") @patch("devx.ci.wait_for_checks.query_job_status") def test_no_jobs_found_timeout(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: """No matching jobs at all → returns 3.""" mock_query.return_value = [] # monotonic calls: deadline=0, while-check=0 (enter), sleep-calc=0, while-check=200 (exit) mock_mono.side_effect = [0.0, 0.0, 0.0, 200.0] code = poll_until_complete("https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10) assert code == 3 @patch("devx.ci.wait_for_checks.time.sleep") @patch("devx.ci.wait_for_checks.time.monotonic") @patch("devx.ci.wait_for_checks.query_job_status") def test_in_progress_then_success(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: """First poll in_progress, second poll success → returns 0.""" mock_query.side_effect = [ [{"name": "molecule-tests (1)", "status": "in_progress", "conclusion": None}], [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}], ] # monotonic: deadline=0, while=0 (enter), sleep-calc=0, while=5 (enter), success→return mock_mono.side_effect = [0.0, 0.0, 0.0, 5.0] code = poll_until_complete("https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10) assert code == 0 class TestMain: @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=0) def test_success_exit_code(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r"]) assert result.exit_code == 0 @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=1) def test_failure_exit_code(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r"]) assert result.exit_code == 1 @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=2) def test_timeout_exit_code(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r"]) assert result.exit_code == 2 @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=3) def test_api_error_exit_code(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r"]) assert result.exit_code == 3 def test_missing_job_name(self) -> None: runner = CliRunner() result = runner.invoke(main, ["--repo", "o/r"]) assert result.exit_code != 0 def test_missing_repo(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("GITHUB_REPOSITORY", raising=False) runner = CliRunner() result = runner.invoke(main, ["--job-name", "molecule-tests"]) assert result.exit_code != 0 @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=0) def test_repo_from_env(self, mock_poll: MagicMock, mock_token: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("GITHUB_REPOSITORY", "oblachno-oss/grm") runner = CliRunner() result = runner.invoke(main, ["--job-name", "molecule-tests"]) assert result.exit_code == 0 args, kwargs = mock_poll.call_args assert args[2] == "oblachno-oss/grm" @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=0) def test_invalid_timeout(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r", "--timeout", "0"]) assert result.exit_code != 0 @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=0) def test_invalid_interval(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r", "--poll-interval", "0"]) assert result.exit_code != 0 @patch("devx.ci.wait_for_checks.get_ci_token", side_effect=__import__("click").ClickException("no token")) def test_no_token_exit_3(self, mock_token: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r"]) assert result.exit_code == 3