92 lines
3.2 KiB
Python
Executable File
92 lines
3.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Create an idempotent Gitea pull request or conflict issue."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
BASE_URL = os.environ['GITEA_URL'].rstrip('/') + '/api/v1'
|
|
REPOSITORY = os.environ['CUSTOM_REPOSITORY']
|
|
TOKEN = os.environ['GITEA_TOKEN']
|
|
TARGET = os.environ['INTEGRATION_BRANCH']
|
|
HEAD = os.environ['UPDATE_BRANCH']
|
|
|
|
|
|
def request(method: str, path: str, payload: dict | None = None):
|
|
data = json.dumps(payload).encode() if payload is not None else None
|
|
req = urllib.request.Request(
|
|
f'{BASE_URL}{path}', data=data, method=method,
|
|
headers={
|
|
'Authorization': f'token {TOKEN}',
|
|
'Accept': 'application/json',
|
|
'Content-Type': 'application/json',
|
|
})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as response:
|
|
return json.load(response)
|
|
except urllib.error.HTTPError as exc:
|
|
detail = exc.read().decode(errors='replace')
|
|
raise RuntimeError(f'Gitea API returned {exc.code}: {detail}') from exc
|
|
except urllib.error.URLError as exc:
|
|
raise RuntimeError(f'Could not reach the Gitea API: {exc.reason}') from exc
|
|
|
|
|
|
def open_pulls():
|
|
query = urllib.parse.urlencode({'state': 'open', 'limit': 50})
|
|
return request('GET', f'/repos/{REPOSITORY}/pulls?{query}')
|
|
|
|
|
|
def ensure_pull_request(upstream_sha: str) -> None:
|
|
for pull in open_pulls():
|
|
if pull['head']['ref'] == HEAD and pull['base']['ref'] == TARGET:
|
|
print(f"Using existing pull request: {pull['html_url']}")
|
|
return
|
|
pull = request('POST', f'/repos/{REPOSITORY}/pulls', {
|
|
'title': f'Upstream Tryton 8.0 update ({upstream_sha[:12]})',
|
|
'head': HEAD,
|
|
'base': TARGET,
|
|
'body': (
|
|
'Automated proposal from the private official mirror.\n\n'
|
|
'Review the upstream changes and wait for all required checks. '
|
|
'This pull request must be merged manually.'),
|
|
})
|
|
print(f"Created pull request: {pull['html_url']}")
|
|
|
|
|
|
def ensure_conflict_issue(conflicts: str) -> None:
|
|
title = 'Upstream Tryton 8.0 synchronization conflict'
|
|
query = urllib.parse.urlencode({'state': 'open', 'type': 'issues', 'limit': 50})
|
|
for issue in request('GET', f'/repos/{REPOSITORY}/issues?{query}'):
|
|
if issue['title'] == title:
|
|
print(f"Conflict issue already open: {issue['html_url']}")
|
|
return
|
|
issue = request('POST', f'/repos/{REPOSITORY}/issues', {
|
|
'title': title,
|
|
'body': (
|
|
'The scheduled synchronization could not merge the latest '
|
|
'official 8.0 changes. No branch was overwritten.\n\n'
|
|
f'Conflicting paths:\n```\n{conflicts or "unknown"}\n```'),
|
|
})
|
|
print(f"Created conflict issue: {issue['html_url']}")
|
|
|
|
|
|
def main() -> None:
|
|
if len(sys.argv) != 3:
|
|
raise SystemExit('usage: upstream_pr.py pull-request SHA | conflict PATHS')
|
|
if sys.argv[1] == 'pull-request':
|
|
ensure_pull_request(sys.argv[2])
|
|
elif sys.argv[1] == 'conflict':
|
|
ensure_conflict_issue(sys.argv[2])
|
|
else:
|
|
raise SystemExit('usage: upstream_pr.py pull-request SHA | conflict PATHS')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|