From e91780169e11d6d2b4db5ce053835ac4a490af19 Mon Sep 17 00:00:00 2001 From: meeks Date: Fri, 17 Jul 2026 07:31:05 +0200 Subject: [PATCH] refactor: extract focused clients from GiteaClient (Slices 1-6) - Create gitea/issues_client.py with IssuesClient class (9 methods) - Create gitea/prs_client.py with PullRequestsClient class (17 methods) - Create gitea/files_client.py with FilesClient class (4 methods) - Create gitea/notifications_client.py with NotificationsClient class (2 methods) - Create gitea/repos_client.py with ReposClient class (2 methods) - Create gitea/__init__.py to export all client classes - Remove delegation methods from GiteaClient (now ~70 lines) - Update all callers to use sub-clients (client.issues, client.prs, etc.) - Update test files to mock sub-client attributes GiteaClient is now a facade that provides access to focused sub-clients: - repos: Repository operations (ReposClient) - issues: Issue operations (IssuesClient) - prs: Pull request operations (PullRequestsClient) - files: File and git ref operations (FilesClient) - notifications: Notification operations (NotificationsClient) Refs: #godclass-refactor --- core/dispatcher.py | 54 ++-- core/orchestrator.py | 18 +- gitea/__init__.py | 35 +++ gitea/client.py | 504 +++------------------------------- gitea/files_client.py | 116 ++++++++ gitea/issues_client.py | 248 +++++++++++++++++ gitea/notifications_client.py | 78 ++++++ gitea/prs_client.py | 462 +++++++++++++++++++++++++++++++ gitea/repos_client.py | 72 +++++ gitea/tools/file_tools.py | 16 +- gitea/tools/git_tools.py | 2 +- gitea/tools/issue_tools.py | 18 +- gitea/tools/pr_tools.py | 61 ++-- gitea/workspace.py | 85 ++++-- main.py | 2 +- tests/test_client.py | 64 +++-- tests/test_file_tools.py | 91 +++--- tests/test_git_tools.py | 17 +- tests/test_issue_tools.py | 88 +++--- tests/test_pr_tools.py | 122 ++++---- tests/test_workspace.py | 55 ++-- 21 files changed, 1480 insertions(+), 728 deletions(-) create mode 100644 gitea/__init__.py create mode 100644 gitea/files_client.py create mode 100644 gitea/issues_client.py create mode 100644 gitea/notifications_client.py create mode 100644 gitea/prs_client.py create mode 100644 gitea/repos_client.py diff --git a/core/dispatcher.py b/core/dispatcher.py index 25b36b2..1fa6793 100644 --- a/core/dispatcher.py +++ b/core/dispatcher.py @@ -43,7 +43,7 @@ def _find_pr_for_issue_helper( """Find an open pull request that addresses the given issue number.""" owner, repo_name = repo_full_name.split("/") try: - prs = client.list_repo_pull_requests(owner, repo_name) + prs = client.prs.list_repo_pull_requests(owner, repo_name) for pr in prs: ref = pr.head.get("ref", "") if pr.head else "" if re.search(rf"(? str: try: - pr_detail = self.client.get_pull_request( + pr_detail = self.client.prs.get_pull_request( self.owner, self.repo_name, self.item.task_number ) except Exception as e: logger.warning(f"Error fetching PR #{self.item.task_number} detail: {e}") return f"FAILED: Could not fetch details for PR #{self.item.task_number}." - is_own_pr = pr_detail.user and pr_detail.user.login == self.ai_username + is_own_pr = bool(pr_detail.user and pr_detail.user.login == self.ai_username) is_requested_reviewer = any( r.login == self.ai_username for r in pr_detail.requested_reviewers ) @@ -412,7 +416,7 @@ class PRTaskProcessor(TaskProcessor): pr_comments = [] try: - pr_comments = self.client.get_pull_request_comments( + pr_comments = self.client.prs.get_pull_request_comments( self.owner, self.repo_name, self.item.task_number ) except Exception as e: @@ -490,7 +494,7 @@ class IssueTaskProcessor(TaskProcessor): comments: list[CommentModel] = [] try: - comments = self.client.get_issue_comments( + comments = self.client.issues.get_issue_comments( self.owner, self.repo_name, issue_number ) except Exception as e: @@ -558,7 +562,7 @@ class IssueTaskProcessor(TaskProcessor): is_wip = title_upper.startswith("WIP:") or "[WIP]" in title_upper try: - reviews = self.client.get_pr_reviews( + reviews = self.client.prs.get_pr_reviews( self.owner, self.repo_name, existing_pr.number ) has_request_changes = any( @@ -578,7 +582,7 @@ class IssueTaskProcessor(TaskProcessor): # Check comments on issue and PR issue_comments = [] try: - issue_comments = self.client.get_issue_comments( + issue_comments = self.client.issues.get_issue_comments( self.owner, self.repo_name, self.item.task_number ) except Exception as e: @@ -590,7 +594,7 @@ class IssueTaskProcessor(TaskProcessor): pr_comments = [] if existing_pr: try: - pr_comments = self.client.get_pull_request_comments( + pr_comments = self.client.prs.get_pull_request_comments( self.owner, self.repo_name, existing_pr.number ) except Exception as e: @@ -639,7 +643,7 @@ class IssueTaskProcessor(TaskProcessor): else "No PR comments yet." ) try: - reviews = self.client.get_pr_reviews( + reviews = self.client.prs.get_pr_reviews( self.owner, self.repo_name, existing_pr.number ) reviews_str = ( @@ -703,7 +707,7 @@ class IssueTaskProcessor(TaskProcessor): f"\n" f"" ) - self.client.add_comment( + self.client.issues.add_comment( self.owner, self.repo_name, self.item.task_number, comment_body ) return f"POSTED_COMMENT: PROPOSE_PLAN comment posted to issue #{self.item.task_number}." @@ -718,7 +722,7 @@ class IssueTaskProcessor(TaskProcessor): f"\n" f"" ) - self.client.add_comment( + self.client.issues.add_comment( self.owner, self.repo_name, self.item.task_number, comment_body ) return f"POSTED_COMMENT: ANSWER_QUESTION comment posted to issue #{self.item.task_number}." @@ -727,10 +731,10 @@ class IssueTaskProcessor(TaskProcessor): comment = coord_tools.arguments.get( "comment", "Closing the issue as resolved." ) - self.client.add_comment( + self.client.issues.add_comment( self.owner, self.repo_name, self.item.task_number, comment ) - self.client.close_issue( + self.client.issues.close_issue( self.owner, self.repo_name, self.item.task_number ) return f"CLOSED_ISSUE: Issue #{self.item.task_number} closed." @@ -805,7 +809,7 @@ class IssueTaskProcessor(TaskProcessor): pr_description = ( f"Work in progress for issue #{self.item.task_number}." ) - pr_to_use = self.client.create_pull_request( + pr_to_use = self.client.prs.create_pull_request( self.owner, self.repo_name, head=branch_name, @@ -821,7 +825,7 @@ class IssueTaskProcessor(TaskProcessor): start_comment = ( f"Started work on PR #{pr_to_use.number} ({pr_link})." ) - self.client.add_comment( + self.client.issues.add_comment( self.owner, self.repo_name, self.item.task_number, @@ -919,7 +923,7 @@ class AgentDispatcher: results: list[str] = [] # Get authenticated username for reviewer filter try: - user = self._client.get_authenticated_user() + user = self._client.repos.get_authenticated_user() except Exception as e: raise RuntimeError("No authenticated user found.") from e @@ -973,7 +977,7 @@ class AgentDispatcher: return _find_pr_for_issue_helper(self._client, repo_full_name, issue_number) def _is_awaiting_reply(self, comments: list[CommentModel]) -> bool: - user = self._client.get_authenticated_user() + user = self._client.repos.get_authenticated_user() if not user or not user.login: raise RuntimeError("No authenticated user found.") return _is_awaiting_reply_helper(comments, user.login) @@ -982,7 +986,7 @@ class AgentDispatcher: pr_info = item.task_info if not isinstance(pr_info, PullRequestModel): raise TypeError("Expected task_info to be a PullRequestModel") - user = self._client.get_authenticated_user() + user = self._client.repos.get_authenticated_user() if not user or not user.login: raise RuntimeError("No authenticated user found.") processor = PRTaskProcessor( @@ -1002,7 +1006,7 @@ class AgentDispatcher: issue_info = item.task_info if not isinstance(issue_info, IssueModel): raise TypeError("Expected task_info to be an IssueModel") - user = self._client.get_authenticated_user() + user = self._client.repos.get_authenticated_user() if not user or not user.login: raise RuntimeError("No authenticated user found.") processor = IssueTaskProcessor( diff --git a/core/orchestrator.py b/core/orchestrator.py index 30fdebd..329e302 100644 --- a/core/orchestrator.py +++ b/core/orchestrator.py @@ -90,7 +90,9 @@ class AgentOrchestrator: f"Polling unread notifications since: {last_checked or 'beginning'}" ) - notifications = self._client.list_unread_notifications(since=last_checked) + notifications = self._client.notifications.list_unread_notifications( + since=last_checked + ) if not notifications: logger.info("No new notifications found.") @@ -165,7 +167,9 @@ class AgentOrchestrator: f"Skipping notification {notification_id} for {repo_full_name}#{task_number}. Reason: {reason}" ) if notification_id is not None: - self._client.mark_notification_as_read(notification_id) + self._client.notifications.mark_notification_as_read( + notification_id + ) logger.info( f"Marked skipped Gitea notification thread {notification_id} as read." ) @@ -174,7 +178,7 @@ class AgentOrchestrator: # Route based on decided action if notification_tools.action == "PROCESS_ISSUE": try: - issue = self._client.get_issue(owner, repo_name, task_number) + issue = self._client.issues.get_issue(owner, repo_name, task_number) if issue.repository is None: issue = issue.model_copy( update={"repository": RepositoryModel(**repo_info)} @@ -195,7 +199,9 @@ class AgentOrchestrator: ) elif notification_tools.action == "PROCESS_PR": try: - pr = self._client.get_pull_request(owner, repo_name, task_number) + pr = self._client.prs.get_pull_request( + owner, repo_name, task_number + ) if pr.repository is None: pr = pr.model_copy( update={"repository": RepositoryModel(**repo_info)} @@ -252,7 +258,9 @@ class AgentOrchestrator: f"Completed {item.task_type} #{item.task_number}: {result[:200]}" ) if item.notification_id is not None: - self._client.mark_notification_as_read(item.notification_id) + self._client.notifications.mark_notification_as_read( + item.notification_id + ) logger.info( f"Marked Gitea notification thread {item.notification_id} as read." ) diff --git a/gitea/__init__.py b/gitea/__init__.py new file mode 100644 index 0000000..6e8a9bb --- /dev/null +++ b/gitea/__init__.py @@ -0,0 +1,35 @@ +"""Gitea API client package.""" + +from .client import GiteaClient +from .files_client import FilesClient +from .issues_client import IssuesClient +from .notifications_client import NotificationsClient +from .prs_client import PullRequestsClient +from .repos_client import ReposClient +from .models import ( + CommentModel, + GiteaConfig, + IssueModel, + LabelModel, + PullRequestFileModel, + PullRequestModel, + RepositoryModel, + UserModel, +) + +__all__ = [ + "FilesClient", + "GiteaClient", + "IssuesClient", + "NotificationsClient", + "PullRequestsClient", + "ReposClient", + "CommentModel", + "GiteaConfig", + "IssueModel", + "LabelModel", + "PullRequestFileModel", + "PullRequestModel", + "RepositoryModel", + "UserModel", +] diff --git a/gitea/client.py b/gitea/client.py index e3b4174..a402304 100644 --- a/gitea/client.py +++ b/gitea/client.py @@ -1,25 +1,28 @@ import httpx -import json -import base64 import logging -from typing import Any, Optional +from typing import Any logger: logging.Logger = logging.getLogger("gitea.client") from .config import GITEA_URL, GITEA_TOKEN, GITEA_ORG_FILTER -from .models import ( - UserModel, - LabelModel, - RepositoryModel, - IssueModel, - PullRequestModel, - CommentModel, - PullRequestFileModel, -) +from .files_client import FilesClient +from .issues_client import IssuesClient +from .notifications_client import NotificationsClient +from .prs_client import PullRequestsClient +from .repos_client import ReposClient class GiteaClient: - """HTTP client for Gitea API v1.""" + """HTTP client for Gitea API v1. + + This is a facade class that provides access to focused sub-clients + for different API domains: + - repos: Repository operations (ReposClient) + - issues: Issue operations (IssuesClient) + - prs: Pull request operations (PullRequestsClient) + - files: File and git ref operations (FilesClient) + - notifications: Notification operations (NotificationsClient) + """ def __init__(self) -> None: self.base_url: str = GITEA_URL.rstrip("/") @@ -28,6 +31,25 @@ class GiteaClient: "Accept": "application/json", } self.client: httpx.Client = httpx.Client(headers=self.headers) + self.repos: ReposClient = ReposClient( + self.base_url, self.client, GITEA_ORG_FILTER + ) + self.issues: IssuesClient = IssuesClient( + self.base_url, + self.client, + get_user=self.repos.get_authenticated_user, + get_repos=self.repos.list_all_user_repos, + ) + self.prs: PullRequestsClient = PullRequestsClient( + self.base_url, + self.client, + get_user=self.repos.get_authenticated_user, + get_repos=self.repos.list_all_user_repos, + ) + self.files: FilesClient = FilesClient(self.base_url, self.client) + self.notifications: NotificationsClient = NotificationsClient( + self.base_url, self.client, GITEA_ORG_FILTER + ) def close(self) -> None: """Close the underlying HTTP client.""" @@ -44,459 +66,3 @@ class GiteaClient: self.client.close() except Exception: pass - - def get_authenticated_user(self) -> UserModel: - try: - response = self.client.get(f"{self.base_url}/api/v1/user") - response.raise_for_status() - return UserModel(**response.json()) - except Exception as e: - logger.error(f"Error getting authenticated user: {e}", exc_info=True) - raise RuntimeError(f"Could not get authenticated user: {e}") from e - - def list_all_user_repos(self) -> list[RepositoryModel]: - try: - url = f"{self.base_url}/api/v1/user/repos" - response = self.client.get(url) - response.raise_for_status() - repos: list[dict[str, Any]] = response.json() - # Filter to ONLY configured organization repos, include mirrors - seen: set[str] = set() - result: list[RepositoryModel] = [] - for r in repos: - full_name = r.get("full_name", "") - if ( - full_name - and full_name not in seen - and (r.get("owner") or {}).get("login") == GITEA_ORG_FILTER - ): - seen.add(full_name) - result.append(RepositoryModel(**r)) - return result - except Exception as e: - logger.error(f"Error listing user repos: {e}", exc_info=True) - return [] - - def list_repo_issues( - self, owner: str, repo: str, state: str = "open" - ) -> list[IssueModel]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?type=issues&state={state}" - response = self.client.get(url) - response.raise_for_status() - return [IssueModel(**item) for item in response.json()] - - def list_repo_pull_requests( - self, owner: str, repo: str, state: str = "open" - ) -> list[PullRequestModel]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state={state}" - response = self.client.get(url) - if response.status_code == 404: - return [] - response.raise_for_status() - return [PullRequestModel(**item) for item in response.json()] - - def get_pull_request( - self, owner: str, repo: str, pull_number: int - ) -> PullRequestModel: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" - response = self.client.get(url) - response.raise_for_status() - return PullRequestModel(**response.json()) - - def get_issue(self, owner: str, repo: str, issue_number: int) -> IssueModel: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}" - response = self.client.get(url) - response.raise_for_status() - return IssueModel(**response.json()) - - def close_issue(self, owner: str, repo: str, issue_number: int) -> IssueModel: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}" - data: dict[str, str] = {"state": "closed"} - response = self.client.patch(url, json=data) - response.raise_for_status() - return IssueModel(**response.json()) - - def close_pull_request( - self, owner: str, repo: str, pull_number: int - ) -> PullRequestModel: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" - data: dict[str, str] = {"state": "closed"} - response = self.client.patch(url, json=data) - response.raise_for_status() - return PullRequestModel(**response.json()) - - def get_issue_comments( - self, owner: str, repo: str, issue_number: int - ) -> list[CommentModel]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments" - response = self.client.get(url) - response.raise_for_status() - return [CommentModel(**item) for item in response.json()] - - def get_pull_request_comments( - self, owner: str, repo: str, pull_number: int - ) -> list[CommentModel]: - url = ( - f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pull_number}/comments" - ) - response = self.client.get(url) - response.raise_for_status() - return [CommentModel(**item) for item in response.json()] - - def get_pull_request_diff(self, owner: str, repo: str, pull_number: int) -> str: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/diff" - response = self.client.get(url) - response.raise_for_status() - return response.text - - def get_pull_request_patch(self, owner: str, repo: str, pull_number: int) -> str: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/patch" - response = self.client.get(url) - response.raise_for_status() - return response.text - - def get_pull_request_files( - self, owner: str, repo: str, pull_number: int - ) -> list[PullRequestFileModel]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/files" - response = self.client.get(url) - response.raise_for_status() - return [PullRequestFileModel(**item) for item in response.json()] - - def list_assigned_issues(self, owner: str = "", repo: str = "") -> list[IssueModel]: - try: - user = self.get_authenticated_user() - if not user: - return [] - username: str = user.login - if owner and repo: - response = self.client.get( - f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?assignee={username}&state=open&type=issues", - ) - response.raise_for_status() - return [IssueModel(**item) for item in response.json()] - all_issues: list[IssueModel] = [] - repos = self.list_all_user_repos() - for r in repos: - repo_owner = r.owner - repo_name = r.name - resp = self.client.get( - f"{self.base_url}/api/v1/repos/{repo_owner}/{repo_name}/issues?assignee={username}&state=open&type=issues", - ) - if resp.status_code == 200: - for item in resp.json(): - issue = IssueModel(**item) - # Backfill repository if Gitea omitted it - if issue.repository is None: - issue = issue.model_copy(update={"repository": r}) - all_issues.append(issue) - return all_issues - except Exception as e: - logger.error(f"Error listing assigned issues: {e}", exc_info=True) - return [] - - def list_assigned_pull_requests( - self, owner: str = "", repo: str = "" - ) -> list[PullRequestModel]: - """List all pull requests assigned to or authored by the authenticated user.""" - try: - user = self.get_authenticated_user() - if not user: - return [] - username: str = user.login - if owner and repo: - response = self.client.get( - f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state=open", - ) - if response.status_code == 404: - return [] - response.raise_for_status() - all_prs: list[PullRequestModel] = [ - PullRequestModel(**pr) for pr in response.json() - ] - return [ - pr - for pr in all_prs - if (pr.assignee and pr.assignee.login == username) - or (pr.user and pr.user.login == username) - ] - all_prs: list[PullRequestModel] = [] - repos = self.list_all_user_repos() - for r in repos: - repo_owner = r.owner - repo_name = r.name - resp = self.client.get( - f"{self.base_url}/api/v1/repos/{repo_owner}/{repo_name}/pulls?state=open", - ) - if resp.status_code == 200: - for pr_data in resp.json(): - pr = PullRequestModel(**pr_data) - if (pr.assignee and pr.assignee.login == username) or ( - pr.user and pr.user.login == username - ): - # Backfill repository if Gitea omitted it - if pr.repository is None: - pr = pr.model_copy(update={"repository": r}) - all_prs.append(pr) - return all_prs - except Exception as e: - logger.error(f"Error listing assigned pull requests: {e}", exc_info=True) - return [] - - def create_pull_request( - self, - owner: str, - repo: str, - head: str, - base: str, - title: str, - description: str = "", - ) -> PullRequestModel: - try: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls" - data: dict[str, str] = { - "title": title, - "body": description, - "head": head, - "base": base, - } - response = self.client.post(url, json=data) - response.raise_for_status() - return PullRequestModel(**response.json()) - except Exception as e: - logger.error(f"Error creating pull request: {e}", exc_info=True) - raise - - def update_pull_request( - self, - owner: str, - repo: str, - pull_number: int, - title: str | None = None, - body: str | None = None, - state: str | None = None, - ) -> PullRequestModel: - try: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" - data: dict[str, Any] = {} - if title is not None: - data["title"] = title - if body is not None: - data["body"] = body - if state is not None: - data["state"] = state - response = self.client.patch(url, json=data) - response.raise_for_status() - return PullRequestModel(**response.json()) - except Exception as e: - logger.error(f"Error updating pull request: {e}", exc_info=True) - raise - - def create_pr_via_tea( - self, owner: str, repo: str, title: str, description: str, head: str, base: str - ) -> PullRequestModel: - return self.create_pull_request(owner, repo, head, base, title, description) - - def approve_pr( - self, owner: str, repo: str, pr_number: int, comment: str - ) -> dict[str, Any]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" - data: dict[str, Any] = {"event": "APPROVED", "body": comment} - response = self.client.post(url, json=data) - response.raise_for_status() - return response.json() - - def request_changes_pr( - self, owner: str, repo: str, pr_number: int, comment: str - ) -> dict[str, Any]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" - data: dict[str, Any] = {"event": "REQUEST_CHANGES", "body": comment} - response = self.client.post(url, json=data) - response.raise_for_status() - return response.json() - - def get_pr_reviews( - self, owner: str, repo: str, pr_number: int - ) -> list[dict[str, Any]]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" - response = self.client.get(url) - if response.status_code == 404: - return [] - response.raise_for_status() - return response.json() - - def dismiss_review_pr( - self, owner: str, repo: str, pr_number: int, review_id: int, message: str - ) -> dict[str, Any]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews/{review_id}/dismissals" - data: dict[str, str] = {"message": message} - response = self.client.post(url, json=data) - response.raise_for_status() - return response.json() - - def assign_issue( - self, owner: str, repo: str, issue_number: int, username: str - ) -> IssueModel: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}" - data: dict[str, list[str]] = {"assignees": [username]} - response = self.client.patch(url, json=data) - response.raise_for_status() - return IssueModel(**response.json()) - - def create_issue( - self, - owner: str, - repo: str, - title: str, - body: str, - labels: list[str] | None = None, - assignees: list[str] | None = None, - ) -> IssueModel: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues" - data: dict[str, Any] = {"title": title, "body": body} - if labels: - data["labels"] = labels - if assignees: - data["assignees"] = assignees - response = self.client.post(url, json=data) - response.raise_for_status() - return IssueModel(**response.json()) - - def add_comment( - self, owner: str, repo: str, issue_number: int, body: str - ) -> CommentModel: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments" - data: dict[str, str] = {"body": body} - response = self.client.post(url, json=data) - response.raise_for_status() - return CommentModel(**response.json()) - - def add_label( - self, owner: str, repo: str, issue_number: int, label: str - ) -> LabelModel: - url = ( - f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/labels" - ) - data: list[str] = [label] - response = self.client.post(url, json=data) - response.raise_for_status() - return LabelModel(**response.json()) - - def add_label_pr( - self, owner: str, repo: str, pr_number: int, label: str - ) -> LabelModel: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pr_number}/labels" - data: list[str] = [label] - response = self.client.post(url, json=data) - response.raise_for_status() - return LabelModel(**response.json()) - - def update_ref(self, owner: str, repo: str, ref: str, sha: str) -> dict[str, Any]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/git/ref/{ref}" - data: dict[str, str] = {"sha": sha} - response = self.client.post(url, json=data) - response.raise_for_status() - return response.json() - - def create_ref(self, owner: str, repo: str, ref: str, sha: str) -> dict[str, Any]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/git/refs" - data: dict[str, str] = {"ref": ref, "sha": sha} - response = self.client.post(url, json=data) - response.raise_for_status() - return response.json() - - def update_file( - self, owner: str, repo: str, path: str, message: str, content: str, branch: str - ) -> dict[str, Any]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}" - data: dict[str, str] = { - "message": message, - "content": base64.b64encode(content.encode()).decode(), - "branch": branch, - "new_branch": f"{branch}-update-{path}", - } - response = self.client.put(url, json=data) - response.raise_for_status() - return response.json() - - def get_file_content( - self, owner: str, repo: str, path: str, ref: str = "master" - ) -> str | list[str]: - url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}" - params: dict[str, str] = {"ref": ref} - response = self.client.get(url, params=params) - response.raise_for_status() - data = response.json() - if isinstance(data, list): - return [ - item.get("content", "") for item in data if item.get("type") == "file" - ] - return ( - base64.b64decode(data.get("content", "")).decode() - if data.get("content") - else "" - ) - - def list_unread_notifications( - self, since: Optional[str] = None - ) -> list[dict[str, Any]]: - try: - url = f"{self.base_url}/api/v1/notifications" - params: dict[str, str] = {"all": "false"} - if since: - params["since"] = since - response = self.client.get(url, params=params) - response.raise_for_status() - notifications: list[dict[str, Any]] = response.json() - - result: list[dict[str, Any]] = [] - for n in notifications: - repo_info = n.get("repository") or {} - owner_info = repo_info.get("owner") or {} - owner_login = owner_info.get("login", "") - if owner_login == GITEA_ORG_FILTER: - result.append(n) - return result - except Exception as e: - logger.error(f"Error listing unread notifications: {e}", exc_info=True) - return [] - - def mark_notification_as_read(self, thread_id: int) -> bool: - try: - url = f"{self.base_url}/api/v1/notifications/threads/{thread_id}" - response = self.client.patch(url) - response.raise_for_status() - return True - except Exception as e: - logger.error( - f"Error marking notification thread {thread_id} as read: {e}", - exc_info=True, - ) - return False - - def merge_pull_request( - self, - owner: str, - repo: str, - pull_number: int, - style: str = "squash", - title: str = "", - message: str = "", - ) -> bool: - try: - url = ( - f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/merge" - ) - data: dict[str, Any] = { - "Do": style, - "MergeTitleField": title, - "MergeMessageField": message, - } - response = self.client.post(url, json=data) - response.raise_for_status() - return True - except Exception as e: - logger.error( - f"Error merging pull request {pull_number}: {e}", exc_info=True - ) - raise diff --git a/gitea/files_client.py b/gitea/files_client.py new file mode 100644 index 0000000..2acb8bd --- /dev/null +++ b/gitea/files_client.py @@ -0,0 +1,116 @@ +"""Files client for Gitea API operations.""" + +import base64 +import logging +from typing import Any + +import httpx + + +logger: logging.Logger = logging.getLogger("gitea.files_client") + + +class FilesClient: + """HTTP client for Gitea Files and Git Refs API operations.""" + + def __init__(self, base_url: str, client: httpx.Client) -> None: + """Initialize the FilesClient. + + Args: + base_url: The base URL for the Gitea API. + client: The httpx client for making requests. + """ + self.base_url: str = base_url + self.client: httpx.Client = client + + def update_file( + self, owner: str, repo: str, path: str, message: str, content: str, branch: str + ) -> dict[str, Any]: + """Update a file in a repository. + + Args: + owner: Repository owner. + repo: Repository name. + path: File path. + message: Commit message. + content: File content. + branch: Branch name. + + Returns: + The API response. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}" + data: dict[str, str] = { + "message": message, + "content": base64.b64encode(content.encode()).decode(), + "branch": branch, + "new_branch": f"{branch}-update-{path}", + } + response = self.client.put(url, json=data) + response.raise_for_status() + return response.json() + + def get_file_content( + self, owner: str, repo: str, path: str, ref: str = "master" + ) -> str | list[str]: + """Get the content of a file or directory. + + Args: + owner: Repository owner. + repo: Repository name. + path: File or directory path. + ref: Git reference (branch, tag, commit). + + Returns: + File content as string, or list of file names if path is a directory. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/contents/{path}" + params: dict[str, str] = {"ref": ref} + response = self.client.get(url, params=params) + response.raise_for_status() + data = response.json() + if isinstance(data, list): + return [ + item.get("content", "") for item in data if item.get("type") == "file" + ] + return ( + base64.b64decode(data.get("content", "")).decode() + if data.get("content") + else "" + ) + + def update_ref(self, owner: str, repo: str, ref: str, sha: str) -> dict[str, Any]: + """Update a git reference. + + Args: + owner: Repository owner. + repo: Repository name. + ref: Reference name (e.g., heads/main). + sha: New SHA for the reference. + + Returns: + The API response. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/git/ref/{ref}" + data: dict[str, str] = {"sha": sha} + response = self.client.post(url, json=data) + response.raise_for_status() + return response.json() + + def create_ref(self, owner: str, repo: str, ref: str, sha: str) -> dict[str, Any]: + """Create a new git reference. + + Args: + owner: Repository owner. + repo: Repository name. + ref: Reference name (e.g., refs/heads/new-branch). + sha: SHA for the reference. + + Returns: + The API response. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/git/refs" + data: dict[str, str] = {"ref": ref, "sha": sha} + response = self.client.post(url, json=data) + response.raise_for_status() + return response.json() diff --git a/gitea/issues_client.py b/gitea/issues_client.py new file mode 100644 index 0000000..6af4a7a --- /dev/null +++ b/gitea/issues_client.py @@ -0,0 +1,248 @@ +"""Issues client for Gitea API operations.""" + +import logging +from typing import Any, Callable, Optional + +import httpx + +from .models import ( + CommentModel, + IssueModel, + LabelModel, + RepositoryModel, + UserModel, +) + +logger: logging.Logger = logging.getLogger("gitea.issues_client") + + +class IssuesClient: + """HTTP client for Gitea Issues API operations.""" + + def __init__( + self, + base_url: str, + client: httpx.Client, + get_user: Callable[[], UserModel] | None = None, + get_repos: Callable[[], list[RepositoryModel]] | None = None, + ) -> None: + """Initialize the IssuesClient. + + Args: + base_url: The base URL for the Gitea API. + client: The httpx client for making requests. + get_user: Optional callable to get the authenticated user. + get_repos: Optional callable to get all user repos. + """ + self.base_url: str = base_url + self.client: httpx.Client = client + self._get_user: Callable[[], UserModel] | None = get_user + self._get_repos: Callable[[], list[RepositoryModel]] | None = get_repos + + def list_repo_issues( + self, owner: str, repo: str, state: str = "open" + ) -> list[IssueModel]: + """List issues for a repository. + + Args: + owner: Repository owner. + repo: Repository name. + state: Issue state filter (open, closed, all). + + Returns: + List of issues matching the criteria. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?type=issues&state={state}" + response = self.client.get(url) + response.raise_for_status() + return [IssueModel(**item) for item in response.json()] + + def get_issue(self, owner: str, repo: str, issue_number: int) -> IssueModel: + """Get a specific issue. + + Args: + owner: Repository owner. + repo: Repository name. + issue_number: Issue number. + + Returns: + The requested issue. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}" + response = self.client.get(url) + response.raise_for_status() + return IssueModel(**response.json()) + + def close_issue(self, owner: str, repo: str, issue_number: int) -> IssueModel: + """Close an issue. + + Args: + owner: Repository owner. + repo: Repository name. + issue_number: Issue number. + + Returns: + The updated issue. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}" + data: dict[str, str] = {"state": "closed"} + response = self.client.patch(url, json=data) + response.raise_for_status() + return IssueModel(**response.json()) + + def get_issue_comments( + self, owner: str, repo: str, issue_number: int + ) -> list[CommentModel]: + """Get comments on an issue. + + Args: + owner: Repository owner. + repo: Repository name. + issue_number: Issue number. + + Returns: + List of comments on the issue. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments" + response = self.client.get(url) + response.raise_for_status() + return [CommentModel(**item) for item in response.json()] + + def list_assigned_issues(self, owner: str = "", repo: str = "") -> list[IssueModel]: + """List issues assigned to the authenticated user. + + Args: + owner: Optional repository owner to filter by. + repo: Optional repository name to filter by. + + Returns: + List of issues assigned to the authenticated user. + """ + try: + if self._get_user is None or self._get_repos is None: + logger.error("get_user and get_repos callables are required") + return [] + + user = self._get_user() + if not user: + return [] + username: str = user.login + if owner and repo: + response = self.client.get( + f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues?assignee={username}&state=open&type=issues", + ) + response.raise_for_status() + return [IssueModel(**item) for item in response.json()] + all_issues: list[IssueModel] = [] + repos = self._get_repos() + for r in repos: + repo_owner = r.owner + repo_name = r.name + resp = self.client.get( + f"{self.base_url}/api/v1/repos/{repo_owner}/{repo_name}/issues?assignee={username}&state=open&type=issues", + ) + if resp.status_code == 200: + for item in resp.json(): + issue = IssueModel(**item) + # Backfill repository if Gitea omitted it + if issue.repository is None: + issue = issue.model_copy(update={"repository": r}) + all_issues.append(issue) + return all_issues + except Exception as e: + logger.error(f"Error listing assigned issues: {e}", exc_info=True) + return [] + + def assign_issue( + self, owner: str, repo: str, issue_number: int, username: str + ) -> IssueModel: + """Assign an issue to a user. + + Args: + owner: Repository owner. + repo: Repository name. + issue_number: Issue number. + username: Username to assign. + + Returns: + The updated issue. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}" + data: dict[str, list[str]] = {"assignees": [username]} + response = self.client.patch(url, json=data) + response.raise_for_status() + return IssueModel(**response.json()) + + def create_issue( + self, + owner: str, + repo: str, + title: str, + body: str, + labels: list[str] | None = None, + assignees: list[str] | None = None, + ) -> IssueModel: + """Create a new issue. + + Args: + owner: Repository owner. + repo: Repository name. + title: Issue title. + body: Issue body/description. + labels: Optional list of label IDs. + assignees: Optional list of usernames to assign. + + Returns: + The created issue. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues" + data: dict[str, Any] = {"title": title, "body": body} + if labels: + data["labels"] = labels + if assignees: + data["assignees"] = assignees + response = self.client.post(url, json=data) + response.raise_for_status() + return IssueModel(**response.json()) + + def add_comment( + self, owner: str, repo: str, issue_number: int, body: str + ) -> CommentModel: + """Add a comment to an issue. + + Args: + owner: Repository owner. + repo: Repository name. + issue_number: Issue number. + body: Comment body. + + Returns: + The created comment. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments" + data: dict[str, str] = {"body": body} + response = self.client.post(url, json=data) + response.raise_for_status() + return CommentModel(**response.json()) + + def add_label( + self, owner: str, repo: str, issue_number: int, label: str + ) -> LabelModel: + """Add a label to an issue. + + Args: + owner: Repository owner. + repo: Repository name. + issue_number: Issue number. + label: Label name or ID. + + Returns: + The added label. + """ + url = ( + f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_number}/labels" + ) + data: list[str] = [label] + response = self.client.post(url, json=data) + response.raise_for_status() + return LabelModel(**response.json()) diff --git a/gitea/notifications_client.py b/gitea/notifications_client.py new file mode 100644 index 0000000..1461b8c --- /dev/null +++ b/gitea/notifications_client.py @@ -0,0 +1,78 @@ +"""Notifications client for Gitea API operations.""" + +import logging +from typing import Any, Optional + +import httpx + + +logger: logging.Logger = logging.getLogger("gitea.notifications_client") + + +class NotificationsClient: + """HTTP client for Gitea Notifications API operations.""" + + def __init__(self, base_url: str, client: httpx.Client, org_filter: str) -> None: + """Initialize the NotificationsClient. + + Args: + base_url: The base URL for the Gitea API. + client: The httpx client for making requests. + org_filter: Organization filter for notifications. + """ + self.base_url: str = base_url + self.client: httpx.Client = client + self.org_filter: str = org_filter + + def list_unread_notifications( + self, since: Optional[str] = None + ) -> list[dict[str, Any]]: + """List unread notifications. + + Args: + since: Optional ISO 8601 timestamp to filter notifications after. + + Returns: + List of unread notifications for the configured organization. + """ + try: + url = f"{self.base_url}/api/v1/notifications" + params: dict[str, str] = {"all": "false"} + if since: + params["since"] = since + response = self.client.get(url, params=params) + response.raise_for_status() + notifications: list[dict[str, Any]] = response.json() + + result: list[dict[str, Any]] = [] + for n in notifications: + repo_info = n.get("repository") or {} + owner_info = repo_info.get("owner") or {} + owner_login = owner_info.get("login", "") + if owner_login == self.org_filter: + result.append(n) + return result + except Exception as e: + logger.error(f"Error listing unread notifications: {e}", exc_info=True) + return [] + + def mark_notification_as_read(self, thread_id: int) -> bool: + """Mark a notification as read. + + Args: + thread_id: Notification thread ID. + + Returns: + True if successful, False otherwise. + """ + try: + url = f"{self.base_url}/api/v1/notifications/threads/{thread_id}" + response = self.client.patch(url) + response.raise_for_status() + return True + except Exception as e: + logger.error( + f"Error marking notification thread {thread_id} as read: {e}", + exc_info=True, + ) + return False diff --git a/gitea/prs_client.py b/gitea/prs_client.py new file mode 100644 index 0000000..aaf867d --- /dev/null +++ b/gitea/prs_client.py @@ -0,0 +1,462 @@ +"""Pull Requests client for Gitea API operations.""" + +import logging +from typing import Any, Callable + +import httpx + +from .models import ( + CommentModel, + LabelModel, + PullRequestFileModel, + PullRequestModel, + RepositoryModel, + UserModel, +) + +logger: logging.Logger = logging.getLogger("gitea.prs_client") + + +class PullRequestsClient: + """HTTP client for Gitea Pull Requests API operations.""" + + def __init__( + self, + base_url: str, + client: httpx.Client, + get_user: Callable[[], UserModel] | None = None, + get_repos: Callable[[], list[RepositoryModel]] | None = None, + ) -> None: + """Initialize the PullRequestsClient. + + Args: + base_url: The base URL for the Gitea API. + client: The httpx client for making requests. + get_user: Optional callable to get the authenticated user. + get_repos: Optional callable to get all user repos. + """ + self.base_url: str = base_url + self.client: httpx.Client = client + self._get_user: Callable[[], UserModel] | None = get_user + self._get_repos: Callable[[], list[RepositoryModel]] | None = get_repos + + def list_repo_pull_requests( + self, owner: str, repo: str, state: str = "open" + ) -> list[PullRequestModel]: + """List pull requests for a repository. + + Args: + owner: Repository owner. + repo: Repository name. + state: PR state filter (open, closed, all). + + Returns: + List of pull requests matching the criteria. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state={state}" + response = self.client.get(url) + if response.status_code == 404: + return [] + response.raise_for_status() + return [PullRequestModel(**item) for item in response.json()] + + def get_pull_request( + self, owner: str, repo: str, pull_number: int + ) -> PullRequestModel: + """Get a specific pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pull_number: Pull request number. + + Returns: + The requested pull request. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" + response = self.client.get(url) + response.raise_for_status() + return PullRequestModel(**response.json()) + + def close_pull_request( + self, owner: str, repo: str, pull_number: int + ) -> PullRequestModel: + """Close a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pull_number: Pull request number. + + Returns: + The updated pull request. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" + data: dict[str, str] = {"state": "closed"} + response = self.client.patch(url, json=data) + response.raise_for_status() + return PullRequestModel(**response.json()) + + def get_pull_request_comments( + self, owner: str, repo: str, pull_number: int + ) -> list[CommentModel]: + """Get comments on a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pull_number: Pull request number. + + Returns: + List of comments on the pull request. + """ + url = ( + f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pull_number}/comments" + ) + response = self.client.get(url) + response.raise_for_status() + return [CommentModel(**item) for item in response.json()] + + def get_pull_request_diff(self, owner: str, repo: str, pull_number: int) -> str: + """Get the diff for a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pull_number: Pull request number. + + Returns: + The diff as a string. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/diff" + response = self.client.get(url) + response.raise_for_status() + return response.text + + def get_pull_request_patch(self, owner: str, repo: str, pull_number: int) -> str: + """Get the patch for a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pull_number: Pull request number. + + Returns: + The patch as a string. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/patch" + response = self.client.get(url) + response.raise_for_status() + return response.text + + def get_pull_request_files( + self, owner: str, repo: str, pull_number: int + ) -> list[PullRequestFileModel]: + """Get the files changed in a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pull_number: Pull request number. + + Returns: + List of files changed in the pull request. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/files" + response = self.client.get(url) + response.raise_for_status() + return [PullRequestFileModel(**item) for item in response.json()] + + def list_assigned_pull_requests( + self, owner: str = "", repo: str = "" + ) -> list[PullRequestModel]: + """List all pull requests assigned to or authored by the authenticated user. + + Args: + owner: Optional repository owner to filter by. + repo: Optional repository name to filter by. + + Returns: + List of pull requests assigned to or authored by the user. + """ + try: + if self._get_user is None or self._get_repos is None: + logger.error("get_user and get_repos callables are required") + return [] + + user = self._get_user() + if not user: + return [] + username: str = user.login + if owner and repo: + response = self.client.get( + f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls?state=open", + ) + if response.status_code == 404: + return [] + response.raise_for_status() + all_prs: list[PullRequestModel] = [ + PullRequestModel(**pr) for pr in response.json() + ] + return [ + pr + for pr in all_prs + if (pr.assignee and pr.assignee.login == username) + or (pr.user and pr.user.login == username) + ] + all_prs: list[PullRequestModel] = [] + repos = self._get_repos() + for r in repos: + repo_owner = r.owner + repo_name = r.name + resp = self.client.get( + f"{self.base_url}/api/v1/repos/{repo_owner}/{repo_name}/pulls?state=open", + ) + if resp.status_code == 200: + for pr_data in resp.json(): + pr = PullRequestModel(**pr_data) + if (pr.assignee and pr.assignee.login == username) or ( + pr.user and pr.user.login == username + ): + # Backfill repository if Gitea omitted it + if pr.repository is None: + pr = pr.model_copy(update={"repository": r}) + all_prs.append(pr) + return all_prs + except Exception as e: + logger.error(f"Error listing assigned pull requests: {e}", exc_info=True) + return [] + + def create_pull_request( + self, + owner: str, + repo: str, + head: str, + base: str, + title: str, + description: str = "", + ) -> PullRequestModel: + """Create a new pull request. + + Args: + owner: Repository owner. + repo: Repository name. + head: Head branch name. + base: Base branch name. + title: Pull request title. + description: Pull request description. + + Returns: + The created pull request. + """ + try: + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls" + data: dict[str, str] = { + "title": title, + "body": description, + "head": head, + "base": base, + } + response = self.client.post(url, json=data) + response.raise_for_status() + return PullRequestModel(**response.json()) + except Exception as e: + logger.error(f"Error creating pull request: {e}", exc_info=True) + raise + + def update_pull_request( + self, + owner: str, + repo: str, + pull_number: int, + title: str | None = None, + body: str | None = None, + state: str | None = None, + ) -> PullRequestModel: + """Update a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pull_number: Pull request number. + title: Optional new title. + body: Optional new body. + state: Optional new state. + + Returns: + The updated pull request. + """ + try: + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}" + data: dict[str, Any] = {} + if title is not None: + data["title"] = title + if body is not None: + data["body"] = body + if state is not None: + data["state"] = state + response = self.client.patch(url, json=data) + response.raise_for_status() + return PullRequestModel(**response.json()) + except Exception as e: + logger.error(f"Error updating pull request: {e}", exc_info=True) + raise + + def create_pr_via_tea( + self, owner: str, repo: str, title: str, description: str, head: str, base: str + ) -> PullRequestModel: + """Create a pull request (alias for create_pull_request). + + Args: + owner: Repository owner. + repo: Repository name. + title: Pull request title. + description: Pull request description. + head: Head branch name. + base: Base branch name. + + Returns: + The created pull request. + """ + return self.create_pull_request(owner, repo, head, base, title, description) + + def approve_pr( + self, owner: str, repo: str, pr_number: int, comment: str + ) -> dict[str, Any]: + """Approve a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pr_number: Pull request number. + comment: Review comment. + + Returns: + The review response. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" + data: dict[str, Any] = {"event": "APPROVED", "body": comment} + response = self.client.post(url, json=data) + response.raise_for_status() + return response.json() + + def request_changes_pr( + self, owner: str, repo: str, pr_number: int, comment: str + ) -> dict[str, Any]: + """Request changes on a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pr_number: Pull request number. + comment: Review comment. + + Returns: + The review response. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" + data: dict[str, Any] = {"event": "REQUEST_CHANGES", "body": comment} + response = self.client.post(url, json=data) + response.raise_for_status() + return response.json() + + def get_pr_reviews( + self, owner: str, repo: str, pr_number: int + ) -> list[dict[str, Any]]: + """Get reviews for a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pr_number: Pull request number. + + Returns: + List of reviews. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews" + response = self.client.get(url) + if response.status_code == 404: + return [] + response.raise_for_status() + return response.json() + + def dismiss_review_pr( + self, owner: str, repo: str, pr_number: int, review_id: int, message: str + ) -> dict[str, Any]: + """Dismiss a review on a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pr_number: Pull request number. + review_id: Review ID to dismiss. + message: Dismissal message. + + Returns: + The dismissal response. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pr_number}/reviews/{review_id}/dismissals" + data: dict[str, str] = {"message": message} + response = self.client.post(url, json=data) + response.raise_for_status() + return response.json() + + def add_label_pr( + self, owner: str, repo: str, pr_number: int, label: str + ) -> LabelModel: + """Add a label to a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pr_number: Pull request number. + label: Label name or ID. + + Returns: + The added label. + """ + url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{pr_number}/labels" + data: list[str] = [label] + response = self.client.post(url, json=data) + response.raise_for_status() + return LabelModel(**response.json()) + + def merge_pull_request( + self, + owner: str, + repo: str, + pull_number: int, + style: str = "squash", + title: str = "", + message: str = "", + ) -> bool: + """Merge a pull request. + + Args: + owner: Repository owner. + repo: Repository name. + pull_number: Pull request number. + style: Merge style (squash, merge, rebase). + title: Optional merge commit title. + message: Optional merge commit message. + + Returns: + True if merge was successful. + """ + try: + url = ( + f"{self.base_url}/api/v1/repos/{owner}/{repo}/pulls/{pull_number}/merge" + ) + data: dict[str, Any] = { + "Do": style, + "MergeTitleField": title, + "MergeMessageField": message, + } + response = self.client.post(url, json=data) + response.raise_for_status() + return True + except Exception as e: + logger.error( + f"Error merging pull request {pull_number}: {e}", exc_info=True + ) + raise diff --git a/gitea/repos_client.py b/gitea/repos_client.py new file mode 100644 index 0000000..2f5cbbc --- /dev/null +++ b/gitea/repos_client.py @@ -0,0 +1,72 @@ +"""Repositories client for Gitea API operations.""" + +import logging +from typing import Any + +import httpx + +from .models import RepositoryModel, UserModel + + +logger: logging.Logger = logging.getLogger("gitea.repos_client") + + +class ReposClient: + """HTTP client for Gitea Repositories API operations.""" + + def __init__(self, base_url: str, client: httpx.Client, org_filter: str) -> None: + """Initialize the ReposClient. + + Args: + base_url: The base URL for the Gitea API. + client: The httpx client for making requests. + org_filter: Organization filter for repositories. + """ + self.base_url: str = base_url + self.client: httpx.Client = client + self.org_filter: str = org_filter + + def list_all_user_repos(self) -> list[RepositoryModel]: + """List all repositories for the authenticated user. + + Returns: + List of repositories belonging to the configured organization. + """ + try: + url = f"{self.base_url}/api/v1/user/repos" + response = self.client.get(url) + response.raise_for_status() + repos: list[dict[str, Any]] = response.json() + # Filter to ONLY configured organization repos, include mirrors + seen: set[str] = set() + result: list[RepositoryModel] = [] + for r in repos: + full_name = r.get("full_name", "") + if ( + full_name + and full_name not in seen + and (r.get("owner") or {}).get("login") == self.org_filter + ): + seen.add(full_name) + result.append(RepositoryModel(**r)) + return result + except Exception as e: + logger.error(f"Error listing user repos: {e}", exc_info=True) + return [] + + def get_authenticated_user(self) -> UserModel: + """Get the authenticated user. + + Returns: + The authenticated user. + + Raises: + RuntimeError: If the user cannot be retrieved. + """ + try: + response = self.client.get(f"{self.base_url}/api/v1/user") + response.raise_for_status() + return UserModel(**response.json()) + except Exception as e: + logger.error(f"Error getting authenticated user: {e}", exc_info=True) + raise RuntimeError(f"Could not get authenticated user: {e}") from e diff --git a/gitea/tools/file_tools.py b/gitea/tools/file_tools.py index 9821988..a2be5c9 100644 --- a/gitea/tools/file_tools.py +++ b/gitea/tools/file_tools.py @@ -50,7 +50,7 @@ class FileTools: limit: Maximum number of lines to return (default 250). """ try: - content = self._client.get_file_content(owner, repo, path) + content = self._client.files.get_file_content(owner, repo, path) raw: str = "\n".join(content) if isinstance(content, list) else content return self._paginate_lines(raw, offset, limit) except Exception as e: @@ -73,22 +73,26 @@ class FileTools: limit: Maximum number of lines to return (default 250). """ try: - content = self._client.get_file_content(owner, repo, path, ref) + content = self._client.files.get_file_content(owner, repo, path, ref) raw: str = "\n".join(content) if isinstance(content, list) else content return self._paginate_lines(raw, offset, limit) except Exception as e: return f"Error getting file content: {str(e)}" - def commit_file(self, owner: str, repo: str, path: str, message: str, content: str, branch: str) -> str: + def commit_file( + self, owner: str, repo: str, path: str, message: str, content: str, branch: str + ) -> str: try: - self._client.update_file(owner, repo, path, message, content, branch) + self._client.files.update_file(owner, repo, path, message, content, branch) return f"File '{path}' committed successfully to {owner}/{repo}." except Exception as e: return f"Error committing file: {str(e)}" - def update_file(self, owner: str, repo: str, path: str, message: str, content: str, branch: str) -> str: + def update_file( + self, owner: str, repo: str, path: str, message: str, content: str, branch: str + ) -> str: try: - self._client.update_file(owner, repo, path, message, content, branch) + self._client.files.update_file(owner, repo, path, message, content, branch) return f"File '{path}' updated in {owner}/{repo}." except Exception as e: return f"Error updating file: {str(e)}" diff --git a/gitea/tools/git_tools.py b/gitea/tools/git_tools.py index e1e9478..d9a5a83 100644 --- a/gitea/tools/git_tools.py +++ b/gitea/tools/git_tools.py @@ -10,7 +10,7 @@ class GitTools: def create_branch(self, owner: str, repo: str, ref: str, sha: str) -> str: try: - self._client.create_ref(owner, repo, ref, sha) + self._client.files.create_ref(owner, repo, ref, sha) return f"Branch '{ref}' created successfully in {owner}/{repo}." except Exception as e: return f"Error creating branch: {str(e)}" diff --git a/gitea/tools/issue_tools.py b/gitea/tools/issue_tools.py index 5420329..e3fea65 100644 --- a/gitea/tools/issue_tools.py +++ b/gitea/tools/issue_tools.py @@ -17,14 +17,14 @@ class IssueTools: def get_issue(self, owner: str, repo: str, issue_number: int) -> str: try: - issue: IssueModel = self._client.get_issue(owner, repo, issue_number) + issue: IssueModel = self._client.issues.get_issue(owner, repo, issue_number) return issue.model_dump_json(indent=2) except Exception as e: return f"Error getting issue: {str(e)}" def close_issue(self, owner: str, repo: str, issue_number: int) -> str: try: - self._client.close_issue(owner, repo, issue_number) + self._client.issues.close_issue(owner, repo, issue_number) return f"Issue #{issue_number} closed successfully." except Exception as e: return f"Error closing issue: {str(e)}" @@ -44,7 +44,7 @@ class IssueTools: offset: Zero-based comment index to start from (default 0). """ try: - comments: list[CommentModel] = self._client.get_issue_comments( + comments: list[CommentModel] = self._client.issues.get_issue_comments( owner, repo, issue_number ) total: int = len(comments) @@ -62,12 +62,12 @@ class IssueTools: def list_assigned_issues(self) -> list[dict[str, Any]]: try: - repos = self._client.list_all_user_repos() + repos = self._client.repos.list_all_user_repos() all_issues: list[dict[str, Any]] = [] for repo in repos: owner = repo.owner repo_name = repo.name - issues = self._client.list_assigned_issues(owner, repo_name) + issues = self._client.issues.list_assigned_issues(owner, repo_name) if issues: all_issues.extend( [ @@ -84,7 +84,7 @@ class IssueTools: def list_issues(self, owner: str, repo: str, state: str = "open") -> str: try: - issues = self._client.list_repo_issues(owner, repo, state) + issues = self._client.issues.list_repo_issues(owner, repo, state) if not issues: return f"No issues in {owner}/{repo}." summary = [f"#{issue.number}: {issue.title}" for issue in issues] @@ -102,7 +102,7 @@ class IssueTools: assignees: list[str] | None = None, ) -> str: try: - issue = self._client.create_issue( + issue = self._client.issues.create_issue( owner, repo, title, body, labels, assignees ) return f"Issue #{issue.number} created successfully in {owner}/{repo}." @@ -113,7 +113,7 @@ class IssueTools: self, owner: str, repo: str, issue_number: int, label: str ) -> str: try: - self._client.add_label(owner, repo, issue_number, label) + self._client.issues.add_label(owner, repo, issue_number, label) return f"Label '{label}' added to issue #{issue_number}." except Exception as e: return f"Error adding label to issue #{issue_number}: {e}" @@ -122,7 +122,7 @@ class IssueTools: self, owner: str, repo: str, issue_number: int, body: str ) -> str: try: - self._client.add_comment(owner, repo, issue_number, body) + self._client.issues.add_comment(owner, repo, issue_number, body) return f"Comment added to issue #{issue_number}." except Exception as e: return f"Error adding comment to issue #{issue_number}: {e}" diff --git a/gitea/tools/pr_tools.py b/gitea/tools/pr_tools.py index 65668f6..c46b15f 100644 --- a/gitea/tools/pr_tools.py +++ b/gitea/tools/pr_tools.py @@ -42,14 +42,16 @@ class PRTools: def get_pull_request(self, owner: str, repo: str, pull_number: int) -> str: try: - pr: PullRequestModel = self._client.get_pull_request(owner, repo, pull_number) + pr: PullRequestModel = self._client.prs.get_pull_request( + owner, repo, pull_number + ) return pr.model_dump_json(indent=2) except Exception as e: return f"Error getting pull request: {str(e)}" def close_pull_request(self, owner: str, repo: str, pull_number: int) -> str: try: - self._client.close_pull_request(owner, repo, pull_number) + self._client.prs.close_pull_request(owner, repo, pull_number) return f"Pull request #{pull_number} closed successfully." except Exception as e: return f"Error closing pull request: {str(e)}" @@ -69,7 +71,7 @@ class PRTools: offset: Zero-based comment index to start from (default 0). """ try: - comments: list[CommentModel] = self._client.get_pull_request_comments( + comments: list[CommentModel] = self._client.prs.get_pull_request_comments( owner, repo, pull_number ) total: int = len(comments) @@ -87,14 +89,21 @@ class PRTools: def list_assigned_pull_requests(self) -> list[dict[str, Any]]: try: - repos = self._client.list_all_user_repos() + repos = self._client.repos.list_all_user_repos() all_prs: list[dict[str, Any]] = [] for repo_info in repos: repo_owner = repo_info.owner repo_name = repo_info.name - prs = self._client.list_assigned_pull_requests(repo_owner, repo_name) + prs = self._client.prs.list_assigned_pull_requests( + repo_owner, repo_name + ) if prs: - all_prs.extend([pr.model_dump() if hasattr(pr, 'model_dump') else pr for pr in prs]) + all_prs.extend( + [ + pr.model_dump() if hasattr(pr, "model_dump") else pr + for pr in prs + ] + ) return all_prs except Exception as e: logger.error(f"Error listing assigned pull requests: {e}", exc_info=True) @@ -102,7 +111,7 @@ class PRTools: def list_pull_requests(self, owner: str, repo: str, state: str = "open") -> str: try: - prs = self._client.list_repo_pull_requests(owner, repo, state) + prs = self._client.prs.list_repo_pull_requests(owner, repo, state) if not prs: return f"No PRs in {owner}/{repo}." summary = [f"#{pr.number}: {pr.title}" for pr in prs] @@ -110,9 +119,19 @@ class PRTools: except Exception as e: return f"Error listing PRs: {str(e)}" - def create_pull_request(self, owner: str, repo: str, head: str, base: str, title: str, description: str = "") -> str: + def create_pull_request( + self, + owner: str, + repo: str, + head: str, + base: str, + title: str, + description: str = "", + ) -> str: try: - pr = self._client.create_pr_via_tea(owner, repo, title, description, head, base) + pr = self._client.prs.create_pr_via_tea( + owner, repo, title, description, head, base + ) return pr.model_dump_json(indent=2) except Exception as e: return f"Error creating PR: {str(e)}" @@ -127,14 +146,16 @@ class PRTools: state: str | None = None, ) -> str: try: - pr = self._client.update_pull_request(owner, repo, pull_number, title, body, state) + pr = self._client.prs.update_pull_request( + owner, repo, pull_number, title, body, state + ) return pr.model_dump_json(indent=2) except Exception as e: return f"Error updating PR #{pull_number}: {str(e)}" def add_label_to_pr(self, owner: str, repo: str, pr_number: int, label: str) -> str: try: - self._client.add_label_pr(owner, repo, pr_number, label) + self._client.prs.add_label_pr(owner, repo, pr_number, label) return f"Label '{label}' added to PR #{pr_number}." except Exception as e: return f"Error adding label to PR #{pr_number}: {e}" @@ -155,7 +176,7 @@ class PRTools: Increment by max_chars to page through a large diff. """ try: - diff: str = self._client.get_pull_request_diff(owner, repo, pull_number) + diff: str = self._client.prs.get_pull_request_diff(owner, repo, pull_number) return _truncate_diff(diff, max_chars, char_offset) except Exception as e: return f"Error getting PR diff: {str(e)}" @@ -176,21 +197,27 @@ class PRTools: Increment by max_chars to page through a large patch. """ try: - patch: str = self._client.get_pull_request_patch(owner, repo, pull_number) + patch: str = self._client.prs.get_pull_request_patch( + owner, repo, pull_number + ) return _truncate_diff(patch, max_chars, char_offset) except Exception as e: return f"Error getting PR patch: {str(e)}" - def approve_pull_request(self, owner: str, repo: str, pull_number: int, comment: str) -> str: + def approve_pull_request( + self, owner: str, repo: str, pull_number: int, comment: str + ) -> str: try: - self._client.approve_pr(owner, repo, pull_number, comment) + self._client.prs.approve_pr(owner, repo, pull_number, comment) return f"Approved PR #{pull_number}." except Exception as e: return f"Error approving PR: {str(e)}" - def request_changes(self, owner: str, repo: str, pull_number: int, comment: str) -> str: + def request_changes( + self, owner: str, repo: str, pull_number: int, comment: str + ) -> str: try: - self._client.request_changes_pr(owner, repo, pull_number, comment) + self._client.prs.request_changes_pr(owner, repo, pull_number, comment) return f"Requested changes on PR #{pull_number}." except Exception as e: return f"Error requesting changes: {str(e)}" diff --git a/gitea/workspace.py b/gitea/workspace.py index b729899..0673a6b 100644 --- a/gitea/workspace.py +++ b/gitea/workspace.py @@ -21,7 +21,7 @@ class WorkspaceManager: def _configure_repo_user(self, repo_path: Path) -> None: try: client = GiteaClient() - user = client.get_authenticated_user() + user = client.repos.get_authenticated_user() if not user or not user.login: raise RuntimeError("No authenticated user found.") username: str = user.login @@ -32,19 +32,29 @@ class WorkspaceManager: # Configure extraHeader locally for the repo subprocess.run( - ["git", "-C", str(repo_path), "config", "http.extraHeader", f"Authorization: Basic {auth_b64}"], - check=True, capture_output=True + [ + "git", + "-C", + str(repo_path), + "config", + "http.extraHeader", + f"Authorization: Basic {auth_b64}", + ], + check=True, + capture_output=True, ) name: str = user.full_name or user.login email: str = user.email or f"{user.login}@noreply.gitea" subprocess.run( ["git", "-C", str(repo_path), "config", "user.name", name], - check=True, capture_output=True + check=True, + capture_output=True, ) subprocess.run( ["git", "-C", str(repo_path), "config", "user.email", email], - check=True, capture_output=True + check=True, + capture_output=True, ) except Exception as e: logger.error(f"Error configuring local git user: {e}") @@ -64,59 +74,84 @@ class WorkspaceManager: auth_url = self._get_authenticated_url(repo_full_name) subprocess.run( ["git", "-C", str(repo_path), "remote", "set-url", "origin", auth_url], - check=True, capture_output=True, + check=True, + capture_output=True, ) self._configure_repo_user(repo_path) # Check for any uncommitted changes or untracked files status_res = subprocess.run( ["git", "-C", str(repo_path), "status", "--porcelain"], - check=True, capture_output=True, text=True + check=True, + capture_output=True, + text=True, ) if status_res.stdout.strip(): - logger.info(f"Uncommitted changes detected in {repo_path}. Stashing before sanitization.") + logger.info( + f"Uncommitted changes detected in {repo_path}. Stashing before sanitization." + ) subprocess.run( - ["git", "-C", str(repo_path), "stash", "push", "-u", "-m", "Auto-backup before agent sanitization"], - check=True, capture_output=True + [ + "git", + "-C", + str(repo_path), + "stash", + "push", + "-u", + "-m", + "Auto-backup before agent sanitization", + ], + check=True, + capture_output=True, ) subprocess.run( ["git", "-C", str(repo_path), "reset", "--hard", "HEAD"], - check=True, capture_output=True, + check=True, + capture_output=True, ) subprocess.run( ["git", "-C", str(repo_path), "clean", "-fdx"], - check=True, capture_output=True, + check=True, + capture_output=True, ) try: subprocess.run( ["git", "-C", str(repo_path), "checkout", "main"], - check=True, capture_output=True, + check=True, + capture_output=True, ) except subprocess.CalledProcessError: subprocess.run( ["git", "-C", str(repo_path), "checkout", "master"], - check=True, capture_output=True, + check=True, + capture_output=True, ) try: subprocess.run( ["git", "-C", str(repo_path), "pull", "origin", "main"], - check=True, capture_output=True, + check=True, + capture_output=True, ) except subprocess.CalledProcessError: subprocess.run( ["git", "-C", str(repo_path), "pull", "origin", "master"], - check=True, capture_output=True, + check=True, + capture_output=True, ) except Exception as e: logger.error(f"Error during sanitization: {e}", exc_info=True) - raise RuntimeError(f"Failed to sanitize repository {repo_full_name} at {repo_path}: {e}") from e + raise RuntimeError( + f"Failed to sanitize repository {repo_full_name} at {repo_path}: {e}" + ) from e def clone_repo(self, repo_full_name: str, clone_url: str | None = None) -> Path: repo_path: Path = self.get_repo_path(repo_full_name) if repo_path.exists(): if not (repo_path / ".git").exists(): - new_path: Path = repo_path.parent / f"{repo_full_name.replace('/', '_')}_old" + new_path: Path = ( + repo_path.parent / f"{repo_full_name.replace('/', '_')}_old" + ) if new_path.exists(): shutil.rmtree(new_path) repo_path.rename(new_path) @@ -126,7 +161,7 @@ class WorkspaceManager: auth_url = self._get_authenticated_url(repo_full_name) client = GiteaClient() - user = client.get_authenticated_user() + user = client.repos.get_authenticated_user() if not user or not user.login: raise RuntimeError("No authenticated user found.") username: str = user.login @@ -135,8 +170,16 @@ class WorkspaceManager: auth_bytes: bytes = auth_str.encode("utf-8") auth_b64: str = base64.b64encode(auth_bytes).decode("utf-8") subprocess.run( - ["git", "clone", "-c", f"http.extraHeader=Authorization: Basic {auth_b64}", auth_url, str(repo_path)], - check=True, capture_output=True + [ + "git", + "clone", + "-c", + f"http.extraHeader=Authorization: Basic {auth_b64}", + auth_url, + str(repo_path), + ], + check=True, + capture_output=True, ) self._configure_repo_user(repo_path) return repo_path diff --git a/main.py b/main.py index 13e52f1..a483e71 100644 --- a/main.py +++ b/main.py @@ -61,7 +61,7 @@ async def main() -> None: # Initialize Gitea components client: GiteaClient = GiteaClient() try: - user = client.get_authenticated_user() + user = client.repos.get_authenticated_user() if not user or not user.login: raise RuntimeError("No authenticated user found.") logger.info(f"Authenticated as user: {user.login}") diff --git a/tests/test_client.py b/tests/test_client.py index 2f782ee..b005e31 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -11,7 +11,7 @@ def test_gitea_client_list_repo_issues() -> None: mock_get.return_value = mock_response # Test default parameter ("open") - client.list_repo_issues("owner", "repo") + client.issues.list_repo_issues("owner", "repo") mock_get.assert_called_once() args, _ = mock_get.call_args assert "type=issues" in args[0] @@ -20,7 +20,7 @@ def test_gitea_client_list_repo_issues() -> None: mock_get.reset_mock() # Test custom parameter ("closed") - client.list_repo_issues("owner", "repo", state="closed") + client.issues.list_repo_issues("owner", "repo", state="closed") mock_get.assert_called_once() args, _ = mock_get.call_args assert "type=issues" in args[0] @@ -36,7 +36,7 @@ def test_gitea_client_list_repo_pull_requests() -> None: mock_get.return_value = mock_response # Test default parameter ("open") - client.list_repo_pull_requests("owner", "repo") + client.prs.list_repo_pull_requests("owner", "repo") mock_get.assert_called_once() args, _ = mock_get.call_args assert "state=open" in args[0] @@ -44,7 +44,7 @@ def test_gitea_client_list_repo_pull_requests() -> None: mock_get.reset_mock() # Test custom parameter ("closed") - client.list_repo_pull_requests("owner", "repo", state="closed") + client.prs.list_repo_pull_requests("owner", "repo", state="closed") mock_get.assert_called_once() args, _ = mock_get.call_args assert "state=closed" in args[0] @@ -55,14 +55,17 @@ def test_gitea_client_list_assigned_issues() -> None: user_mock: MagicMock = MagicMock() user_mock.login = "testuser" - with patch.object(client, "get_authenticated_user", return_value=user_mock), \ - patch("httpx.Client.get") as mock_get: + with ( + patch.object(client.repos, "get_authenticated_user", return_value=user_mock), + patch.object(client.issues, "_get_user", return_value=user_mock), + patch("httpx.Client.get") as mock_get, + ): mock_response: MagicMock = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = [] mock_get.return_value = mock_response - client.list_assigned_issues("owner", "repo") + client.issues.list_assigned_issues("owner", "repo") mock_get.assert_called_once() args, _ = mock_get.call_args assert "type=issues" in args[0] @@ -74,18 +77,36 @@ def test_gitea_client_list_assigned_pull_requests() -> None: user_mock: MagicMock = MagicMock() user_mock.login = "testuser" - with patch.object(client, "get_authenticated_user", return_value=user_mock), \ - patch("httpx.Client.get") as mock_get: + with ( + patch.object(client.repos, "get_authenticated_user", return_value=user_mock), + patch.object(client.prs, "_get_user", return_value=user_mock), + patch("httpx.Client.get") as mock_get, + ): mock_response: MagicMock = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = [ - {"number": 1, "title": "PR 1", "assignee": {"login": "testuser"}, "user": {"login": "otheruser"}}, - {"number": 2, "title": "PR 2", "assignee": None, "user": {"login": "testuser"}}, - {"number": 3, "title": "PR 3", "assignee": {"login": "otheruser"}, "user": {"login": "otheruser"}} + { + "number": 1, + "title": "PR 1", + "assignee": {"login": "testuser"}, + "user": {"login": "otheruser"}, + }, + { + "number": 2, + "title": "PR 2", + "assignee": None, + "user": {"login": "testuser"}, + }, + { + "number": 3, + "title": "PR 3", + "assignee": {"login": "otheruser"}, + "user": {"login": "otheruser"}, + }, ] mock_get.return_value = mock_response - res = client.list_assigned_pull_requests("owner", "repo") + res = client.prs.list_assigned_pull_requests("owner", "repo") mock_get.assert_called_once() assert len(res) == 2 numbers = [pr.number for pr in res] @@ -106,7 +127,7 @@ def test_gitea_client_list_unread_notifications() -> None: mock_get.return_value = mock_response # Test without since - res = client.list_unread_notifications() + res = client.notifications.list_unread_notifications() mock_get.assert_called_once() _, kwargs = mock_get.call_args assert kwargs.get("params") == {"all": "false"} @@ -116,20 +137,23 @@ def test_gitea_client_list_unread_notifications() -> None: mock_get.reset_mock() # Test with since - res = client.list_unread_notifications(since="2026-06-30T21:41:16+02:00") + res = client.notifications.list_unread_notifications( + since="2026-06-30T21:41:16+02:00" + ) mock_get.assert_called_once() _, kwargs = mock_get.call_args - assert kwargs.get("params") == {"all": "false", "since": "2026-06-30T21:41:16+02:00"} + assert kwargs.get("params") == { + "all": "false", + "since": "2026-06-30T21:41:16+02:00", + } import pytest + def test_gitea_client_get_authenticated_user_failure() -> None: client: GiteaClient = GiteaClient() with patch("httpx.Client.get") as mock_get: mock_get.side_effect = Exception("Connection error") with pytest.raises(RuntimeError, match="Could not get authenticated user"): - client.get_authenticated_user() - - - + client.repos.get_authenticated_user() diff --git a/tests/test_file_tools.py b/tests/test_file_tools.py index 2f7809d..1f3d3e3 100644 --- a/tests/test_file_tools.py +++ b/tests/test_file_tools.py @@ -3,19 +3,28 @@ from gitea.client import GiteaClient from gitea.tools.file_tools import FileTools -def test_get_file_content_string_success() -> None: +def _create_mock_client() -> MagicMock: + """Create a mock GiteaClient with sub-client attributes.""" mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_file_content.return_value = "file content here" + mock_client.files = MagicMock() + return mock_client + + +def test_get_file_content_string_success() -> None: + mock_client = _create_mock_client() + mock_client.files.get_file_content.return_value = "file content here" file_tools: FileTools = FileTools(mock_client) res: str = file_tools.get_file_content("owner", "repo", "path/to/file") assert res == "1: file content here" - mock_client.get_file_content.assert_called_once_with("owner", "repo", "path/to/file") + mock_client.files.get_file_content.assert_called_once_with( + "owner", "repo", "path/to/file" + ) def test_get_file_content_list_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_file_content.return_value = ["line1", "line2"] + mock_client = _create_mock_client() + mock_client.files.get_file_content.return_value = ["line1", "line2"] file_tools: FileTools = FileTools(mock_client) res: str = file_tools.get_file_content("owner", "repo", "path/to/file") @@ -23,8 +32,8 @@ def test_get_file_content_list_success() -> None: def test_get_file_content_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_file_content.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.files.get_file_content.side_effect = Exception("API Error") file_tools: FileTools = FileTools(mock_client) res: str = file_tools.get_file_content("owner", "repo", "path/to/file") @@ -32,66 +41,86 @@ def test_get_file_content_failure() -> None: def test_get_file_content_with_ref_string_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_file_content.return_value = "file content here" + mock_client = _create_mock_client() + mock_client.files.get_file_content.return_value = "file content here" file_tools: FileTools = FileTools(mock_client) - res: str = file_tools.get_file_content_with_ref("owner", "repo", "path/to/file", "main") + res: str = file_tools.get_file_content_with_ref( + "owner", "repo", "path/to/file", "main" + ) assert res == "1: file content here" - mock_client.get_file_content.assert_called_once_with("owner", "repo", "path/to/file", "main") + mock_client.files.get_file_content.assert_called_once_with( + "owner", "repo", "path/to/file", "main" + ) def test_get_file_content_with_ref_list_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_file_content.return_value = ["line1", "line2"] + mock_client = _create_mock_client() + mock_client.files.get_file_content.return_value = ["line1", "line2"] file_tools: FileTools = FileTools(mock_client) - res: str = file_tools.get_file_content_with_ref("owner", "repo", "path/to/file", "main") + res: str = file_tools.get_file_content_with_ref( + "owner", "repo", "path/to/file", "main" + ) assert res == "1: line1\n2: line2" def test_get_file_content_with_ref_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_file_content.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.files.get_file_content.side_effect = Exception("API Error") file_tools: FileTools = FileTools(mock_client) - res: str = file_tools.get_file_content_with_ref("owner", "repo", "path/to/file", "main") + res: str = file_tools.get_file_content_with_ref( + "owner", "repo", "path/to/file", "main" + ) assert "Error getting file content: API Error" in res def test_commit_file_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.update_file.return_value = {} + mock_client = _create_mock_client() + mock_client.files.update_file.return_value = {} file_tools: FileTools = FileTools(mock_client) - res: str = file_tools.commit_file("owner", "repo", "path/to/file", "msg", "content", "branch") + res: str = file_tools.commit_file( + "owner", "repo", "path/to/file", "msg", "content", "branch" + ) assert "committed successfully" in res - mock_client.update_file.assert_called_once_with("owner", "repo", "path/to/file", "msg", "content", "branch") + mock_client.files.update_file.assert_called_once_with( + "owner", "repo", "path/to/file", "msg", "content", "branch" + ) def test_commit_file_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.update_file.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.files.update_file.side_effect = Exception("API Error") file_tools: FileTools = FileTools(mock_client) - res: str = file_tools.commit_file("owner", "repo", "path/to/file", "msg", "content", "branch") + res: str = file_tools.commit_file( + "owner", "repo", "path/to/file", "msg", "content", "branch" + ) assert "Error committing file: API Error" in res def test_update_file_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.update_file.return_value = {} + mock_client = _create_mock_client() + mock_client.files.update_file.return_value = {} file_tools: FileTools = FileTools(mock_client) - res: str = file_tools.update_file("owner", "repo", "path/to/file", "msg", "content", "branch") + res: str = file_tools.update_file( + "owner", "repo", "path/to/file", "msg", "content", "branch" + ) assert "updated in" in res - mock_client.update_file.assert_called_once_with("owner", "repo", "path/to/file", "msg", "content", "branch") + mock_client.files.update_file.assert_called_once_with( + "owner", "repo", "path/to/file", "msg", "content", "branch" + ) def test_update_file_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.update_file.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.files.update_file.side_effect = Exception("API Error") file_tools: FileTools = FileTools(mock_client) - res: str = file_tools.update_file("owner", "repo", "path/to/file", "msg", "content", "branch") + res: str = file_tools.update_file( + "owner", "repo", "path/to/file", "msg", "content", "branch" + ) assert "Error updating file: API Error" in res diff --git a/tests/test_git_tools.py b/tests/test_git_tools.py index 261293a..8a5c368 100644 --- a/tests/test_git_tools.py +++ b/tests/test_git_tools.py @@ -3,19 +3,26 @@ from gitea.client import GiteaClient from gitea.tools.git_tools import GitTools -def test_create_branch_success() -> None: +def _create_mock_client() -> MagicMock: + """Create a mock GiteaClient with sub-client attributes.""" mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.create_ref.return_value = {} + mock_client.files = MagicMock() + return mock_client + + +def test_create_branch_success() -> None: + mock_client = _create_mock_client() + mock_client.files.create_ref.return_value = {} git_tools: GitTools = GitTools(mock_client) res: str = git_tools.create_branch("owner", "repo", "ref", "sha") assert res == "Branch 'ref' created successfully in owner/repo." - mock_client.create_ref.assert_called_once_with("owner", "repo", "ref", "sha") + mock_client.files.create_ref.assert_called_once_with("owner", "repo", "ref", "sha") def test_create_branch_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.create_ref.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.files.create_ref.side_effect = Exception("API Error") git_tools: GitTools = GitTools(mock_client) res: str = git_tools.create_branch("owner", "repo", "ref", "sha") diff --git a/tests/test_issue_tools.py b/tests/test_issue_tools.py index 8b2a001..ca6b078 100644 --- a/tests/test_issue_tools.py +++ b/tests/test_issue_tools.py @@ -6,10 +6,18 @@ from gitea.models import IssueModel, CommentModel, LabelModel, RepositoryModel from gitea.tools.issue_tools import IssueTools -def test_get_issue_success() -> None: +def _create_mock_client() -> MagicMock: + """Create a mock GiteaClient with sub-client attributes.""" mock_client: MagicMock = MagicMock(spec=GiteaClient) + mock_client.issues = MagicMock() + mock_client.repos = MagicMock() + return mock_client + + +def test_get_issue_success() -> None: + mock_client = _create_mock_client() issue: IssueModel = IssueModel(number=1, title="Test Issue", state="open") - mock_client.get_issue.return_value = issue + mock_client.issues.get_issue.return_value = issue issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.get_issue("owner", "repo", 1) @@ -17,12 +25,12 @@ def test_get_issue_success() -> None: data: dict[str, Any] = json.loads(res) assert data["number"] == 1 assert data["title"] == "Test Issue" - mock_client.get_issue.assert_called_once_with("owner", "repo", 1) + mock_client.issues.get_issue.assert_called_once_with("owner", "repo", 1) def test_get_issue_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_issue.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.issues.get_issue.side_effect = Exception("API Error") issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.get_issue("owner", "repo", 1) @@ -30,18 +38,18 @@ def test_get_issue_failure() -> None: def test_close_issue_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.close_issue.return_value = IssueModel(number=1, state="closed") + mock_client = _create_mock_client() + mock_client.issues.close_issue.return_value = IssueModel(number=1, state="closed") issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.close_issue("owner", "repo", 1) assert res == "Issue #1 closed successfully." - mock_client.close_issue.assert_called_once_with("owner", "repo", 1) + mock_client.issues.close_issue.assert_called_once_with("owner", "repo", 1) def test_close_issue_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.close_issue.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.issues.close_issue.side_effect = Exception("API Error") issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.close_issue("owner", "repo", 1) @@ -49,9 +57,9 @@ def test_close_issue_failure() -> None: def test_get_issue_comments_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) + mock_client = _create_mock_client() comment: CommentModel = CommentModel(id=123, body="Comment body") - mock_client.get_issue_comments.return_value = [comment] + mock_client.issues.get_issue_comments.return_value = [comment] issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.get_issue_comments("owner", "repo", 1) @@ -61,8 +69,8 @@ def test_get_issue_comments_success() -> None: def test_get_issue_comments_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_issue_comments.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.issues.get_issue_comments.side_effect = Exception("API Error") issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.get_issue_comments("owner", "repo", 1) @@ -70,23 +78,23 @@ def test_get_issue_comments_failure() -> None: def test_list_assigned_issues_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) + mock_client = _create_mock_client() repo: RepositoryModel = RepositoryModel(name="repo1", owner="owner1") issue: IssueModel = IssueModel(number=1, title="Test Issue") - mock_client.list_all_user_repos.return_value = [repo] - mock_client.list_assigned_issues.return_value = [issue] + mock_client.repos.list_all_user_repos.return_value = [repo] + mock_client.issues.list_assigned_issues.return_value = [issue] issue_tools: IssueTools = IssueTools(mock_client) res: list[dict[str, Any]] = issue_tools.list_assigned_issues() assert len(res) == 1 assert res[0]["number"] == 1 - mock_client.list_all_user_repos.assert_called_once() - mock_client.list_assigned_issues.assert_called_once_with("owner1", "repo1") + mock_client.repos.list_all_user_repos.assert_called_once() + mock_client.issues.list_assigned_issues.assert_called_once_with("owner1", "repo1") def test_list_assigned_issues_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.list_all_user_repos.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.repos.list_all_user_repos.side_effect = Exception("API Error") issue_tools: IssueTools = IssueTools(mock_client) res: list[dict[str, Any]] = issue_tools.list_assigned_issues() @@ -94,9 +102,9 @@ def test_list_assigned_issues_failure() -> None: def test_list_issues_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) + mock_client = _create_mock_client() issue: IssueModel = IssueModel(number=1, title="Test Issue") - mock_client.list_repo_issues.return_value = [issue] + mock_client.issues.list_repo_issues.return_value = [issue] issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.list_issues("owner", "repo") @@ -104,8 +112,8 @@ def test_list_issues_success() -> None: def test_list_issues_empty() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.list_repo_issues.return_value = [] + mock_client = _create_mock_client() + mock_client.issues.list_repo_issues.return_value = [] issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.list_issues("owner", "repo") @@ -113,8 +121,8 @@ def test_list_issues_empty() -> None: def test_list_issues_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.list_repo_issues.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.issues.list_repo_issues.side_effect = Exception("API Error") issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.list_issues("owner", "repo") @@ -122,23 +130,23 @@ def test_list_issues_failure() -> None: def test_create_issue_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) + mock_client = _create_mock_client() issue: IssueModel = IssueModel(number=2) - mock_client.create_issue.return_value = issue + mock_client.issues.create_issue.return_value = issue issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.create_issue( "owner", "repo", "Title", "Body", ["label1"], ["assignee1"] ) assert res == "Issue #2 created successfully in owner/repo." - mock_client.create_issue.assert_called_once_with( + mock_client.issues.create_issue.assert_called_once_with( "owner", "repo", "Title", "Body", ["label1"], ["assignee1"] ) def test_create_issue_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.create_issue.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.issues.create_issue.side_effect = Exception("API Error") issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.create_issue("owner", "repo", "Title", "Body") @@ -146,8 +154,8 @@ def test_create_issue_failure() -> None: def test_add_label_to_issue_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.add_label.return_value = LabelModel(name="bug") + mock_client = _create_mock_client() + mock_client.issues.add_label.return_value = LabelModel(name="bug") issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.add_label_to_issue("owner", "repo", 1, "bug") @@ -155,8 +163,8 @@ def test_add_label_to_issue_success() -> None: def test_add_label_to_issue_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.add_label.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.issues.add_label.side_effect = Exception("API Error") issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.add_label_to_issue("owner", "repo", 1, "bug") @@ -164,8 +172,8 @@ def test_add_label_to_issue_failure() -> None: def test_add_comment_to_issue_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.add_comment.return_value = CommentModel(id=1) + mock_client = _create_mock_client() + mock_client.issues.add_comment.return_value = CommentModel(id=1) issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.add_comment_to_issue("owner", "repo", 1, "body") @@ -173,8 +181,8 @@ def test_add_comment_to_issue_success() -> None: def test_add_comment_to_issue_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.add_comment.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.issues.add_comment.side_effect = Exception("API Error") issue_tools: IssueTools = IssueTools(mock_client) res: str = issue_tools.add_comment_to_issue("owner", "repo", 1, "body") diff --git a/tests/test_pr_tools.py b/tests/test_pr_tools.py index a0e23a0..2a35903 100644 --- a/tests/test_pr_tools.py +++ b/tests/test_pr_tools.py @@ -6,10 +6,18 @@ from gitea.models import PullRequestModel, CommentModel, RepositoryModel from gitea.tools.pr_tools import PRTools -def test_get_pull_request_success() -> None: +def _create_mock_client() -> MagicMock: + """Create a mock GiteaClient with sub-client attributes.""" mock_client: MagicMock = MagicMock(spec=GiteaClient) + mock_client.prs = MagicMock() + mock_client.repos = MagicMock() + return mock_client + + +def test_get_pull_request_success() -> None: + mock_client = _create_mock_client() pr: PullRequestModel = PullRequestModel(number=1, title="Test PR", state="open") - mock_client.get_pull_request.return_value = pr + mock_client.prs.get_pull_request.return_value = pr pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.get_pull_request("owner", "repo", 1) @@ -17,12 +25,12 @@ def test_get_pull_request_success() -> None: data: dict[str, Any] = json.loads(res) assert data["number"] == 1 assert data["title"] == "Test PR" - mock_client.get_pull_request.assert_called_once_with("owner", "repo", 1) + mock_client.prs.get_pull_request.assert_called_once_with("owner", "repo", 1) def test_get_pull_request_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_pull_request.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.prs.get_pull_request.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.get_pull_request("owner", "repo", 1) @@ -30,18 +38,20 @@ def test_get_pull_request_failure() -> None: def test_close_pull_request_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.close_pull_request.return_value = PullRequestModel(number=1, state="closed") + mock_client = _create_mock_client() + mock_client.prs.close_pull_request.return_value = PullRequestModel( + number=1, state="closed" + ) pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.close_pull_request("owner", "repo", 1) assert res == "Pull request #1 closed successfully." - mock_client.close_pull_request.assert_called_once_with("owner", "repo", 1) + mock_client.prs.close_pull_request.assert_called_once_with("owner", "repo", 1) def test_close_pull_request_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.close_pull_request.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.prs.close_pull_request.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.close_pull_request("owner", "repo", 1) @@ -49,9 +59,9 @@ def test_close_pull_request_failure() -> None: def test_get_pull_request_comments_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) + mock_client = _create_mock_client() comment: CommentModel = CommentModel(id=123, body="Comment body") - mock_client.get_pull_request_comments.return_value = [comment] + mock_client.prs.get_pull_request_comments.return_value = [comment] pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.get_pull_request_comments("owner", "repo", 1) @@ -61,8 +71,8 @@ def test_get_pull_request_comments_success() -> None: def test_get_pull_request_comments_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_pull_request_comments.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.prs.get_pull_request_comments.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.get_pull_request_comments("owner", "repo", 1) @@ -70,23 +80,25 @@ def test_get_pull_request_comments_failure() -> None: def test_list_assigned_pull_requests_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) + mock_client = _create_mock_client() repo: RepositoryModel = RepositoryModel(name="repo1", owner="owner1") pr: PullRequestModel = PullRequestModel(number=1, title="Test PR") - mock_client.list_all_user_repos.return_value = [repo] - mock_client.list_assigned_pull_requests.return_value = [pr] + mock_client.repos.list_all_user_repos.return_value = [repo] + mock_client.prs.list_assigned_pull_requests.return_value = [pr] pr_tools: PRTools = PRTools(mock_client) res: list[dict[str, Any]] = pr_tools.list_assigned_pull_requests() assert len(res) == 1 assert res[0]["number"] == 1 - mock_client.list_all_user_repos.assert_called_once() - mock_client.list_assigned_pull_requests.assert_called_once_with("owner1", "repo1") + mock_client.repos.list_all_user_repos.assert_called_once() + mock_client.prs.list_assigned_pull_requests.assert_called_once_with( + "owner1", "repo1" + ) def test_list_assigned_pull_requests_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.list_all_user_repos.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.repos.list_all_user_repos.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: list[dict[str, Any]] = pr_tools.list_assigned_pull_requests() @@ -94,9 +106,9 @@ def test_list_assigned_pull_requests_failure() -> None: def test_list_pull_requests_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) + mock_client = _create_mock_client() pr: PullRequestModel = PullRequestModel(number=1, title="Test PR") - mock_client.list_repo_pull_requests.return_value = [pr] + mock_client.prs.list_repo_pull_requests.return_value = [pr] pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.list_pull_requests("owner", "repo") @@ -104,8 +116,8 @@ def test_list_pull_requests_success() -> None: def test_list_pull_requests_empty() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.list_repo_pull_requests.return_value = [] + mock_client = _create_mock_client() + mock_client.prs.list_repo_pull_requests.return_value = [] pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.list_pull_requests("owner", "repo") @@ -113,8 +125,8 @@ def test_list_pull_requests_empty() -> None: def test_list_pull_requests_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.list_repo_pull_requests.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.prs.list_repo_pull_requests.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.list_pull_requests("owner", "repo") @@ -122,20 +134,24 @@ def test_list_pull_requests_failure() -> None: def test_create_pull_request_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) + mock_client = _create_mock_client() pr: PullRequestModel = PullRequestModel(number=2, title="Title") - mock_client.create_pr_via_tea.return_value = pr + mock_client.prs.create_pr_via_tea.return_value = pr pr_tools: PRTools = PRTools(mock_client) - res: str = pr_tools.create_pull_request("owner", "repo", "head", "base", "Title", "Desc") + res: str = pr_tools.create_pull_request( + "owner", "repo", "head", "base", "Title", "Desc" + ) data: dict[str, Any] = json.loads(res) assert data["number"] == 2 - mock_client.create_pr_via_tea.assert_called_once_with("owner", "repo", "Title", "Desc", "head", "base") + mock_client.prs.create_pr_via_tea.assert_called_once_with( + "owner", "repo", "Title", "Desc", "head", "base" + ) def test_create_pull_request_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.create_pr_via_tea.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.prs.create_pr_via_tea.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.create_pull_request("owner", "repo", "head", "base", "Title") @@ -143,8 +159,8 @@ def test_create_pull_request_failure() -> None: def test_add_label_to_pr_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.add_label_pr.return_value = {} + mock_client = _create_mock_client() + mock_client.prs.add_label_pr.return_value = {} pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.add_label_to_pr("owner", "repo", 1, "bug") @@ -152,8 +168,8 @@ def test_add_label_to_pr_success() -> None: def test_add_label_to_pr_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.add_label_pr.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.prs.add_label_pr.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.add_label_to_pr("owner", "repo", 1, "bug") @@ -161,8 +177,8 @@ def test_add_label_to_pr_failure() -> None: def test_get_pull_request_diff_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_pull_request_diff.return_value = "diff content" + mock_client = _create_mock_client() + mock_client.prs.get_pull_request_diff.return_value = "diff content" pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.get_pull_request_diff("owner", "repo", 1) @@ -170,8 +186,8 @@ def test_get_pull_request_diff_success() -> None: def test_get_pull_request_diff_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_pull_request_diff.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.prs.get_pull_request_diff.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.get_pull_request_diff("owner", "repo", 1) @@ -179,8 +195,8 @@ def test_get_pull_request_diff_failure() -> None: def test_get_pull_request_patch_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_pull_request_patch.return_value = "patch content" + mock_client = _create_mock_client() + mock_client.prs.get_pull_request_patch.return_value = "patch content" pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.get_pull_request_patch("owner", "repo", 1) @@ -188,8 +204,8 @@ def test_get_pull_request_patch_success() -> None: def test_get_pull_request_patch_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.get_pull_request_patch.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.prs.get_pull_request_patch.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.get_pull_request_patch("owner", "repo", 1) @@ -197,8 +213,8 @@ def test_get_pull_request_patch_failure() -> None: def test_approve_pull_request_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.approve_pr.return_value = {} + mock_client = _create_mock_client() + mock_client.prs.approve_pr.return_value = {} pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.approve_pull_request("owner", "repo", 1, "good") @@ -206,8 +222,8 @@ def test_approve_pull_request_success() -> None: def test_approve_pull_request_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.approve_pr.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.prs.approve_pr.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.approve_pull_request("owner", "repo", 1, "good") @@ -215,8 +231,8 @@ def test_approve_pull_request_failure() -> None: def test_request_changes_success() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.request_changes_pr.return_value = {} + mock_client = _create_mock_client() + mock_client.prs.request_changes_pr.return_value = {} pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.request_changes("owner", "repo", 1, "bad") @@ -224,8 +240,8 @@ def test_request_changes_success() -> None: def test_request_changes_failure() -> None: - mock_client: MagicMock = MagicMock(spec=GiteaClient) - mock_client.request_changes_pr.side_effect = Exception("API Error") + mock_client = _create_mock_client() + mock_client.prs.request_changes_pr.side_effect = Exception("API Error") pr_tools: PRTools = PRTools(mock_client) res: str = pr_tools.request_changes("owner", "repo", 1, "bad") diff --git a/tests/test_workspace.py b/tests/test_workspace.py index 8e5c707..60b9f35 100644 --- a/tests/test_workspace.py +++ b/tests/test_workspace.py @@ -9,25 +9,25 @@ from gitea.workspace import WorkspaceManager @patch("gitea.workspace.subprocess.run") @patch("gitea.workspace.GiteaClient") def test_workspace_manager_configure_repo_user( - mock_client_class: MagicMock, - mock_run: MagicMock + mock_client_class: MagicMock, mock_run: MagicMock ) -> None: mock_client = MagicMock() mock_client_class.return_value = mock_client + mock_client.repos = MagicMock() mock_user = MagicMock() mock_user.full_name = "Agent Tester" mock_user.login = "agent-test" mock_user.email = "agent-test@example.com" - mock_client.get_authenticated_user.return_value = mock_user + mock_client.repos.get_authenticated_user.return_value = mock_user workspace = WorkspaceManager() repo_path = Path("/tmp/mock-repo") - + workspace._configure_repo_user(repo_path) - + assert mock_run.call_count >= 3 calls = [c[0][0] for c in mock_run.call_args_list] - + assert any("http.extraHeader" in call for call in calls) assert any("user.name" in call for call in calls) assert any("user.email" in call for call in calls) @@ -36,27 +36,27 @@ def test_workspace_manager_configure_repo_user( @patch("gitea.workspace.subprocess.run") @patch("gitea.workspace.GiteaClient") def test_workspace_manager_clone_repo( - mock_client_class: MagicMock, - mock_run: MagicMock + mock_client_class: MagicMock, mock_run: MagicMock ) -> None: mock_client = MagicMock() mock_client_class.return_value = mock_client + mock_client.repos = MagicMock() mock_user = MagicMock() mock_user.full_name = "Agent Tester" mock_user.login = "agent-test" mock_user.email = "agent-test@example.com" - mock_client.get_authenticated_user.return_value = mock_user + mock_client.repos.get_authenticated_user.return_value = mock_user workspace = WorkspaceManager() - + with patch.object(workspace, "_configure_repo_user") as mock_configure: with patch.object(workspace, "get_repo_path") as mock_get_path: mock_repo_path = MagicMock(spec=Path) mock_repo_path.exists.return_value = False mock_get_path.return_value = mock_repo_path - + workspace.clone_repo("meeks/repo1") - + mock_run.assert_called_once() args = mock_run.call_args[0][0] assert "clone" in args @@ -64,13 +64,15 @@ def test_workspace_manager_clone_repo( mock_configure.assert_called_once_with(mock_repo_path) +@patch("gitea.workspace.subprocess.run") @patch("gitea.workspace.GiteaClient") def test_workspace_manager_fails_if_no_authenticated_user( - mock_client_class: MagicMock + mock_client_class: MagicMock, mock_run: MagicMock ) -> None: mock_client = MagicMock() mock_client_class.return_value = mock_client - mock_client.get_authenticated_user.return_value = None + mock_client.repos = MagicMock() + mock_client.repos.get_authenticated_user.return_value = None workspace = WorkspaceManager() with pytest.raises(RuntimeError, match="No authenticated user found."): @@ -80,15 +82,17 @@ def test_workspace_manager_fails_if_no_authenticated_user( workspace._configure_repo_user(Path("/tmp/mock-repo")) +@patch("gitea.workspace.subprocess.run") @patch("gitea.workspace.GiteaClient") def test_workspace_manager_fails_if_authenticated_user_has_no_login( - mock_client_class: MagicMock + mock_client_class: MagicMock, mock_run: MagicMock ) -> None: mock_client = MagicMock() mock_client_class.return_value = mock_client + mock_client.repos = MagicMock() mock_user = MagicMock() mock_user.login = "" - mock_client.get_authenticated_user.return_value = mock_user + mock_client.repos.get_authenticated_user.return_value = mock_user workspace = WorkspaceManager() with pytest.raises(RuntimeError, match="No authenticated user found."): @@ -101,17 +105,17 @@ def test_workspace_manager_fails_if_authenticated_user_has_no_login( @patch("gitea.workspace.subprocess.run") @patch("gitea.workspace.GiteaClient") def test_workspace_manager_sanitize_repo_no_changes( - mock_client_class: MagicMock, - mock_run: MagicMock + mock_client_class: MagicMock, mock_run: MagicMock ) -> None: # Setup Gitea client mock mock_client = MagicMock() mock_client_class.return_value = mock_client + mock_client.repos = MagicMock() mock_user = MagicMock() mock_user.full_name = "Agent Tester" mock_user.login = "agent-test" mock_user.email = "agent-test@example.com" - mock_client.get_authenticated_user.return_value = mock_user + mock_client.repos.get_authenticated_user.return_value = mock_user # Mock subprocess.run for status check and others def mock_run_side_effect(args: list[str], **kwargs: Any) -> MagicMock: @@ -144,17 +148,17 @@ def test_workspace_manager_sanitize_repo_no_changes( @patch("gitea.workspace.subprocess.run") @patch("gitea.workspace.GiteaClient") def test_workspace_manager_sanitize_repo_with_changes( - mock_client_class: MagicMock, - mock_run: MagicMock + mock_client_class: MagicMock, mock_run: MagicMock ) -> None: # Setup Gitea client mock mock_client = MagicMock() mock_client_class.return_value = mock_client + mock_client.repos = MagicMock() mock_user = MagicMock() mock_user.full_name = "Agent Tester" mock_user.login = "agent-test" mock_user.email = "agent-test@example.com" - mock_client.get_authenticated_user.return_value = mock_user + mock_client.repos.get_authenticated_user.return_value = mock_user # Mock subprocess.run to show modified files def mock_run_side_effect(args: list[str], **kwargs: Any) -> MagicMock: @@ -183,20 +187,21 @@ def test_workspace_manager_sanitize_repo_with_changes( @patch("gitea.workspace.subprocess.run") @patch("gitea.workspace.GiteaClient") def test_workspace_manager_sanitize_repo_fails( - mock_client_class: MagicMock, - mock_run: MagicMock + mock_client_class: MagicMock, mock_run: MagicMock ) -> None: # Setup Gitea client mock mock_client = MagicMock() mock_client_class.return_value = mock_client + mock_client.repos = MagicMock() mock_user = MagicMock() mock_user.full_name = "Agent Tester" mock_user.login = "agent-test" mock_user.email = "agent-test@example.com" - mock_client.get_authenticated_user.return_value = mock_user + mock_client.repos.get_authenticated_user.return_value = mock_user # Mock remote set-url to fail import subprocess + mock_run.side_effect = subprocess.CalledProcessError(1, "git remote set-url") workspace = WorkspaceManager()