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}
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: