GRM-20: feat: replace inline workflow scripts with tested Python modules
This commit is contained in:
Emil Simeonov
@@ -9,31 +9,15 @@ jobs:
|
||||
if: github.event.label.name == 'ready-to-merge'
|
||||
runs-on: docker
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- name: Install dependencies
|
||||
run: python3 -m pip install requests
|
||||
- name: Squash merge with task ID
|
||||
env:
|
||||
GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
|
||||
run: |
|
||||
BRANCH="${{ github.head_ref }}"
|
||||
TASK_ID=$(echo "$BRANCH" | grep -oE 'GRM-[0-9]+' || echo "")
|
||||
if [ -z "$TASK_ID" ]; then
|
||||
echo "ERROR: No task ID (GRM-N) found in branch name '$BRANCH'"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
PR_TITLE="${{ github.event.pull_request.title }}"
|
||||
|
||||
# Validate PR title follows conventional commits so squash merge message is valid
|
||||
if ! echo "$PR_TITLE" | grep -qE '^(feat|fix|chore|docs|style|refactor|perf|test|ci|build|revert|BREAKING CHANGE)(\(.+\))?: .+'; then
|
||||
echo "ERROR: PR title must follow conventional commit format."
|
||||
echo " Expected: <type>: <description>"
|
||||
echo " Got: $PR_TITLE"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
MERGE_TITLE="${TASK_ID}: ${PR_TITLE}"
|
||||
|
||||
curl -X POST \
|
||||
"https://git.oblachno.oblachno.fyi/api/v1/repos/${{ github.repository }}/pulls/${{ github.event.number }}/merge" \
|
||||
-H "Authorization: token ${GITEA_TOKEN}" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d "{\"Do\": \"squash\", \"MergeTitleField\": \"${MERGE_TITLE}\"}"
|
||||
python3 scripts/auto_merge.py \
|
||||
"${{ github.head_ref }}" \
|
||||
"${{ github.event.pull_request.title }}" \
|
||||
"${{ github.repository }}" \
|
||||
"${{ github.event.number }}"
|
||||
|
||||
@@ -11,57 +11,12 @@ jobs:
|
||||
- uses: actions/checkout@v4
|
||||
with:
|
||||
fetch-depth: 0
|
||||
- name: Install dependencies
|
||||
run: python3 -m pip install requests
|
||||
- name: Update Vikunja task
|
||||
env:
|
||||
VIKUNJA_TOKEN: ${{ secrets.VIKUNJA_TOKEN }}
|
||||
run: |
|
||||
MERGE_MSG=$(git log -1 --pretty=%B)
|
||||
TASK_ID=$(echo "$MERGE_MSG" | grep -oE 'GRM-[0-9]+' | head -1)
|
||||
|
||||
if [ -z "$TASK_ID" ]; then
|
||||
echo "No task ID in commit message, skipping Vikunja update."
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Resolve GRM-N identifier to Vikunja numeric task ID.
|
||||
# Vikunja's filter API does not support filtering by 'identifier' field,
|
||||
# so we list all tasks and filter client-side by project_id=6 and identifier.
|
||||
VIKUNJA_TASK_ID=$(curl -s \
|
||||
"https://work.oblachno.oblachno.fyi/api/v1/tasks/all?per_page=50" \
|
||||
-H "Authorization: Bearer ${VIKUNJA_TOKEN}" | \
|
||||
python3 -c "
|
||||
import sys, json
|
||||
tasks = json.load(sys.stdin)
|
||||
matches = [t for t in tasks if t.get('project_id') == 6 and t.get('identifier') == '${TASK_ID}']
|
||||
print(matches[0]['id'] if matches else '')
|
||||
")
|
||||
|
||||
if [ -z "$VIKUNJA_TASK_ID" ]; then
|
||||
echo "ERROR: Could not find Vikunja task for ${TASK_ID} in project 6"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
CONV_MSG=$(echo "$MERGE_MSG" | head -1 | sed -E 's/^GRM-[0-9]+: //')
|
||||
COMMIT_SHA=$(git rev-parse HEAD)
|
||||
|
||||
# Post HTML comment to Vikunja
|
||||
python3 -c "
|
||||
import json, os
|
||||
html = '<p><strong>${TASK_ID}</strong>: ${CONV_MSG}</p><p>Commit: <code>${COMMIT_SHA}</code></p>'
|
||||
print(json.dumps({'content': html}))
|
||||
" > /tmp/vikunja_comment.json
|
||||
|
||||
curl -X POST \
|
||||
"https://work.oblachno.oblachno.fyi/api/v1/tasks/${VIKUNJA_TASK_ID}/comments" \
|
||||
-H "Authorization: Bearer ${VIKUNJA_TOKEN}" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d @/tmp/vikunja_comment.json
|
||||
|
||||
# Mark task as done
|
||||
curl -X PUT \
|
||||
"https://work.oblachno.oblachno.fyi/api/v1/tasks/${VIKUNJA_TASK_ID}" \
|
||||
-H "Authorization: Bearer ${VIKUNJA_TOKEN}" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"done": true}'
|
||||
|
||||
echo "Vikunja task ${TASK_ID} (ID ${VIKUNJA_TASK_ID}) updated and marked done."
|
||||
python3 scripts/post_merge.py \
|
||||
"$(git log -1 --pretty=%B)" \
|
||||
--commit-sha "$(git rev-parse HEAD)"
|
||||
|
||||
@@ -10,31 +10,14 @@ jobs:
|
||||
runs-on: docker
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- name: Install build tools
|
||||
run: |
|
||||
python3 -m pip install build twine requests
|
||||
- name: Build and publish release
|
||||
env:
|
||||
GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
|
||||
PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
|
||||
run: |
|
||||
python3 -m venv .venv
|
||||
.venv/bin/pip install build twine
|
||||
|
||||
# Build package
|
||||
.venv/bin/python -m build
|
||||
|
||||
# Publish to PyPI (only if PYPI_TOKEN secret is configured)
|
||||
if [ -n "${PYPI_TOKEN}" ]; then
|
||||
.venv/bin/twine upload dist/* -u __token__ -p "${PYPI_TOKEN}"
|
||||
echo "Published to PyPI."
|
||||
else
|
||||
echo "PYPI_TOKEN not set — skipping PyPI publish."
|
||||
fi
|
||||
|
||||
# Create Gitea release
|
||||
TAG="${{ github.ref_name }}"
|
||||
curl -X POST \
|
||||
"https://git.oblachno.oblachno.fyi/api/v1/repos/${{ github.repository }}/releases" \
|
||||
-H "Authorization: token ${GITEA_TOKEN}" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d "{\"tag_name\": \"${TAG}\", \"name\": \"${TAG}\", \"body\": \"Release ${TAG}\\n\\nSee CHANGELOG.md for details.\", \"draft\": false, \"prerelease\": false}"
|
||||
|
||||
echo "Gitea release ${TAG} created."
|
||||
python3 scripts/publish.py \
|
||||
"${{ github.ref_name }}" \
|
||||
"${{ github.repository }}"
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Auto-merge PR by extracting task ID from branch and validating PR title.
|
||||
|
||||
Usage:
|
||||
GITEA_TOKEN=<token> python3 scripts/auto_merge.py <branch> <pr_title> <repo> <pr_number>
|
||||
"""
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
|
||||
import requests
|
||||
|
||||
GITEA_API = "https://git.oblachno.oblachno.fyi/api/v1"
|
||||
TASK_ID_RE = re.compile(r"GRM-\d+")
|
||||
CONVENTIONAL_RE = re.compile(
|
||||
r"^(feat|fix|chore|docs|style|refactor|perf|test|ci|build|revert|BREAKING CHANGE)(\(.+\))?: .+"
|
||||
)
|
||||
|
||||
|
||||
def extract_task_id(branch: str) -> str:
|
||||
"""Extract GRM-N task identifier from branch name."""
|
||||
match = TASK_ID_RE.search(branch)
|
||||
return match.group(0) if match else ""
|
||||
|
||||
|
||||
def validate_pr_title(pr_title: str) -> None:
|
||||
"""Raise SystemExit if PR title does not follow conventional commits."""
|
||||
if not CONVENTIONAL_RE.match(pr_title):
|
||||
print("ERROR: PR title must follow conventional commit format.")
|
||||
print(" Expected: <type>: <description>")
|
||||
print(f" Got: {pr_title}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def merge_pr(token: str, repo: str, pr_number: str, merge_title: str) -> None:
|
||||
"""Call Gitea API to squash-merge the PR."""
|
||||
url = f"{GITEA_API}/repos/{repo}/pulls/{pr_number}/merge"
|
||||
headers = {
|
||||
"Authorization": f"token {token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
payload = {"Do": "squash", "MergeTitleField": merge_title}
|
||||
response = requests.post(url, headers=headers, json=payload, timeout=30)
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
def main(args: list[str] | None = None) -> None: # pragma: no cover
|
||||
argv = args if args is not None else sys.argv
|
||||
parser = argparse.ArgumentParser(description="Auto-merge a PR with task ID")
|
||||
parser.add_argument("branch", help="Source branch name")
|
||||
parser.add_argument("pr_title", help="Pull request title")
|
||||
parser.add_argument("repo", help="Repository full name (owner/repo)")
|
||||
parser.add_argument("pr_number", help="Pull request number")
|
||||
parsed = parser.parse_args(argv[1:])
|
||||
|
||||
token = os.environ.get("GITEA_TOKEN", "")
|
||||
if not token:
|
||||
print("ERROR: GITEA_TOKEN is not set.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
task_id = extract_task_id(parsed.branch)
|
||||
if not task_id:
|
||||
print(f"ERROR: No task ID (GRM-N) found in branch name '{parsed.branch}'")
|
||||
sys.exit(1)
|
||||
|
||||
validate_pr_title(parsed.pr_title)
|
||||
|
||||
merge_title = f"{task_id}: {parsed.pr_title}"
|
||||
merge_pr(token, parsed.repo, parsed.pr_number, merge_title)
|
||||
print(f"PR #{parsed.pr_number} squash-merged with title: {merge_title}")
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
main()
|
||||
@@ -0,0 +1,111 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Update Vikunja task after a merge to master.
|
||||
|
||||
Usage:
|
||||
VIKUNJA_TOKEN=<token> python3 scripts/post_merge.py <commit_msg>
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
|
||||
import requests
|
||||
|
||||
VIKUNJA_API = "https://work.oblachno.oblachno.fyi/api/v1"
|
||||
TASK_ID_RE = re.compile(r"GRM-\d+")
|
||||
PROJECT_ID = 6
|
||||
|
||||
|
||||
def extract_task_id(commit_msg: str) -> str:
|
||||
"""Extract GRM-N task identifier from the first line of commit message."""
|
||||
first_line = commit_msg.split("\n")[0]
|
||||
match = TASK_ID_RE.search(first_line)
|
||||
return match.group(0) if match else ""
|
||||
|
||||
|
||||
def extract_conventional_msg(commit_msg: str) -> str:
|
||||
"""Strip the GRM-N prefix from the commit subject."""
|
||||
first_line = commit_msg.split("\n")[0]
|
||||
return re.sub(r"^GRM-\d+:\s*", "", first_line)
|
||||
|
||||
|
||||
def resolve_task_id(token: str, task_id: str) -> int:
|
||||
"""Resolve GRM-N identifier to Vikunja numeric task ID."""
|
||||
url = f"{VIKUNJA_API}/tasks/all"
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
params = {"per_page": 50}
|
||||
response = requests.get(url, headers=headers, params=params, timeout=30)
|
||||
response.raise_for_status()
|
||||
tasks = response.json()
|
||||
matches = [
|
||||
t for t in tasks
|
||||
if t.get("project_id") == PROJECT_ID and t.get("identifier") == task_id
|
||||
]
|
||||
if not matches:
|
||||
print(f"ERROR: Could not find Vikunja task for {task_id} in project {PROJECT_ID}")
|
||||
sys.exit(1)
|
||||
return int(matches[0]["id"])
|
||||
|
||||
|
||||
def post_comment(token: str, task_id: int, html: str) -> None:
|
||||
"""Post an HTML comment to a Vikunja task."""
|
||||
url = f"{VIKUNJA_API}/tasks/{task_id}/comments"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
payload = {"comment": html}
|
||||
response = requests.post(url, headers=headers, json=payload, timeout=30)
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
def mark_task_done(token: str, task_id: int) -> None:
|
||||
"""Mark a Vikunja task as done."""
|
||||
url = f"{VIKUNJA_API}/tasks/{task_id}"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
payload = {"done": True}
|
||||
response = requests.put(url, headers=headers, json=payload, timeout=30)
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
def build_comment(task_id: str, conv_msg: str, commit_sha: str) -> str:
|
||||
"""Build HTML comment body for Vikunja."""
|
||||
return (
|
||||
f"<p><strong>{task_id}</strong>: {conv_msg}</p>"
|
||||
f"<p>Commit: <code>{commit_sha}</code></p>"
|
||||
)
|
||||
|
||||
|
||||
def main(args: list[str] | None = None) -> None: # pragma: no cover
|
||||
argv = args if args is not None else sys.argv
|
||||
parser = argparse.ArgumentParser(description="Update Vikunja task after merge")
|
||||
parser.add_argument("commit_msg", help="Full merge commit message")
|
||||
parser.add_argument("--commit-sha", default="", help="Commit SHA")
|
||||
parsed = parser.parse_args(argv[1:])
|
||||
|
||||
token = os.environ.get("VIKUNJA_TOKEN", "")
|
||||
if not token:
|
||||
print("ERROR: VIKUNJA_TOKEN is not set.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
task_id = extract_task_id(parsed.commit_msg)
|
||||
if not task_id:
|
||||
print("No task ID in commit message, skipping Vikunja update.")
|
||||
return
|
||||
|
||||
vikunja_task_id = resolve_task_id(token, task_id)
|
||||
conv_msg = extract_conventional_msg(parsed.commit_msg)
|
||||
commit_sha = parsed.commit_sha or "unknown"
|
||||
html = build_comment(task_id, conv_msg, commit_sha)
|
||||
|
||||
post_comment(token, vikunja_task_id, html)
|
||||
mark_task_done(token, vikunja_task_id)
|
||||
print(f"Vikunja task {task_id} (ID {vikunja_task_id}) updated and marked done.")
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
main()
|
||||
@@ -0,0 +1,93 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Build package, optionally publish to PyPI, and create Gitea release.
|
||||
|
||||
Usage:
|
||||
GITEA_TOKEN=<token> [PYPI_TOKEN=<token>] python3 scripts/publish.py <tag> <repo>
|
||||
"""
|
||||
import argparse
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import requests
|
||||
|
||||
GITEA_API = "https://git.oblachno.oblachno.fyi/api/v1"
|
||||
|
||||
|
||||
def build_package() -> None:
|
||||
"""Build the Python package using python -m build."""
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "build"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
print("ERROR: Package build failed.")
|
||||
print(result.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def publish_to_pypi(token: str) -> None:
|
||||
"""Publish built packages to PyPI using twine."""
|
||||
result = subprocess.run(
|
||||
[
|
||||
sys.executable, "-m", "twine", "upload", "dist/*",
|
||||
"-u", "__token__", "-p", token,
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
print("ERROR: PyPI publish failed.")
|
||||
print(result.stderr)
|
||||
sys.exit(1)
|
||||
print("Published to PyPI.")
|
||||
|
||||
|
||||
def create_gitea_release(token: str, repo: str, tag: str) -> None:
|
||||
"""Create a Gitea release for the given tag."""
|
||||
url = f"{GITEA_API}/repos/{repo}/releases"
|
||||
headers = {
|
||||
"Authorization": f"token {token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
payload = {
|
||||
"tag_name": tag,
|
||||
"name": tag,
|
||||
"body": f"Release {tag}\n\nSee CHANGELOG.md for details.",
|
||||
"draft": False,
|
||||
"prerelease": False,
|
||||
}
|
||||
response = requests.post(url, headers=headers, json=payload, timeout=30)
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
def main(args: list[str] | None = None) -> None: # pragma: no cover
|
||||
argv = args if args is not None else sys.argv
|
||||
parser = argparse.ArgumentParser(description="Build and publish release")
|
||||
parser.add_argument("tag", help="Git tag (e.g. v1.0.0)")
|
||||
parser.add_argument("repo", help="Repository full name (owner/repo)")
|
||||
parsed = parser.parse_args(argv[1:])
|
||||
|
||||
gitea_token = os.environ.get("GITEA_TOKEN", "")
|
||||
if not gitea_token:
|
||||
print("ERROR: GITEA_TOKEN is not set.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
pypi_token = os.environ.get("PYPI_TOKEN", "")
|
||||
|
||||
build_package()
|
||||
|
||||
if pypi_token:
|
||||
publish_to_pypi(pypi_token)
|
||||
else:
|
||||
print("PYPI_TOKEN not set — skipping PyPI publish.")
|
||||
|
||||
create_gitea_release(gitea_token, parsed.repo, parsed.tag)
|
||||
print(f"Gitea release {parsed.tag} created.")
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
main()
|
||||
@@ -0,0 +1,114 @@
|
||||
"""Unit tests for scripts/auto_merge.py."""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from scripts.auto_merge import (
|
||||
CONVENTIONAL_RE,
|
||||
GITEA_API,
|
||||
TASK_ID_RE,
|
||||
extract_task_id,
|
||||
main,
|
||||
merge_pr,
|
||||
validate_pr_title,
|
||||
)
|
||||
|
||||
|
||||
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:")
|
||||
|
||||
|
||||
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("fix: resolve timeout")
|
||||
|
||||
def test_valid_title_with_scope_passes(self) -> None:
|
||||
validate_pr_title("feat(cli): add --url option")
|
||||
|
||||
def test_invalid_title_exits(self) -> None:
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
validate_pr_title("random message")
|
||||
assert exc.value.code == 1
|
||||
|
||||
|
||||
class TestMergePr:
|
||||
@patch("scripts.auto_merge.requests.post")
|
||||
def test_successful_merge(self, mock_post: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_post.return_value = mock_response
|
||||
merge_pr("tok", "owner/repo", "7", "GRM-19: fix: bug")
|
||||
mock_post.assert_called_once()
|
||||
args, kwargs = mock_post.call_args
|
||||
assert kwargs["headers"]["Authorization"] == "token tok"
|
||||
assert kwargs["json"]["Do"] == "squash"
|
||||
assert kwargs["json"]["MergeTitleField"] == "GRM-19: fix: bug"
|
||||
assert GITEA_API in args[0]
|
||||
|
||||
@patch("scripts.auto_merge.requests.post")
|
||||
def test_merge_raises_on_http_error(self, mock_post: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status.side_effect = requests.HTTPError("500")
|
||||
mock_post.return_value = mock_response
|
||||
with pytest.raises(requests.HTTPError):
|
||||
merge_pr("tok", "owner/repo", "7", "title")
|
||||
|
||||
|
||||
class TestMain:
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": "tok"})
|
||||
@patch("scripts.auto_merge.merge_pr")
|
||||
def test_successful_flow(self, mock_merge: MagicMock, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
main(["auto_merge.py", "GRM-19-fix-bug", "fix: resolve timeout", "owner/repo", "7"])
|
||||
mock_merge.assert_called_once_with("tok", "owner/repo", "7", "GRM-19: fix: resolve timeout")
|
||||
captured = capsys.readouterr()
|
||||
assert "squash-merged" in captured.out
|
||||
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": ""}, clear=True)
|
||||
def test_missing_token_exits(self) -> None:
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
main(["auto_merge.py", "branch", "title", "repo", "1"])
|
||||
assert exc.value.code == 1
|
||||
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": "tok"})
|
||||
def test_missing_task_id_exits(self) -> None:
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
main(["auto_merge.py", "feature-no-id", "fix: bug", "repo", "1"])
|
||||
assert exc.value.code == 1
|
||||
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": "tok"})
|
||||
def test_invalid_pr_title_exits(self) -> None:
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
main(["auto_merge.py", "GRM-19-fix", "random title", "repo", "1"])
|
||||
assert exc.value.code == 1
|
||||
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": "tok"})
|
||||
@patch("scripts.auto_merge.merge_pr")
|
||||
def test_merge_pr_failure_propagates(self, mock_merge: MagicMock) -> None:
|
||||
mock_merge.side_effect = requests.HTTPError("500")
|
||||
with pytest.raises(requests.HTTPError):
|
||||
main(["auto_merge.py", "GRM-19-fix", "fix: bug", "repo", "1"])
|
||||
@@ -0,0 +1,188 @@
|
||||
"""Unit tests for scripts/post_merge.py."""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from scripts.post_merge import (
|
||||
PROJECT_ID,
|
||||
VIKUNJA_API,
|
||||
build_comment,
|
||||
extract_conventional_msg,
|
||||
extract_task_id,
|
||||
main,
|
||||
mark_task_done,
|
||||
post_comment,
|
||||
resolve_task_id,
|
||||
)
|
||||
|
||||
|
||||
class TestExtractTaskId:
|
||||
def test_extracts_from_first_line(self) -> None:
|
||||
assert extract_task_id("GRM-19: fix: resolve bug\n\nBody") == "GRM-19"
|
||||
|
||||
def test_returns_empty_when_missing(self) -> None:
|
||||
assert extract_task_id("fix: resolve bug") == ""
|
||||
|
||||
|
||||
class TestExtractConventionalMsg:
|
||||
def test_strips_task_id_prefix(self) -> None:
|
||||
assert extract_conventional_msg("GRM-19: fix: resolve bug") == "fix: resolve bug"
|
||||
|
||||
def test_returns_unchanged_without_prefix(self) -> None:
|
||||
assert extract_conventional_msg("fix: resolve bug") == "fix: resolve bug"
|
||||
|
||||
|
||||
class TestBuildComment:
|
||||
def test_html_format(self) -> None:
|
||||
html = build_comment("GRM-19", "fix: bug", "abc123")
|
||||
assert "<strong>GRM-19</strong>" in html
|
||||
assert "fix: bug" in html
|
||||
assert "<code>abc123</code>" in html
|
||||
|
||||
|
||||
class TestResolveTaskId:
|
||||
@patch("scripts.post_merge.requests.get")
|
||||
def test_found(self, mock_get: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = [
|
||||
{"id": 42, "project_id": PROJECT_ID, "identifier": "GRM-19"},
|
||||
]
|
||||
mock_get.return_value = mock_response
|
||||
assert resolve_task_id("tok", "GRM-19") == 42
|
||||
mock_get.assert_called_once()
|
||||
|
||||
@patch("scripts.post_merge.requests.get")
|
||||
def test_not_found_exits(self, mock_get: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = []
|
||||
mock_get.return_value = mock_response
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
resolve_task_id("tok", "GRM-99")
|
||||
assert exc.value.code == 1
|
||||
|
||||
@patch("scripts.post_merge.requests.get")
|
||||
def test_wrong_project_filtered(self, mock_get: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = [
|
||||
{"id": 42, "project_id": 999, "identifier": "GRM-19"},
|
||||
]
|
||||
mock_get.return_value = mock_response
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
resolve_task_id("tok", "GRM-19")
|
||||
assert exc.value.code == 1
|
||||
|
||||
@patch("scripts.post_merge.requests.get")
|
||||
def test_http_error_propagates(self, mock_get: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status.side_effect = requests.HTTPError("500")
|
||||
mock_get.return_value = mock_response
|
||||
with pytest.raises(requests.HTTPError):
|
||||
resolve_task_id("tok", "GRM-19")
|
||||
|
||||
|
||||
class TestPostComment:
|
||||
@patch("scripts.post_merge.requests.post")
|
||||
def test_success(self, mock_post: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_post.return_value = mock_response
|
||||
post_comment("tok", 42, "<p>hi</p>")
|
||||
args, kwargs = mock_post.call_args
|
||||
assert VIKUNJA_API in args[0]
|
||||
assert kwargs["json"]["comment"] == "<p>hi</p>"
|
||||
|
||||
@patch("scripts.post_merge.requests.post")
|
||||
def test_http_error_propagates(self, mock_post: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status.side_effect = requests.HTTPError("500")
|
||||
mock_post.return_value = mock_response
|
||||
with pytest.raises(requests.HTTPError):
|
||||
post_comment("tok", 42, "html")
|
||||
|
||||
|
||||
class TestMarkTaskDone:
|
||||
@patch("scripts.post_merge.requests.put")
|
||||
def test_success(self, mock_put: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_put.return_value = mock_response
|
||||
mark_task_done("tok", 42)
|
||||
args, kwargs = mock_put.call_args
|
||||
assert kwargs["json"]["done"] is True
|
||||
|
||||
@patch("scripts.post_merge.requests.put")
|
||||
def test_http_error_propagates(self, mock_put: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status.side_effect = requests.HTTPError("500")
|
||||
mock_put.return_value = mock_response
|
||||
with pytest.raises(requests.HTTPError):
|
||||
mark_task_done("tok", 42)
|
||||
|
||||
|
||||
class TestMain:
|
||||
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
|
||||
@patch("scripts.post_merge.resolve_task_id")
|
||||
@patch("scripts.post_merge.post_comment")
|
||||
@patch("scripts.post_merge.mark_task_done")
|
||||
def test_full_flow(
|
||||
self, mock_mark: MagicMock, mock_post: MagicMock, mock_resolve: MagicMock, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
mock_resolve.return_value = 267
|
||||
main(["post_merge.py", "GRM-20: fix: resolve bug\n\nBody", "--commit-sha", "abc123"])
|
||||
mock_resolve.assert_called_once_with("tok", "GRM-20")
|
||||
mock_post.assert_called_once()
|
||||
mock_mark.assert_called_once_with("tok", 267)
|
||||
captured = capsys.readouterr()
|
||||
assert "updated and marked done" in captured.out
|
||||
|
||||
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
|
||||
@patch("scripts.post_merge.resolve_task_id")
|
||||
@patch("scripts.post_merge.post_comment")
|
||||
@patch("scripts.post_merge.mark_task_done")
|
||||
def test_no_commit_sha(self, mock_mark: MagicMock, mock_post: MagicMock, mock_resolve: MagicMock) -> None:
|
||||
mock_resolve.return_value = 267
|
||||
main(["post_merge.py", "GRM-20: fix: resolve bug"])
|
||||
mock_post.assert_called_once()
|
||||
args, _ = mock_post.call_args
|
||||
assert "unknown" in args[2]
|
||||
|
||||
@patch.dict("os.environ", {"VIKUNJA_TOKEN": ""}, clear=True)
|
||||
def test_missing_token_exits(self) -> None:
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
main(["post_merge.py", "GRM-20: fix: bug"])
|
||||
assert exc.value.code == 1
|
||||
|
||||
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
|
||||
def test_no_task_id_skips(self, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
main(["post_merge.py", "fix: resolve bug"])
|
||||
captured = capsys.readouterr()
|
||||
assert "skipping Vikunja update" in captured.out
|
||||
|
||||
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
|
||||
@patch("scripts.post_merge.resolve_task_id")
|
||||
def test_resolve_failure_propagates(self, mock_resolve: MagicMock) -> None:
|
||||
mock_resolve.side_effect = SystemExit(1)
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
main(["post_merge.py", "GRM-20: fix: bug"])
|
||||
assert exc.value.code == 1
|
||||
|
||||
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
|
||||
@patch("scripts.post_merge.resolve_task_id")
|
||||
@patch("scripts.post_merge.post_comment")
|
||||
def test_post_comment_failure_propagates(self, mock_post: MagicMock, mock_resolve: MagicMock) -> None:
|
||||
mock_resolve.return_value = 267
|
||||
mock_post.side_effect = requests.HTTPError("500")
|
||||
with pytest.raises(requests.HTTPError):
|
||||
main(["post_merge.py", "GRM-20: fix: bug"])
|
||||
|
||||
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
|
||||
@patch("scripts.post_merge.resolve_task_id")
|
||||
@patch("scripts.post_merge.post_comment")
|
||||
@patch("scripts.post_merge.mark_task_done")
|
||||
def test_mark_done_failure_propagates(
|
||||
self, mock_mark: MagicMock, mock_post: MagicMock, mock_resolve: MagicMock
|
||||
) -> None:
|
||||
mock_resolve.return_value = 267
|
||||
mock_mark.side_effect = requests.HTTPError("500")
|
||||
with pytest.raises(requests.HTTPError):
|
||||
main(["post_merge.py", "GRM-20: fix: bug"])
|
||||
@@ -0,0 +1,141 @@
|
||||
"""Unit tests for scripts/publish.py."""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from scripts.publish import (
|
||||
GITEA_API,
|
||||
build_package,
|
||||
create_gitea_release,
|
||||
main,
|
||||
publish_to_pypi,
|
||||
)
|
||||
|
||||
|
||||
class TestBuildPackage:
|
||||
@patch("scripts.publish.subprocess.run")
|
||||
def test_success(self, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock(returncode=0, stderr="")
|
||||
build_package()
|
||||
args, _ = mock_run.call_args
|
||||
assert args[0][1] == "-m"
|
||||
assert args[0][2] == "build"
|
||||
|
||||
@patch("scripts.publish.subprocess.run")
|
||||
def test_failure_exits(self, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock(returncode=1, stderr="build error")
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
build_package()
|
||||
assert exc.value.code == 1
|
||||
|
||||
|
||||
class TestPublishToPypi:
|
||||
@patch("scripts.publish.subprocess.run")
|
||||
def test_success(self, mock_run: MagicMock, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
mock_run.return_value = MagicMock(returncode=0, stderr="")
|
||||
publish_to_pypi("pypi-tok")
|
||||
args, _ = mock_run.call_args
|
||||
assert "twine" in args[0]
|
||||
assert "pypi-tok" in args[0]
|
||||
captured = capsys.readouterr()
|
||||
assert "Published to PyPI" in captured.out
|
||||
|
||||
@patch("scripts.publish.subprocess.run")
|
||||
def test_failure_exits(self, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock(returncode=1, stderr="upload failed")
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
publish_to_pypi("pypi-tok")
|
||||
assert exc.value.code == 1
|
||||
|
||||
|
||||
class TestCreateGiteaRelease:
|
||||
@patch("scripts.publish.requests.post")
|
||||
def test_success(self, mock_post: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_post.return_value = mock_response
|
||||
create_gitea_release("tok", "owner/repo", "v1.0.0")
|
||||
args, kwargs = mock_post.call_args
|
||||
assert GITEA_API in args[0]
|
||||
assert kwargs["json"]["tag_name"] == "v1.0.0"
|
||||
assert kwargs["json"]["draft"] is False
|
||||
assert kwargs["json"]["prerelease"] is False
|
||||
|
||||
@patch("scripts.publish.requests.post")
|
||||
def test_http_error_propagates(self, mock_post: MagicMock) -> None:
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status.side_effect = requests.HTTPError("500")
|
||||
mock_post.return_value = mock_response
|
||||
with pytest.raises(requests.HTTPError):
|
||||
create_gitea_release("tok", "owner/repo", "v1.0.0")
|
||||
|
||||
|
||||
class TestMain:
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": "gitea-tok", "PYPI_TOKEN": "pypi-tok"})
|
||||
@patch("scripts.publish.create_gitea_release")
|
||||
@patch("scripts.publish.publish_to_pypi")
|
||||
@patch("scripts.publish.build_package")
|
||||
def test_full_flow_with_pypi(
|
||||
self,
|
||||
mock_build: MagicMock,
|
||||
mock_publish: MagicMock,
|
||||
mock_release: MagicMock,
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
main(["publish.py", "v1.0.0", "owner/repo"])
|
||||
mock_build.assert_called_once()
|
||||
mock_publish.assert_called_once_with("pypi-tok")
|
||||
mock_release.assert_called_once_with("gitea-tok", "owner/repo", "v1.0.0")
|
||||
captured = capsys.readouterr()
|
||||
assert "Gitea release v1.0.0 created" in captured.out
|
||||
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": "gitea-tok"}, clear=True)
|
||||
@patch("scripts.publish.create_gitea_release")
|
||||
@patch("scripts.publish.build_package")
|
||||
def test_without_pypi(
|
||||
self,
|
||||
mock_build: MagicMock,
|
||||
mock_release: MagicMock,
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
main(["publish.py", "v1.0.0", "owner/repo"])
|
||||
mock_build.assert_called_once()
|
||||
mock_release.assert_called_once_with("gitea-tok", "owner/repo", "v1.0.0")
|
||||
captured = capsys.readouterr()
|
||||
assert "PYPI_TOKEN not set" in captured.out
|
||||
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": ""}, clear=True)
|
||||
def test_missing_gitea_token_exits(self) -> None:
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
main(["publish.py", "v1.0.0", "owner/repo"])
|
||||
assert exc.value.code == 1
|
||||
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": "gitea-tok", "PYPI_TOKEN": "pypi-tok"})
|
||||
@patch("scripts.publish.create_gitea_release")
|
||||
@patch("scripts.publish.publish_to_pypi")
|
||||
@patch("scripts.publish.build_package")
|
||||
def test_build_failure_propagates(self, mock_build: MagicMock, *_: MagicMock) -> None:
|
||||
mock_build.side_effect = SystemExit(1)
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
main(["publish.py", "v1.0.0", "owner/repo"])
|
||||
assert exc.value.code == 1
|
||||
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": "gitea-tok", "PYPI_TOKEN": "pypi-tok"})
|
||||
@patch("scripts.publish.create_gitea_release")
|
||||
@patch("scripts.publish.publish_to_pypi")
|
||||
@patch("scripts.publish.build_package")
|
||||
def test_publish_failure_propagates(self, mock_publish: MagicMock, *_: MagicMock) -> None:
|
||||
mock_publish.side_effect = SystemExit(1)
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
main(["publish.py", "v1.0.0", "owner/repo"])
|
||||
assert exc.value.code == 1
|
||||
|
||||
@patch.dict("os.environ", {"GITEA_TOKEN": "gitea-tok", "PYPI_TOKEN": "pypi-tok"})
|
||||
@patch("scripts.publish.create_gitea_release")
|
||||
@patch("scripts.publish.publish_to_pypi")
|
||||
@patch("scripts.publish.build_package")
|
||||
def test_release_failure_propagates(self, mock_release: MagicMock, *_: MagicMock) -> None:
|
||||
mock_release.side_effect = requests.HTTPError("500")
|
||||
with pytest.raises(requests.HTTPError):
|
||||
main(["publish.py", "v1.0.0", "owner/repo"])
|
||||
Reference in New Issue
Block a user