"""Unit tests for scripts/ci/auto_merge.py.""" from unittest.mock import MagicMock, patch import click import pytest from click.testing import CliRunner from devx.ci.auto_merge import ( extract_conventional_msg, extract_task_id, main, read_taskid, run_cmd, validate_pr_title, validate_pr_title_matches_vikunja, ) from devx.exceptions import APIError # -- read_taskid -- class TestReadTaskid: def test_extracts_from_branch_name(self, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def] monkeypatch.chdir(tmp_path) assert read_taskid("DEVX-19-fix-bug") == "DEVX-19" def test_returns_empty_when_no_match(self, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def] monkeypatch.chdir(tmp_path) assert read_taskid("feature-branch") == "" def test_warns_on_stale_taskid_file(self, tmp_path, monkeypatch, capsys) -> None: # type: ignore[no-untyped-def] monkeypatch.chdir(tmp_path) (tmp_path / ".taskid").write_text("DEVX-60\n") # Branch name takes priority, stale .taskid should produce deprecation warning assert read_taskid("DEVX-19-fix-bug") == "DEVX-19" captured = capsys.readouterr() combined = captured.out + captured.err assert "WARNING" in combined assert "deprecated" in combined assert "DEVX-60" in combined assert "DEVX-19" in combined def test_no_warning_when_taskid_file_absent(self, tmp_path, monkeypatch, capsys) -> None: # type: ignore[no-untyped-def] monkeypatch.chdir(tmp_path) assert read_taskid("DEVX-42-test") == "DEVX-42" captured = capsys.readouterr() assert "WARNING" not in captured.out # -- extract_task_id (legacy fallback) -- class TestExtractTaskId: def test_extracts_from_branch(self) -> None: assert extract_task_id("DEVX-19-fix-bug") == "DEVX-19" assert extract_task_id("DEVX-123") == "DEVX-123" def test_returns_empty_when_no_match(self) -> None: assert extract_task_id("feature-branch") == "" # -- validate_pr_title -- class TestValidatePrTitle: def test_valid_title(self) -> None: validate_pr_title("DEVX-19: Add new feature", "DEVX-19") def test_missing_colon(self) -> None: with pytest.raises(click.ClickException, match="format"): validate_pr_title("DEVX-19 Add new feature", "DEVX-19") def test_task_id_mismatch(self) -> None: with pytest.raises(click.ClickException, match="mismatch"): validate_pr_title("DEVX-20: Add feature", "DEVX-19") def test_no_task_id_in_title(self) -> None: with pytest.raises(click.ClickException, match="format"): validate_pr_title("Add new feature", "DEVX-19") # -- validate_pr_title_matches_vikunja -- class TestValidatePrTitleMatchesVikunja: @patch.dict("os.environ", {}, clear=True) def test_raises_when_no_token(self) -> None: """Should raise ClickException when VIKUNJA_TOKEN is not set.""" with pytest.raises(click.ClickException, match="VIKUNJA_TOKEN is not set"): validate_pr_title_matches_vikunja("DEVX-19: test", "DEVX-19") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.VikunjaClient") def test_matches(self, mock_client_cls: MagicMock) -> None: mock_client = MagicMock() mock_client.list_project_tasks.return_value = [ {"id": 1, "identifier": "DEVX-19", "title": "Add new feature"}, ] mock_client_cls.return_value = mock_client validate_pr_title_matches_vikunja("DEVX-19: Add new feature", "DEVX-19") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.VikunjaClient") def test_mismatch_raises(self, mock_client_cls: MagicMock) -> None: mock_client = MagicMock() mock_client.list_project_tasks.return_value = [ {"id": 1, "identifier": "DEVX-19", "title": "Different title"}, ] mock_client_cls.return_value = mock_client with pytest.raises(click.ClickException, match="does not match"): validate_pr_title_matches_vikunja("DEVX-19: Add new feature", "DEVX-19") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.VikunjaClient") def test_task_not_found_raises(self, mock_client_cls: MagicMock) -> None: """When Vikunja task is not found and token is set, raises ClickException.""" mock_client = MagicMock() mock_client.list_project_tasks.return_value = [] mock_client_cls.return_value = mock_client with pytest.raises(click.ClickException, match="Could not find"): validate_pr_title_matches_vikunja("DEVX-99: test", "DEVX-99") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.VikunjaClient") def test_task_found_on_second_page(self, mock_client_cls: MagicMock) -> None: """Pagination: task found on page 2.""" mock_client = MagicMock() page1 = [{"id": i, "identifier": f"DEVX-{i}", "title": f"Title {i}"} for i in range(50)] page2 = [{"id": 100, "identifier": "DEVX-99", "title": "Found me"}] mock_client.list_project_tasks.side_effect = [page1, page2] mock_client_cls.return_value = mock_client # Should not raise — title matches validate_pr_title_matches_vikunja("DEVX-99: Found me", "DEVX-99") @patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.VikunjaClient") def test_task_not_found_partial_page_raises(self, mock_client_cls: MagicMock) -> None: """Pagination stops when page has fewer than DEFAULT_PER_PAGE results. Task not found raises.""" mock_client = MagicMock() mock_client.list_project_tasks.return_value = [ {"id": 1, "identifier": "DEVX-1", "title": "Title 1"}, ] mock_client_cls.return_value = mock_client with pytest.raises(click.ClickException, match="Could not find"): validate_pr_title_matches_vikunja("DEVX-99: test", "DEVX-99") # -- extract_conventional_msg -- class TestExtractConventionalMsg: def test_finds_conventional(self) -> None: commits = [ {"commit": {"message": "fix: resolve timeout"}}, {"commit": {"message": "merge branch"}}, ] assert extract_conventional_msg(commits) == "fix: resolve timeout" def test_finds_latest_conventional(self) -> None: commits = [ {"commit": {"message": "merge branch"}}, {"commit": {"message": "feat: add feature"}}, ] assert extract_conventional_msg(commits) == "feat: add feature" def test_falls_back_to_newest(self) -> None: commits = [ {"commit": {"message": "random message"}}, ] assert extract_conventional_msg(commits) == "random message" def test_empty_commits(self) -> None: assert extract_conventional_msg([]) == "" def test_multiline_message(self) -> None: commits = [ {"commit": {"message": "feat: add feature\n\nBody text."}}, ] assert extract_conventional_msg(commits) == "feat: add feature" # -- run_cmd -- class TestRunCmd: def test_success(self) -> None: result = run_cmd(["echo", "hello"]) assert result.returncode == 0 def test_failure_raises(self) -> None: with pytest.raises(click.ClickException, match="Command failed"): run_cmd(["false"]) def test_failure_no_check(self) -> None: result = run_cmd(["false"], check=False) assert result.returncode != 0 # -- main (integration) -- class TestMain: @patch.dict("os.environ", {"REPO_TOKEN": "tok", "VIKUNJA_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("devx.ci.auto_merge.GiteaClient") def test_full_merge_flow( self, mock_client_cls: MagicMock, _mock_validate: MagicMock, tmp_path, monkeypatch ) -> None: # type: ignore[no-untyped-def] monkeypatch.chdir(tmp_path) mock_client = MagicMock() mock_client.get_pr_commits.return_value = [ {"commit": {"message": "fix: resolve timeout"}}, ] mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke( main, ["DEVX-19-fix-bug", "DEVX-19: Fix timeout", "owner/repo", "7"], ) assert result.exit_code == 0, result.output mock_client.merge_pr.assert_called_once_with(7, "DEVX-19: fix: resolve timeout") @patch.dict("os.environ", {"REPO_TOKEN": ""}, clear=True) def test_no_token_raises(self) -> None: runner = CliRunner() result = runner.invoke(main, ["DEVX-19-fix", "DEVX-19: test", "owner/repo", "7"]) assert result.exit_code != 0 assert "REPO_TOKEN" in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.GiteaClient") def test_no_task_id_raises(self, mock_client_cls: MagicMock, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def] monkeypatch.chdir(tmp_path) # No DEVX-N in branch name runner = CliRunner() result = runner.invoke(main, ["feature-branch", "DEVX-19: test", "owner/repo", "7"]) assert result.exit_code != 0 assert "No task ID" in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.GiteaClient") def test_invalid_pr_title_raises(self, mock_client_cls: MagicMock, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def] monkeypatch.chdir(tmp_path) runner = CliRunner() result = runner.invoke(main, ["DEVX-19-fix", "Bad title", "owner/repo", "7"]) assert result.exit_code != 0 assert "format" in result.output.lower() @patch.dict("os.environ", {"REPO_TOKEN": "tok", "VIKUNJA_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("devx.ci.auto_merge.GiteaClient") def test_merge_behind_master_raises_no_rebase( self, mock_client_cls: MagicMock, _mock_validate: MagicMock, tmp_path, monkeypatch ) -> None: # type: ignore[no-untyped-def] """When branch is behind master, auto-merge should NOT rebase. Auto-rebasing creates a feedback loop: the force-push triggers a new pull_request synchronize event, which starts a new CI run, which runs auto-merge again, which rebases again, etc. """ monkeypatch.chdir(tmp_path) mock_client = MagicMock() mock_client.get_pr_commits.return_value = [ {"commit": {"message": "fix: resolve timeout"}}, ] mock_client.merge_pr.side_effect = APIError(405, "HEAD branch is behind master") mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke( main, ["DEVX-19-fix-bug", "DEVX-19: Fix timeout", "owner/repo", "7"], ) assert result.exit_code != 0 assert "behind master" in result.output.lower() assert "rebase manually" in result.output.lower() # Must NOT have called merge_pr twice (no retry after rebase) assert mock_client.merge_pr.call_count == 1 @patch.dict("os.environ", {"REPO_TOKEN": "tok", "VIKUNJA_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("devx.ci.auto_merge.GiteaClient") def test_merge_failure_raises( self, mock_client_cls: MagicMock, _mock_validate: MagicMock, tmp_path, monkeypatch ) -> None: # type: ignore[no-untyped-def] monkeypatch.chdir(tmp_path) mock_client = MagicMock() mock_client.get_pr_commits.return_value = [ {"commit": {"message": "fix: resolve timeout"}}, ] mock_client.merge_pr.side_effect = APIError(409, "Conflict") mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke( main, ["DEVX-19-fix-bug", "DEVX-19: Fix timeout", "owner/repo", "7"], ) assert result.exit_code != 0 assert "Merge failed" in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok", "VIKUNJA_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("devx.ci.auto_merge.GiteaClient") def test_no_conventional_msg_raises( self, mock_client_cls: MagicMock, _mock_validate: MagicMock, tmp_path, monkeypatch ) -> None: # type: ignore[no-untyped-def] """When no conventional commit message is found in PR commits, raises.""" monkeypatch.chdir(tmp_path) mock_client = MagicMock() mock_client.get_pr_commits.return_value = [] mock_client_cls.return_value = mock_client runner = CliRunner() result = runner.invoke( main, ["DEVX-19-fix-bug", "DEVX-19: Fix timeout", "owner/repo", "7"], ) assert result.exit_code != 0 assert "conventional commit" in result.output.lower() @patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True) def test_invalid_pr_number_raises(self, tmp_path, monkeypatch) -> None: """Non-integer PR number should raise.""" monkeypatch.chdir(tmp_path) runner = CliRunner() result = runner.invoke(main, ["DEVX-19-fix", "DEVX-19: Test", "owner/repo", "not-a-number"]) assert result.exit_code != 0 assert "PR number must be an integer" in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True) def test_invalid_repo_format_raises(self, tmp_path, monkeypatch) -> None: """Repo without owner/name should raise.""" monkeypatch.chdir(tmp_path) runner = CliRunner() result = runner.invoke(main, ["DEVX-19-fix", "DEVX-19: Test", "invalidrepo", "7"]) assert result.exit_code != 0 assert "owner/name" in result.output @patch.dict("os.environ", {"REPO_TOKEN": "tok", "VIKUNJA_TOKEN": "tok"}, clear=True) @patch("devx.ci.auto_merge.validate_pr_title_matches_vikunja") @patch("devx.ci.auto_merge.GiteaClient") def test_merge_behind_master_does_not_force_push( self, mock_client_cls: MagicMock, _mock_validate: MagicMock, tmp_path, monkeypatch ) -> None: # type: ignore[no-untyped-def] """Verify no git commands are run when branch is behind master.""" monkeypatch.chdir(tmp_path) mock_client = MagicMock() mock_client.get_pr_commits.return_value = [ {"commit": {"message": "fix: resolve timeout"}}, ] mock_client.merge_pr.side_effect = APIError(405, "HEAD branch is behind master") mock_client_cls.return_value = mock_client with patch("devx.ci.auto_merge.run_cmd") as mock_run: runner = CliRunner() result = runner.invoke( main, ["DEVX-19-fix-bug", "DEVX-19: Fix timeout", "owner/repo", "7"], ) assert result.exit_code != 0 # No git commands should be run (no rebase, no push) mock_run.assert_not_called() def test_main_module_block() -> None: """Test that the __main__ block can be executed.""" import devx.ci.auto_merge as am with open(am.__file__) as f: source = f.read() source = source.replace('if __name__ == "__main__":\n main()\n', "") namespace = dict(am.__dict__) exec(compile(source, am.__file__, "exec"), namespace) # Verify main is callable assert callable(namespace["main"])