mirror of
https://github.com/qodo-ai/pr-agent.git
synced 2025-07-02 20:00:41 +08:00
Merge pull request #1787 from nicholasgribanov/feature/gitea-forgejo-support
#1657 add gitea/forgejo support
This commit is contained in:
Tal
185
tests/e2e_tests/test_gitea_app.py
Normal file
185
tests/e2e_tests/test_gitea_app.py
Normal file
@ -0,0 +1,185 @@
|
||||
import os
|
||||
import time
|
||||
import requests
|
||||
from datetime import datetime
|
||||
|
||||
from pr_agent.config_loader import get_settings
|
||||
from pr_agent.log import get_logger, setup_logger
|
||||
from tests.e2e_tests.e2e_utils import (FILE_PATH,
|
||||
IMPROVE_START_WITH_REGEX_PATTERN,
|
||||
NEW_FILE_CONTENT, NUM_MINUTES,
|
||||
PR_HEADER_START_WITH, REVIEW_START_WITH)
|
||||
|
||||
log_level = os.environ.get("LOG_LEVEL", "INFO")
|
||||
setup_logger(log_level)
|
||||
logger = get_logger()
|
||||
|
||||
def test_e2e_run_gitea_app():
|
||||
repo_name = 'pr-agent-tests'
|
||||
owner = 'codiumai'
|
||||
base_branch = "main"
|
||||
new_branch = f"gitea_app_e2e_test-{datetime.now().strftime('%Y-%m-%d-%H-%M-%S')}"
|
||||
get_settings().config.git_provider = "gitea"
|
||||
|
||||
headers = None
|
||||
pr_number = None
|
||||
|
||||
try:
|
||||
gitea_url = get_settings().get("GITEA.URL", None)
|
||||
gitea_token = get_settings().get("GITEA.TOKEN", None)
|
||||
|
||||
if not gitea_url:
|
||||
logger.error("GITEA.URL is not set in the configuration")
|
||||
logger.info("Please set GITEA.URL in .env file or environment variables")
|
||||
assert False, "GITEA.URL is not set in the configuration"
|
||||
|
||||
if not gitea_token:
|
||||
logger.error("GITEA.TOKEN is not set in the configuration")
|
||||
logger.info("Please set GITEA.TOKEN in .env file or environment variables")
|
||||
assert False, "GITEA.TOKEN is not set in the configuration"
|
||||
|
||||
headers = {
|
||||
'Authorization': f'token {gitea_token}',
|
||||
'Content-Type': 'application/json',
|
||||
'Accept': 'application/json'
|
||||
}
|
||||
|
||||
logger.info(f"Creating a new branch {new_branch} from {base_branch}")
|
||||
|
||||
response = requests.get(
|
||||
f"{gitea_url}/api/v1/repos/{owner}/{repo_name}/branches/{base_branch}",
|
||||
headers=headers
|
||||
)
|
||||
response.raise_for_status()
|
||||
base_branch_data = response.json()
|
||||
base_commit_sha = base_branch_data['commit']['id']
|
||||
|
||||
branch_data = {
|
||||
'ref': f"refs/heads/{new_branch}",
|
||||
'sha': base_commit_sha
|
||||
}
|
||||
response = requests.post(
|
||||
f"{gitea_url}/api/v1/repos/{owner}/{repo_name}/git/refs",
|
||||
headers=headers,
|
||||
json=branch_data
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
logger.info(f"Updating file {FILE_PATH} in branch {new_branch}")
|
||||
|
||||
import base64
|
||||
file_content_encoded = base64.b64encode(NEW_FILE_CONTENT.encode()).decode()
|
||||
|
||||
try:
|
||||
response = requests.get(
|
||||
f"{gitea_url}/api/v1/repos/{owner}/{repo_name}/contents/{FILE_PATH}?ref={new_branch}",
|
||||
headers=headers
|
||||
)
|
||||
response.raise_for_status()
|
||||
existing_file = response.json()
|
||||
file_sha = existing_file.get('sha')
|
||||
|
||||
file_data = {
|
||||
'message': 'Update cli_pip.py',
|
||||
'content': file_content_encoded,
|
||||
'sha': file_sha,
|
||||
'branch': new_branch
|
||||
}
|
||||
except:
|
||||
file_data = {
|
||||
'message': 'Add cli_pip.py',
|
||||
'content': file_content_encoded,
|
||||
'branch': new_branch
|
||||
}
|
||||
|
||||
response = requests.put(
|
||||
f"{gitea_url}/api/v1/repos/{owner}/{repo_name}/contents/{FILE_PATH}",
|
||||
headers=headers,
|
||||
json=file_data
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
logger.info(f"Creating a pull request from {new_branch} to {base_branch}")
|
||||
pr_data = {
|
||||
'title': f'Test PR from {new_branch}',
|
||||
'body': 'update cli_pip.py',
|
||||
'head': new_branch,
|
||||
'base': base_branch
|
||||
}
|
||||
response = requests.post(
|
||||
f"{gitea_url}/api/v1/repos/{owner}/{repo_name}/pulls",
|
||||
headers=headers,
|
||||
json=pr_data
|
||||
)
|
||||
response.raise_for_status()
|
||||
pr = response.json()
|
||||
pr_number = pr['number']
|
||||
|
||||
for i in range(NUM_MINUTES):
|
||||
logger.info(f"Waiting for the PR to get all the tool results...")
|
||||
time.sleep(60)
|
||||
|
||||
response = requests.get(
|
||||
f"{gitea_url}/api/v1/repos/{owner}/{repo_name}/issues/{pr_number}/comments",
|
||||
headers=headers
|
||||
)
|
||||
response.raise_for_status()
|
||||
comments = response.json()
|
||||
|
||||
if len(comments) >= 5:
|
||||
valid_review = False
|
||||
for comment in comments:
|
||||
if comment['body'].startswith('## PR Reviewer Guide 🔍'):
|
||||
valid_review = True
|
||||
break
|
||||
if valid_review:
|
||||
break
|
||||
else:
|
||||
logger.error("REVIEW feedback is invalid")
|
||||
raise Exception("REVIEW feedback is invalid")
|
||||
else:
|
||||
logger.info(f"Waiting for the PR to get all the tool results. {i + 1} minute(s) passed")
|
||||
else:
|
||||
assert False, f"After {NUM_MINUTES} minutes, the PR did not get all the tool results"
|
||||
|
||||
logger.info(f"Cleaning up: closing PR and deleting branch {new_branch}")
|
||||
|
||||
close_data = {'state': 'closed'}
|
||||
response = requests.patch(
|
||||
f"{gitea_url}/api/v1/repos/{owner}/{repo_name}/pulls/{pr_number}",
|
||||
headers=headers,
|
||||
json=close_data
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
response = requests.delete(
|
||||
f"{gitea_url}/api/v1/repos/{owner}/{repo_name}/git/refs/heads/{new_branch}",
|
||||
headers=headers
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
logger.info(f"Succeeded in running e2e test for Gitea app on the PR")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to run e2e test for Gitea app: {e}")
|
||||
raise
|
||||
finally:
|
||||
try:
|
||||
if headers is None or gitea_url is None:
|
||||
return
|
||||
|
||||
if pr_number is not None:
|
||||
requests.patch(
|
||||
f"{gitea_url}/api/v1/repos/{owner}/{repo_name}/pulls/{pr_number}",
|
||||
headers=headers,
|
||||
json={'state': 'closed'}
|
||||
)
|
||||
|
||||
requests.delete(
|
||||
f"{gitea_url}/api/v1/repos/{owner}/{repo_name}/git/refs/heads/{new_branch}",
|
||||
headers=headers
|
||||
)
|
||||
except Exception as cleanup_error:
|
||||
logger.error(f"Failed to clean up after test: {cleanup_error}")
|
||||
|
||||
if __name__ == '__main__':
|
||||
test_e2e_run_gitea_app()
|
||||
126
tests/unittest/test_gitea_provider.py
Normal file
126
tests/unittest/test_gitea_provider.py
Normal file
@ -0,0 +1,126 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from pr_agent.algo.types import EDIT_TYPE
|
||||
from pr_agent.git_providers.gitea_provider import GiteaProvider
|
||||
|
||||
|
||||
class TestGiteaProvider:
|
||||
"""Unit-tests for GiteaProvider following project style (explicit object construction, minimal patching)."""
|
||||
|
||||
def _provider(self):
|
||||
"""Create provider instance with patched settings and avoid real HTTP calls."""
|
||||
with patch('pr_agent.git_providers.gitea_provider.get_settings') as mock_get_settings, \
|
||||
patch('requests.get') as mock_get:
|
||||
settings = MagicMock()
|
||||
settings.get.side_effect = lambda k, d=None: {
|
||||
'GITEA.URL': 'https://gitea.example.com',
|
||||
'GITEA.TOKEN': 'test-token'
|
||||
}.get(k, d)
|
||||
mock_get_settings.return_value = settings
|
||||
# Stub the PR fetch triggered during provider initialization
|
||||
pr_resp = MagicMock()
|
||||
pr_resp.json.return_value = {
|
||||
'title': 'stub',
|
||||
'body': 'stub',
|
||||
'head': {'ref': 'main'},
|
||||
'user': {'id': 1}
|
||||
}
|
||||
pr_resp.raise_for_status = MagicMock()
|
||||
mock_get.return_value = pr_resp
|
||||
return GiteaProvider('https://gitea.example.com/owner/repo/pulls/123')
|
||||
|
||||
# ---------------- URL parsing ----------------
|
||||
def test_parse_pr_url_valid(self):
|
||||
owner, repo, pr_num = GiteaProvider._parse_pr_url('https://gitea.example.com/owner/repo/pulls/123')
|
||||
assert (owner, repo, pr_num) == ('owner', 'repo', '123')
|
||||
|
||||
def test_parse_pr_url_invalid(self):
|
||||
with pytest.raises(ValueError):
|
||||
GiteaProvider._parse_pr_url('https://gitea.example.com/owner/repo')
|
||||
|
||||
# ---------------- simple getters ----------------
|
||||
def test_get_files(self):
|
||||
provider = self._provider()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = [{'filename': 'a.txt'}, {'filename': 'b.txt'}]
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
with patch('requests.get', return_value=mock_resp) as mock_get:
|
||||
assert provider.get_files() == ['a.txt', 'b.txt']
|
||||
mock_get.assert_called_once()
|
||||
|
||||
def test_get_diff_files(self):
|
||||
provider = self._provider()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = [
|
||||
{'filename': 'f1', 'previous_filename': 'old_f1', 'status': 'renamed', 'patch': ''},
|
||||
{'filename': 'f2', 'status': 'added', 'patch': ''},
|
||||
{'filename': 'f3', 'status': 'deleted', 'patch': ''},
|
||||
{'filename': 'f4', 'status': 'modified', 'patch': ''}
|
||||
]
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
with patch('requests.get', return_value=mock_resp):
|
||||
res = provider.get_diff_files()
|
||||
assert [f.edit_type for f in res] == [EDIT_TYPE.RENAMED, EDIT_TYPE.ADDED, EDIT_TYPE.DELETED,
|
||||
EDIT_TYPE.MODIFIED]
|
||||
|
||||
# ---------------- publishing methods ----------------
|
||||
def test_publish_description(self):
|
||||
provider = self._provider()
|
||||
mock_resp = MagicMock();
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
with patch('requests.patch', return_value=mock_resp) as mock_patch:
|
||||
provider.publish_description('t', 'b');
|
||||
mock_patch.assert_called_once()
|
||||
|
||||
def test_publish_comment(self):
|
||||
provider = self._provider()
|
||||
mock_resp = MagicMock();
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
with patch('requests.post', return_value=mock_resp) as mock_post:
|
||||
provider.publish_comment('c');
|
||||
mock_post.assert_called_once()
|
||||
|
||||
def test_publish_inline_comment(self):
|
||||
provider = self._provider()
|
||||
mock_resp = MagicMock();
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
with patch('requests.post', return_value=mock_resp) as mock_post:
|
||||
provider.publish_inline_comment('body', 'file', '10');
|
||||
mock_post.assert_called_once()
|
||||
|
||||
# ---------------- labels & reactions ----------------
|
||||
def test_get_pr_labels(self):
|
||||
provider = self._provider()
|
||||
mock_resp = MagicMock();
|
||||
mock_resp.raise_for_status = MagicMock();
|
||||
mock_resp.json.return_value = [{'name': 'l1'}]
|
||||
with patch('requests.get', return_value=mock_resp):
|
||||
assert provider.get_pr_labels() == ['l1']
|
||||
|
||||
def test_add_eyes_reaction(self):
|
||||
provider = self._provider()
|
||||
mock_resp = MagicMock();
|
||||
mock_resp.raise_for_status = MagicMock();
|
||||
mock_resp.json.return_value = {'id': 7}
|
||||
with patch('requests.post', return_value=mock_resp):
|
||||
assert provider.add_eyes_reaction(1) == 7
|
||||
|
||||
# ---------------- commit messages & url helpers ----------------
|
||||
def test_get_commit_messages(self):
|
||||
provider = self._provider()
|
||||
mock_resp = MagicMock();
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
mock_resp.json.return_value = [
|
||||
{'commit': {'message': 'm1'}}, {'commit': {'message': 'm2'}}]
|
||||
with patch('requests.get', return_value=mock_resp):
|
||||
assert provider.get_commit_messages() == ['m1', 'm2']
|
||||
|
||||
def test_git_url_helpers(self):
|
||||
provider = self._provider()
|
||||
issues_url = 'https://gitea.example.com/owner/repo/pulls/3'
|
||||
assert provider.get_git_repo_url(issues_url) == 'https://gitea.example.com/owner/repo.git'
|
||||
prefix, suffix = provider.get_canonical_url_parts('https://gitea.example.com/owner/repo.git', 'dev')
|
||||
assert prefix == 'https://gitea.example.com/owner/repo/src/branch/dev'
|
||||
assert suffix == ''
|
||||
Reference in New Issue
Block a user