Public Access
emil
e3fa9b7c95
Post-merge / detect-type (push) Successful in 10s
Post-merge / validate-commit-msg (push) Successful in 13s
Build Images / detect-type (push) Successful in 43s
Post-merge / vikunja (push) Successful in 16s
Post-merge / sync-wiki (push) Successful in 33s
Post-merge / release (push) Successful in 37s
Post-merge / configure-repo (push) Successful in 13s
Post-merge / badges (push) Successful in 48s
Post-merge / publish (push) Successful in 20s
Build Images / build-and-push (push) Successful in 3m7s
Build Images / cleanup (push) Successful in 3m18s
465 lines
19 KiB
Python
465 lines
19 KiB
Python
"""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._shared import run_cmd
|
|
from devx.ci.auto_merge import (
|
|
extract_conventional_msg,
|
|
extract_task_id,
|
|
main,
|
|
read_taskid,
|
|
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
|
|
|
|
def test_no_warning_when_taskid_file_matches_branch(self, tmp_path, monkeypatch, capsys) -> None: # type: ignore[no-untyped-def]
|
|
"""No warning when .taskid file content matches the branch task ID."""
|
|
monkeypatch.chdir(tmp_path)
|
|
(tmp_path / ".taskid").write_text("DEVX-19\n")
|
|
assert read_taskid("DEVX-19-fix-bug") == "DEVX-19"
|
|
captured = capsys.readouterr()
|
|
assert "WARNING" not in captured.out
|
|
|
|
def test_no_warning_when_taskid_file_empty(self, tmp_path, monkeypatch, capsys) -> None: # type: ignore[no-untyped-def]
|
|
"""No warning when .taskid file exists but is empty."""
|
|
monkeypatch.chdir(tmp_path)
|
|
(tmp_path / ".taskid").write_text("\n")
|
|
assert read_taskid("DEVX-19-fix-bug") == "DEVX-19"
|
|
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"
|
|
|
|
def test_prefers_feat_over_refactor(self) -> None:
|
|
"""When both feat and refactor commits exist, feat wins."""
|
|
commits = [
|
|
{"commit": {"message": "refactor: add find_task_by_identifier"}},
|
|
{"commit": {"message": "fix: remove hardcoded fallbacks"}},
|
|
{"commit": {"message": "feat: add manual review support"}},
|
|
]
|
|
assert extract_conventional_msg(commits) == "feat: add manual review support"
|
|
|
|
def test_prefers_fix_over_docs(self) -> None:
|
|
commits = [
|
|
{"commit": {"message": "docs: update README"}},
|
|
{"commit": {"message": "fix: resolve bug"}},
|
|
]
|
|
assert extract_conventional_msg(commits) == "fix: resolve bug"
|
|
|
|
def test_scope_in_prefix(self) -> None:
|
|
commits = [
|
|
{"commit": {"message": "refactor(ci): cleanup code"}},
|
|
{"commit": {"message": "feat(api): add endpoint"}},
|
|
]
|
|
assert extract_conventional_msg(commits) == "feat(api): add endpoint"
|
|
|
|
def test_strips_task_id_prefix(self) -> None:
|
|
"""Commit messages with a task ID prefix should have it stripped."""
|
|
commits = [
|
|
{"commit": {"message": "DEVX-12: fix: resolve timeout"}},
|
|
]
|
|
assert extract_conventional_msg(commits) == "fix: resolve timeout"
|
|
|
|
def test_strips_task_id_prefix_fallback(self) -> None:
|
|
"""Fallback to newest commit should also strip task ID prefix."""
|
|
commits = [
|
|
{"commit": {"message": "DEVX-12: random message"}},
|
|
]
|
|
assert extract_conventional_msg(commits) == "random message"
|
|
|
|
|
|
# -- 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", {"CI_GITEA_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", {"CI_GITEA_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 "CI_GITEA_TOKEN" in result.output
|
|
|
|
@patch.dict("os.environ", {"CI_GITEA_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", {"CI_GITEA_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", {"CI_GITEA_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_auto_rebases(
|
|
self, mock_client_cls: MagicMock, _mock_validate: MagicMock, tmp_path, monkeypatch
|
|
) -> None: # type: ignore[no-untyped-def]
|
|
"""When branch is behind master, auto-merge rebases via Gitea API.
|
|
|
|
The rebase triggers a new CI run. The next auto-merge attempt will
|
|
find the branch up-to-date and merge successfully.
|
|
"""
|
|
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 "auto-rebasing" in result.output.lower()
|
|
# Should have called update_pr_branch to trigger server-side rebase
|
|
mock_client.update_pr_branch.assert_called_once_with(7, style="rebase")
|
|
# Must NOT have called merge_pr twice (no immediate retry)
|
|
assert mock_client.merge_pr.call_count == 1
|
|
|
|
@patch.dict("os.environ", {"CI_GITEA_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_rebase_failure_raises(
|
|
self, mock_client_cls: MagicMock, _mock_validate: MagicMock, tmp_path, monkeypatch
|
|
) -> None: # type: ignore[no-untyped-def]
|
|
"""When auto-rebase fails, raise with manual rebase instructions."""
|
|
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.update_pr_branch.side_effect = APIError(409, "Conflict during rebase")
|
|
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 "auto-rebase failed" in result.output.lower()
|
|
assert "rebase manually" in result.output.lower()
|
|
|
|
@patch.dict("os.environ", {"CI_GITEA_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", {"CI_GITEA_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", {"CI_GITEA_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", {"CI_GITEA_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", {"CI_GITEA_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_run_git_commands(
|
|
self, mock_client_cls: MagicMock, _mock_validate: MagicMock, tmp_path, monkeypatch
|
|
) -> None: # type: ignore[no-untyped-def]
|
|
"""When behind master, auto-merge uses API rebase — no local git commands."""
|
|
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._shared.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 local git commands should be run (rebase is via API)
|
|
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"])
|