"""Unit tests for scripts/ci/auto_merge.py.""" import http from unittest.mock import MagicMock, patch import click import pytest from click.testing import CliRunner from gitea_runner_manager.config import CONVENTIONAL_RE, TASK_ID_RE from gitea_runner_manager.exceptions import APIError from scripts.ci.auto_merge import ( PR_TITLE_RE, extract_conventional_msg, extract_task_id, has_approval_review, has_ready_to_merge_label, main, validate_pr_title, validate_pr_title_matches_vikunja, wait_for_ci, ) CI_SUCCESS = "success" CI_PENDING = "pending" CI_FAILURE = "failure" def _status(context: str, status: str, updated_at: str = "2026-01-01T00:00:00Z") -> dict[str, str]: return {"context": context, "status": status, "updated_at": updated_at} def _commit(message: str) -> dict[str, dict[str, str]]: return {"commit": {"message": message}} class TestRegexes: def test_task_id_re_matches(self) -> None: assert TASK_ID_RE.search("GRM-19-fix-bug") assert TASK_ID_RE.search("feature/GRM-42") def test_task_id_re_no_match(self) -> None: assert not TASK_ID_RE.search("feature-no-id") def test_conventional_re_matches(self) -> None: assert CONVENTIONAL_RE.match("feat: add feature") assert CONVENTIONAL_RE.match("fix(api): handle timeout") def test_conventional_re_rejects(self) -> None: assert not CONVENTIONAL_RE.match("random message") assert not CONVENTIONAL_RE.match("feat:") def test_pr_title_re_matches(self) -> None: assert PR_TITLE_RE.match("GRM-19: Some task title") assert PR_TITLE_RE.match("GRM-42: double space title") def test_pr_title_re_rejects(self) -> None: assert not PR_TITLE_RE.match("fix: resolve timeout") assert not PR_TITLE_RE.match("GRM-19:No space after colon") assert not PR_TITLE_RE.match("random message") class TestExtractTaskId: def test_extracts_from_branch(self) -> None: assert extract_task_id("GRM-19-fix-bug") == "GRM-19" def test_extracts_from_feature_branch(self) -> None: assert extract_task_id("feature/GRM-42-add-x") == "GRM-42" def test_returns_empty_when_missing(self) -> None: assert extract_task_id("feature-no-id") == "" class TestValidatePrTitle: def test_valid_title_passes(self) -> None: validate_pr_title("GRM-19: Some task title", "GRM-19") def test_valid_title_with_scope_passes(self) -> None: validate_pr_title("GRM-42: Add --url option", "GRM-42") def test_invalid_format_raises(self) -> None: with pytest.raises(click.ClickException) as exc: validate_pr_title("random message", "GRM-19") assert "GRM-N" in str(exc.value) def test_invalid_format_conventional_raises(self) -> None: with pytest.raises(click.ClickException) as exc: validate_pr_title("fix: resolve timeout", "GRM-19") assert "GRM-N" in str(exc.value) def test_task_id_mismatch_raises(self) -> None: with pytest.raises(click.ClickException) as exc: validate_pr_title("GRM-42: Some task title", "GRM-19") assert "mismatch" in str(exc.value) class TestExtractConventionalMsg: def test_returns_newest_conventional(self) -> None: """Iterates in reverse — picks the newest conventional commit.""" commits = [ _commit("fix: resolve timeout"), _commit("random message"), _commit("feat: add thing"), ] assert extract_conventional_msg(commits) == "feat: add thing" def test_falls_back_to_newest_commit(self) -> None: commits = [ _commit("another random"), _commit("random message"), ] assert extract_conventional_msg(commits) == "random message" def test_uses_first_line_only(self) -> None: commits = [_commit("fix: resolve timeout\n\nBody text here")] assert extract_conventional_msg(commits) == "fix: resolve timeout" def test_empty_commits_returns_empty(self) -> None: assert extract_conventional_msg([]) == "" def test_commit_with_scope(self) -> None: commits = [_commit("feat(api): new endpoint")] assert extract_conventional_msg(commits) == "feat(api): new endpoint" def test_missing_commit_key(self) -> None: commits = [{}] # type: ignore[list-item] assert extract_conventional_msg(commits) == "" class TestHasReadyToMergeLabel: def test_label_present(self) -> None: client = MagicMock() client.get_pr_labels.return_value = [{"name": "bug"}, {"name": "ready-to-merge"}] assert has_ready_to_merge_label(client, "5") is True def test_label_absent(self) -> None: client = MagicMock() client.get_pr_labels.return_value = [{"name": "bug"}] assert has_ready_to_merge_label(client, "5") is False def test_no_labels(self) -> None: client = MagicMock() client.get_pr_labels.return_value = [] assert has_ready_to_merge_label(client, "5") is False class TestHasApprovalReview: def test_has_approved(self) -> None: client = MagicMock() client.get_pr_reviews.return_value = [{"state": "APPROVED"}, {"state": "COMMENT"}] assert has_approval_review(client, "5") is True def test_no_approved(self) -> None: client = MagicMock() client.get_pr_reviews.return_value = [{"state": "COMMENT"}, {"state": "REQUEST_CHANGES"}] assert has_approval_review(client, "5") is False def test_no_reviews(self) -> None: client = MagicMock() client.get_pr_reviews.return_value = [] assert has_approval_review(client, "5") is False class TestValidatePrTitleMatchesVikunja: @patch("scripts.ci.auto_merge.get_vikunja_task_title", return_value="") @patch.dict("os.environ", {}, clear=True) def test_no_vikunja_token_skips(self, mock_get: MagicMock) -> None: """Should skip validation when VIKUNJA_TOKEN is not set.""" validate_pr_title_matches_vikunja("GRM-19: Some title", "GRM-19") @patch("scripts.ci.auto_merge.get_vikunja_task_title", return_value="Some task title") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) def test_matching_title_passes(self, mock_get: MagicMock) -> None: validate_pr_title_matches_vikunja("GRM-19: Some task title", "GRM-19") @patch("scripts.ci.auto_merge.get_vikunja_task_title", return_value="Some task title") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) def test_mismatched_title_raises(self, mock_get: MagicMock) -> None: with pytest.raises(click.ClickException) as exc: validate_pr_title_matches_vikunja("GRM-19: Different title", "GRM-19") assert "does not match" in str(exc.value) @patch("scripts.ci.auto_merge.get_vikunja_task_title", return_value="") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) def test_task_not_found_skips(self, mock_get: MagicMock) -> None: """Should skip validation when Vikunja task is not found.""" validate_pr_title_matches_vikunja("GRM-19: Some title", "GRM-19") class TestGetVikunjaTaskTitle: @patch.dict("os.environ", {}, clear=True) def test_no_token_returns_empty(self) -> None: from scripts.ci.auto_merge import get_vikunja_task_title assert get_vikunja_task_title("GRM-19") == "" @patch("scripts.ci.auto_merge.VikunjaClient") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) def test_finds_task(self, mock_client_cls: MagicMock) -> None: from scripts.ci.auto_merge import get_vikunja_task_title mock_client = MagicMock() mock_client.list_project_tasks.return_value = [ {"identifier": "GRM-19", "title": "Some task title"}, ] mock_client_cls.return_value = mock_client assert get_vikunja_task_title("GRM-19") == "Some task title" @patch("scripts.ci.auto_merge.VikunjaClient") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) def test_task_not_found_returns_empty(self, mock_client_cls: MagicMock) -> None: from scripts.ci.auto_merge import get_vikunja_task_title mock_client = MagicMock() mock_client.list_project_tasks.return_value = [ {"identifier": "GRM-20", "title": "Other task"}, ] mock_client_cls.return_value = mock_client assert get_vikunja_task_title("GRM-19") == "" @patch("scripts.ci.auto_merge.VikunjaClient") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) def test_paginates_to_find_task(self, mock_client_cls: MagicMock) -> None: from scripts.ci.auto_merge import get_vikunja_task_title mock_client = MagicMock() # First page: full page of 50 tasks, no match; second page: match page1 = [{"identifier": f"GRM-{i}", "title": f"task {i}"} for i in range(50)] page2 = [{"identifier": "GRM-99", "title": "Found task"}] mock_client.list_project_tasks.side_effect = [page1, page2] mock_client_cls.return_value = mock_client assert get_vikunja_task_title("GRM-99") == "Found task" @patch("scripts.ci.auto_merge.VikunjaClient") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) def test_empty_pages_returns_empty(self, mock_client_cls: MagicMock) -> None: from scripts.ci.auto_merge import get_vikunja_task_title mock_client = MagicMock() mock_client.list_project_tasks.return_value = [] mock_client_cls.return_value = mock_client assert get_vikunja_task_title("GRM-19") == "" class TestWaitForCi: def test_all_pass_immediately(self) -> None: client = MagicMock() client.get_commit_status.return_value = [ _status("CI / quality (pull_request)", CI_SUCCESS), _status("CI / molecule-tests (0) (pull_request)", CI_SUCCESS), ] assert wait_for_ci(client, "abc123", max_wait=10) is True def test_waits_then_passes(self) -> None: client = MagicMock() client.get_commit_status.side_effect = [ [_status("CI / quality (pull_request)", CI_PENDING)], [_status("CI / quality (pull_request)", CI_SUCCESS)], ] with patch("scripts.ci.auto_merge.time.sleep"): assert wait_for_ci(client, "abc123", max_wait=10, poll_interval=5) is True def test_fails_on_failed_check(self) -> None: client = MagicMock() client.get_commit_status.return_value = [ _status("CI / quality (pull_request)", CI_SUCCESS), _status("CI / molecule-tests (0) (pull_request)", CI_FAILURE), ] assert wait_for_ci(client, "abc123", max_wait=10) is False def test_times_out(self) -> None: client = MagicMock() client.get_commit_status.return_value = [ _status("CI / quality (pull_request)", CI_PENDING), ] with patch("scripts.ci.auto_merge.time.sleep"): assert wait_for_ci(client, "abc123", max_wait=5) is False def test_no_statuses_waits(self) -> None: client = MagicMock() client.get_commit_status.side_effect = [ [], [_status("CI / quality (pull_request)", CI_SUCCESS)], ] with patch("scripts.ci.auto_merge.time.sleep"): assert wait_for_ci(client, "abc123", max_wait=10, poll_interval=5) is True def test_ignores_non_ci_contexts(self) -> None: client = MagicMock() client.get_commit_status.return_value = [ _status("Auto-merge / merge (pull_request)", CI_PENDING), _status("CI / quality (pull_request)", CI_SUCCESS), ] assert wait_for_ci(client, "abc123", max_wait=10) is True def test_deduplicates_by_latest(self) -> None: client = MagicMock() client.get_commit_status.return_value = [ _status("CI / quality (pull_request)", CI_PENDING, "2026-01-01T00:00:00Z"), _status("CI / quality (pull_request)", CI_SUCCESS, "2026-01-01T00:01:00Z"), ] assert wait_for_ci(client, "abc123", max_wait=10) is True def test_only_non_ci_contexts_waits_then_ci_appears(self) -> None: client = MagicMock() client.get_commit_status.side_effect = [ [_status("Auto-merge / merge (pull_request)", CI_PENDING)], [_status("CI / quality (pull_request)", CI_SUCCESS)], ] with patch("scripts.ci.auto_merge.time.sleep"): assert wait_for_ci(client, "abc123", max_wait=10, poll_interval=5) is True def _mock_pr(sha: str = "abc123def456") -> dict[str, object]: return {"head": {"sha": sha}} def _mock_ci_passing() -> list[dict[str, str]]: return [_status("CI / quality (pull_request)", CI_SUCCESS)] def _mock_commits() -> list[dict[str, dict[str, str]]]: return [_commit("fix: resolve timeout")] class TestMain: @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("scripts.ci.auto_merge.has_approval_review", return_value=True) @patch("scripts.ci.auto_merge.GiteaClient") def test_successful_flow_with_label_arg( self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock ) -> None: mock_client = MagicMock() mock_client_cls.return_value = mock_client mock_client.get_pr.return_value = _mock_pr() mock_client.get_commit_status.return_value = _mock_ci_passing() mock_client.get_pr_commits.return_value = _mock_commits() runner = CliRunner() result = runner.invoke( main, ["GRM-19-fix-bug", "GRM-19: Some task title", "owner/repo", "7", "ready-to-merge"], ) assert result.exit_code == 0 assert "squash-merged" in result.output mock_client.merge_pr.assert_called_once_with("7", "GRM-19 fix: resolve timeout") @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("scripts.ci.auto_merge.has_approval_review", return_value=True) @patch("scripts.ci.auto_merge.GiteaClient") def test_successful_flow_label_fallback( self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock ) -> None: """Label not passed via arg, but PR has ready-to-merge via API.""" mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client.get_pr.return_value = _mock_pr() mock_client.get_commit_status.return_value = _mock_ci_passing() mock_client.get_pr_commits.return_value = _mock_commits() mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke( main, ["GRM-19-fix-bug", "GRM-19: Some task title", "owner/repo", "7"], ) assert result.exit_code == 0 assert "squash-merged" in result.output mock_client.merge_pr.assert_called_once_with("7", "GRM-19 fix: resolve timeout") @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("scripts.ci.auto_merge.has_approval_review", return_value=True) @patch("scripts.ci.auto_merge.GiteaClient") def test_wrong_label_skips_merge( self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock ) -> None: """Label is not ready-to-merge and PR doesn't have it via API either.""" mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "bug"}] mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke( main, ["GRM-19-fix-bug", "GRM-19: Some task title", "owner/repo", "7", "bug"], ) assert result.exit_code == 0 assert "skipping" in result.output mock_client.merge_pr.assert_not_called() @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("scripts.ci.auto_merge.has_approval_review", return_value=True) @patch("scripts.ci.auto_merge.GiteaClient") def test_empty_label_falls_back_to_api( self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock ) -> None: """Gitea Actions doesn't populate label name, but API shows ready-to-merge.""" mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client.get_pr.return_value = _mock_pr() mock_client.get_commit_status.return_value = _mock_ci_passing() mock_client.get_pr_commits.return_value = _mock_commits() mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke( main, ["GRM-19-fix-bug", "GRM-19: Some task title", "owner/repo", "7", ""], ) assert result.exit_code == 0 assert "squash-merged" in result.output @patch.dict("os.environ", {"REPO_TOKEN": ""}, clear=True) def test_missing_token_exits(self) -> None: runner = CliRunner() result = runner.invoke(main, ["branch", "title", "owner/repo", "1"]) assert result.exit_code == 1 assert "REPO_TOKEN" in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.GiteaClient") def test_missing_task_id_exits(self, mock_client_cls: MagicMock) -> None: mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke(main, ["feature-no-id", "GRM-19: bug", "owner/repo", "1"]) assert result.exit_code == 1 assert "task ID" in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.GiteaClient") def test_invalid_pr_title_exits(self, mock_client_cls: MagicMock) -> None: mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke(main, ["GRM-19-fix", "random title", "owner/repo", "1"]) assert result.exit_code == 1 assert "GRM-N" in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.GiteaClient") def test_pr_title_task_id_mismatch_exits(self, mock_client_cls: MagicMock) -> None: """PR title has a different task ID than the branch.""" mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke(main, ["GRM-19-fix", "GRM-42: Some task title", "owner/repo", "1"]) assert result.exit_code == 1 assert "mismatch" in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("scripts.ci.auto_merge.has_approval_review", return_value=False) @patch("scripts.ci.auto_merge.GiteaClient") def test_no_approval_review_blocks_merge( self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock ) -> None: """No APPROVE review — merge should be blocked.""" mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"]) assert result.exit_code == 1 assert "APPROVE review" in result.output mock_client.merge_pr.assert_not_called() @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("scripts.ci.auto_merge.has_approval_review", return_value=True) @patch("scripts.ci.auto_merge.GiteaClient") def test_empty_commits_exits( self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock ) -> None: """PR has no commits — cannot extract conventional message.""" mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client.get_pr.return_value = _mock_pr() mock_client.get_commit_status.return_value = _mock_ci_passing() mock_client.get_pr_commits.return_value = [] mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"]) assert result.exit_code == 1 assert "conventional commit" in result.output mock_client.merge_pr.assert_not_called() @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("scripts.ci.auto_merge.has_approval_review", return_value=True) @patch("scripts.ci.auto_merge.GiteaClient") def test_merge_pr_failure_raises_click( self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock ) -> None: mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client.get_pr.return_value = _mock_pr() mock_client.get_commit_status.return_value = _mock_ci_passing() mock_client.get_pr_commits.return_value = _mock_commits() mock_client.merge_pr.side_effect = APIError(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error") mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"]) assert result.exit_code == 1 assert "HTTP" in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("scripts.ci.auto_merge.has_approval_review", return_value=True) @patch("scripts.ci.auto_merge.GiteaClient") def test_merge_pr_json_parse_failure( self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock ) -> None: mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client.get_pr.return_value = _mock_pr() mock_client.get_commit_status.return_value = _mock_ci_passing() mock_client.get_pr_commits.return_value = _mock_commits() mock_client.merge_pr.side_effect = APIError(http.HTTPStatus.BAD_GATEWAY, "bad gateway") mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"]) assert result.exit_code == 1 assert str(http.HTTPStatus.BAD_GATEWAY) in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("scripts.ci.auto_merge.has_approval_review", return_value=True) @patch("scripts.ci.auto_merge.GiteaClient") def test_ci_failure_blocks_merge( self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock ) -> None: """CI checks fail — merge should not be attempted.""" mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client.get_pr.return_value = _mock_pr() mock_client.get_commit_status.return_value = [ _status("CI / quality (pull_request)", CI_FAILURE), ] mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"]) assert result.exit_code == 1 assert "CI checks did not pass" in result.output mock_client.merge_pr.assert_not_called() @patch.dict("os.environ", {"REPO_TOKEN": "tok"}) @patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("scripts.ci.auto_merge.has_approval_review", return_value=True) @patch("scripts.ci.auto_merge.GiteaClient") def test_no_sha_proceeds_without_wait( self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock ) -> None: """PR head SHA missing — should proceed without waiting.""" mock_client = MagicMock() mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}] mock_client.get_pr.return_value = {"head": {}} mock_client.get_pr_commits.return_value = _mock_commits() mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"]) assert result.exit_code == 0 assert "squash-merged" in result.output