From fefddda715180841a3ef116381a778203072c9ee Mon Sep 17 00:00:00 2001 From: Emil Simeonov Date: Fri, 19 Jun 2026 21:00:21 +0200 Subject: [PATCH] GRM-20: refactor(scripts): centralize constants, API clients, and HTTP status codes - 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 --- scripts/auto_merge.py | 44 ++-- scripts/configure_repo.py | 131 +++-------- scripts/post_merge.py | 83 ++----- scripts/publish.py | 44 ++-- scripts/validate_commit_msg.py | 12 +- src/gitea_runner_manager/api_clients.py | 153 +++++++++++++ src/gitea_runner_manager/config.py | 39 ++++ src/gitea_runner_manager/exceptions.py | 9 + tests/unit/test_api_clients.py | 264 +++++++++++++++++++++++ tests/unit/test_auto_merge.py | 71 ++---- tests/unit/test_config.py | 54 +++++ tests/unit/test_configure_repo.py | 276 ++++-------------------- tests/unit/test_post_merge.py | 179 ++++++--------- tests/unit/test_publish.py | 71 +++--- tests/unit/test_validate_commit_msg.py | 3 +- 15 files changed, 751 insertions(+), 682 deletions(-) create mode 100644 src/gitea_runner_manager/api_clients.py create mode 100644 src/gitea_runner_manager/config.py create mode 100644 tests/unit/test_api_clients.py create mode 100644 tests/unit/test_config.py diff --git a/scripts/auto_merge.py b/scripts/auto_merge.py index eef181c..d385b1a 100644 --- a/scripts/auto_merge.py +++ b/scripts/auto_merge.py @@ -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 diff --git a/scripts/configure_repo.py b/scripts/configure_repo.py index 3e0fc7d..518209f 100644 --- a/scripts/configure_repo.py +++ b/scripts/configure_repo.py @@ -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) diff --git a/scripts/post_merge.py b/scripts/post_merge.py index 099c746..4b20cef 100644 --- a/scripts/post_merge.py +++ b/scripts/post_merge.py @@ -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"

{task_id}: {conv_msg}

Commit: {commit_sha}

" @@ -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( _( diff --git a/scripts/publish.py b/scripts/publish.py index 7f7c01d..4908715 100644 --- a/scripts/publish.py +++ b/scripts/publish.py @@ -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 diff --git a/scripts/validate_commit_msg.py b/scripts/validate_commit_msg.py index 9cf2396..de0b362 100644 --- a/scripts/validate_commit_msg.py +++ b/scripts/validate_commit_msg.py @@ -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" diff --git a/src/gitea_runner_manager/api_clients.py b/src/gitea_runner_manager/api_clients.py new file mode 100644 index 0000000..30f5d6d --- /dev/null +++ b/src/gitea_runner_manager/api_clients.py @@ -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) diff --git a/src/gitea_runner_manager/config.py b/src/gitea_runner_manager/config.py new file mode 100644 index 0000000..cbc7c8b --- /dev/null +++ b/src/gitea_runner_manager/config.py @@ -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", +} diff --git a/src/gitea_runner_manager/exceptions.py b/src/gitea_runner_manager/exceptions.py index 230c6bd..e246799 100644 --- a/src/gitea_runner_manager/exceptions.py +++ b/src/gitea_runner_manager/exceptions.py @@ -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}") diff --git a/tests/unit/test_api_clients.py b/tests/unit/test_api_clients.py new file mode 100644 index 0000000..23e4806 --- /dev/null +++ b/tests/unit/test_api_clients.py @@ -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, "

hi

") + client._session.request.assert_called_once_with( + "POST", + "https://work.example.com/tasks/42/comments", + timeout=DEFAULT_TIMEOUT, + json={"comment": "

hi

"}, + ) + + 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) diff --git a/tests/unit/test_auto_merge.py b/tests/unit/test_auto_merge.py index d860cd4..7ab4314 100644 --- a/tests/unit/test_auto_merge.py +++ b/tests/unit/test_auto_merge.py @@ -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 diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..20876f9 --- /dev/null +++ b/tests/unit/test_config.py @@ -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" diff --git a/tests/unit/test_configure_repo.py b/tests/unit/test_configure_repo.py index 50f2e76..51b68b4 100644 --- a/tests/unit/test_configure_repo.py +++ b/tests/unit/test_configure_repo.py @@ -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) diff --git a/tests/unit/test_post_merge.py b/tests/unit/test_post_merge.py index 8a95732..e3cfdf9 100644 --- a/tests/unit/test_post_merge.py +++ b/tests/unit/test_post_merge.py @@ -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, "

hi

") - args, kwargs = mock_post.call_args - assert VIKUNJA_API in args[0] - assert kwargs["json"]["comment"] == "

hi

" - - @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 diff --git a/tests/unit/test_publish.py b/tests/unit/test_publish.py index 15db66a..25613f1 100644 --- a/tests/unit/test_publish.py +++ b/tests/unit/test_publish.py @@ -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 diff --git a/tests/unit/test_validate_commit_msg.py b/tests/unit/test_validate_commit_msg.py index 3ac53c8..ce231a8 100644 --- a/tests/unit/test_validate_commit_msg.py +++ b/tests/unit/test_validate_commit_msg.py @@ -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: