GRM-35: feat: fix 12 critical workflow gaps in release pipeline

Addresses all 12 critical gaps in the automated semantic versioning, tagging, and release workflow.

Closes GRM-35
This commit is contained in:
2026-06-21 17:44:27 +00:00
parent ea6276ff38
commit 996a8dc806
14 changed files with 808 additions and 143 deletions
+1
View File
@@ -14,6 +14,7 @@ jobs:
- name: Squash merge with task ID
env:
REPO_TOKEN: ${{ secrets.REPO_TOKEN }}
VIKUNJA_TOKEN: ${{ secrets.VIKUNJA_TOKEN }}
PYTHONPATH: src
run: |
python3 scripts/auto_merge.py \
+34 -1
View File
@@ -26,9 +26,42 @@ jobs:
run: |
. .venv/bin/activate
python3 scripts/check_test_speed.py --max-seconds 10
- name: Release dry-run validation
run: |
. .venv/bin/activate
PYTHONPATH=src python3 scripts/release.py --dry-run || true
detect-changes:
runs-on: docker
outputs:
ansible-changed: ${{ steps.detect.outputs.ansible-changed }}
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Detect changed paths
id: detect
run: |
if [ "${{ github.event_name }}" = "pull_request" ]; then
BASE="origin/master"
HEAD="${{ github.event.pull_request.head.sha }}"
else
BASE="HEAD~1"
HEAD="HEAD"
fi
# Check if any Ansible-related files changed
ANSIBLE_CHANGED=$(git diff --name-only "$BASE" "$HEAD" -- ansible/ .ansible-lint 2>/dev/null | head -1)
if [ -n "$ANSIBLE_CHANGED" ]; then
echo "ansible-changed=true" >> "$GITHUB_OUTPUT"
echo "Ansible files changed — molecule tests will run."
else
echo "ansible-changed=false" >> "$GITHUB_OUTPUT"
echo "No Ansible files changed — skipping molecule tests."
fi
molecule-tests:
needs: quality
needs: [quality, detect-changes]
if: needs.detect-changes.outputs.ansible-changed == 'true'
runs-on: docker
strategy:
matrix:
+16
View File
@@ -26,6 +26,11 @@ jobs:
- name: Install build tools
run: |
python3 -m pip install --break-system-packages build twine requests python-dotenv click
- name: Validate PYPI_TOKEN
run: |
if [ -z "${{ secrets.PYPI_TOKEN }}" ]; then
echo "::warning::PYPI_TOKEN is not set — package will be built but not published to PyPI."
fi
- name: Build and publish release
env:
REPO_TOKEN: ${{ secrets.REPO_TOKEN }}
@@ -35,3 +40,14 @@ jobs:
python3 scripts/publish.py \
"${{ github.ref_name }}" \
"${{ github.repository }}"
- name: Notify on failure
if: failure()
env:
REPO_TOKEN: ${{ secrets.REPO_TOKEN }}
PYTHONPATH: src
run: |
python3 scripts/notify_failure.py \
--repo "${{ github.repository }}" \
--run-id "${{ github.run_id }}" \
--workflow "publish" \
--commit "${{ github.sha }}"
+11
View File
@@ -34,3 +34,14 @@ jobs:
PYTHONPATH: src
run: |
python3 scripts/release.py
- name: Notify on failure
if: failure()
env:
REPO_TOKEN: ${{ secrets.REPO_TOKEN }}
PYTHONPATH: src
run: |
python3 scripts/notify_failure.py \
--repo "${{ github.repository }}" \
--run-id "${{ github.run_id }}" \
--workflow "release" \
--commit "${{ github.sha }}"
+43 -10
View File
@@ -16,13 +16,24 @@ make test-all # pytest-cov + molecule
- **Python CLI** (`src/gitea_runner_manager/`) — Click-based CLI that delegates to Ansible
- **Ansible Role** (`ansible/roles/gitea-runner/`) — Idempotent role for rootless Docker runner setup
- **CI Scripts** (`scripts/`) — Automation for auto-merge, post-merge, release, publishing, molecule distribution, PR reviews
- **CI Scripts** (`scripts/`) — Automation for auto-merge, post-merge, release, publishing, molecule distribution, PR reviews, failure notifications
- **Versioning** (`cliff.toml`) — git-cliff configuration for automated semver versioning from conventional commits
## PR Workflow (Mandatory)
Every change to master goes through this workflow. No exceptions.
### Branch Protection (Required Gitea Settings)
Configure the following branch protection rules for `master` in Gitea repo settings:
- **Require pull request**: No direct pushes to master
- **Require approval review**: At least 1 `APPROVE` review before merge
- **Require status checks**: CI quality + molecule tests must pass
- **Block force pushes**: No history rewriting on master
The auto-merge workflow enforces the APPROVE review check programmatically
as a defense-in-depth measure, but branch protection is the primary gate.
### 1. Create Vikunja Task
Create a task in Vikunja project 6 to get a `GRM-N` identifier.
@@ -46,7 +57,7 @@ docs: update README
```
### 5. Push and Create PR
- **PR title format**: `GRM-N: <vikunja task title>` (colon-separated)
- **PR title format**: `GRM-N: <vikunja task title>` (must match the Vikunja task title exactly)
- PR body: summary of changes, `Closes GRM-N`
- Add `ready-to-merge` label **only after review is complete**
@@ -92,29 +103,51 @@ REPO_TOKEN=<token> python3 scripts/review_pr.py <pr_number> <owner/repo> \
```
Then add the `ready-to-merge` label. The auto-merge workflow will:
1. Wait for all CI checks to pass
2. Squash-merge with title: `GRM-N <conventional commit message>` (space-separated)
3. The post-merge workflow marks the Vikunja task as done
4. The release workflow automatically versions, tags, and publishes (see below)
1. **Validate** PR title format and match against Vikunja task title
2. **Check** that at least one APPROVE review exists
3. Wait for all CI checks to pass
4. Squash-merge with title: `GRM-N <conventional commit message>` (space-separated)
5. The post-merge workflow marks the Vikunja task as done
6. The release workflow automatically versions, tags, and publishes (see below)
### CI Path Filtering
The CI workflow includes a `detect-changes` job that checks whether any files
under `ansible/` or `.ansible-lint` have changed. If no Ansible files are
changed, molecule tests are skipped — this prevents non-Ansible changes
(e.g., Python scripts, workflow YAML, docs) from being blocked by molecule
test infrastructure flakiness.
### Automated Release Pipeline
After a PR is merged to master, the release pipeline runs automatically:
1. **Release workflow** (`.gitea/workflows/release.yml`):
- Triggers on push to master (skips `chore(release):` commits to avoid loops)
- Triggers on push to master
- Runs `scripts/release.py` which uses **git-cliff** to:
- Calculate the next semver version from conventional commits since the last tag
- Update `__version__` in `src/gitea_runner_manager/__init__.py` (single source of truth)
- Create a `chore(release): prepare for vX.Y.Z` commit
- Create an annotated tag `vX.Y.Z` with the changelog as the tag message
- Push the commit and tag to master
- Update `CHANGELOG.md` with the new version section
- Commit with `release: vX.Y.Z` prefix (cleaner than `chore(release):`)
- Create an annotated tag `vX.Y.Z` on the release commit
- Push both the commit and tag to master
- Loops are prevented by `has_unreleased_changes` — after a release commit is tagged, the next run finds no unreleased changes and exits
- On failure, creates a Gitea issue via `scripts/notify_failure.py`
2. **Publish workflow** (`.gitea/workflows/publish.yml`):
- Triggers on tag push (`v*`)
- Validates `PYPI_TOKEN` is set (warns if missing)
- Builds the Python package
- Optionally publishes to PyPI (if `PYPI_TOKEN` is set)
- Creates a Gitea release with git-cliff-generated release notes
- On failure, creates a Gitea issue via `scripts/notify_failure.py`
### git-cliff Commit Preprocessing
Merge commits on master have the format `GRM-N <conventional commit>`. The
`GRM-N ` prefix is not a valid conventional commit prefix, so `cliff.toml`
includes a `commit_preprocessors` entry that strips it before parsing. This
ensures all merged work appears in the changelog.
### Version Bumping Rules (git-cliff)
+5
View File
@@ -38,6 +38,11 @@ topo_order_commits = true
sort_commits = "oldest"
recurse_submodules = false
commit_preprocessors = [
# Strip GRM-N task ID prefix from merge commits so git-cliff sees conventional commits
{ pattern = "^GRM-\\d+\\s+", replace = "" },
]
commit_parsers = [
{ message = "^feat", group = "<!-- 0 -->Features" },
{ message = "^fix", group = "<!-- 1 -->Bug Fixes" },
+75 -18
View File
@@ -22,8 +22,15 @@ from typing import Any
import click
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
from gitea_runner_manager.api_clients import GiteaClient
from gitea_runner_manager.config import CONVENTIONAL_RE, GITEA_API_URL, TASK_ID_RE
from gitea_runner_manager.api_clients import GiteaClient, VikunjaClient
from gitea_runner_manager.config import (
CONVENTIONAL_RE,
DEFAULT_PER_PAGE,
GITEA_API_URL,
TASK_ID_RE,
VIKUNJA_API_URL,
VIKUNJA_PROJECT_ID,
)
from gitea_runner_manager.exceptions import APIError
from gitea_runner_manager.i18n import _
@@ -61,15 +68,62 @@ def validate_pr_title(pr_title: str, task_id: str) -> None:
if not pr_title.startswith(f"{task_id}:"):
raise click.ClickException(
_(
"Oops! PR title task ID mismatch.\n"
" Branch task ID: {task_id}\n"
" PR title: {pr_title}",
"Oops! PR title task ID mismatch.\n Branch task ID: {task_id}\n PR title: {pr_title}",
task_id=task_id,
pr_title=pr_title,
)
)
def get_vikunja_task_title(task_id: str) -> str:
"""Fetch the Vikunja task title for the given GRM-N identifier.
Returns empty string if VIKUNJA_TOKEN is not set (skip validation).
"""
token = os.environ.get("VIKUNJA_TOKEN", "")
if not token:
return ""
client = VikunjaClient(VIKUNJA_API_URL, token)
page = 1
while True:
tasks = client.list_project_tasks(VIKUNJA_PROJECT_ID, page=page, per_page=DEFAULT_PER_PAGE)
if not tasks:
break
matches = [t for t in tasks if t.get("identifier") == task_id]
if matches:
return str(matches[0].get("title", ""))
if len(tasks) < DEFAULT_PER_PAGE:
break
page += 1
return ""
def validate_pr_title_matches_vikunja(pr_title: str, task_id: str) -> None:
"""Validate that PR title matches the Vikunja task title.
Skips validation if VIKUNJA_TOKEN is not set.
"""
vikunja_title = get_vikunja_task_title(task_id)
if not vikunja_title:
click.echo(_("Warning: could not fetch Vikunja task title, skipping title match validation."))
return
expected = f"{task_id}: {vikunja_title}"
if pr_title != expected:
raise click.ClickException(
_(
"PR title does not match Vikunja task title.\n Expected: {expected}\n Got: {pr_title}",
expected=expected,
pr_title=pr_title,
)
)
def has_approval_review(client: GiteaClient, pr_number: str) -> bool:
"""Check whether the PR has at least one APPROVE review."""
reviews = client.get_pr_reviews(pr_number)
return any(r.get("state") == "APPROVED" for r in reviews)
def extract_conventional_msg(commits: list[dict[str, Any]]) -> str:
"""Extract the conventional commit message from PR commits.
@@ -128,15 +182,9 @@ def wait_for_ci(
pending = [ctx for ctx, s in ci_statuses.items() if s.get("status") in ("pending", "waiting")]
if not pending:
# All CI checks are complete — check if they all succeeded.
failed = [
ctx
for ctx, s in ci_statuses.items()
if s.get("status") not in ("success", "ok")
]
failed = [ctx for ctx, s in ci_statuses.items() if s.get("status") not in ("success", "ok")]
if failed:
click.echo(
_("CI checks failed: {failed}", failed=", ".join(sorted(failed)))
)
click.echo(_("CI checks failed: {failed}", failed=", ".join(sorted(failed))))
return False
click.echo(_("All CI checks passed."))
return True
@@ -184,6 +232,18 @@ def main(branch: str, pr_title: str, repo: str, pr_number: str, label_name: str)
)
validate_pr_title(pr_title, task_id)
validate_pr_title_matches_vikunja(pr_title, task_id)
# Enforce APPROVE review before merge (Gap 2 fix)
if not has_approval_review(client, pr_number):
raise click.ClickException(
_(
"Cannot merge: PR #{pr_number} has no APPROVE review. "
"Please review and approve before adding the ready-to-merge label.",
pr_number=pr_number,
)
)
click.echo(_("PR has at least one APPROVE review."))
# Wait for CI checks to complete before attempting merge.
pr = client.get_pr(pr_number)
@@ -201,9 +261,7 @@ def main(branch: str, pr_title: str, repo: str, pr_number: str, label_name: str)
commits = client.get_pr_commits(pr_number)
conv_msg = extract_conventional_msg(commits)
if not conv_msg:
raise click.ClickException(
_("Could not extract conventional commit message from PR commits.")
)
raise click.ClickException(_("Could not extract conventional commit message from PR commits."))
merge_title = f"{task_id} {conv_msg}"
try:
@@ -211,8 +269,7 @@ def main(branch: str, pr_title: str, repo: str, pr_number: str, label_name: str)
except APIError as e:
raise click.ClickException(
_(
"Merge failed with HTTP {status}: {message}\n"
"Please check the PR is ready and you have merge rights.",
"Merge failed with HTTP {status}: {message}\nPlease check the PR is ready and you have merge rights.",
status=e.status,
message=e.message,
)
+74
View File
@@ -0,0 +1,74 @@
#!/usr/bin/env python3
"""Create a Gitea issue when a CI workflow fails.
Used by the release and publish workflows to alert on failures that would
otherwise go unnoticed in the Actions tab.
Usage:
REPO_TOKEN=<token> python3 scripts/notify_failure.py \
--repo <owner/repo> \
--run-id <run_id> \
--workflow <workflow_name> \
--commit <commit_sha>
"""
from __future__ import annotations
import os
import click
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
from gitea_runner_manager.api_clients import GiteaClient
from gitea_runner_manager.config import GITEA_API_URL
from gitea_runner_manager.exceptions import APIError
from gitea_runner_manager.i18n import _
load_dotenv(override=True)
@click.command()
@click.option("--repo", required=True, help="Repository in owner/name format.")
@click.option("--run-id", required=True, help="CI run ID.")
@click.option("--workflow", required=True, help="Workflow name.")
@click.option("--commit", required=True, help="Commit SHA.")
def main(repo: str, run_id: str, workflow: str, commit: str) -> None:
token = os.environ.get("REPO_TOKEN", "")
if not token:
raise click.ClickException(_("ERROR: REPO_TOKEN is not set."))
owner, repo_name = repo.split("/")
client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
title = f"[CI] {workflow} workflow failed (run #{run_id})"
body = (
f"The **{workflow}** workflow failed.\n\n"
f"- **Run ID**: #{run_id}\n"
f"- **Commit**: `{commit[:8]}`\n"
f"- **Check the logs**: {GITEA_API_URL.replace('/api/v1', '')}/"
f"{repo}/actions/runs/{run_id}\n\n"
f"Please investigate and fix the issue."
)
try:
issue = client.create_issue(title=title, body=body, labels=["bug"])
except APIError as e:
# If labels don't exist, retry without labels
if e.status == 404:
issue = client.create_issue(title=title, body=body)
else:
raise click.ClickException(
_("Failed to create issue: HTTP {status}{message}", status=e.status, message=e.message)
) from None
click.echo(
_(
"Created issue #{issue_id}: {title}",
issue_id=issue.get("id", "?"),
title=title,
)
)
if __name__ == "__main__": # pragma: no cover
main()
+112 -49
View File
@@ -3,12 +3,19 @@
Uses git-cliff to determine the next semver version from conventional commits
since the last tag. Updates ``__version__`` in ``__init__.py`` (the single
source of truth, read by setuptools via ``dynamic = ["version"]``), creates a
release commit, tags it with the changelog as the tag message, and pushes the
tag to trigger the publish workflow.
source of truth, read by setuptools via ``dynamic = ["version"]``) and
``CHANGELOG.md``, commits them with a ``release:`` prefix, tags the commit
with the changelog as the tag message, and pushes both to trigger the publish
workflow.
The ``release:`` prefix (instead of ``chore(release):``) keeps the history
clean while still being descriptive. Loops are prevented by the
``has_unreleased_changes`` check — after a release commit is tagged, the next
run finds no unreleased changes and exits.
This script is idempotent: if there are no new conventional commits since the
last tag, it exits with a message and does nothing.
last tag, it exits with a message and does nothing. If the tag already exists
(e.g., from a partial previous run), it skips tag creation and only pushes.
Usage:
REPO_TOKEN=<token> python3 scripts/release.py [--dry-run]
@@ -27,6 +34,7 @@ from gitea_runner_manager.i18n import _
load_dotenv(override=True)
INIT_FILE = "src/gitea_runner_manager/__init__.py"
CHANGELOG_FILE = "CHANGELOG.md"
CLIFF_CONFIG = "cliff.toml"
@@ -57,6 +65,12 @@ def get_latest_tag() -> str:
return result.stdout.strip()
def tag_exists(tag: str) -> bool:
"""Check if a git tag already exists."""
result = run_cmd(["git", "tag", "-l", tag], check=False)
return bool(result.stdout.strip())
def get_bumped_version() -> str:
"""Use git-cliff to calculate the next version from conventional commits."""
result = run_cmd(["git-cliff", "--bumped-version", "--config", CLIFF_CONFIG])
@@ -83,20 +97,25 @@ def get_changelog(new_version: str) -> str:
return result.stdout.strip()
def has_unreleased_changes() -> bool:
"""Check if there are conventional commits since the last tag."""
result = run_cmd(
["git-cliff", "--bumped-version", "--config", CLIFF_CONFIG],
check=False,
)
if result.returncode != 0:
return False
def has_unreleased_changes(bumped_version: str | None = None) -> bool:
"""Check if there are conventional commits since the last tag.
If ``bumped_version`` is provided (from a prior git-cliff call), reuses it
to avoid a duplicate subprocess invocation.
"""
if bumped_version is None:
result = run_cmd(
["git-cliff", "--bumped-version", "--config", CLIFF_CONFIG],
check=False,
)
if result.returncode != 0:
return False
bumped_version = result.stdout.strip().lstrip("v")
latest = get_latest_tag()
if not latest:
return True
bumped = result.stdout.strip().lstrip("v")
current = latest.lstrip("v")
return bumped != current
return bumped_version != current
def update_init_version(new_version: str) -> None:
@@ -116,32 +135,68 @@ def update_init_version(new_version: str) -> None:
f.write(updated)
def create_release_commit(new_version: str) -> bool:
"""Stage version file and create a release commit.
def update_changelog(changelog: str) -> None:
"""Prepend the new changelog section to CHANGELOG.md.
Returns True if a commit was created, False if there were no changes
(e.g., version file already at the target version).
If the file doesn't exist, create it with the changelog as the sole content.
If it exists, insert the new version section after the header (before the
first existing version section).
"""
run_cmd(["git", "add", INIT_FILE])
# Check if there are staged changes
try:
with open(CHANGELOG_FILE) as f:
existing = f.read()
except FileNotFoundError:
with open(CHANGELOG_FILE, "w") as f:
f.write(changelog + "\n")
return
# Find the first version section header (## [...] or ## [unreleased])
match = re.search(r"^## \[", existing, flags=re.MULTILINE)
if match:
# Insert before the first version section
pos = match.start()
updated = existing[:pos] + changelog + "\n\n" + existing[pos:]
else:
# No version sections found — append
updated = existing.rstrip() + "\n\n" + changelog + "\n"
with open(CHANGELOG_FILE, "w") as f:
f.write(updated)
def commit_release_changes(new_version: str) -> bool:
"""Stage version file and changelog, then create a release commit.
Uses ``release:`` prefix (not ``chore(release):``) for clarity.
Returns True if a commit was created, False if there were no staged changes.
"""
run_cmd(["git", "add", INIT_FILE, CHANGELOG_FILE])
status = run_cmd(["git", "diff", "--cached", "--quiet"], check=False)
if status.returncode == 0:
# No staged changes — version file already at target
click.echo(_("Version file already at v{version}, skipping commit.", version=new_version))
click.echo(_("No staged changes — version and changelog already up to date."))
return False
run_cmd(["git", "commit", "-m", f"chore(release): prepare for v{new_version}"])
run_cmd(["git", "commit", "-m", f"release: v{new_version}"])
return True
def create_and_push_tag(new_version: str, changelog: str, dry_run: bool) -> None:
"""Create an annotated tag with the changelog as message and push it."""
def create_and_push_tag(new_version: str, changelog: str, dry_run: bool) -> bool:
"""Create an annotated tag with the changelog as message and push it.
Returns True if the tag was created/pushed, False if it already existed.
"""
tag = f"v{new_version}"
if tag_exists(tag):
click.echo(_("Tag {tag} already exists, skipping creation.", tag=tag))
if not dry_run:
# Ensure the existing tag is pushed
run_cmd(["git", "push", "origin", tag], check=False)
return False
tag_msg = f"Release v{new_version}\n\n{changelog}"
run_cmd(["git", "tag", "-a", tag, "-m", tag_msg])
if dry_run:
click.echo(_("[dry-run] Would push tag {tag}", tag=tag))
return
click.echo(_("[dry-run] Would create tag: {tag}", tag=tag))
return True
run_cmd(["git", "tag", "-a", tag, "-m", tag_msg])
run_cmd(["git", "push", "origin", tag])
return True
@click.command()
@@ -152,19 +207,14 @@ def main(dry_run: bool) -> None:
if branch != "master":
raise click.ClickException(_("Release must be run on master, currently on '{branch}'.", branch=branch))
# Skip release commits to avoid infinite loops
last_msg = run_cmd(["git", "log", "-1", "--pretty=%B"]).stdout.strip()
if last_msg.startswith("chore(release):"):
click.echo(_("Last commit is a release commit. Nothing to do."))
return
# Calculate next version (single git-cliff call — Gap 7 fix)
new_version = get_bumped_version()
# Check for unreleased changes
if not has_unreleased_changes():
# Check for unreleased changes (reuses the version we just calculated)
if not has_unreleased_changes(bumped_version=new_version):
click.echo(_("No unreleased changes found. Nothing to release."))
return
# Calculate next version
new_version = get_bumped_version()
current_tag = get_latest_tag()
click.echo(
_(
@@ -182,7 +232,9 @@ def main(dry_run: bool) -> None:
if dry_run:
click.echo(_("\n[dry-run] Changelog:\n{changelog}", changelog=changelog))
click.echo(_("[dry-run] Would update {init}", init=INIT_FILE))
click.echo(_("[dry-run] Would create commit: chore(release): prepare for v{version}", version=new_version))
click.echo(_("[dry-run] Would update {changelog_file}", changelog_file=CHANGELOG_FILE))
click.echo(_("[dry-run] Would commit: release: v{version}", version=new_version))
click.echo(_("[dry-run] Would push commit to master"))
click.echo(_("[dry-run] Would create tag: v{version}", version=new_version))
return
@@ -190,24 +242,35 @@ def main(dry_run: bool) -> None:
update_init_version(new_version)
click.echo(_("Updated version in {init}", init=INIT_FILE))
# Create release commit (may be skipped if version unchanged)
committed = create_release_commit(new_version)
# Update CHANGELOG.md (Gap 3 fix)
update_changelog(changelog)
click.echo(_("Updated {changelog_file}", changelog_file=CHANGELOG_FILE))
# Commit version + changelog (Gap 11: use 'release:' prefix, not 'chore(release):')
committed = commit_release_changes(new_version)
if committed:
click.echo(_("Created release commit."))
# Push commit to master
run_cmd(["git", "push", "origin", "master"])
click.echo(_("Pushed release commit to master."))
else:
click.echo(_("Skipping commit push — version unchanged."))
click.echo(_("Skipping commit push — no staged changes."))
# Create and push tag
create_and_push_tag(new_version, changelog, dry_run)
click.echo(
_(
"Nice! Release v{version} tagged and pushed. The publish workflow will be triggered.",
version=new_version,
# Create and push tag (Gap 4: handles existing tag)
created = create_and_push_tag(new_version, changelog, dry_run)
if created:
click.echo(
_(
"Nice! Release v{version} tagged and pushed. The publish workflow will be triggered.",
version=new_version,
)
)
else:
click.echo(
_(
"Tag v{version} already existed. Publish workflow should already have been triggered.",
version=new_version,
)
)
)
if __name__ == "__main__": # pragma: no cover
+13
View File
@@ -105,6 +105,14 @@ class GiteaClient:
return None
return self.create_label(name, color, description)
def create_issue(self, title: str, body: str = "", labels: list[str] | None = None) -> dict[str, Any]:
"""Create a new issue in the repository."""
payload: dict[str, Any] = {"title": title, "body": body}
if labels:
payload["labels"] = labels
r = self._request("POST", "/issues", json=payload)
return r.json()
# -- pulls / releases --
def get_pr_labels(self, pr_number: str | int) -> list[dict[str, Any]]:
@@ -136,6 +144,11 @@ class GiteaClient:
r = self._request("GET", f"/pulls/{pr_number}/commits")
return r.json()
def get_pr_reviews(self, pr_number: str | int) -> list[dict[str, Any]]:
"""Fetch reviews posted on a pull request."""
r = self._request("GET", f"/pulls/{pr_number}/reviews")
return r.json()
def create_review(
self,
pr_number: str | int,
+39
View File
@@ -258,6 +258,45 @@ class TestGiteaClient:
timeout=DEFAULT_TIMEOUT,
)
def test_get_pr_reviews(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response([{"id": 1, "state": "APPROVED"}]))
result = client.get_pr_reviews(7)
assert len(result) == 1
assert result[0]["state"] == "APPROVED"
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/pulls/7/reviews",
timeout=DEFAULT_TIMEOUT,
)
def test_create_issue(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 42, "title": "bug"}))
result = client.create_issue(title="bug", body="description", labels=["bug"])
assert result["id"] == 42
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/issues",
json={"title": "bug", "body": "description", "labels": ["bug"]},
timeout=DEFAULT_TIMEOUT,
)
def test_create_issue_no_labels(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 43, "title": "bug"}))
result = client.create_issue(title="bug", body="description")
assert result["id"] == 43
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/issues",
json={"title": "bug", "body": "description"},
timeout=DEFAULT_TIMEOUT,
)
def test_create_review_comment(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 42}))
+162 -9
View File
@@ -13,9 +13,11 @@ from scripts.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,
)
@@ -142,6 +144,104 @@ class TestHasReadyToMergeLabel:
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.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.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.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.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.auto_merge import get_vikunja_task_title
assert get_vikunja_task_title("GRM-19") == ""
@patch("scripts.auto_merge.VikunjaClient")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
def test_finds_task(self, mock_client_cls: MagicMock) -> None:
from scripts.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.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.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.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.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.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.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()
@@ -225,8 +325,12 @@ def _mock_commits() -> list[dict[str, dict[str, str]]]:
class TestMain:
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.auto_merge.has_approval_review", return_value=True)
@patch("scripts.auto_merge.GiteaClient")
def test_successful_flow_with_label_arg(self, mock_client_cls: MagicMock) -> None:
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()
@@ -242,8 +346,12 @@ class TestMain:
mock_client.merge_pr.assert_called_once_with("7", "GRM-19 fix: resolve timeout")
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.auto_merge.has_approval_review", return_value=True)
@patch("scripts.auto_merge.GiteaClient")
def test_successful_flow_label_fallback(self, mock_client_cls: MagicMock) -> None:
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"}]
@@ -261,8 +369,12 @@ class TestMain:
mock_client.merge_pr.assert_called_once_with("7", "GRM-19 fix: resolve timeout")
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.auto_merge.has_approval_review", return_value=True)
@patch("scripts.auto_merge.GiteaClient")
def test_wrong_label_skips_merge(self, mock_client_cls: MagicMock) -> None:
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"}]
@@ -277,8 +389,12 @@ class TestMain:
mock_client.merge_pr.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.auto_merge.has_approval_review", return_value=True)
@patch("scripts.auto_merge.GiteaClient")
def test_empty_label_falls_back_to_api(self, mock_client_cls: MagicMock) -> None:
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"}]
@@ -336,8 +452,29 @@ class TestMain:
assert "mismatch" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.auto_merge.has_approval_review", return_value=False)
@patch("scripts.auto_merge.GiteaClient")
def test_empty_commits_exits(self, mock_client_cls: MagicMock) -> None:
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.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.auto_merge.has_approval_review", return_value=True)
@patch("scripts.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"}]
@@ -352,8 +489,12 @@ class TestMain:
mock_client.merge_pr.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.auto_merge.has_approval_review", return_value=True)
@patch("scripts.auto_merge.GiteaClient")
def test_merge_pr_failure_raises_click(self, mock_client_cls: MagicMock) -> None:
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()
@@ -367,8 +508,12 @@ class TestMain:
assert "HTTP" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.auto_merge.has_approval_review", return_value=True)
@patch("scripts.auto_merge.GiteaClient")
def test_merge_pr_json_parse_failure(self, mock_client_cls: MagicMock) -> None:
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()
@@ -382,8 +527,12 @@ class TestMain:
assert str(http.HTTPStatus.BAD_GATEWAY) in result.output
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.auto_merge.has_approval_review", return_value=True)
@patch("scripts.auto_merge.GiteaClient")
def test_ci_failure_blocks_merge(self, mock_client_cls: MagicMock) -> None:
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"}]
@@ -399,8 +548,12 @@ class TestMain:
mock_client.merge_pr.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.auto_merge.has_approval_review", return_value=True)
@patch("scripts.auto_merge.GiteaClient")
def test_no_sha_proceeds_without_wait(self, mock_client_cls: MagicMock) -> None:
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"}]
+104
View File
@@ -0,0 +1,104 @@
"""Unit tests for scripts/notify_failure.py."""
import http
from unittest.mock import MagicMock, patch
from click.testing import CliRunner
from gitea_runner_manager.exceptions import APIError
from scripts.notify_failure import main
class TestNotifyFailure:
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.notify_failure.GiteaClient")
def test_creates_issue_with_labels(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.create_issue.return_value = {"id": 42}
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
[
"--repo",
"owner/repo",
"--run-id",
"123",
"--workflow",
"release",
"--commit",
"abc123def456",
],
)
assert result.exit_code == 0
assert "issue #42" in result.output
mock_client.create_issue.assert_called_once()
call_kwargs = mock_client.create_issue.call_args
assert "release" in call_kwargs.kwargs["title"]
assert call_kwargs.kwargs["labels"] == ["bug"]
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.notify_failure.GiteaClient")
def test_creates_issue_without_labels_on_404(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.create_issue.side_effect = [
APIError(http.HTTPStatus.NOT_FOUND, "label not found"),
{"id": 43},
]
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
[
"--repo",
"owner/repo",
"--run-id",
"124",
"--workflow",
"publish",
"--commit",
"def789",
],
)
assert result.exit_code == 0
assert "issue #43" in result.output
assert mock_client.create_issue.call_count == 2
# Second call should not have labels
second_call = mock_client.create_issue.call_args_list[1]
assert "labels" not in second_call.kwargs or second_call.kwargs.get("labels") is None
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.notify_failure.GiteaClient")
def test_api_error_raises(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.create_issue.side_effect = APIError(http.HTTPStatus.FORBIDDEN, "forbidden")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
[
"--repo",
"owner/repo",
"--run-id",
"125",
"--workflow",
"release",
"--commit",
"abc",
],
)
assert result.exit_code == 1
assert "403" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": ""}, clear=True)
def test_missing_token_exits(self) -> None:
runner = CliRunner()
result = runner.invoke(
main,
["--repo", "owner/repo", "--run-id", "1", "--workflow", "release", "--commit", "abc"],
)
assert result.exit_code == 1
assert "REPO_TOKEN" in result.output
+119 -56
View File
@@ -7,14 +7,16 @@ import pytest
from click.testing import CliRunner
from scripts.release import (
commit_release_changes,
create_and_push_tag,
create_release_commit,
get_bumped_version,
get_changelog,
get_latest_tag,
has_unreleased_changes,
main,
run_cmd,
tag_exists,
update_changelog,
update_init_version,
)
@@ -52,6 +54,18 @@ class TestGetLatestTag:
assert get_latest_tag() == ""
class TestTagExists:
@patch("scripts.release.run_cmd")
def test_exists(self, mock_run_cmd: MagicMock) -> None:
mock_run_cmd.return_value = MagicMock(returncode=0, stdout="v0.1.0\n")
assert tag_exists("v0.1.0") is True
@patch("scripts.release.run_cmd")
def test_not_exists(self, mock_run_cmd: MagicMock) -> None:
mock_run_cmd.return_value = MagicMock(returncode=0, stdout="")
assert tag_exists("v0.2.0") is False
class TestGetBumpedVersion:
@patch("scripts.release.run_cmd")
def test_returns_version(self, mock_run_cmd: MagicMock) -> None:
@@ -83,27 +97,35 @@ class TestGetChangelog:
class TestHasUnreleasedChanges:
@patch("scripts.release.get_latest_tag")
def test_with_bumped_version_no_changes(self, mock_latest: MagicMock) -> None:
mock_latest.return_value = "v0.2.0"
assert has_unreleased_changes(bumped_version="0.2.0") is False
@patch("scripts.release.get_latest_tag")
def test_with_bumped_version_has_changes(self, mock_latest: MagicMock) -> None:
mock_latest.return_value = "v0.2.0"
assert has_unreleased_changes(bumped_version="0.3.0") is True
@patch("scripts.release.get_latest_tag")
def test_with_bumped_version_no_tags(self, mock_latest: MagicMock) -> None:
mock_latest.return_value = ""
assert has_unreleased_changes(bumped_version="0.1.0") is True
@patch("scripts.release.get_latest_tag")
@patch("scripts.release.run_cmd")
def test_no_changes(self, mock_run_cmd: MagicMock, mock_latest: MagicMock) -> None:
def test_without_bumped_version_no_changes(self, mock_run_cmd: MagicMock, mock_latest: MagicMock) -> None:
mock_run_cmd.return_value = MagicMock(returncode=0, stdout="v0.2.0\n")
mock_latest.return_value = "v0.2.0"
assert has_unreleased_changes() is False
@patch("scripts.release.get_latest_tag")
@patch("scripts.release.run_cmd")
def test_has_changes(self, mock_run_cmd: MagicMock, mock_latest: MagicMock) -> None:
def test_without_bumped_version_has_changes(self, mock_run_cmd: MagicMock, mock_latest: MagicMock) -> None:
mock_run_cmd.return_value = MagicMock(returncode=0, stdout="v0.3.0\n")
mock_latest.return_value = "v0.2.0"
assert has_unreleased_changes() is True
@patch("scripts.release.get_latest_tag")
@patch("scripts.release.run_cmd")
def test_no_tags_returns_true(self, mock_run_cmd: MagicMock, mock_latest: MagicMock) -> None:
mock_run_cmd.return_value = MagicMock(returncode=0, stdout="v0.2.0\n")
mock_latest.return_value = ""
assert has_unreleased_changes() is True
@patch("scripts.release.run_cmd")
def test_cliff_fails_returns_false(self, mock_run_cmd: MagicMock) -> None:
mock_run_cmd.return_value = MagicMock(returncode=1, stdout="")
@@ -134,47 +156,93 @@ class TestUpdateInitVersion:
update_init_version("0.2.0")
class TestCreateReleaseCommit:
class TestUpdateChangelog:
def test_creates_new_file(self, tmp_path, monkeypatch) -> None:
changelog_file = tmp_path / "CHANGELOG.md"
monkeypatch.setattr("scripts.release.CHANGELOG_FILE", str(changelog_file))
update_changelog("## [0.2.0] - 2026-06-21\n\n### Features\n- new thing")
content = changelog_file.read_text()
assert "## [0.2.0]" in content
assert "new thing" in content
def test_prepends_to_existing(self, tmp_path, monkeypatch) -> None:
changelog_file = tmp_path / "CHANGELOG.md"
changelog_file.write_text("# Changelog\n\n## [0.1.0] - 2026-06-20\n\n### Features\n- old thing\n")
monkeypatch.setattr("scripts.release.CHANGELOG_FILE", str(changelog_file))
update_changelog("## [0.2.0] - 2026-06-21\n\n### Features\n- new thing")
content = changelog_file.read_text()
assert "# Changelog" in content
# New version should be before old version
assert content.index("0.2.0") < content.index("0.1.0")
assert "new thing" in content
assert "old thing" in content
def test_appends_when_no_version_sections(self, tmp_path, monkeypatch) -> None:
changelog_file = tmp_path / "CHANGELOG.md"
changelog_file.write_text("# Changelog\n\nSome intro text.\n")
monkeypatch.setattr("scripts.release.CHANGELOG_FILE", str(changelog_file))
update_changelog("## [0.2.0] - 2026-06-21\n\n### Features\n- new thing")
content = changelog_file.read_text()
assert "Some intro text" in content
assert "## [0.2.0]" in content
class TestCommitReleaseChanges:
@patch("scripts.release.run_cmd")
def test_commits(self, mock_run_cmd: MagicMock) -> None:
def test_commits_when_changes(self, mock_run_cmd: MagicMock) -> None:
# git diff --cached --quiet returns 1 (changes exist)
mock_run_cmd.return_value = MagicMock(returncode=1, stdout="", stderr="")
result = create_release_commit("0.2.0")
result = commit_release_changes("0.2.0")
assert result is True
calls = [c.args[0] for c in mock_run_cmd.call_args_list]
assert ["git", "add", "src/gitea_runner_manager/__init__.py"] in calls
assert ["git", "commit", "-m", "chore(release): prepare for v0.2.0"] in calls
assert ["git", "add", "src/gitea_runner_manager/__init__.py", "CHANGELOG.md"] in calls
assert ["git", "commit", "-m", "release: v0.2.0"] in calls
@patch("scripts.release.run_cmd")
def test_skips_when_no_changes(self, mock_run_cmd: MagicMock) -> None:
# git diff --cached --quiet returns 0 (no changes)
mock_run_cmd.return_value = MagicMock(returncode=0, stdout="", stderr="")
result = create_release_commit("0.1.0")
result = commit_release_changes("0.1.0")
assert result is False
calls = [c.args[0] for c in mock_run_cmd.call_args_list]
assert ["git", "commit", "-m", "chore(release): prepare for v0.1.0"] not in calls
assert ["git", "commit", "-m", "release: v0.1.0"] not in calls
class TestCreateAndPushTag:
@patch("scripts.release.tag_exists", return_value=False)
@patch("scripts.release.run_cmd")
def test_creates_tag(self, mock_run_cmd: MagicMock) -> None:
create_and_push_tag("0.2.0", "changelog", dry_run=True)
first_call = mock_run_cmd.call_args_list[0].args[0]
assert first_call[:3] == ["git", "tag", "-a"]
assert first_call[3] == "v0.2.0"
assert "Release v0.2.0" in first_call[5]
def test_creates_tag(self, mock_run_cmd: MagicMock, mock_tag_exists: MagicMock) -> None:
create_and_push_tag("0.2.0", "changelog", dry_run=False)
calls = [c.args[0] for c in mock_run_cmd.call_args_list]
assert ["git", "tag", "-a", "v0.2.0", "-m", "Release v0.2.0\n\nchangelog"] in calls
assert ["git", "push", "origin", "v0.2.0"] in calls
@patch("scripts.release.tag_exists", return_value=False)
@patch("scripts.release.run_cmd")
def test_dry_run_no_push(self, mock_run_cmd: MagicMock) -> None:
create_and_push_tag("0.2.0", "changelog", dry_run=True)
def test_dry_run_no_push(self, mock_run_cmd: MagicMock, mock_tag_exists: MagicMock) -> None:
result = create_and_push_tag("0.2.0", "changelog", dry_run=True)
assert result is True
for call in mock_run_cmd.call_args_list:
assert call.args[0][0:2] != ["git", "push"]
assert call.args[0][0:2] != ["git", "tag"]
@patch("scripts.release.tag_exists", return_value=True)
@patch("scripts.release.run_cmd")
def test_pushes_tag(self, mock_run_cmd: MagicMock) -> None:
create_and_push_tag("0.2.0", "changelog", dry_run=False)
last_call = mock_run_cmd.call_args_list[-1].args[0]
assert last_call == ["git", "push", "origin", "v0.2.0"]
def test_tag_exists_skips_creation(self, mock_run_cmd: MagicMock, mock_tag_exists: MagicMock) -> None:
result = create_and_push_tag("0.1.0", "changelog", dry_run=False)
assert result is False
# Should not create tag, but should ensure it's pushed
calls = [c.args[0] for c in mock_run_cmd.call_args_list]
assert ["git", "tag", "-a"] not in [c[:3] for c in calls]
assert ["git", "push", "origin", "v0.1.0"] in calls
@patch("scripts.release.tag_exists", return_value=True)
@patch("scripts.release.run_cmd")
def test_tag_exists_dry_run_no_push(self, mock_run_cmd: MagicMock, mock_tag_exists: MagicMock) -> None:
result = create_and_push_tag("0.1.0", "changelog", dry_run=True)
assert result is False
# No git commands at all in dry-run when tag exists
mock_run_cmd.assert_not_called()
class TestMain:
@@ -189,33 +257,19 @@ class TestMain:
@patch.dict("os.environ", {})
@patch("scripts.release.has_unreleased_changes", return_value=False)
@patch("scripts.release.get_bumped_version", return_value="0.2.0")
@patch("scripts.release.run_cmd")
def test_no_unreleased_changes(self, mock_run_cmd: MagicMock, mock_has: MagicMock) -> None:
mock_run_cmd.side_effect = [
MagicMock(returncode=0, stdout="master\n", stderr=""), # branch check
MagicMock(returncode=0, stdout="GRM-34 feat: something\n", stderr=""), # last commit msg
]
def test_no_unreleased_changes(self, mock_run_cmd: MagicMock, mock_bumped: MagicMock, mock_has: MagicMock) -> None:
mock_run_cmd.return_value = MagicMock(returncode=0, stdout="master\n", stderr="")
runner = CliRunner()
result = runner.invoke(main, [])
assert result.exit_code == 0
assert "No unreleased changes" in result.output
@patch.dict("os.environ", {})
@patch("scripts.release.run_cmd")
def test_skips_release_commit(self, mock_run_cmd: MagicMock) -> None:
"""Skip when last commit is a chore(release): commit to avoid loops."""
mock_run_cmd.side_effect = [
MagicMock(returncode=0, stdout="master\n", stderr=""), # branch check
MagicMock(returncode=0, stdout="chore(release): prepare for v0.2.0\n", stderr=""), # last commit msg
]
runner = CliRunner()
result = runner.invoke(main, [])
assert result.exit_code == 0
assert "release commit" in result.output
@patch.dict("os.environ", {})
@patch("scripts.release.create_and_push_tag")
@patch("scripts.release.create_release_commit")
@patch("scripts.release.commit_release_changes")
@patch("scripts.release.update_changelog")
@patch("scripts.release.update_init_version")
@patch("scripts.release.get_changelog", return_value="")
@patch("scripts.release.get_latest_tag", return_value="v0.1.0")
@@ -230,6 +284,7 @@ class TestMain:
mock_latest: MagicMock,
mock_changelog: MagicMock,
mock_update_init: MagicMock,
mock_update_changelog: MagicMock,
mock_commit: MagicMock,
mock_tag: MagicMock,
) -> None:
@@ -241,7 +296,8 @@ class TestMain:
@patch.dict("os.environ", {})
@patch("scripts.release.create_and_push_tag")
@patch("scripts.release.create_release_commit")
@patch("scripts.release.commit_release_changes")
@patch("scripts.release.update_changelog")
@patch("scripts.release.update_init_version")
@patch("scripts.release.get_changelog", return_value="changelog")
@patch("scripts.release.get_latest_tag", return_value="v0.1.0")
@@ -256,6 +312,7 @@ class TestMain:
mock_latest: MagicMock,
mock_changelog: MagicMock,
mock_update_init: MagicMock,
mock_update_changelog: MagicMock,
mock_commit: MagicMock,
mock_tag: MagicMock,
) -> None:
@@ -265,12 +322,14 @@ class TestMain:
assert result.exit_code == 0
assert "[dry-run]" in result.output
mock_update_init.assert_not_called()
mock_update_changelog.assert_not_called()
mock_commit.assert_not_called()
mock_tag.assert_not_called()
@patch.dict("os.environ", {})
@patch("scripts.release.create_and_push_tag")
@patch("scripts.release.create_release_commit", return_value=True)
@patch("scripts.release.create_and_push_tag", return_value=True)
@patch("scripts.release.commit_release_changes", return_value=True)
@patch("scripts.release.update_changelog")
@patch("scripts.release.update_init_version")
@patch("scripts.release.get_changelog", return_value="changelog")
@patch("scripts.release.get_latest_tag", return_value="v0.1.0")
@@ -285,6 +344,7 @@ class TestMain:
mock_latest: MagicMock,
mock_changelog: MagicMock,
mock_update_init: MagicMock,
mock_update_changelog: MagicMock,
mock_commit: MagicMock,
mock_tag: MagicMock,
) -> None:
@@ -294,19 +354,21 @@ class TestMain:
assert result.exit_code == 0
assert "Bumping version" in result.output
mock_update_init.assert_called_once_with("0.2.0")
mock_update_changelog.assert_called_once_with("changelog")
mock_commit.assert_called_once_with("0.2.0")
mock_tag.assert_called_once_with("0.2.0", "changelog", False)
@patch.dict("os.environ", {})
@patch("scripts.release.create_and_push_tag")
@patch("scripts.release.create_release_commit", return_value=False)
@patch("scripts.release.create_and_push_tag", return_value=False)
@patch("scripts.release.commit_release_changes", return_value=False)
@patch("scripts.release.update_changelog")
@patch("scripts.release.update_init_version")
@patch("scripts.release.get_changelog", return_value="changelog")
@patch("scripts.release.get_latest_tag", return_value="v0.1.0")
@patch("scripts.release.get_bumped_version", return_value="0.1.0")
@patch("scripts.release.has_unreleased_changes", return_value=True)
@patch("scripts.release.run_cmd")
def test_full_flow_no_commit(
def test_full_flow_tag_exists(
self,
mock_run_cmd: MagicMock,
mock_has: MagicMock,
@@ -314,13 +376,14 @@ class TestMain:
mock_latest: MagicMock,
mock_changelog: MagicMock,
mock_update_init: MagicMock,
mock_update_changelog: MagicMock,
mock_commit: MagicMock,
mock_tag: MagicMock,
) -> None:
"""When version is unchanged, skip commit but still tag."""
"""When tag already exists, still update files but report existing tag."""
mock_run_cmd.return_value = MagicMock(returncode=0, stdout="master\n", stderr="")
runner = CliRunner()
result = runner.invoke(main, [])
assert result.exit_code == 0
assert "Skipping commit push" in result.output
assert "already existed" in result.output
mock_tag.assert_called_once_with("0.1.0", "changelog", False)