Compare commits
2 Commits
02de35e1b0
...
b4a90227e7
Author | SHA1 | Date | |
---|---|---|---|
b4a90227e7 | |||
98f6e9e1bc |
|
@ -467,6 +467,109 @@ def get_diff(cwd: Path, base_branch: str, current_branch: str) -> str:
|
|||
return result.stdout.strip()
|
||||
|
||||
|
||||
def contains_only_ruff_changes(
|
||||
cwd: Path,
|
||||
base_branch: str,
|
||||
current_branch: str,
|
||||
) -> bool:
|
||||
"""Check if the branch contains only ruff formatting changes.
|
||||
|
||||
Args:
|
||||
cwd: The current working directory (repository path).
|
||||
base_branch: The name of the base branch to compare against.
|
||||
current_branch: The name of the current branch to check.
|
||||
|
||||
Returns:
|
||||
True if the branch contains only ruff formatting changes, False otherwise.
|
||||
"""
|
||||
# Get all commit messages to check if they're only ruff-related
|
||||
commit_messages = get_commit_messages(cwd, base_branch, current_branch)
|
||||
|
||||
# If no commits, then no changes at all
|
||||
if not commit_messages:
|
||||
return True
|
||||
|
||||
# Check if all commits are ruff-related
|
||||
ruff_related_patterns = [
|
||||
'ruff',
|
||||
'format',
|
||||
'lint',
|
||||
'auto-fix',
|
||||
'autofix',
|
||||
'code style',
|
||||
'formatting',
|
||||
]
|
||||
|
||||
all_ruff_commits = True
|
||||
for message in commit_messages:
|
||||
message_lower = message.lower()
|
||||
# Skip empty messages
|
||||
if not message.strip():
|
||||
continue
|
||||
# Check if this commit message indicates ruff/formatting changes
|
||||
is_ruff_commit = any(
|
||||
pattern in message_lower for pattern in ruff_related_patterns
|
||||
)
|
||||
if not is_ruff_commit:
|
||||
all_ruff_commits = False
|
||||
break
|
||||
|
||||
if not all_ruff_commits:
|
||||
return False
|
||||
|
||||
# Additional check: get the actual diff and analyze it
|
||||
# This helps catch cases where commit messages might be misleading
|
||||
try:
|
||||
diff_result = subprocess.run(
|
||||
['git', 'diff', f'{base_branch}..{current_branch}'],
|
||||
check=True,
|
||||
cwd=cwd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
diff_content = diff_result.stdout
|
||||
|
||||
# If no diff content, then only ruff changes
|
||||
if not diff_content.strip():
|
||||
return True
|
||||
|
||||
# Analyze the diff to see if it contains only whitespace/formatting changes
|
||||
lines = diff_content.split('\n')
|
||||
substantive_changes = 0
|
||||
|
||||
for line in lines:
|
||||
# Skip diff headers and file markers
|
||||
if (
|
||||
line.startswith('diff --git')
|
||||
or line.startswith('index ')
|
||||
or line.startswith('+++')
|
||||
or line.startswith('---')
|
||||
):
|
||||
continue
|
||||
# Skip context lines (no + or -)
|
||||
if line.startswith(' ') or not line.strip():
|
||||
continue
|
||||
# Count added/removed lines
|
||||
if line.startswith('+') or line.startswith('-'):
|
||||
# Remove the +/- prefix and check if it's substantive
|
||||
content = line[1:].strip()
|
||||
# Skip empty lines
|
||||
if not content:
|
||||
continue
|
||||
# This is a substantive change if it's not just whitespace
|
||||
if content and not content.isspace():
|
||||
substantive_changes += 1
|
||||
|
||||
# If we have very few substantive changes and all commits are ruff-related,
|
||||
# it's likely only formatting changes
|
||||
return substantive_changes <= 2 # Allow for minimal formatting adjustments
|
||||
|
||||
except subprocess.CalledProcessError:
|
||||
logger.exception('Failed to get diff for ruff-only check')
|
||||
# If we can't get the diff, fall back to commit message analysis
|
||||
return all_ruff_commits
|
||||
|
||||
|
||||
def push_changes(
|
||||
repository_config: RepositoryConfig,
|
||||
cwd: Path,
|
||||
|
@ -480,6 +583,14 @@ def push_changes(
|
|||
logger.info('No commits made on branch %s, skipping push', branch_name)
|
||||
return IssueResolution(False)
|
||||
|
||||
# Check if the branch contains only ruff formatting changes
|
||||
if contains_only_ruff_changes(cwd, repository_config.base_branch, branch_name):
|
||||
logger.info(
|
||||
'Branch %s contains only ruff formatting changes, skipping MR creation to avoid ruff-only PRs',
|
||||
branch_name,
|
||||
)
|
||||
return IssueResolution(False)
|
||||
|
||||
# Get commit messages for PR description
|
||||
commit_messages = get_commit_messages(
|
||||
cwd,
|
||||
|
@ -725,10 +836,10 @@ def solve_issues_in_repository(
|
|||
return
|
||||
|
||||
for issue in issues:
|
||||
issue_url = issue.html_url
|
||||
issue_number = str(issue.number)
|
||||
issue_description = issue.body
|
||||
title = issue.title
|
||||
issue_url = issue.get('web_url')
|
||||
issue_number = issue.get('number')
|
||||
issue_description = issue.get('body', '')
|
||||
title = issue.get('title', f'Issue {issue_number}')
|
||||
if seen_issues_db.has_seen(issue_url):
|
||||
logger.info('Skipping already processed issue #%s: %s', issue_number, title)
|
||||
else:
|
||||
|
|
|
@ -8,8 +8,6 @@ import argparse
|
|||
import logging
|
||||
import time
|
||||
|
||||
import requests
|
||||
|
||||
from . import RepositoryConfig, secrets, solve_issues_in_repository
|
||||
from .gitea_client import GiteaClient
|
||||
from .seen_issues_db import SeenIssuesDB
|
||||
|
@ -71,8 +69,7 @@ def main():
|
|||
core.EVALUATOR_MODEL = args.evaluator_model
|
||||
|
||||
seen_issues_db = SeenIssuesDB()
|
||||
session = requests.Session()
|
||||
client = GiteaClient(session, gitea_url=args.gitea_url, token=secrets.gitea_token())
|
||||
client = GiteaClient(args.gitea_url, secrets.gitea_token())
|
||||
|
||||
if args.repo:
|
||||
repositories = [args.repo]
|
||||
|
|
|
@ -1,25 +1,10 @@
|
|||
import logging
|
||||
from collections.abc import Iterator
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
|
||||
from .models import (
|
||||
GiteaIssue,
|
||||
GiteaLabel,
|
||||
GiteaUser,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Module-level constants
|
||||
API_VERSION_PATH = '/api/v1'
|
||||
DEFAULT_CONTENT_TYPE = 'application/json'
|
||||
AIDER_LABEL_NAME = 'aider'
|
||||
SUCCESS_CONCLUSION = 'success'
|
||||
CONFLICT_STATUS_CODE = 409
|
||||
UNPROCESSABLE_ENTITY_STATUS_CODE = 422
|
||||
|
||||
|
||||
class GiteaClient:
|
||||
"""Client for interacting with the Gitea API.
|
||||
|
@ -29,191 +14,121 @@ class GiteaClient:
|
|||
|
||||
Read more about the Gitea API here: https://gitea.com/api/swagger
|
||||
|
||||
Follows the standardized client format:
|
||||
1. Constructor takes a requests.Session object
|
||||
2. All secrets are provided via keyword arguments
|
||||
3. ROOT_URL constant field for constructing URLs
|
||||
|
||||
Attributes:
|
||||
gitea_url (str): The base URL for the Gitea API endpoints.
|
||||
session (requests.Session): HTTP session for making API requests.
|
||||
ROOT_URL (str): The base URL for the Gitea API endpoints.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
session: requests.Session,
|
||||
*,
|
||||
gitea_url: str,
|
||||
token: str = '',
|
||||
) -> None:
|
||||
def __init__(self, gitea_url: str, token: str) -> None:
|
||||
"""Initialize a new Gitea API client.
|
||||
|
||||
Args:
|
||||
session: HTTP session object to use for requests.
|
||||
gitea_url: Base URL for the Gitea instance (without '/api/v1').
|
||||
token: Authentication token for the Gitea API. If empty, requests will be unauthenticated.
|
||||
gitea_url (str): Base URL for the Gitea instance (without '/api/v1').
|
||||
token (str): Authentication token for the Gitea API. If empty, requests will be unauthenticated.
|
||||
|
||||
Raises:
|
||||
AssertionError: If gitea_url ends with '/api/v1'.
|
||||
"""
|
||||
assert not gitea_url.endswith(API_VERSION_PATH)
|
||||
self.session = session
|
||||
self.ROOT_URL = gitea_url + API_VERSION_PATH
|
||||
self.session.headers['Content-Type'] = DEFAULT_CONTENT_TYPE
|
||||
assert not gitea_url.endswith('/api/v1')
|
||||
self.gitea_url = gitea_url + '/api/v1'
|
||||
self.session = requests.Session()
|
||||
self.session.headers['Content-Type'] = 'application/json'
|
||||
if token:
|
||||
self.session.headers['Authorization'] = f'token {token}'
|
||||
|
||||
def get_default_branch_sha(self, owner: str, repo: str, branch_name: str) -> str:
|
||||
def get_default_branch_sha(self, owner: str, repo: str, branch: str) -> str:
|
||||
"""Retrieve the commit SHA of the specified branch.
|
||||
|
||||
Args:
|
||||
owner: Owner of the repository.
|
||||
repo: Name of the repository.
|
||||
branch_name: Name of the branch.
|
||||
owner (str): Owner of the repository.
|
||||
repo (str): Name of the repository.
|
||||
branch (str): Name of the branch.
|
||||
|
||||
Returns:
|
||||
The commit SHA of the specified branch.
|
||||
str: The commit SHA of the specified branch.
|
||||
|
||||
Raises:
|
||||
requests.HTTPError: If the API request fails.
|
||||
"""
|
||||
api_url = f'{self.ROOT_URL}/repos/{owner}/{repo}/branches/{branch_name}'
|
||||
response = self.session.get(api_url)
|
||||
url = f'{self.gitea_url}/repos/{owner}/{repo}/branches/{branch}'
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
branch_data = response.json()
|
||||
return branch_data['commit']['sha']
|
||||
data = response.json()
|
||||
return data['commit']['sha']
|
||||
|
||||
def create_branch(
|
||||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
new_branch_name: str,
|
||||
commit_sha: str,
|
||||
) -> bool:
|
||||
def create_branch(self, owner: str, repo: str, new_branch: str, sha: str) -> bool:
|
||||
"""Create a new branch from the provided SHA.
|
||||
|
||||
Args:
|
||||
owner: Owner of the repository.
|
||||
repo: Name of the repository.
|
||||
new_branch_name: Name of the new branch to create.
|
||||
commit_sha: Commit SHA to use as the starting point for the new branch.
|
||||
owner (str): Owner of the repository.
|
||||
repo (str): Name of the repository.
|
||||
new_branch (str): Name of the new branch to create.
|
||||
sha (str): Commit SHA to use as the starting point for the new branch.
|
||||
|
||||
Returns:
|
||||
True if the branch was created successfully, False if the branch already exists.
|
||||
bool: True if the branch was created successfully, False if the branch already exists.
|
||||
|
||||
Raises:
|
||||
requests.HTTPError: If the API request fails for reasons other than branch already existing.
|
||||
"""
|
||||
api_url = f'{self.ROOT_URL}/repos/{owner}/{repo}/git/refs'
|
||||
request_payload = {'ref': f'refs/heads/{new_branch_name}', 'sha': commit_sha}
|
||||
response = self.session.post(api_url, json=request_payload)
|
||||
if response.status_code == UNPROCESSABLE_ENTITY_STATUS_CODE:
|
||||
logger.warning('Branch %s already exists.', new_branch_name)
|
||||
url = f'{self.gitea_url}/repos/{owner}/{repo}/git/refs'
|
||||
json_data = {'ref': f'refs/heads/{new_branch}', 'sha': sha}
|
||||
response = self.session.post(url, json=json_data)
|
||||
if response.status_code == 422:
|
||||
logger.warning('Branch %s already exists.', new_branch)
|
||||
return False
|
||||
response.raise_for_status()
|
||||
return True
|
||||
|
||||
def get_issues(self, owner: str, repo: str) -> list[GiteaIssue]:
|
||||
def get_issues(self, owner: str, repo: str) -> list[dict[str, str]]:
|
||||
"""Download issues from the specified repository and filter those with the 'aider' label.
|
||||
|
||||
Args:
|
||||
owner: Owner of the repository.
|
||||
repo: Name of the repository.
|
||||
owner (str): Owner of the repository.
|
||||
repo (str): Name of the repository.
|
||||
|
||||
Returns:
|
||||
A list of GiteaIssue objects, filtered to only include issues with the 'aider' label.
|
||||
list: A list of issue dictionaries, filtered to only include issues with the 'aider' label.
|
||||
|
||||
Raises:
|
||||
requests.HTTPError: If the API request fails.
|
||||
"""
|
||||
api_url = f'{self.ROOT_URL}/repos/{owner}/{repo}/issues'
|
||||
response = self.session.get(api_url)
|
||||
url = f'{self.gitea_url}/repos/{owner}/{repo}/issues'
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
issues_data = response.json()
|
||||
|
||||
issues = response.json()
|
||||
# Filter to only include issues marked with the "aider" label.
|
||||
filtered_issues = [
|
||||
issue_data
|
||||
for issue_data in issues_data
|
||||
if any(
|
||||
label_data.get('name') == AIDER_LABEL_NAME
|
||||
for label_data in issue_data.get('labels', [])
|
||||
)
|
||||
issues = [
|
||||
issue
|
||||
for issue in issues
|
||||
if any(label.get('name') == 'aider' for label in issue.get('labels', []))
|
||||
]
|
||||
|
||||
# Convert to dataclass objects
|
||||
gitea_issues = []
|
||||
for issue_data in filtered_issues:
|
||||
labels = [
|
||||
GiteaLabel(
|
||||
id=label_data['id'],
|
||||
name=label_data['name'],
|
||||
color=label_data['color'],
|
||||
description=label_data.get('description', ''),
|
||||
)
|
||||
for label_data in issue_data.get('labels', [])
|
||||
]
|
||||
|
||||
user = GiteaUser(
|
||||
login=issue_data['user']['login'],
|
||||
id=issue_data['user']['id'],
|
||||
full_name=issue_data['user'].get('full_name', ''),
|
||||
email=issue_data['user'].get('email', ''),
|
||||
avatar_url=issue_data['user'].get('avatar_url', ''),
|
||||
)
|
||||
|
||||
assignees = [
|
||||
GiteaUser(
|
||||
login=assignee_data['login'],
|
||||
id=assignee_data['id'],
|
||||
full_name=assignee_data.get('full_name', ''),
|
||||
email=assignee_data.get('email', ''),
|
||||
avatar_url=assignee_data.get('avatar_url', ''),
|
||||
)
|
||||
for assignee_data in issue_data.get('assignees', [])
|
||||
]
|
||||
|
||||
gitea_issue = GiteaIssue(
|
||||
id=issue_data['id'],
|
||||
number=issue_data['number'],
|
||||
title=issue_data['title'],
|
||||
body=issue_data.get('body', ''),
|
||||
state=issue_data['state'],
|
||||
labels=labels,
|
||||
user=user,
|
||||
assignees=assignees,
|
||||
html_url=issue_data['html_url'],
|
||||
created_at=issue_data['created_at'],
|
||||
updated_at=issue_data['updated_at'],
|
||||
)
|
||||
gitea_issues.append(gitea_issue)
|
||||
|
||||
return gitea_issues
|
||||
return issues
|
||||
|
||||
def iter_user_repositories(
|
||||
self,
|
||||
owner_name: str,
|
||||
owner: str,
|
||||
only_those_with_issues: bool = False,
|
||||
) -> Iterator[str]:
|
||||
"""Get a list of repositories for a given user.
|
||||
|
||||
Args:
|
||||
owner_name: The owner of the repositories.
|
||||
only_those_with_issues: If True, only return repositories with issues enabled.
|
||||
owner (str): The owner of the repositories.
|
||||
only_those_with_issues (bool): If True, only return repositories with issues enabled.
|
||||
|
||||
Returns:
|
||||
An iterator of repository names.
|
||||
Iterator[str]: An iterator of repository names.
|
||||
"""
|
||||
api_url = f'{self.ROOT_URL}/user/repos'
|
||||
response = self.session.get(api_url)
|
||||
url = f'{self.gitea_url}/user/repos'
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
|
||||
for repository_data in response.json():
|
||||
if only_those_with_issues and not repository_data['has_issues']:
|
||||
for repo in response.json():
|
||||
if only_those_with_issues and not repo['has_issues']:
|
||||
continue
|
||||
if repository_data['owner']['login'].lower() != owner_name.lower():
|
||||
if repo['owner']['login'].lower() != owner.lower():
|
||||
continue
|
||||
yield repository_data['name']
|
||||
yield repo['name']
|
||||
|
||||
def create_pull_request(
|
||||
self,
|
||||
|
@ -242,61 +157,54 @@ class GiteaClient:
|
|||
Raises:
|
||||
requests.HTTPError: If the API request fails.
|
||||
"""
|
||||
api_url = f'{self.ROOT_URL}/repos/{owner}/{repo}/pulls'
|
||||
request_payload = {
|
||||
url = f'{self.gitea_url}/repos/{owner}/{repo}/pulls'
|
||||
json_data = {
|
||||
'title': title,
|
||||
'body': body,
|
||||
'head': head,
|
||||
'base': base,
|
||||
}
|
||||
|
||||
response = self.session.post(api_url, json=request_payload)
|
||||
response = self.session.post(url, json=json_data)
|
||||
# If a pull request for this head/base already exists, return it instead of crashing
|
||||
if response.status_code == CONFLICT_STATUS_CODE:
|
||||
if response.status_code == 409:
|
||||
logger.warning(
|
||||
'Pull request already exists for head %s and base %s',
|
||||
head,
|
||||
base,
|
||||
)
|
||||
existing_pull_requests = self.get_pull_requests(owner, repo)
|
||||
for existing_pr in existing_pull_requests:
|
||||
prs = self.get_pull_requests(owner, repo)
|
||||
for pr in prs:
|
||||
if (
|
||||
existing_pr.get('head', {}).get('ref') == head
|
||||
and existing_pr.get('base', {}).get('ref') == base
|
||||
pr.get('head', {}).get('ref') == head
|
||||
and pr.get('base', {}).get('ref') == base
|
||||
):
|
||||
return existing_pr
|
||||
return pr
|
||||
# fallback to raise if we can’t find it
|
||||
response.raise_for_status()
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def get_failed_pipelines(
|
||||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
pull_request_number: str,
|
||||
) -> list[int]:
|
||||
def get_failed_pipelines(self, owner: str, repo: str, pr_number: str) -> list[int]:
|
||||
"""Fetch pipeline runs for a PR and return IDs of failed runs."""
|
||||
api_url = f'{self.ROOT_URL}/repos/{owner}/{repo}/actions/runs'
|
||||
response = self.session.get(api_url)
|
||||
url = f'{self.gitea_url}/repos/{owner}/{repo}/actions/runs'
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
workflow_runs = response.json().get('workflow_runs', [])
|
||||
failed_run_ids = []
|
||||
for workflow_run in workflow_runs:
|
||||
runs = response.json().get('workflow_runs', [])
|
||||
failed = []
|
||||
for run in runs:
|
||||
if any(
|
||||
pull_request.get('number') == int(pull_request_number)
|
||||
for pull_request in workflow_run.get('pull_requests', [])
|
||||
pr.get('number') == int(pr_number)
|
||||
for pr in run.get('pull_requests', [])
|
||||
):
|
||||
if workflow_run.get('conclusion') != SUCCESS_CONCLUSION:
|
||||
failed_run_ids.append(workflow_run.get('id'))
|
||||
return failed_run_ids
|
||||
if run.get('conclusion') not in ('success',):
|
||||
failed.append(run.get('id'))
|
||||
return failed
|
||||
|
||||
def get_pipeline_log(self, owner: str, repo: str, workflow_run_id: int) -> str:
|
||||
def get_pipeline_log(self, owner: str, repo: str, run_id: int) -> str:
|
||||
"""Download the logs for a pipeline run."""
|
||||
api_url = (
|
||||
f'{self.ROOT_URL}/repos/{owner}/{repo}/actions/runs/{workflow_run_id}/logs'
|
||||
)
|
||||
response = self.session.get(api_url)
|
||||
url = f'{self.gitea_url}/repos/{owner}/{repo}/actions/runs/{run_id}/logs'
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
return response.text
|
||||
|
||||
|
@ -304,12 +212,10 @@ class GiteaClient:
|
|||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
pull_request_state: str = 'open',
|
||||
) -> list[dict[str, Any]]:
|
||||
state: str = 'open',
|
||||
) -> list[dict]:
|
||||
"""Fetch pull requests for a repository."""
|
||||
api_url = (
|
||||
f'{self.ROOT_URL}/repos/{owner}/{repo}/pulls?state={pull_request_state}'
|
||||
)
|
||||
response = self.session.get(api_url)
|
||||
url = f'{self.gitea_url}/repos/{owner}/{repo}/pulls?state={state}'
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
|
|
@ -1,111 +0,0 @@
|
|||
"""Data models for Gitea API responses.
|
||||
|
||||
This module contains dataclasses that represent the structure of data
|
||||
returned from the Gitea API, providing type safety and better code organization.
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class GiteaUser:
|
||||
"""Represents a Gitea user."""
|
||||
|
||||
login: str # User's login name
|
||||
id: int # User's ID
|
||||
full_name: str # User's full name
|
||||
email: str # User's email address
|
||||
avatar_url: str # URL to user's avatar image
|
||||
|
||||
|
||||
@dataclass
|
||||
class GiteaLabel:
|
||||
"""Represents a label in Gitea."""
|
||||
|
||||
id: int # Label ID
|
||||
name: str # Label name
|
||||
color: str # Label color (hex code)
|
||||
description: str # Label description
|
||||
|
||||
|
||||
@dataclass
|
||||
class GiteaCommit:
|
||||
"""Represents a commit in Gitea."""
|
||||
|
||||
sha: str # Commit SHA hash
|
||||
url: str # API URL for the commit
|
||||
message: str # Commit message
|
||||
author: GiteaUser # Commit author
|
||||
|
||||
|
||||
@dataclass
|
||||
class GiteaBranch:
|
||||
"""Represents a branch in Gitea."""
|
||||
|
||||
name: str # Branch name
|
||||
commit: GiteaCommit # Latest commit on the branch
|
||||
protected: bool # Whether the branch is protected
|
||||
|
||||
|
||||
@dataclass
|
||||
class GiteaRepository:
|
||||
"""Represents a repository in Gitea."""
|
||||
|
||||
id: int # Repository ID
|
||||
name: str # Repository name
|
||||
full_name: str # Full repository name (owner/repo)
|
||||
owner: GiteaUser # Repository owner
|
||||
description: str # Repository description
|
||||
clone_url: str # URL for cloning the repository
|
||||
has_issues: bool # Whether issues are enabled
|
||||
default_branch: str # Default branch name
|
||||
|
||||
|
||||
@dataclass
|
||||
class GiteaIssue:
|
||||
"""Represents an issue in Gitea."""
|
||||
|
||||
id: int # Issue ID
|
||||
number: int # Issue number
|
||||
title: str # Issue title
|
||||
body: str # Issue description/body
|
||||
state: str # Issue state (open, closed, etc.)
|
||||
labels: list[GiteaLabel] # List of labels attached to the issue
|
||||
user: GiteaUser # User who created the issue
|
||||
assignees: list[GiteaUser] # List of assigned users
|
||||
html_url: str # Web URL for the issue
|
||||
created_at: str # ISO datetime when issue was created
|
||||
updated_at: str # ISO datetime when issue was last updated
|
||||
|
||||
|
||||
@dataclass
|
||||
class GiteaPullRequest:
|
||||
"""Represents a pull request in Gitea."""
|
||||
|
||||
id: int # Pull request ID
|
||||
number: int # Pull request number
|
||||
title: str # Pull request title
|
||||
body: str # Pull request description
|
||||
state: str # Pull request state (open, closed, merged)
|
||||
user: GiteaUser # User who created the pull request
|
||||
head: dict[str, Any] # Head branch information
|
||||
base: dict[str, Any] # Base branch information
|
||||
html_url: str # Web URL for the pull request
|
||||
created_at: str # ISO datetime when PR was created
|
||||
updated_at: str # ISO datetime when PR was last updated
|
||||
merged_at: str | None # ISO datetime when PR was merged (if applicable)
|
||||
|
||||
|
||||
@dataclass
|
||||
class GiteaWorkflowRun:
|
||||
"""Represents a GitHub Actions workflow run in Gitea."""
|
||||
|
||||
id: int # Workflow run ID
|
||||
name: str # Workflow name
|
||||
status: str # Run status (queued, in_progress, completed)
|
||||
conclusion: str | None # Run conclusion (success, failure, cancelled, etc.)
|
||||
workflow_id: int # ID of the workflow
|
||||
pull_requests: list[dict[str, Any]] # Associated pull requests
|
||||
created_at: str # ISO datetime when run was created
|
||||
updated_at: str # ISO datetime when run was last updated
|
189
test/test_contains_only_ruff_changes.py
Normal file
189
test/test_contains_only_ruff_changes.py
Normal file
|
@ -0,0 +1,189 @@
|
|||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from aider_gitea import contains_only_ruff_changes
|
||||
|
||||
|
||||
class TestContainsOnlyRuffChanges:
|
||||
"""Test the contains_only_ruff_changes function."""
|
||||
|
||||
def setup_test_repo(self) -> Path:
|
||||
"""Create a test git repository."""
|
||||
temp_dir = Path(tempfile.mkdtemp())
|
||||
|
||||
# Initialize git repo
|
||||
subprocess.run(['git', 'init'], cwd=temp_dir, check=True)
|
||||
subprocess.run(
|
||||
['git', 'config', 'user.name', 'Test User'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
subprocess.run(
|
||||
['git', 'config', 'user.email', 'test@example.com'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
# Create initial content and commit
|
||||
test_file = temp_dir / 'test.py'
|
||||
test_file.write_text('def hello():\n print("hello")\n')
|
||||
subprocess.run(['git', 'add', '.'], cwd=temp_dir, check=True)
|
||||
subprocess.run(
|
||||
['git', 'commit', '-m', 'Initial commit'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
# Rename master to main for consistency
|
||||
subprocess.run(
|
||||
['git', 'branch', '-m', 'master', 'main'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
return temp_dir
|
||||
|
||||
def test_no_commits_returns_true(self):
|
||||
"""Test that branches with no commits return True."""
|
||||
temp_dir = self.setup_test_repo()
|
||||
|
||||
# Create a new branch but don't commit anything
|
||||
subprocess.run(
|
||||
['git', 'checkout', '-b', 'test-branch'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
result = contains_only_ruff_changes(temp_dir, 'main', 'test-branch')
|
||||
assert result is True
|
||||
|
||||
def test_only_ruff_commits_returns_true(self):
|
||||
"""Test that branches with only ruff-related commits return True."""
|
||||
temp_dir = self.setup_test_repo()
|
||||
|
||||
# Create a new branch and make ruff-related commits
|
||||
subprocess.run(
|
||||
['git', 'checkout', '-b', 'test-branch'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
# Modify the file with formatting changes
|
||||
test_file = temp_dir / 'test.py'
|
||||
test_file.write_text('def hello():\n print("hello")\n\n') # Added newline
|
||||
subprocess.run(['git', 'add', '.'], cwd=temp_dir, check=True)
|
||||
subprocess.run(
|
||||
['git', 'commit', '-m', 'ruff format changes'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
# Another ruff commit
|
||||
test_file.write_text(
|
||||
'def hello():\n print("hello")\n',
|
||||
) # Removed newline again
|
||||
subprocess.run(['git', 'add', '.'], cwd=temp_dir, check=True)
|
||||
subprocess.run(
|
||||
['git', 'commit', '-m', 'auto-fix lint issues'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
result = contains_only_ruff_changes(temp_dir, 'main', 'test-branch')
|
||||
assert result is True
|
||||
|
||||
def test_substantive_changes_returns_false(self):
|
||||
"""Test that branches with substantive changes return False."""
|
||||
temp_dir = self.setup_test_repo()
|
||||
|
||||
# Create a new branch and make substantive changes
|
||||
subprocess.run(
|
||||
['git', 'checkout', '-b', 'test-branch'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
# Add a new function (substantive change)
|
||||
test_file = temp_dir / 'test.py'
|
||||
test_file.write_text(
|
||||
'def hello():\n print("hello")\n\ndef goodbye():\n print("goodbye")\n',
|
||||
)
|
||||
subprocess.run(['git', 'add', '.'], cwd=temp_dir, check=True)
|
||||
subprocess.run(
|
||||
['git', 'commit', '-m', 'Add goodbye function'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
result = contains_only_ruff_changes(temp_dir, 'main', 'test-branch')
|
||||
assert result is False
|
||||
|
||||
def test_mixed_commits_returns_false(self):
|
||||
"""Test that branches with both ruff and substantive commits return False."""
|
||||
temp_dir = self.setup_test_repo()
|
||||
|
||||
# Create a new branch
|
||||
subprocess.run(
|
||||
['git', 'checkout', '-b', 'test-branch'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
# First, a substantive change
|
||||
test_file = temp_dir / 'test.py'
|
||||
test_file.write_text(
|
||||
'def hello():\n print("hello world")\n',
|
||||
) # Changed string
|
||||
subprocess.run(['git', 'add', '.'], cwd=temp_dir, check=True)
|
||||
subprocess.run(
|
||||
['git', 'commit', '-m', 'Update greeting message'],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
# Then, a ruff change
|
||||
test_file.write_text(
|
||||
'def hello():\n print("hello world")\n\n',
|
||||
) # Added newline
|
||||
subprocess.run(['git', 'add', '.'], cwd=temp_dir, check=True)
|
||||
subprocess.run(['git', 'commit', '-m', 'ruff format'], cwd=temp_dir, check=True)
|
||||
|
||||
result = contains_only_ruff_changes(temp_dir, 'main', 'test-branch')
|
||||
assert result is False
|
||||
|
||||
def test_ruff_keywords_in_commit_messages(self):
|
||||
"""Test various ruff-related keywords in commit messages."""
|
||||
temp_dir = self.setup_test_repo()
|
||||
|
||||
ruff_messages = [
|
||||
'ruff format',
|
||||
'Ruff after aider',
|
||||
'auto-fix lint issues',
|
||||
'code style formatting',
|
||||
'Apply formatting changes',
|
||||
'Lint fixes',
|
||||
]
|
||||
|
||||
for i, message in enumerate(ruff_messages):
|
||||
# Create a branch for each test
|
||||
branch_name = f'test-branch-{i}'
|
||||
subprocess.run(['git', 'checkout', 'main'], cwd=temp_dir, check=True)
|
||||
subprocess.run(
|
||||
['git', 'checkout', '-b', branch_name],
|
||||
cwd=temp_dir,
|
||||
check=True,
|
||||
)
|
||||
|
||||
# Make a minor formatting change that's always different
|
||||
test_file = temp_dir / 'test.py'
|
||||
content = test_file.read_text()
|
||||
# Add a unique comment to make each change different
|
||||
content += f'# formatting change {i}\n'
|
||||
test_file.write_text(content)
|
||||
|
||||
subprocess.run(['git', 'add', '.'], cwd=temp_dir, check=True)
|
||||
subprocess.run(['git', 'commit', '-m', message], cwd=temp_dir, check=True)
|
||||
|
||||
result = contains_only_ruff_changes(temp_dir, 'main', branch_name)
|
||||
assert result is True, f'Failed for message: {message}'
|
|
@ -1,18 +1,11 @@
|
|||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import requests
|
||||
|
||||
from aider_gitea.gitea_client import GiteaClient
|
||||
|
||||
|
||||
class TestGiteaClientPRLabels:
|
||||
def setup_method(self):
|
||||
session = requests.Session()
|
||||
self.client = GiteaClient(
|
||||
session,
|
||||
gitea_url='https://gitea.example.com',
|
||||
token='fake_token',
|
||||
)
|
||||
self.client = GiteaClient('https://gitea.example.com', 'fake_token')
|
||||
|
||||
@patch('requests.Session.post')
|
||||
def test_create_pull_request_with_labels(self, mock_post):
|
||||
|
|
Loading…
Reference in New Issue
Block a user