michael/coding-agent-gitea
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: extract focused clients from GiteaClient (Slices 1-6)

Browse Source 
- 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
This commit is contained in:
meeks
2026-07-17 07:31:05 +02:00
parent 3c94c3cfac
commit e91780169e
21 changed files with 1480 additions and 728 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/dispatcher.py
+29 -25
View File
@@ -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"(?<!\d){issue_number}(?!\d)", ref):
@@ -195,7 +195,7 @@ class PRTaskProcessor(TaskProcessor):
pr_details = pr_info.model_dump_json(indent=2)
pr_diff = ""
try:
pr_diff = self.client.get_pull_request_diff(
pr_diff = self.client.prs.get_pull_request_diff(
self.owner, self.repo_name, pr_number
)
except Exception as e:
@@ -204,7 +204,7 @@ class PRTaskProcessor(TaskProcessor):
pr_files: list[PullRequestFileModel] = []
try:
pr_files = self.client.get_pull_request_files(
pr_files = self.client.prs.get_pull_request_files(
self.owner, self.repo_name, pr_number
)
except Exception as e:
@@ -220,7 +220,7 @@ class PRTaskProcessor(TaskProcessor):
comments: list[CommentModel] = []
try:
comments = self.client.get_pull_request_comments(
comments = self.client.prs.get_pull_request_comments(
self.owner, self.repo_name, pr_number
)
if not isinstance(comments, list):
@@ -232,7 +232,9 @@ class PRTaskProcessor(TaskProcessor):
reviews: list[dict[str, Any]] = []
try:
reviews = self.client.get_pr_reviews(self.owner, self.repo_name, pr_number)
reviews = self.client.prs.get_pr_reviews(
self.owner, self.repo_name, pr_number
)
if not isinstance(reviews, list):
reviews = []
except Exception as e:
@@ -302,8 +304,10 @@ class PRTaskProcessor(TaskProcessor):
issues_details = []
for issue_num in linked_issues:
try:
issue = self.client.get_issue(self.owner, self.repo_name, issue_num)
issue_comments = self.client.get_issue_comments(
issue = self.client.issues.get_issue(
self.owner, self.repo_name, issue_num
)
issue_comments = self.client.issues.get_issue_comments(
self.owner, self.repo_name, issue_num
)
comments_list = (
@@ -392,14 +396,14 @@ class PRTaskProcessor(TaskProcessor):
async def process(self, attempt_limit: int) -> 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"<!-- agent:plan-proposal -->\n"
f"<!-- agent:awaiting-reply -->"
)
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"<!-- agent:question-response -->\n"
f"<!-- agent:awaiting-reply -->"
)
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(
core/orchestrator.py
+13 -5
View File
@@ -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."
)
gitea/__init__.py
+35
View File
@@ -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",
]
gitea/client.py
+35 -469
View File
@@ -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
gitea/files_client.py
+116
View File
@@ -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()
gitea/issues_client.py
+248
View File
@@ -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())
gitea/notifications_client.py
+78
View File
@@ -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
gitea/prs_client.py
+462
View File
@@ -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
gitea/repos_client.py
+72
View File
@@ -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
gitea/tools/file_tools.py
+10 -6
View File
@@ -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)}"
gitea/tools/git_tools.py
+1 -1
View File
@@ -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)}"
gitea/tools/issue_tools.py
+9 -9
View File
@@ -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}"
gitea/tools/pr_tools.py
+44 -17
View File
@@ -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)}"
gitea/workspace.py
+64 -21
View File
@@ -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
main.py
+1 -1
View File
@@ -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}")
tests/test_client.py
+44 -20
View File
@@ -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()
tests/test_file_tools.py
+60 -31
View File
@@ -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
tests/test_git_tools.py
+12 -5
View File
@@ -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")
tests/test_issue_tools.py
+48 -40
View File
@@ -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")
tests/test_pr_tools.py
+69 -53
View File
@@ -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")
tests/test_workspace.py
+24 -19
View File
@@ -9,16 +9,16 @@ 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")
@@ -36,16 +36,16 @@ 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()
@@ -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()