oblachno-oss/grm
3
0
Fork 0
Code Issues 33 Pull Requests Actions Packages Projects Releases 23 Wiki Activity

GRM-20: refactor(scripts): centralize constants, API clients, and HTTP status codes

Browse Source 
- Add shared config.py with API URLs, regexes, timeouts, pagination
- Add GiteaClient and VikunjaClient in api_clients.py with pooled sessions
- Add APIError exception for unified HTTP error handling
- Refactor all scripts to use shared modules and http.HTTPStatus
- Rewrite unit tests to mock clients and use HTTPStatus constants
- Add tests for api_clients and config modules
- Achieve 100% test coverage
This commit is contained in:
Emil Simeonov
2026-06-19 21:00:21 +02:00
parent 790b5c3769
commit fefddda715
15 changed files with 751 additions and 682 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/auto_merge.py
+12 -32
View File
@@ -6,19 +6,14 @@ Usage:
"""
import os
import re
import click
import requests
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.exceptions import APIError
from gitea_runner_manager.i18n import _
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."""
@@ -39,18 +34,6 @@ def validate_pr_title(pr_title: str) -> None:
)
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()
@click.command()
@click.argument("branch")
@click.argument("pr_title")
@@ -73,21 +56,18 @@ def main(branch: str, pr_title: str, repo: str, pr_number: str) -> None:
validate_pr_title(pr_title)
merge_title = f"{task_id}: {pr_title}"
owner, repo_name = repo.split("/")
client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
try:
merge_pr(token, repo, pr_number, merge_title)
except requests.HTTPError as e:
response = e.response
status = response.status_code if response else 0
try:
body = response.json() if response else {}
message = body.get("message", str(e))
except Exception:
message = str(e)
client.merge_pr(pr_number, merge_title)
except APIError as e:
raise click.ClickException(
_(
"Merge failed with HTTP {status}: {message}\nPlease check the PR is ready and you have merge rights.",
status=status,
message=message,
"Merge failed with HTTP {status}: {message}\n"
"Please check the PR is ready and you have merge rights.",
status=e.status,
message=e.message,
)
) from None
scripts/configure_repo.py
+26 -105
View File
@@ -7,119 +7,36 @@ Usage:
import http
import os
from typing import cast
import click
import requests
from gitea_runner_manager.api_clients import GiteaClient
from gitea_runner_manager.config import (
BRANCH_PROTECTION_CONFIG,
GITEA_API_URL,
LABEL_CONFIG,
REPO_NAME,
REPO_OWNER,
)
from gitea_runner_manager.exceptions import APIError
from gitea_runner_manager.i18n import _
GITEA_API = "https://git.oblachno.oblachno.fyi/api/v1"
OWNER = "oblachno-oss"
REPO = "grm"
BRANCH_PROTECTION_CONFIG = {
"branch_name": "master",
"enable_push": False,
"enable_status_check": True,
"status_check_contexts": ["lint", "unit-tests", "molecule-tests"],
"required_approvals": 1,
"dismiss_stale_approvals": True,
"block_on_outdated_branch": True,
"block_on_rejected_reviews": True,
"block_on_official_review_requests": True,
}
LABEL_CONFIG = {
"name": "ready-to-merge",
"color": "2ecc71",
"description": "Auto-merge PR when all CI checks pass",
}
class GiteaRepoConfig:
"""Configure a Gitea repository: branch protection and labels."""
def __init__(self, base_url: str, token: str, owner: str, repo: str) -> None:
self._base_url = base_url.rstrip("/")
self._owner = owner
self._repo = repo
self._session = requests.Session()
self._session.headers.update(
{
"Authorization": f"token {token}",
"Content-Type": "application/json",
}
)
def _url(self, path: str) -> str:
return f"{self._base_url}/repos/{self._owner}/{self._repo}{path}"
def list_branch_protections(self) -> list[dict]:
r = self._session.get(self._url("/branch_protections"))
r.raise_for_status()
return r.json()
def create_branch_protection(self, config: dict) -> dict:
r = self._session.post(self._url("/branch_protections"), json=config)
r.raise_for_status()
return r.json()
def update_branch_protection(self, protection_id: int, config: dict) -> dict:
r = self._session.patch(self._url(f"/branch_protections/{protection_id}"), json=config)
r.raise_for_status()
return r.json()
def ensure_branch_protection(self, branch: str, config: dict) -> dict:
"""Idempotent: create or update branch protection for the given branch."""
existing = self.list_branch_protections()
for p in existing:
if p.get("branch_name") == branch:
protection_id = p["id"]
update_config = {k: v for k, v in config.items() if k != "branch_name"}
return self.update_branch_protection(protection_id, update_config)
return self.create_branch_protection(config)
def list_labels(self) -> list[dict]:
r = self._session.get(self._url("/labels"))
r.raise_for_status()
return r.json()
def create_label(self, name: str, color: str, description: str = "") -> dict:
r = self._session.post(
self._url("/labels"),
json={"name": name, "color": color, "description": description},
)
r.raise_for_status()
return r.json()
def ensure_label(self, name: str, color: str, description: str = "") -> dict | None:
"""Idempotent: create label if it doesn't already exist."""
labels = self.list_labels()
for label in labels:
if label["name"] == name:
return None # already exists
return self.create_label(name, color, description)
def _handle_http_error(e: requests.HTTPError) -> None:
def _handle_http_error(e: APIError) -> None:
"""Raise a user-friendly Click exception for HTTP errors."""
response = e.response
status = response.status_code if response else 0
if status == http.HTTPStatus.FORBIDDEN:
if e.status == http.HTTPStatus.FORBIDDEN:
raise click.ClickException(
_(
"HTTP {status} Forbidden — your token lacks admin rights.\n"
"Make sure the token belongs to a repo owner or organisation admin.\n"
"Alternatively, configure branch protection manually in Settings → Branches.",
status=status,
status=e.status,
)
)
try:
body = response.json() if response else {}
message = body.get("message", str(e))
except Exception:
message = str(e)
raise click.ClickException(_("HTTP error: {status} — {message}", status=status, message=message))
raise click.ClickException(
_("HTTP error: {status} — {message}", status=e.status, message=e.message)
)
def main() -> None:
@@ -127,11 +44,11 @@ def main() -> None:
if not token:
raise click.ClickException(_("ERROR: GITEA_ADMIN_TOKEN is not set."))
cfg = GiteaRepoConfig(GITEA_API, token, OWNER, REPO)
client = GiteaClient(GITEA_API_URL, token, REPO_OWNER, REPO_NAME)
try:
click.echo(_("Configuring branch protection for {branch}...", branch="master"))
cfg.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
client.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
click.echo(_(" - Direct pushes: BLOCKED (require PR)"))
click.echo(
_(
@@ -142,13 +59,17 @@ def main() -> None:
click.echo(_(" - Dismiss stale approvals: yes"))
click.echo(_(" - Block outdated branches: yes"))
click.echo(_(" - Block rejected reviews: yes"))
checks = ", ".join(BRANCH_PROTECTION_CONFIG["status_check_contexts"])
checks = ", ".join(cast(list[str], BRANCH_PROTECTION_CONFIG["status_check_contexts"]))
click.echo(_(" - Required status checks: {checks}", checks=checks))
click.echo("")
label_name = LABEL_CONFIG["name"]
label_name = cast(str, LABEL_CONFIG["name"])
click.echo(_("Creating {label} label...", label=label_name))
result = cfg.ensure_label(**LABEL_CONFIG)
result = client.ensure_label(
name=cast(str, LABEL_CONFIG["name"]),
color=cast(str, LABEL_CONFIG["color"]),
description=cast(str, LABEL_CONFIG["description"]),
)
if result is None:
click.echo(_(" Label '{label}' already exists.", label=label_name))
else:
@@ -156,7 +77,7 @@ def main() -> None:
click.echo("")
click.echo(_("Repository configuration complete."))
except requests.HTTPError as e:
except APIError as e:
_handle_http_error(e)
scripts/post_merge.py
+22 -61
View File
@@ -9,14 +9,12 @@ import os
import re
import click
import requests
from gitea_runner_manager.api_clients import VikunjaClient
from gitea_runner_manager.config import DEFAULT_PER_PAGE, TASK_ID_RE, VIKUNJA_API_URL, VIKUNJA_PROJECT_ID
from gitea_runner_manager.exceptions import APIError
from gitea_runner_manager.i18n import _
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."""
@@ -31,68 +29,24 @@ def extract_conventional_msg(commit_msg: str) -> str:
return re.sub(r"^GRM-\d+:\s*", "", first_line)
def _handle_http_error(e: requests.HTTPError) -> None:
"""Raise a user-friendly Click exception for HTTP errors."""
response = e.response
status = response.status_code if response else 0
try:
body = response.json() if response else {}
message = body.get("message", str(e))
except Exception:
message = str(e)
raise click.ClickException(
_(
"Vikunja API error: HTTP {status} — {message}",
status=status,
message=message,
)
)
def resolve_task_id(token: str, task_id: str) -> int:
def resolve_task_id(client: VikunjaClient, 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]
tasks = client.list_tasks(per_page=DEFAULT_PER_PAGE)
matches = [
t for t in tasks
if t.get("project_id") == VIKUNJA_PROJECT_ID and t.get("identifier") == task_id
]
if not matches:
raise click.ClickException(
_(
"Could not find Vikunja task for {task_id} in project {project_id}.",
task_id=task_id,
project_id=PROJECT_ID,
project_id=VIKUNJA_PROJECT_ID,
)
)
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><p>Commit: <code>{commit_sha}</code></p>"
@@ -111,17 +65,24 @@ def main(commit_msg: str, commit_sha: str) -> None:
click.echo(_("No task ID in commit message, skipping Vikunja update. All good — nothing to do here!"))
return
client = VikunjaClient(VIKUNJA_API_URL, token)
vikunja_task_id = 0
try:
vikunja_task_id = resolve_task_id(token, task_id)
vikunja_task_id = resolve_task_id(client, task_id)
conv_msg = extract_conventional_msg(commit_msg)
sha = commit_sha or "unknown"
html = build_comment(task_id, conv_msg, sha)
post_comment(token, vikunja_task_id, html)
mark_task_done(token, vikunja_task_id)
except requests.HTTPError as e:
_handle_http_error(e)
client.post_comment(vikunja_task_id, html)
client.update_task(vikunja_task_id, done=True)
except APIError as e:
raise click.ClickException(
_(
"Vikunja API error: HTTP {status} — {message}",
status=e.status,
message=e.message,
)
) from None
click.echo(
_(
scripts/publish.py
+12 -32
View File
@@ -10,12 +10,12 @@ import subprocess
import sys
import click
import requests
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 _
GITEA_API = "https://git.oblachno.oblachno.fyi/api/v1"
def build_package() -> None:
"""Build the Python package using python -m build."""
@@ -62,24 +62,6 @@ def publish_to_pypi(token: str) -> None:
click.echo(_("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()
@click.command()
@click.argument("tag")
@click.argument("repo")
@@ -97,21 +79,19 @@ def main(tag: str, repo: str) -> None:
else:
click.echo(_("PYPI_TOKEN not set — skipping PyPI publish. No worries, we'll just create the Gitea release."))
owner, repo_name = repo.split("/")
client = GiteaClient(GITEA_API_URL, gitea_token, owner, repo_name)
try:
create_gitea_release(gitea_token, repo, tag)
except requests.HTTPError as e:
response = e.response
status = response.status_code if response else 0
try:
body = response.json() if response else {}
message = body.get("message", str(e))
except Exception:
message = str(e)
client.create_release(
tag=tag,
body=f"Release {tag}\n\nSee CHANGELOG.md for details.",
)
except APIError as e:
raise click.ClickException(
_(
"Release creation failed with HTTP {status}: {message}",
status=status,
message=message,
status=e.status,
message=e.message,
)
) from None
scripts/validate_commit_msg.py
+5 -7
View File
@@ -12,12 +12,10 @@ import subprocess
import click
from gitea_runner_manager.config import CONVENTIONAL_RE
from gitea_runner_manager.i18n import _
CONVENTIONAL_RE = re.compile(
r"^(feat|fix|chore|docs|style|refactor|perf|test|ci|build|revert|BREAKING CHANGE)(\(.+\))?: .+"
)
TASK_ID_RE = re.compile(r"^GRM-\d+:")
MASTER_TASK_ID_RE = re.compile(r"^GRM-\d+:")
def first_line(text: str) -> str:
@@ -47,7 +45,7 @@ def main(commit_msg_file: str) -> None:
subject = first_line(msg)
if branch == "master":
if not TASK_ID_RE.match(subject):
if not MASTER_TASK_ID_RE.match(subject):
raise click.ClickException(
_(
"Oops! Master branch commits must start with a task ID.\n"
@@ -56,7 +54,7 @@ def main(commit_msg_file: str) -> None:
subject=subject,
)
)
remainder = TASK_ID_RE.sub("", subject).strip()
remainder = MASTER_TASK_ID_RE.sub("", subject).strip()
if not CONVENTIONAL_RE.match(remainder):
raise click.ClickException(
_(
@@ -68,7 +66,7 @@ def main(commit_msg_file: str) -> None:
)
return
if TASK_ID_RE.match(subject):
if MASTER_TASK_ID_RE.match(subject):
raise click.ClickException(
_(
"Oops! Do not include task ID (GRM-N) in feature branch commits.\n"
src/gitea_runner_manager/api_clients.py
+153
View File
@@ -0,0 +1,153 @@
"""Reusable HTTP API clients for Gitea and Vikunja."""
from __future__ import annotations
import logging
from typing import Any
import requests
from .config import DEFAULT_TIMEOUT
from .exceptions import APIError
logger = logging.getLogger("grm")
def _parse_error(e: requests.HTTPError) -> tuple[int, str]:
"""Extract status code and message from an HTTPError response."""
response = e.response
status = response.status_code if response else 0
try:
body: dict[str, Any] = response.json() if response else {}
message: str = body.get("message", str(e))
except Exception:
message = str(e)
return status, message
class GiteaClient:
"""Low-level Gitea REST API client with connection pooling."""
def __init__(self, base_url: str, token: str, owner: str, repo: str) -> None:
self._base_url = base_url.rstrip("/")
self._owner = owner
self._repo = repo
self._session = requests.Session()
self._session.headers.update(
{
"Authorization": f"token {token}",
"Content-Type": "application/json",
}
)
def _url(self, path: str) -> str:
return f"{self._base_url}/repos/{self._owner}/{self._repo}{path}"
def _request(self, method: str, path: str, **kwargs: Any) -> requests.Response:
url = self._url(path)
try:
response = self._session.request(method, url, timeout=DEFAULT_TIMEOUT, **kwargs)
response.raise_for_status()
except requests.HTTPError as e:
status, message = _parse_error(e)
raise APIError(status, message) from e
return response
# -- branch protection --
def list_branch_protections(self) -> list[dict[str, Any]]:
r = self._request("GET", "/branch_protections")
return r.json()
def create_branch_protection(self, config: dict[str, Any]) -> dict[str, Any]:
r = self._request("POST", "/branch_protections", json=config)
return r.json()
def update_branch_protection(self, protection_id: int, config: dict[str, Any]) -> dict[str, Any]:
r = self._request("PATCH", f"/branch_protections/{protection_id}", json=config)
return r.json()
def ensure_branch_protection(self, branch: str, config: dict[str, Any]) -> dict[str, Any]:
"""Idempotent: create or update branch protection for the given branch."""
existing = self.list_branch_protections()
for p in existing:
if p.get("branch_name") == branch:
protection_id = p["id"]
update_config = {k: v for k, v in config.items() if k != "branch_name"}
return self.update_branch_protection(protection_id, update_config)
return self.create_branch_protection(config)
# -- labels --
def list_labels(self) -> list[dict[str, Any]]:
r = self._request("GET", "/labels")
return r.json()
def create_label(self, name: str, color: str, description: str = "") -> dict[str, Any]:
r = self._request(
"POST",
"/labels",
json={"name": name, "color": color, "description": description},
)
return r.json()
def ensure_label(self, name: str, color: str, description: str = "") -> dict[str, Any] | None:
"""Idempotent: create label if it doesn't already exist."""
labels = self.list_labels()
for label in labels:
if label["name"] == name:
return None
return self.create_label(name, color, description)
# -- pulls / releases --
def merge_pr(self, pr_number: str | int, merge_title: str) -> None:
payload = {"Do": "squash", "MergeTitleField": merge_title}
self._request("POST", f"/pulls/{pr_number}/merge", json=payload)
def create_release(
self,
tag: str,
name: str = "",
body: str = "",
draft: bool = False,
prerelease: bool = False,
) -> dict[str, Any]:
payload = {
"tag_name": tag,
"name": name or tag,
"body": body,
"draft": draft,
"prerelease": prerelease,
}
r = self._request("POST", "/releases", json=payload)
return r.json()
class VikunjaClient:
"""Low-level Vikunja REST API client with connection pooling."""
def __init__(self, base_url: str, token: str) -> None:
self._base_url = base_url.rstrip("/")
self._session = requests.Session()
self._session.headers.update({"Authorization": f"Bearer {token}"})
def _request(self, method: str, path: str, **kwargs: Any) -> requests.Response:
url = f"{self._base_url}{path}"
try:
response = self._session.request(method, url, timeout=DEFAULT_TIMEOUT, **kwargs)
response.raise_for_status()
except requests.HTTPError as e:
status, message = _parse_error(e)
raise APIError(status, message) from e
return response
def list_tasks(self, **params: Any) -> list[dict[str, Any]]:
r = self._request("GET", "/tasks", params=params)
return r.json()
def post_comment(self, task_id: int, comment: str) -> None:
self._request("POST", f"/tasks/{task_id}/comments", json={"comment": comment})
def update_task(self, task_id: int, **fields: Any) -> None:
self._request("POST", f"/tasks/{task_id}", json=fields)
src/gitea_runner_manager/config.py
+39
View File
@@ -0,0 +1,39 @@
"""Shared configuration constants for GRM scripts and API clients."""
from __future__ import annotations
import re
GITEA_API_URL = "https://git.oblachno.oblachno.fyi/api/v1"
VIKUNJA_API_URL = "https://work.oblachno.oblachno.fyi/api/v1"
REPO_OWNER = "oblachno-oss"
REPO_NAME = "grm"
VIKUNJA_PROJECT_ID = 6
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)(\(.+\))?: .+"
)
DEFAULT_TIMEOUT = 30
DEFAULT_PER_PAGE = 50
BRANCH_PROTECTION_CONFIG: dict[str, object] = {
"branch_name": "master",
"enable_push": False,
"enable_status_check": True,
"status_check_contexts": ["lint", "unit-tests", "molecule-tests"],
"required_approvals": 1,
"dismiss_stale_approvals": True,
"block_on_outdated_branch": True,
"block_on_rejected_reviews": True,
"block_on_official_review_requests": True,
}
LABEL_CONFIG: dict[str, object] = {
"name": "ready-to-merge",
"color": "2ecc71",
"description": "Auto-merge PR when all CI checks pass",
}
src/gitea_runner_manager/exceptions.py
+9
View File
@@ -11,3 +11,12 @@ class AnsibleError(GRMError):
"""Raised when an Ansible command fails."""
pass
class APIError(GRMError):
"""Raised when a REST API call returns an HTTP error."""
def __init__(self, status: int, message: str) -> None:
self.status = status
self.message = message
super().__init__(f"HTTP {status}: {message}")
tests/unit/test_api_clients.py
+264
View File
@@ -0,0 +1,264 @@
"""Unit tests for api_clients module."""
import http
from unittest.mock import MagicMock
import pytest
import requests
from gitea_runner_manager.api_clients import GiteaClient, VikunjaClient, _parse_error
from gitea_runner_manager.config import (
BRANCH_PROTECTION_CONFIG,
DEFAULT_PER_PAGE,
DEFAULT_TIMEOUT,
VIKUNJA_PROJECT_ID,
)
from gitea_runner_manager.exceptions import APIError
def _mock_response(json_data: object | None = None, raise_on_status: bool = False) -> MagicMock:
mock = MagicMock()
if json_data is not None:
mock.json.return_value = json_data
if raise_on_status:
mock.raise_for_status.side_effect = requests.HTTPError(str(http.HTTPStatus.INTERNAL_SERVER_ERROR))
return mock
class TestParseError:
def test_json_parse_fallback(self) -> None:
mock_response = MagicMock()
mock_response.status_code = http.HTTPStatus.BAD_GATEWAY
mock_response.json = MagicMock(side_effect=ValueError("not json"))
err = requests.HTTPError(str(http.HTTPStatus.BAD_GATEWAY), response=mock_response)
status, message = _parse_error(err)
assert status == http.HTTPStatus.BAD_GATEWAY
assert str(http.HTTPStatus.BAD_GATEWAY) in message
def test_no_response(self) -> None:
err = requests.HTTPError("connection failed")
err.response = None # type: ignore[assignment]
status, message = _parse_error(err)
assert status == 0
assert "connection failed" in message
class TestGiteaClient:
def test_init_sets_headers(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
assert client._base_url == "https://git.example.com"
assert client._session.headers["Authorization"] == "token tok"
assert client._session.headers["Content-Type"] == "application/json"
def test_url_constructs_path(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
assert client._url("/labels") == ("https://git.example.com/repos/owner/repo/labels")
def test_url_strips_trailing_slash(self) -> None:
client = GiteaClient("https://git.example.com/", "tok", "owner", "repo")
assert client._url("/labels") == ("https://git.example.com/repos/owner/repo/labels")
def test_list_labels(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response([{"name": "bug", "color": "ff0000"}]))
result = client.list_labels()
assert len(result) == 1
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/labels",
timeout=DEFAULT_TIMEOUT,
)
def test_list_labels_raises_api_error(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response(raise_on_status=True))
with pytest.raises(APIError):
client.list_labels()
def test_http_error_json_parse_fallback(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.status_code = http.HTTPStatus.BAD_GATEWAY
# Make json() itself raise so the except block in _parse_error is hit
mock_response.json = MagicMock(side_effect=ValueError("not json"))
mock_response.raise_for_status.side_effect = requests.HTTPError(str(http.HTTPStatus.BAD_GATEWAY))
client._session.request = MagicMock(return_value=mock_response)
with pytest.raises(APIError) as exc_info:
client.list_labels()
assert str(http.HTTPStatus.BAD_GATEWAY) in str(exc_info.value)
def test_create_label(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"name": "ready-to-merge", "color": "2ecc71"}))
result = client.create_label("ready-to-merge", "2ecc71", "Auto-merge label")
assert result["name"] == "ready-to-merge"
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/labels",
timeout=DEFAULT_TIMEOUT,
json={"name": "ready-to-merge", "color": "2ecc71", "description": "Auto-merge label"},
)
def test_ensure_label_creates_when_not_exists(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client.list_labels = MagicMock(return_value=[])
client.create_label = MagicMock(return_value={"name": "ready-to-merge", "color": "2ecc71"})
result = client.ensure_label("ready-to-merge", "2ecc71", "desc")
assert result is not None
assert result["name"] == "ready-to-merge"
client.create_label.assert_called_once_with("ready-to-merge", "2ecc71", "desc")
def test_ensure_label_returns_none_when_exists(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client.list_labels = MagicMock(return_value=[{"name": "ready-to-merge", "color": "2ecc71"}])
client.create_label = MagicMock()
result = client.ensure_label("ready-to-merge", "2ecc71", "desc")
assert result is None
client.create_label.assert_not_called()
def test_list_branch_protections(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(
return_value=_mock_response(
[
{"id": 1, "branch_name": "master"},
{"id": 2, "branch_name": "develop"},
]
)
)
result = client.list_branch_protections()
assert len(result) == 2
assert result[0]["branch_name"] == "master"
def test_create_branch_protection(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 3, "branch_name": "master"}))
result = client.create_branch_protection(BRANCH_PROTECTION_CONFIG)
assert result["id"] == 3
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/branch_protections",
timeout=DEFAULT_TIMEOUT,
json=BRANCH_PROTECTION_CONFIG,
)
def test_update_branch_protection(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 1, "required_approvals": 2}))
update = {"required_approvals": 2}
result = client.update_branch_protection(1, update)
assert result["required_approvals"] == 2
client._session.request.assert_called_once_with(
"PATCH",
"https://git.example.com/repos/owner/repo/branch_protections/1",
timeout=DEFAULT_TIMEOUT,
json=update,
)
def test_ensure_branch_protection_creates_when_none_exist(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client.list_branch_protections = MagicMock(return_value=[])
client.create_branch_protection = MagicMock(return_value={"id": 1, "branch_name": "master"})
result = client.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
assert result["id"] == 1
client.create_branch_protection.assert_called_once_with(BRANCH_PROTECTION_CONFIG)
def test_ensure_branch_protection_updates_when_exists(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client.list_branch_protections = MagicMock(return_value=[{"id": 5, "branch_name": "master"}])
client.update_branch_protection = MagicMock(return_value={"id": 5, "required_approvals": 1})
result = client.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
assert result["id"] == 5
expected_update = {k: v for k, v in BRANCH_PROTECTION_CONFIG.items() if k != "branch_name"}
client.update_branch_protection.assert_called_once_with(5, expected_update)
def test_merge_pr(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response())
client.merge_pr(1, "fix: bug")
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/pulls/1/merge",
timeout=DEFAULT_TIMEOUT,
json={"Do": "squash", "MergeTitleField": "fix: bug"},
)
def test_create_release(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 1}))
client.create_release("v1.0.0")
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/releases",
timeout=DEFAULT_TIMEOUT,
json={"tag_name": "v1.0.0", "name": "v1.0.0", "body": "", "draft": False, "prerelease": False},
)
class TestVikunjaClient:
def test_init_sets_headers(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
assert client._base_url == "https://work.example.com"
assert client._session.headers["Authorization"] == "Bearer tok"
def test_list_tasks(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(
return_value=_mock_response([{"id": 1, "identifier": "GRM-19", "project_id": VIKUNJA_PROJECT_ID}])
)
result = client.list_tasks(per_page=DEFAULT_PER_PAGE)
assert len(result) == 1
client._session.request.assert_called_once_with(
"GET",
"https://work.example.com/tasks",
timeout=DEFAULT_TIMEOUT,
params={"per_page": DEFAULT_PER_PAGE},
)
def test_post_comment(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(return_value=_mock_response())
client.post_comment(42, "<p>hi</p>")
client._session.request.assert_called_once_with(
"POST",
"https://work.example.com/tasks/42/comments",
timeout=DEFAULT_TIMEOUT,
json={"comment": "<p>hi</p>"},
)
def test_update_task(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(return_value=_mock_response())
client.update_task(42, done=True)
client._session.request.assert_called_once_with(
"POST",
"https://work.example.com/tasks/42",
timeout=DEFAULT_TIMEOUT,
json={"done": True},
)
def test_http_error_raises_api_error(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(return_value=_mock_response(raise_on_status=True))
with pytest.raises(APIError):
client.list_tasks()
def test_http_error_no_response(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
err = requests.HTTPError("connection failed")
err.response = None # type: ignore[assignment]
client._session.request = MagicMock(side_effect=err)
with pytest.raises(APIError) as exc_info:
client.list_tasks()
assert "connection failed" in str(exc_info.value)
tests/unit/test_auto_merge.py
+23 -48
View File
@@ -1,21 +1,15 @@
"""Unit tests for scripts/auto_merge.py."""
import http
from unittest.mock import MagicMock, patch
import click
import pytest
import requests
from click.testing import CliRunner
from scripts.auto_merge import (
CONVENTIONAL_RE,
GITEA_API,
TASK_ID_RE,
extract_task_id,
main,
merge_pr,
validate_pr_title,
)
from gitea_runner_manager.config import CONVENTIONAL_RE, TASK_ID_RE
from gitea_runner_manager.exceptions import APIError
from scripts.auto_merge import extract_task_id, main, validate_pr_title
class TestRegexes:
@@ -59,32 +53,12 @@ class TestValidatePrTitle:
assert "conventional" in str(exc.value)
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) -> None:
@patch("scripts.auto_merge.GiteaClient")
def test_successful_flow(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
@@ -92,7 +66,8 @@ class TestMain:
)
assert result.exit_code == 0
assert "squash-merged" in result.output
mock_merge.assert_called_once_with("tok", "owner/repo", "7", "GRM-19: fix: resolve timeout")
mock_client_cls.assert_called_once()
mock_client.merge_pr.assert_called_once_with("7", "GRM-19: fix: resolve timeout")
@patch.dict("os.environ", {"GITEA_TOKEN": ""}, clear=True)
def test_missing_token_exits(self) -> None:
@@ -116,23 +91,23 @@ class TestMain:
assert "conventional" in result.output
@patch.dict("os.environ", {"GITEA_TOKEN": "tok"})
@patch("scripts.auto_merge.merge_pr")
def test_merge_pr_failure_raises_click(self, mock_merge: MagicMock) -> None:
mock_merge.side_effect = requests.HTTPError("500")
@patch("scripts.auto_merge.GiteaClient")
def test_merge_pr_failure_raises_click(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.merge_pr.side_effect = APIError(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "fix: bug", "repo", "1"])
result = runner.invoke(main, ["GRM-19-fix", "fix: bug", "owner/repo", "1"])
assert result.exit_code == 1
assert "HTTP" in result.output
@patch.dict("os.environ", {"GITEA_TOKEN": "tok"})
@patch("scripts.auto_merge.merge_pr")
def test_merge_pr_json_parse_failure(self, mock_merge: MagicMock) -> None:
mock_response = MagicMock()
mock_response.status_code = 502
mock_response.json.side_effect = ValueError("not json")
err = requests.HTTPError("502", response=mock_response)
mock_merge.side_effect = err
@patch("scripts.auto_merge.GiteaClient")
def test_merge_pr_json_parse_failure(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.merge_pr.side_effect = APIError(http.HTTPStatus.BAD_GATEWAY, "bad gateway")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "fix: bug", "repo", "1"])
result = runner.invoke(main, ["GRM-19-fix", "fix: bug", "owner/repo", "1"])
assert result.exit_code == 1
assert "502" in result.output
assert str(http.HTTPStatus.BAD_GATEWAY) in result.output
tests/unit/test_config.py
+54
View File
@@ -0,0 +1,54 @@
"""Unit tests for config module constants."""
from gitea_runner_manager.config import (
BRANCH_PROTECTION_CONFIG,
CONVENTIONAL_RE,
DEFAULT_PER_PAGE,
DEFAULT_TIMEOUT,
GITEA_API_URL,
LABEL_CONFIG,
REPO_NAME,
REPO_OWNER,
TASK_ID_RE,
VIKUNJA_API_URL,
VIKUNJA_PROJECT_ID,
)
class TestConfigConstants:
def test_api_urls(self) -> None:
assert GITEA_API_URL == "https://git.oblachno.oblachno.fyi/api/v1"
assert VIKUNJA_API_URL == "https://work.oblachno.oblachno.fyi/api/v1"
def test_project_ids(self) -> None:
assert VIKUNJA_PROJECT_ID == 6
def test_timeouts(self) -> None:
assert DEFAULT_TIMEOUT == 30
assert DEFAULT_PER_PAGE == 50
def test_owner_and_repo(self) -> None:
assert REPO_OWNER == "oblachno-oss"
assert REPO_NAME == "grm"
def test_task_id_re(self) -> None:
assert TASK_ID_RE.search("GRM-1")
assert TASK_ID_RE.search("GRM-123")
assert not TASK_ID_RE.search("GRM-")
assert not TASK_ID_RE.search("other text")
def test_conventional_re(self) -> None:
assert CONVENTIONAL_RE.match("feat: add feature")
assert CONVENTIONAL_RE.match("fix(scope): bug fix")
assert not CONVENTIONAL_RE.match("random message")
assert not CONVENTIONAL_RE.match("feat:")
def test_branch_protection_config(self) -> None:
assert BRANCH_PROTECTION_CONFIG["branch_name"] == "master"
assert BRANCH_PROTECTION_CONFIG["enable_push"] is False
assert BRANCH_PROTECTION_CONFIG["required_approvals"] == 1
assert BRANCH_PROTECTION_CONFIG["status_check_contexts"] == ["lint", "unit-tests", "molecule-tests"]
def test_label_config(self) -> None:
assert LABEL_CONFIG["name"] == "ready-to-merge"
assert LABEL_CONFIG["color"] == "2ecc71"
tests/unit/test_configure_repo.py
+40 -236
View File
@@ -5,202 +5,35 @@ from unittest.mock import MagicMock, patch
import click
import pytest
import requests
from gitea_runner_manager.config import BRANCH_PROTECTION_CONFIG
from gitea_runner_manager.exceptions import APIError
from scripts.configure_repo import (
BRANCH_PROTECTION_CONFIG,
LABEL_CONFIG,
GiteaRepoConfig,
_handle_http_error,
main,
)
class TestGiteaRepoConfig:
def test_init_sets_headers(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
assert cfg._base_url == "https://git.example.com"
assert cfg._owner == "owner"
assert cfg._repo == "repo"
assert cfg._session.headers["Authorization"] == "token tok"
assert cfg._session.headers["Content-Type"] == "application/json"
class TestHandleHttpError:
def test_handle_http_error_403(self) -> None:
err = APIError(http.HTTPStatus.FORBIDDEN, "Forbidden")
with pytest.raises(click.ClickException) as exc:
_handle_http_error(err)
msg = str(exc.value)
assert "admin rights" in msg
assert "Settings → Branches" in msg
def test_url_constructs_path(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
assert cfg._url("/branch_protections") == ("https://git.example.com/repos/owner/repo/branch_protections")
def test_handle_http_error_other(self) -> None:
err = APIError(http.HTTPStatus.INTERNAL_SERVER_ERROR, "Internal Server Error")
with pytest.raises(click.ClickException) as exc:
_handle_http_error(err)
assert str(http.HTTPStatus.INTERNAL_SERVER_ERROR) in str(exc.value)
def test_url_strips_trailing_slash(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com/", "tok", "owner", "repo")
assert cfg._url("/labels") == ("https://git.example.com/repos/owner/repo/labels")
def test_list_branch_protections(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.json.return_value = [
{"id": 1, "branch_name": "master"},
{"id": 2, "branch_name": "develop"},
]
cfg._session.get = MagicMock(return_value=mock_response)
result = cfg.list_branch_protections()
assert len(result) == 2
assert result[0]["branch_name"] == "master"
cfg._session.get.assert_called_once_with("https://git.example.com/repos/owner/repo/branch_protections")
def test_list_branch_protections_raises_on_error(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.raise_for_status.side_effect = requests.HTTPError("500")
cfg._session.get = MagicMock(return_value=mock_response)
with pytest.raises(requests.HTTPError):
cfg.list_branch_protections()
def test_create_branch_protection(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.json.return_value = {"id": 3, "branch_name": "master"}
cfg._session.post = MagicMock(return_value=mock_response)
result = cfg.create_branch_protection(BRANCH_PROTECTION_CONFIG)
assert result["id"] == 3
cfg._session.post.assert_called_once_with(
"https://git.example.com/repos/owner/repo/branch_protections",
json=BRANCH_PROTECTION_CONFIG,
)
def test_create_branch_protection_raises_on_error(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.raise_for_status.side_effect = requests.HTTPError("403")
cfg._session.post = MagicMock(return_value=mock_response)
with pytest.raises(requests.HTTPError):
cfg.create_branch_protection(BRANCH_PROTECTION_CONFIG)
def test_update_branch_protection(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.json.return_value = {"id": 1, "required_approvals": 2}
cfg._session.patch = MagicMock(return_value=mock_response)
update = {"required_approvals": 2}
result = cfg.update_branch_protection(1, update)
assert result["required_approvals"] == 2
cfg._session.patch.assert_called_once_with(
"https://git.example.com/repos/owner/repo/branch_protections/1",
json=update,
)
def test_update_branch_protection_raises_on_error(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.raise_for_status.side_effect = requests.HTTPError("404")
cfg._session.patch = MagicMock(return_value=mock_response)
with pytest.raises(requests.HTTPError):
cfg.update_branch_protection(999, {})
def test_ensure_branch_protection_creates_when_none_exist(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
cfg.list_branch_protections = MagicMock(return_value=[])
cfg.create_branch_protection = MagicMock(return_value={"id": 1, "branch_name": "master"})
result = cfg.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
assert result["id"] == 1
cfg.create_branch_protection.assert_called_once_with(BRANCH_PROTECTION_CONFIG)
def test_ensure_branch_protection_updates_when_exists(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
cfg.list_branch_protections = MagicMock(return_value=[{"id": 5, "branch_name": "master"}])
cfg.update_branch_protection = MagicMock(return_value={"id": 5, "required_approvals": 1})
result = cfg.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
assert result["id"] == 5
# update config should exclude branch_name
expected_update = {k: v for k, v in BRANCH_PROTECTION_CONFIG.items() if k != "branch_name"}
cfg.update_branch_protection.assert_called_once_with(5, expected_update)
def test_ensure_branch_protection_creates_when_other_branches_exist(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
cfg.list_branch_protections = MagicMock(return_value=[{"id": 1, "branch_name": "develop"}])
cfg.create_branch_protection = MagicMock(return_value={"id": 2, "branch_name": "master"})
result = cfg.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
assert result["id"] == 2
cfg.create_branch_protection.assert_called_once_with(BRANCH_PROTECTION_CONFIG)
def test_list_labels(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.json.return_value = [
{"name": "bug", "color": "ff0000"},
{"name": "enhancement", "color": "00ff00"},
]
cfg._session.get = MagicMock(return_value=mock_response)
result = cfg.list_labels()
assert len(result) == 2
cfg._session.get.assert_called_once_with("https://git.example.com/repos/owner/repo/labels")
def test_list_labels_raises_on_error(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.raise_for_status.side_effect = requests.HTTPError("500")
cfg._session.get = MagicMock(return_value=mock_response)
with pytest.raises(requests.HTTPError):
cfg.list_labels()
def test_create_label(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.json.return_value = {"name": "ready-to-merge", "color": "2ecc71"}
cfg._session.post = MagicMock(return_value=mock_response)
result = cfg.create_label("ready-to-merge", "2ecc71", "Auto-merge label")
assert result["name"] == "ready-to-merge"
cfg._session.post.assert_called_once_with(
"https://git.example.com/repos/owner/repo/labels",
json={"name": "ready-to-merge", "color": "2ecc71", "description": "Auto-merge label"},
)
def test_create_label_raises_on_error(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.raise_for_status.side_effect = requests.HTTPError("422")
cfg._session.post = MagicMock(return_value=mock_response)
with pytest.raises(requests.HTTPError):
cfg.create_label("dup", "ffffff")
def test_ensure_label_creates_when_not_exists(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
cfg.list_labels = MagicMock(return_value=[])
cfg.create_label = MagicMock(return_value={"name": "ready-to-merge", "color": "2ecc71"})
result = cfg.ensure_label("ready-to-merge", "2ecc71", "desc")
assert result is not None
assert result["name"] == "ready-to-merge"
cfg.create_label.assert_called_once_with("ready-to-merge", "2ecc71", "desc")
def test_ensure_label_returns_none_when_exists(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
cfg.list_labels = MagicMock(return_value=[{"name": "ready-to-merge", "color": "2ecc71"}])
cfg.create_label = MagicMock()
result = cfg.ensure_label("ready-to-merge", "2ecc71", "desc")
assert result is None
cfg.create_label.assert_not_called()
def test_ensure_label_creates_when_other_labels_exist(self) -> None:
cfg = GiteaRepoConfig("https://git.example.com", "tok", "owner", "repo")
cfg.list_labels = MagicMock(return_value=[{"name": "bug", "color": "ff0000"}])
cfg.create_label = MagicMock(return_value={"name": "ready-to-merge", "color": "2ecc71"})
result = cfg.ensure_label("ready-to-merge", "2ecc71", "desc")
assert result is not None
cfg.create_label.assert_called_once()
def test_handle_http_error_json_parse_fails(self) -> None:
err = APIError(http.HTTPStatus.BAD_GATEWAY, "bad gateway")
with pytest.raises(click.ClickException) as exc:
_handle_http_error(err)
assert str(http.HTTPStatus.BAD_GATEWAY) in str(exc.value)
class TestMain:
@@ -212,73 +45,44 @@ class TestMain:
def test_main_success(self) -> None:
with patch.dict("os.environ", {"GITEA_ADMIN_TOKEN": "tok"}, clear=True):
with patch("scripts.configure_repo.GiteaRepoConfig") as mock_cfg_class:
mock_cfg = MagicMock()
mock_cfg_class.return_value = mock_cfg
with patch("scripts.configure_repo.GiteaClient") as mock_client_cls:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
main()
mock_cfg.ensure_branch_protection.assert_called_once_with("master", BRANCH_PROTECTION_CONFIG)
mock_cfg.ensure_label.assert_called_once_with(**LABEL_CONFIG)
mock_client.ensure_branch_protection.assert_called_once_with("master", BRANCH_PROTECTION_CONFIG)
mock_client.ensure_label.assert_called_once()
def test_main_label_already_exists(self) -> None:
with patch.dict("os.environ", {"GITEA_ADMIN_TOKEN": "tok"}, clear=True):
with patch("scripts.configure_repo.GiteaRepoConfig") as mock_cfg_class:
mock_cfg = MagicMock()
mock_cfg.ensure_label.return_value = None
mock_cfg_class.return_value = mock_cfg
with patch("scripts.configure_repo.GiteaClient") as mock_client_cls:
mock_client = MagicMock()
mock_client.ensure_label.return_value = None
mock_client_cls.return_value = mock_client
main()
mock_cfg.ensure_branch_protection.assert_called_once_with("master", BRANCH_PROTECTION_CONFIG)
mock_cfg.ensure_label.assert_called_once_with(**LABEL_CONFIG)
mock_client.ensure_branch_protection.assert_called_once_with("master", BRANCH_PROTECTION_CONFIG)
mock_client.ensure_label.assert_called_once()
def test_main_api_error(self) -> None:
with patch.dict("os.environ", {"GITEA_ADMIN_TOKEN": "tok"}, clear=True):
with patch("scripts.configure_repo.GiteaRepoConfig") as mock_cfg_class:
mock_cfg = MagicMock()
mock_cfg.ensure_branch_protection.side_effect = requests.HTTPError("403")
mock_cfg_class.return_value = mock_cfg
with patch("scripts.configure_repo.GiteaClient") as mock_client_cls:
mock_client = MagicMock()
mock_client.ensure_branch_protection.side_effect = APIError(http.HTTPStatus.FORBIDDEN, "Forbidden")
mock_client_cls.return_value = mock_client
with pytest.raises(click.ClickException) as exc:
main()
assert "HTTP" in str(exc.value)
def test_handle_http_error_403(self) -> None:
mock_response = MagicMock()
mock_response.status_code = http.HTTPStatus.FORBIDDEN
mock_response.json.return_value = {"message": "Forbidden"}
err = requests.HTTPError("403", response=mock_response)
with pytest.raises(click.ClickException) as exc:
_handle_http_error(err)
msg = str(exc.value)
assert "admin rights" in msg
assert "Settings → Branches" in msg
def test_handle_http_error_other(self) -> None:
mock_response = MagicMock()
mock_response.status_code = http.HTTPStatus.INTERNAL_SERVER_ERROR
mock_response.json.return_value = {"message": "Internal Server Error"}
err = requests.HTTPError("500", response=mock_response)
with pytest.raises(click.ClickException) as exc:
_handle_http_error(err)
assert "500" in str(exc.value)
def test_handle_http_error_json_parse_fails(self) -> None:
mock_response = MagicMock()
mock_response.status_code = http.HTTPStatus.BAD_GATEWAY
mock_response.json.side_effect = ValueError("not json")
err = requests.HTTPError("502", response=mock_response)
with pytest.raises(click.ClickException) as exc:
_handle_http_error(err)
assert "502" in str(exc.value)
def test_main_module_block() -> None:
with patch.dict("os.environ", {"GITEA_ADMIN_TOKEN": "tok"}, clear=True):
with patch("scripts.configure_repo.GiteaRepoConfig") as mock_cfg_class:
mock_cfg = MagicMock()
mock_cfg_class.return_value = mock_cfg
with patch("scripts.configure_repo.GiteaClient") as mock_client_cls:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
import scripts.configure_repo as cr
with open(cr.__file__) as f:
@@ -287,6 +91,6 @@ def test_main_module_block() -> None:
source = source.replace('if __name__ == "__main__":\n main()\n', "")
namespace = dict(cr.__dict__)
exec(compile(source, cr.__file__, "exec"), namespace)
namespace["GiteaRepoConfig"] = mock_cfg_class
namespace["GiteaClient"] = mock_client_cls
namespace["main"]()
mock_cfg.ensure_branch_protection.assert_called_once_with("master", BRANCH_PROTECTION_CONFIG)
mock_client.ensure_branch_protection.assert_called_once_with("master", BRANCH_PROTECTION_CONFIG)
tests/unit/test_post_merge.py
+64 -115
View File
@@ -1,21 +1,19 @@
"""Unit tests for scripts/post_merge.py."""
import http
from unittest.mock import MagicMock, patch
import click
import pytest
import requests
from click.testing import CliRunner
from gitea_runner_manager.config import VIKUNJA_PROJECT_ID
from gitea_runner_manager.exceptions import APIError
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,
)
@@ -45,102 +43,46 @@ class TestBuildComment:
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"},
def test_found(self) -> None:
mock_client = MagicMock()
mock_client.list_tasks.return_value = [
{"id": 42, "project_id": VIKUNJA_PROJECT_ID, "identifier": "GRM-19"},
]
mock_get.return_value = mock_response
assert resolve_task_id("tok", "GRM-19") == 42
mock_get.assert_called_once()
assert resolve_task_id(mock_client, "GRM-19") == 42
mock_client.list_tasks.assert_called_once()
@patch("scripts.post_merge.requests.get")
def test_not_found_raises(self, mock_get: MagicMock) -> None:
mock_response = MagicMock()
mock_response.json.return_value = []
mock_get.return_value = mock_response
def test_not_found_raises(self) -> None:
mock_client = MagicMock()
mock_client.list_tasks.return_value = []
with pytest.raises(click.ClickException) as exc:
resolve_task_id("tok", "GRM-99")
resolve_task_id(mock_client, "GRM-99")
assert "Could not find" in str(exc.value)
@patch("scripts.post_merge.requests.get")
def test_wrong_project_filtered(self, mock_get: MagicMock) -> None:
mock_response = MagicMock()
mock_response.json.return_value = [
def test_wrong_project_filtered(self) -> None:
mock_client = MagicMock()
mock_client.list_tasks.return_value = [
{"id": 42, "project_id": 999, "identifier": "GRM-19"},
]
mock_get.return_value = mock_response
with pytest.raises(click.ClickException) as exc:
resolve_task_id("tok", "GRM-19")
resolve_task_id(mock_client, "GRM-19")
assert "Could not find" in str(exc.value)
@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 TestHandleHttpError:
def test_json_parse_failure(self) -> None:
mock_response = MagicMock()
mock_response.status_code = 502
mock_response.json.side_effect = ValueError("not json")
err = requests.HTTPError("502", response=mock_response)
with pytest.raises(click.ClickException) as exc:
from scripts.post_merge import _handle_http_error
_handle_http_error(err)
assert "502" in str(exc.value)
def test_http_error_propagates(self) -> None:
mock_client = MagicMock()
mock_client.list_tasks.side_effect = APIError(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error")
with pytest.raises(APIError):
resolve_task_id(mock_client, "GRM-19")
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) -> None:
mock_resolve.return_value = 267
@patch("scripts.post_merge.VikunjaClient")
def test_full_flow(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_tasks.return_value = [
{"id": 267, "project_id": VIKUNJA_PROJECT_ID, "identifier": "GRM-20"},
]
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
@@ -148,22 +90,23 @@ class TestMain:
)
assert result.exit_code == 0
assert "updated and marked done" in result.output
mock_resolve.assert_called_once_with("tok", "GRM-20")
mock_post.assert_called_once()
mock_mark.assert_called_once_with("tok", 267)
mock_client.post_comment.assert_called_once()
mock_client.update_task.assert_called_once_with(267, done=True)
@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
@patch("scripts.post_merge.VikunjaClient")
def test_no_commit_sha(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_tasks.return_value = [
{"id": 267, "project_id": VIKUNJA_PROJECT_ID, "identifier": "GRM-20"},
]
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-20: fix: resolve bug"])
assert result.exit_code == 0
mock_post.assert_called_once()
args, _ = mock_post.call_args
assert "unknown" in args[2]
mock_client.post_comment.assert_called_once()
args, _ = mock_client.post_comment.call_args
assert "unknown" in args[1]
@patch.dict("os.environ", {"VIKUNJA_TOKEN": ""}, clear=True)
def test_missing_token_exits(self) -> None:
@@ -180,34 +123,40 @@ class TestMain:
assert "skipping" in result.output
@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 = click.ClickException("not found")
@patch("scripts.post_merge.VikunjaClient")
def test_resolve_failure_propagates(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_tasks.return_value = []
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-20: fix: bug"])
assert result.exit_code == 1
assert "not found" in result.output
assert "Could not find" in result.output
@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_raises_click(self, mock_post: MagicMock, mock_resolve: MagicMock) -> None:
mock_resolve.return_value = 267
mock_post.side_effect = requests.HTTPError("500")
@patch("scripts.post_merge.VikunjaClient")
def test_post_comment_failure_raises_click(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_tasks.return_value = [
{"id": 267, "project_id": VIKUNJA_PROJECT_ID, "identifier": "GRM-20"},
]
mock_client.post_comment.side_effect = APIError(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-20: fix: bug"])
assert result.exit_code == 1
assert "HTTP" in result.output
@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_raises_click(
self, mock_mark: MagicMock, mock_post: MagicMock, mock_resolve: MagicMock
) -> None:
mock_resolve.return_value = 267
mock_mark.side_effect = requests.HTTPError("500")
@patch("scripts.post_merge.VikunjaClient")
def test_mark_done_failure_raises_click(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_tasks.return_value = [
{"id": 267, "project_id": VIKUNJA_PROJECT_ID, "identifier": "GRM-20"},
]
mock_client.post_comment.return_value = None
mock_client.update_task.side_effect = APIError(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-20: fix: bug"])
assert result.exit_code == 1
tests/unit/test_publish.py
+26 -45
View File
@@ -1,16 +1,14 @@
"""Unit tests for scripts/publish.py."""
import http
from unittest.mock import MagicMock, patch
import click
import pytest
import requests
from click.testing import CliRunner
from scripts.publish import (
GITEA_API,
build_package,
create_gitea_release,
main,
publish_to_pypi,
)
@@ -50,37 +48,16 @@ class TestPublishToPypi:
assert "PyPI" in str(exc.value)
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.GiteaClient")
@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,
mock_client_cls: MagicMock,
) -> None:
runner = CliRunner()
result = runner.invoke(main, ["v1.0.0", "owner/repo"])
@@ -88,21 +65,21 @@ class TestMain:
assert "Gitea release v1.0.0 created" in result.output
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")
mock_client_cls.return_value.create_release.assert_called_once()
@patch.dict("os.environ", {"GITEA_TOKEN": "gitea-tok"}, clear=True)
@patch("scripts.publish.create_gitea_release")
@patch("scripts.publish.GiteaClient")
@patch("scripts.publish.build_package")
def test_without_pypi(
self,
mock_build: MagicMock,
mock_release: MagicMock,
mock_client_cls: MagicMock,
) -> None:
runner = CliRunner()
result = runner.invoke(main, ["v1.0.0", "owner/repo"])
assert result.exit_code == 0
mock_build.assert_called_once()
mock_release.assert_called_once_with("gitea-tok", "owner/repo", "v1.0.0")
mock_client_cls.return_value.create_release.assert_called_once()
assert "PYPI_TOKEN not set" in result.output
@patch.dict("os.environ", {"GITEA_TOKEN": ""}, clear=True)
@@ -113,11 +90,11 @@ class TestMain:
assert "GITEA_TOKEN" in result.output
@patch.dict("os.environ", {"GITEA_TOKEN": "gitea-tok", "PYPI_TOKEN": "pypi-tok"})
@patch("scripts.publish.create_gitea_release")
@patch("scripts.publish.GiteaClient")
@patch("scripts.publish.publish_to_pypi")
@patch("scripts.publish.build_package")
def test_build_failure_raises_click(
self, mock_build: MagicMock, mock_publish: MagicMock, mock_release: MagicMock
self, mock_build: MagicMock, mock_publish: MagicMock, mock_client_cls: MagicMock
) -> None:
mock_build.side_effect = click.ClickException("build failed")
runner = CliRunner()
@@ -126,11 +103,11 @@ class TestMain:
assert "build" in result.output
@patch.dict("os.environ", {"GITEA_TOKEN": "gitea-tok", "PYPI_TOKEN": "pypi-tok"})
@patch("scripts.publish.create_gitea_release")
@patch("scripts.publish.GiteaClient")
@patch("scripts.publish.publish_to_pypi")
@patch("scripts.publish.build_package")
def test_publish_failure_raises_click(
self, mock_build: MagicMock, mock_publish: MagicMock, mock_release: MagicMock
self, mock_build: MagicMock, mock_publish: MagicMock, mock_client_cls: MagicMock
) -> None:
mock_publish.side_effect = click.ClickException("publish failed")
runner = CliRunner()
@@ -139,31 +116,35 @@ class TestMain:
assert "publish" in result.output
@patch.dict("os.environ", {"GITEA_TOKEN": "gitea-tok", "PYPI_TOKEN": "pypi-tok"})
@patch("scripts.publish.create_gitea_release")
@patch("scripts.publish.GiteaClient")
@patch("scripts.publish.publish_to_pypi")
@patch("scripts.publish.build_package")
def test_release_failure_raises_click(
self, mock_build: MagicMock, mock_publish: MagicMock, mock_release: MagicMock
self, mock_build: MagicMock, mock_publish: MagicMock, mock_client_cls: MagicMock
) -> None:
mock_release.side_effect = requests.HTTPError("500")
mock_client = MagicMock()
from gitea_runner_manager.exceptions import APIError
mock_client.create_release.side_effect = APIError(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["v1.0.0", "owner/repo"])
assert result.exit_code == 1
assert "HTTP" in result.output
@patch.dict("os.environ", {"GITEA_TOKEN": "gitea-tok", "PYPI_TOKEN": "pypi-tok"})
@patch("scripts.publish.create_gitea_release")
@patch("scripts.publish.GiteaClient")
@patch("scripts.publish.publish_to_pypi")
@patch("scripts.publish.build_package")
def test_release_json_parse_failure(
self, mock_build: MagicMock, mock_publish: MagicMock, mock_release: MagicMock
self, mock_build: MagicMock, mock_publish: MagicMock, mock_client_cls: MagicMock
) -> None:
mock_response = MagicMock()
mock_response.status_code = 502
mock_response.json.side_effect = ValueError("not json")
err = requests.HTTPError("502", response=mock_response)
mock_release.side_effect = err
mock_client = MagicMock()
from gitea_runner_manager.exceptions import APIError
mock_client.create_release.side_effect = APIError(http.HTTPStatus.BAD_GATEWAY, "bad gateway")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["v1.0.0", "owner/repo"])
assert result.exit_code == 1
assert "502" in result.output
assert str(http.HTTPStatus.BAD_GATEWAY) in result.output
tests/unit/test_validate_commit_msg.py
+2 -1
View File
@@ -7,7 +7,8 @@ from unittest.mock import patch
from click.testing import CliRunner
from scripts.validate_commit_msg import CONVENTIONAL_RE, TASK_ID_RE, first_line, get_branch, main
from gitea_runner_manager.config import CONVENTIONAL_RE, TASK_ID_RE
from scripts.validate_commit_msg import first_line, get_branch, main
class TestHelpers: