# coding: utf-8
import ast
import base64
import collections
import contextlib
import datetime
import io
import itertools
import json
import logging
import os
import pprint
import re
import time
from itertools import takewhile
import requests
import werkzeug
from werkzeug.datastructures import Headers
from odoo import api, fields, models, tools
from odoo.exceptions import ValidationError
from odoo.osv import expression
from odoo.tools import OrderedSet
from .. import github, exceptions, controllers, utils
WAIT_FOR_VISIBILITY = [10, 10, 10, 10]
_logger = logging.getLogger(__name__)
class StatusConfiguration(models.Model):
_name = 'runbot_merge.repository.status'
_description = "required statuses on repositories"
_rec_name = 'context'
_log_access = False
context = fields.Char(required=True)
repo_id = fields.Many2one('runbot_merge.repository', required=True, ondelete='cascade')
branch_filter = fields.Char(help="branches this status applies to")
prs = fields.Boolean(string="Applies to pull requests", default=True)
stagings = fields.Boolean(string="Applies to stagings", default=True)
def _for_branch(self, branch):
assert branch._name == 'runbot_merge.branch', \
f'Expected branch, got {branch}'
return self.filtered(lambda st: (
not st.branch_filter
or branch.filtered_domain(ast.literal_eval(st.branch_filter))
))
def _for_pr(self, pr):
assert pr._name == 'runbot_merge.pull_requests', \
f'Expected pull request, got {pr}'
return self._for_branch(pr.target).filtered('prs')
def _for_staging(self, staging):
assert staging._name == 'runbot_merge.stagings', \
f'Expected staging, got {staging}'
return self._for_branch(staging.target).filtered('stagings')
class Repository(models.Model):
_name = _description = 'runbot_merge.repository'
_order = 'sequence, id'
sequence = fields.Integer(default=50, group_operator=None)
name = fields.Char(required=True)
project_id = fields.Many2one('runbot_merge.project', required=True)
status_ids = fields.One2many('runbot_merge.repository.status', 'repo_id', string="Required Statuses")
group_id = fields.Many2one('res.groups', default=lambda self: self.env.ref('base.group_user'))
branch_filter = fields.Char(default='[(1, "=", 1)]', help="Filter branches valid for this repository")
substitutions = fields.Text(
"label substitutions",
help="""sed-style substitution patterns applied to the label on input, one per line.
All substitutions are tentatively applied sequentially to the input.
""")
@api.model
def create(self, vals):
if 'status_ids' in vals:
return super().create(vals)
st = vals.pop('required_statuses', 'legal/cla,ci/runbot')
if st:
vals['status_ids'] = [(0, 0, {'context': c}) for c in st.split(',')]
return super().create(vals)
def write(self, vals):
st = vals.pop('required_statuses', None)
if st:
vals['status_ids'] = [(5, 0, {})] + [(0, 0, {'context': c}) for c in st.split(',')]
return super().write(vals)
def github(self, token_field='github_token'):
return github.GH(self.project_id[token_field], self.name)
def _auto_init(self):
res = super(Repository, self)._auto_init()
tools.create_unique_index(
self._cr, 'runbot_merge_unique_repo', self._table, ['name'])
return res
def _load_pr(self, number):
gh = self.github()
# fetch PR object and handle as *opened*
issue, pr = gh.pr(number)
feedback = self.env['runbot_merge.pull_requests.feedback'].create
if not self.project_id._has_branch(pr['base']['ref']):
_logger.info("Tasked with loading PR %d for un-managed branch %s:%s, ignoring",
number, self.name, pr['base']['ref'])
feedback({
'repository': self.id,
'pull_request': number,
'message': "Branch `{}` is not within my remit, imma just ignore it.".format(pr['base']['ref']),
})
return
# if the PR is already loaded, check... if the heads match?
pr_id = self.env['runbot_merge.pull_requests'].search([
('repository.name', '=', pr['base']['repo']['full_name']),
('number', '=', number),
])
if pr_id:
# TODO: edited, maybe (requires crafting a 'changes' object)
r = controllers.handle_pr(self.env, {
'action': 'synchronize',
'pull_request': pr,
'sender': {'login': self.project_id.github_prefix}
})
feedback({
'repository': pr_id.repository.id,
'pull_request': number,
'message': r,
})
return
feedback({
'repository': self.id,
'pull_request': number,
'message': "%sI didn't know about this PR and had to retrieve "
"its information, you may have to re-approve it as "
"I didn't see previous commands." % pr_id.ping()
})
sender = {'login': self.project_id.github_prefix}
# init the PR to the null commit so we can later synchronise it back
# back to the "proper" head while resetting reviews
controllers.handle_pr(self.env, {
'action': 'opened',
'pull_request': {
**pr,
'head': {**pr['head'], 'sha': '0'*40},
'state': 'open',
},
'sender': sender,
})
# fetch & set up actual head
for st in gh.statuses(pr['head']['sha']):
controllers.handle_status(self.env, st)
# fetch and apply comments
counter = itertools.count()
items = [ # use counter so `comment` and `review` don't get hit during sort
(comment['created_at'], next(counter), False, comment)
for comment in gh.comments(number)
] + [
(review['submitted_at'], next(counter), True, review)
for review in gh.reviews(number)
]
items.sort()
for _, _, is_review, item in items:
if is_review:
controllers.handle_review(self.env, {
'action': 'submitted',
'review': item,
'pull_request': pr,
'repository': {'full_name': self.name},
'sender': sender,
})
else:
controllers.handle_comment(self.env, {
'action': 'created',
'issue': issue,
'comment': item,
'repository': {'full_name': self.name},
'sender': sender,
})
# sync to real head
controllers.handle_pr(self.env, {
'action': 'synchronize',
'pull_request': pr,
'sender': sender,
})
if pr['state'] == 'closed':
# don't go through controller because try_closing does weird things
# for safety / race condition reasons which ends up committing
# and breaks everything
self.env['runbot_merge.pull_requests'].search([
('repository.name', '=', pr['base']['repo']['full_name']),
('number', '=', number),
]).state = 'closed'
def having_branch(self, branch):
branches = self.env['runbot_merge.branch'].search
return self.filtered(lambda r: branch in branches(ast.literal_eval(r.branch_filter)))
def _remap_label(self, label):
for line in filter(None, (self.substitutions or '').splitlines()):
sep = line[0]
_, pattern, repl, flags = line.split(sep)
label = re.sub(
pattern, repl, label,
count=0 if 'g' in flags else 1,
flags=(re.MULTILINE if 'm' in flags.lower() else 0)
| (re.IGNORECASE if 'i' in flags.lower() else 0)
)
return label
class Branch(models.Model):
_name = _description = 'runbot_merge.branch'
_order = 'sequence, name'
name = fields.Char(required=True)
project_id = fields.Many2one('runbot_merge.project', required=True)
active_staging_id = fields.Many2one(
'runbot_merge.stagings', compute='_compute_active_staging', store=True,
help="Currently running staging for the branch."
)
staging_ids = fields.One2many('runbot_merge.stagings', 'target')
split_ids = fields.One2many('runbot_merge.split', 'target')
prs = fields.One2many('runbot_merge.pull_requests', 'target', domain=[
('state', '!=', 'closed'),
('state', '!=', 'merged'),
])
active = fields.Boolean(default=True)
sequence = fields.Integer(group_operator=None)
def _auto_init(self):
res = super(Branch, self)._auto_init()
tools.create_unique_index(
self._cr, 'runbot_merge_unique_branch_per_repo',
self._table, ['name', 'project_id'])
return res
@api.depends('active')
def _compute_display_name(self):
super()._compute_display_name()
for b in self.filtered(lambda b: not b.active):
b.display_name += ' (inactive)'
def write(self, vals):
super().write(vals)
if vals.get('active') is False:
self.active_staging_id.cancel(
"Target branch deactivated by %r.",
self.env.user.login,
)
self.env['runbot_merge.pull_requests.feedback'].create([{
'repository': pr.repository.id,
'pull_request': pr.number,
'message': f'Hey {pr.ping()}the target branch {pr.target.name!r} has been disabled, you may want to close this PR.',
} for pr in self.prs])
return True
@api.depends('staging_ids.active')
def _compute_active_staging(self):
for b in self:
b.active_staging_id = b.with_context(active_test=True).staging_ids
def _ready(self):
self.env.cr.execute("""
SELECT
min(pr.priority) as priority,
array_agg(pr.id) AS match
FROM runbot_merge_pull_requests pr
WHERE pr.target = any(%s)
-- exclude terminal states (so there's no issue when
-- deleting branches & reusing labels)
AND pr.state != 'merged'
AND pr.state != 'closed'
GROUP BY
pr.target,
CASE
WHEN pr.label SIMILAR TO '%%:patch-[[:digit:]]+'
THEN pr.id::text
ELSE pr.label
END
HAVING
bool_or(pr.state = 'ready') or bool_or(pr.priority = 0)
ORDER BY min(pr.priority), min(pr.id)
""", [self.ids])
browse = self.env['runbot_merge.pull_requests'].browse
return [(p, browse(ids)) for p, ids in self.env.cr.fetchall()]
def _stageable(self):
return [
(p, prs)
for p, prs in self._ready()
if not any(prs.mapped('blocked'))
]
def try_staging(self):
""" Tries to create a staging if the current branch does not already
have one. Returns None if the branch already has a staging or there
is nothing to stage, the newly created staging otherwise.
"""
logger = _logger.getChild('cron')
logger.info(
"Checking %s (%s) for staging: %s, skip? %s",
self, self.name,
self.active_staging_id,
bool(self.active_staging_id)
)
if self.active_staging_id:
return
rows = self._stageable()
priority = rows[0][0] if rows else -1
if priority == 0 or priority == 1:
# p=0 take precedence over all else
# p=1 allows merging a fix inside / ahead of a split (e.g. branch
# is broken or widespread false positive) without having to cancel
# the existing staging
batched_prs = [pr_ids for _, pr_ids in takewhile(lambda r: r[0] == priority, rows)]
elif self.split_ids:
split_ids = self.split_ids[0]
logger.info("Found split of PRs %s, re-staging", split_ids.mapped('batch_ids.prs'))
batched_prs = [batch.prs for batch in split_ids.batch_ids]
split_ids.unlink()
else: # p=2
batched_prs = [pr_ids for _, pr_ids in takewhile(lambda r: r[0] == priority, rows)]
if not batched_prs:
return
Batch = self.env['runbot_merge.batch']
staged = Batch
original_heads = {}
meta = {repo: {} for repo in self.project_id.repo_ids.having_branch(self)}
for repo, it in meta.items():
gh = it['gh'] = repo.github()
it['head'] = original_heads[repo] = gh.head(self.name)
# create tmp staging branch
gh.set_ref('tmp.{}'.format(self.name), it['head'])
batch_limit = self.project_id.batch_limit
first = True
for batch in batched_prs:
if len(staged) >= batch_limit:
break
try:
staged |= Batch.stage(meta, batch)
except exceptions.MergeError as e:
pr = e.args[0]
_logger.exception("Failed to merge %s into staging branch", pr.display_name)
if first or isinstance(e, exceptions.Unmergeable):
if len(e.args) > 1 and e.args[1]:
reason = e.args[1]
else:
reason = e.__context__
# if the reason is a json document, assume it's a github
# error and try to extract the error message to give it to
# the user
with contextlib.suppress(Exception):
reason = json.loads(str(reason))['message'].lower()
pr.state = 'error'
self.env['runbot_merge.pull_requests.feedback'].create({
'repository': pr.repository.id,
'pull_request': pr.number,
'message': f'{pr.ping()}unable to stage: {reason}',
})
else:
first = False
if not staged:
return
heads = {}
for repo, it in meta.items():
tree = it['gh'].commit(it['head'])['tree']
# ensures staging branches are unique and always
# rebuilt
r = base64.b64encode(os.urandom(12)).decode('ascii')
trailer = ''
if heads:
trailer = '\n'.join(
'Runbot-dependency: %s:%s' % (repo, h)
for repo, h in heads.items()
if not repo.endswith('^')
)
dummy_head = {'sha': it['head']}
if it['head'] == original_heads[repo]:
# if the repo has not been updated by the staging, create a
# dummy commit to force rebuild
dummy_head = it['gh']('post', 'git/commits', json={
'message': '''force rebuild
uniquifier: %s
For-Commit-Id: %s
%s''' % (r, it['head'], trailer),
'tree': tree['sha'],
'parents': [it['head']],
}).json()
# $repo is the head to check, $repo^ is the head to merge (they
# might be the same)
heads[repo.name + '^'] = it['head']
heads[repo.name] = dummy_head['sha']
self.env.cr.execute(
"INSERT INTO runbot_merge_commit (sha, to_check, statuses) "
"VALUES (%s, true, '{}') "
"ON CONFLICT (sha) DO UPDATE SET to_check=true",
[dummy_head['sha']]
)
# create actual staging object
st = self.env['runbot_merge.stagings'].create({
'target': self.id,
'batch_ids': [(4, batch.id, 0) for batch in staged],
'heads': json.dumps(heads)
})
# create staging branch from tmp
token = self.project_id.github_token
for r in self.project_id.repo_ids.having_branch(self):
it = meta[r]
staging_head = heads[r.name]
_logger.info(
"%s: create staging for %s:%s at %s",
self.project_id.name, r.name, self.name,
staging_head
)
refname = 'staging.{}'.format(self.name)
it['gh'].set_ref(refname, staging_head)
# asserts that the new head is visible through the api
head = it['gh'].head(refname)
assert head == staging_head,\
"[api] updated %s:%s to %s but found %s" % (
r.name, refname,
staging_head, head,
)
i = itertools.count()
@utils.backoff(delays=WAIT_FOR_VISIBILITY, exc=TimeoutError)
def wait_for_visibility():
if self._check_visibility(r, refname, staging_head, token):
_logger.info(
"[repo] updated %s:%s to %s: ok (at %d/%d)",
r.name, refname, staging_head,
next(i), len(WAIT_FOR_VISIBILITY)
)
return
_logger.warning(
"[repo] updated %s:%s to %s: failed (at %d/%d)",
r.name, refname, staging_head,
next(i), len(WAIT_FOR_VISIBILITY)
)
raise TimeoutError("Staged head not updated after %d seconds" % sum(WAIT_FOR_VISIBILITY))
logger.info("Created staging %s (%s) to %s", st, ', '.join(
'%s[%s]' % (batch, batch.prs)
for batch in staged
), st.target.name)
return st
def _check_visibility(self, repo, branch_name, expected_head, token):
""" Checks the repository actual to see if the new / expected head is
now visible
"""
# v1 protocol provides URL for ref discovery: https://github.com/git/git/blob/6e0cc6776106079ed4efa0cc9abace4107657abf/Documentation/technical/http-protocol.txt#L187
# for more complete client this is also the capabilities discovery and
# the "entry point" for the service
url = 'https://github.com/{}.git/info/refs?service=git-upload-pack'.format(repo.name)
with requests.get(url, stream=True, auth=(token, '')) as resp:
if not resp.ok:
return False
for head, ref in parse_refs_smart(resp.raw.read):
if ref != ('refs/heads/' + branch_name):
continue
return head == expected_head
return False
ACL = collections.namedtuple('ACL', 'is_admin is_reviewer is_author')
class PullRequests(models.Model):
_name = _description = 'runbot_merge.pull_requests'
_order = 'number desc'
_rec_name = 'number'
target = fields.Many2one('runbot_merge.branch', required=True, index=True)
repository = fields.Many2one('runbot_merge.repository', required=True)
# NB: check that target & repo have same project & provide project related?
state = fields.Selection([
('opened', 'Opened'),
('closed', 'Closed'),
('validated', 'Validated'),
('approved', 'Approved'),
('ready', 'Ready'),
# staged?
('merged', 'Merged'),
('error', 'Error'),
], default='opened', index=True)
number = fields.Integer(required=True, index=True, group_operator=None)
author = fields.Many2one('res.partner')
head = fields.Char(required=True)
label = fields.Char(
required=True, index=True,
help="Label of the source branch (owner:branchname), used for "
"cross-repository branch-matching"
)
message = fields.Text(required=True)
draft = fields.Boolean(default=False, required=True)
squash = fields.Boolean(default=False)
merge_method = fields.Selection([
('merge', "merge directly, using the PR as merge commit message"),
('rebase-merge', "rebase and merge, using the PR as merge commit message"),
('rebase-ff', "rebase and fast-forward"),
('squash', "squash"),
], default=False)
method_warned = fields.Boolean(default=False)
reviewed_by = fields.Many2one('res.partner')
delegates = fields.Many2many('res.partner', help="Delegate reviewers, not intrinsically reviewers but can review this PR")
priority = fields.Integer(default=2, index=True, group_operator=None)
overrides = fields.Char(required=True, default='{}')
statuses = fields.Text(
compute='_compute_statuses',
help="Copy of the statuses from the HEAD commit, as a Python literal"
)
statuses_full = fields.Text(
compute='_compute_statuses',
help="Compilation of the full status of the PR (commit statuses + overrides), as JSON"
)
status = fields.Char(compute='_compute_statuses')
previous_failure = fields.Char(default='{}')
batch_id = fields.Many2one('runbot_merge.batch', string="Active Batch", compute='_compute_active_batch', store=True)
batch_ids = fields.Many2many('runbot_merge.batch', string="Batches", context={'active_test': False})
staging_id = fields.Many2one(related='batch_id.staging_id', store=True)
commits_map = fields.Char(help="JSON-encoded mapping of PR commits to actually integrated commits. The integration head (either a merge commit or the PR's topmost) is mapped from the 'empty' pr commit (the key is an empty string, because you can't put a null key in json maps).", default='{}')
link_warned = fields.Boolean(
default=False, help="Whether we've already warned that this (ready)"
" PR is linked to an other non-ready PR"
)
blocked = fields.Char(
compute='_compute_is_blocked',
help="PR is not currently stageable for some reason (mostly an issue if status is ready)"
)
url = fields.Char(compute='_compute_url')
github_url = fields.Char(compute='_compute_url')
repo_name = fields.Char(related='repository.name')
message_title = fields.Char(compute='_compute_message_title')
def ping(self, author=True, reviewer=True):
P = self.env['res.partner']
s = ' '.join(
f'@{p.github_login}'
for p in (self.author if author else P) | (self.reviewed_by if reviewer else P)
if p
)
if s:
s += ' '
return s
@api.depends('repository.name', 'number')
def _compute_url(self):
base = werkzeug.urls.url_parse(self.env['ir.config_parameter'].sudo().get_param('web.base.url', 'http://localhost:8069'))
gh_base = werkzeug.urls.url_parse('https://github.com')
for pr in self:
path = f'/{werkzeug.urls.url_quote(pr.repository.name)}/pull/{pr.number}'
pr.url = str(base.join(path))
pr.github_url = str(gh_base.join(path))
@api.depends('message')
def _compute_message_title(self):
for pr in self:
pr.message_title = next(iter(pr.message.splitlines()), '')
@api.depends('repository.name', 'number', 'message')
def _compute_display_name(self):
return super(PullRequests, self)._compute_display_name()
def name_get(self):
name_template = '%(repo_name)s#%(number)d'
if self.env.context.get('pr_include_title'):
name_template += ' (%(message_title)s)'
return [(p.id, name_template % p) for p in self]
@api.model
def name_search(self, name='', args=None, operator='ilike', limit=100):
if not name or operator != 'ilike':
return super().name_search(name, args=args, operator=operator, limit=limit)
bits = [[('label', 'ilike', name)]]
if name.isdigit():
bits.append([('number', '=', name)])
if re.match(r'\w+#\d+$', name):
repo, num = name.rsplit('#', 1)
bits.append(['&', ('repository.name', 'ilike', repo), ('number', '=', int(num))])
else:
bits.append([('repository.name', 'ilike', name)])
domain = expression.OR(bits)
if args:
domain = expression.AND([args, domain])
return self.search(domain, limit=limit).sudo().name_get()
@property
def _approved(self):
return self.state in ('approved', 'ready') or any(
p.priority == 0
for p in (self | self._linked_prs)
)
@property
def _ready(self):
return (self.squash or self.merge_method) and self._approved and self.status == 'success'
@property
def _linked_prs(self):
if re.search(r':patch-\d+', self.label):
return self.browse(())
if self.state == 'merged':
return self.with_context(active_test=False).batch_ids\
.filtered(lambda b: b.staging_id.state == 'success')\
.prs - self
return self.search([
('target', '=', self.target.id),
('label', '=', self.label),
('state', 'not in', ('merged', 'closed')),
]) - self
# missing link to other PRs
@api.depends('priority', 'state', 'squash', 'merge_method', 'batch_id.active', 'label')
def _compute_is_blocked(self):
self.blocked = False
for pr in self:
if pr.state in ('merged', 'closed'):
continue
linked = pr._linked_prs
# check if PRs are configured (single commit or merge method set)
if not (pr.squash or pr.merge_method):
pr.blocked = 'has no merge method'
continue
other_unset = next((p for p in linked if not (p.squash or p.merge_method)), None)
if other_unset:
pr.blocked = "linked PR %s has no merge method" % other_unset.display_name
continue
# check if any PR in the batch is p=0 and none is in error
if any(p.priority == 0 for p in (pr | linked)):
if pr.state == 'error':
pr.blocked = "in error"
other_error = next((p for p in linked if p.state == 'error'), None)
if other_error:
pr.blocked = "linked pr %s in error" % other_error.display_name
# if none is in error then none is blocked because p=0
# "unblocks" the entire batch
continue
if pr.state != 'ready':
pr.blocked = 'not ready'
continue
unready = next((p for p in linked if p.state != 'ready'), None)
if unready:
pr.blocked = 'linked pr %s is not ready' % unready.display_name
continue
def _get_overrides(self):
if self:
return json.loads(self.overrides)
return {}
@api.depends('head', 'repository.status_ids', 'overrides')
def _compute_statuses(self):
Commits = self.env['runbot_merge.commit']
for pr in self:
c = Commits.search([('sha', '=', pr.head)])
st = json.loads(c.statuses or '{}')
statuses = {**st, **pr._get_overrides()}
pr.statuses_full = json.dumps(statuses)
if not statuses:
pr.status = pr.statuses = False
continue
pr.statuses = pprint.pformat(st)
st = 'success'
for ci in pr.repository.status_ids._for_pr(pr):
v = state_(statuses, ci.context) or 'pending'
if v in ('error', 'failure'):
st = 'failure'
break
if v == 'pending':
st = 'pending'
pr.status = st
@api.depends('batch_ids.active')
def _compute_active_batch(self):
for r in self:
r.batch_id = r.batch_ids.filtered(lambda b: b.active)[:1]
def _get_or_schedule(self, repo_name, number, target=None):
repo = self.env['runbot_merge.repository'].search([('name', '=', repo_name)])
if not repo:
return
if target and not repo.project_id._has_branch(target):
self.env['runbot_merge.pull_requests.feedback'].create({
'repository': repo.id,
'pull_request': number,
'message': "I'm sorry. Branch `{}` is not within my remit.".format(target),
})
return
pr = self.search([
('repository', '=', repo.id),
('number', '=', number,)
])
if pr:
return pr
Fetch = self.env['runbot_merge.fetch_job']
if Fetch.search([('repository', '=', repo.id), ('number', '=', number)]):
return
Fetch.create({
'repository': repo.id,
'number': number,
})
def _parse_command(self, commandstring):
for m in re.finditer(
r'(\S+?)(?:([+-])|=(\S*))?(?=\s|$)',
commandstring,
):
name, flag, param = m.groups()
if name == 'r':
name = 'review'
if flag in ('+', '-'):
yield name, flag == '+'
elif name == 'delegate':
if param:
for p in param.split(','):
yield 'delegate', p.lstrip('#@')
elif name == 'override':
if param:
for p in param.split(','):
yield 'override', p
elif name in ('p', 'priority'):
if param in ('0', '1', '2'):
yield ('priority', int(param))
elif any(name == k for k, _ in type(self).merge_method.selection):
yield ('method', name)
else:
yield name, param
def _parse_commands(self, author, comment, login):
"""Parses a command string prefixed by Project::github_prefix.
A command string can contain any number of space-separated commands:
retry
resets a PR in error mode to ready for staging
r(eview)+/-
approves or disapproves a PR (disapproving just cancels an approval)
delegate+/delegate=<users>
adds either PR author or the specified (github) users as
authorised reviewers for this PR. ``<users>`` is a
comma-separated list of github usernames (no @)
p(riority)=2|1|0
sets the priority to normal (2), pressing (1) or urgent (0).
Lower-priority PRs are selected first and batched together.
rebase+/-
Whether the PR should be rebased-and-merged (the default) or just
merged normally.
"""
assert self, "parsing commands must be executed in an actual PR"
(login, name) = (author.github_login, author.display_name) if author else (login, 'not in system')
is_admin, is_reviewer, is_author = self._pr_acl(author)
commands = [
ps
for m in self.repository.project_id._find_commands(comment['body'] or '')
for ps in self._parse_command(m)
]
if not commands:
_logger.info("found no commands in comment of %s (%s) (%s)", author.github_login, author.display_name,
utils.shorten(comment['body'] or '', 50)
)
return 'ok'
Feedback = self.env['runbot_merge.pull_requests.feedback']
if not (is_author or any(cmd == 'override' for cmd, _ in commands)):
# no point even parsing commands
_logger.info("ignoring comment of %s (%s): no ACL to %s",
login, name, self.display_name)
Feedback.create({
'repository': self.repository.id,
'pull_request': self.number,
'message': "I'm sorry, @{}. I'm afraid I can't do that.".format(login)
})
return 'ignored'
applied, ignored = [], []
def reformat(command, param):
if param is None:
pstr = ''
elif isinstance(param, bool):
pstr = '+' if param else '-'
elif isinstance(param, list):
pstr = '=' + ','.join(param)
else:
pstr = '={}'.format(param)
return '%s%s' % (command, pstr)
msgs = []
for command, param in commands:
ok = False
msg = None
if command == 'retry':
if is_author:
if self.state == 'error':
ok = True
self.state = 'ready'
else:
msg = "retry makes no sense when the PR is not in error."
elif command == 'check':
if is_author:
self.env['runbot_merge.fetch_job'].create({
'repository': self.repository.id,
'number': self.number,
})
ok = True
elif command == 'review':
if self.draft:
msg = "draft PRs can not be approved."
elif param and is_reviewer:
oldstate = self.state
newstate = RPLUS.get(self.state)
if not author.email:
msg = "I must know your email before you can review PRs. Please contact an administrator."
elif not newstate:
msg = "this PR is already reviewed, reviewing it again is useless."
else:
self.state = newstate
self.reviewed_by = author
ok = True
_logger.debug(
"r+ on %s by %s (%s->%s) status=%s message? %s",
self.display_name, author.github_login,
oldstate, newstate or oldstate,
self.status, self.status == 'failure'
)
if self.status == 'failure':
# the normal infrastructure is for failure and
# prefixes messages with "I'm sorry"
Feedback.create({
'repository': self.repository.id,
'pull_request': self.number,
'message': "@{} you may want to rebuild or fix this PR as it has failed CI.".format(login),
})
elif not param and is_author:
newstate = RMINUS.get(self.state)
if self.priority == 0 or newstate:
if newstate:
self.state = newstate
if self.priority == 0:
self.priority = 1
Feedback.create({
'repository': self.repository.id,
'pull_request': self.number,
'message': "PR priority reset to 1, as pull requests with priority 0 ignore review state.",
})
self.unstage("unreviewed (r-) by %s", login)
ok = True
else:
msg = "r- makes no sense in the current PR state."
elif command == 'delegate':
if is_reviewer:
ok = True
Partners = self.env['res.partner']
if param is True:
delegate = self.author
else:
delegate = Partners.search([('github_login', '=', param)]) or Partners.create({
'name': param,
'github_login': param,
})
delegate.write({'delegate_reviewer': [(4, self.id, 0)]})
elif command == 'priority':
if is_admin:
ok = True
self.priority = param
if param == 0:
self.target.active_staging_id.cancel(
"P=0 on %s by %s, unstaging target %s",
self.display_name,
author.github_login, self.target.name,
)
elif command == 'method':
if is_reviewer:
self.merge_method = param
ok = True
explanation = next(label for value, label in type(self).merge_method.selection if value == param)
Feedback.create({
'repository': self.repository.id,
'pull_request': self.number,
'message':"Merge method set to %s." % explanation
})
elif command == 'override':
overridable = author.override_rights\
.filtered(lambda r: not r.repository_id or (r.repository_id == self.repository))\
.mapped('context')
if param in overridable:
self.overrides = json.dumps({
**json.loads(self.overrides),
param: {
'state': 'success',
'target_url': comment['html_url'],
'description': f"Overridden by @{author.github_login}",
},
})
c = self.env['runbot_merge.commit'].search([('sha', '=', self.head)])
if c:
c.to_check = True
else:
c.create({'sha': self.head, 'statuses': '{}'})
ok = True
else:
msg = "you are not allowed to override this status."
else:
# ignore unknown commands
continue
_logger.info(
"%s %s(%s) on %s by %s (%s)",
"applied" if ok else "ignored",
command, param, self.display_name,
author.github_login, author.display_name,
)
if ok:
applied.append(reformat(command, param))
else:
ignored.append(reformat(command, param))
msgs.append(msg or "you can't {}.".format(reformat(command, param)))
if msgs:
joiner = ' ' if len(msgs) == 1 else '\n- '
msgs.insert(0, "I'm sorry, @{}:".format(login))
Feedback.create({
'repository': self.repository.id,
'pull_request': self.number,
'message': joiner.join(msgs),
})
msg = []
if applied:
msg.append('applied ' + ' '.join(applied))
if ignored:
ignoredstr = ' '.join(ignored)
msg.append('ignored ' + ignoredstr)
return '\n'.join(msg)
def _pr_acl(self, user):
if not self:
return ACL(False, False, False)
is_admin = self.env['res.partner.review'].search_count([
('partner_id', '=', user.id),
('repository_id', '=', self.repository.id),
('review', '=', True) if self.author != user else ('self_review', '=', True),
]) == 1
is_reviewer = is_admin or self in user.delegate_reviewer
# TODO: should delegate reviewers be able to retry PRs?
is_author = is_reviewer or self.author == user
return ACL(is_admin, is_reviewer, is_author)
def _validate(self, statuses):
# could have two PRs (e.g. one open and one closed) at least
# temporarily on the same head, or on the same head with different
# targets
failed = self.browse(())
for pr in self:
required = pr.repository.status_ids._for_pr(pr).mapped('context')
sts = {**statuses, **pr._get_overrides()}
success = True
for ci in required:
st = state_(sts, ci) or 'pending'
if st == 'success':
continue
success = False
if st in ('error', 'failure'):
failed |= pr
pr._notify_ci_new_failure(ci, to_status(sts.get(ci.strip(), 'pending')))
if success:
oldstate = pr.state
if oldstate == 'opened':
pr.state = 'validated'
elif oldstate == 'approved':
pr.state = 'ready'
return failed
def _notify_ci_new_failure(self, ci, st):
prev = json.loads(self.previous_failure)
if prev.get('state'): # old-style previous-failure
prev = {ci: prev}
if not any(self._statuses_equivalent(st, v) for v in prev.values()):
prev[ci] = st
self.previous_failure = json.dumps(prev)
self._notify_ci_failed(ci)
def _notify_merged(self, gh, payload):
deployment = gh('POST', 'deployments', json={
'ref': self.head, 'environment': 'merge',
'description': "Merge %s into %s" % (self.display_name, self.target.name),
'task': 'merge',
'auto_merge': False,
'required_contexts': [],
}).json()
gh('POST', 'deployments/{}/statuses'.format(deployment['id']), json={
'state': 'success',
'target_url': 'https://github.com/{}/commit/{}'.format(
self.repository.name,
payload['sha'],
),
'description': "Merged %s in %s at %s" % (
self.display_name, self.target.name, payload['sha']
)
})
def _statuses_equivalent(self, a, b):
""" Check if two statuses are *equivalent* meaning the description field
is ignored (check only state and target_url). This is because the
description seems to vary even if the rest does not, and generates
unnecessary notififcations as a result
"""
return a.get('state') == b.get('state') \
and a.get('target_url') == b.get('target_url')
def _notify_ci_failed(self, ci):
# only report an issue of the PR is already approved (r+'d)
if self.state == 'approved':
self.env['runbot_merge.pull_requests.feedback'].create({
'repository': self.repository.id,
'pull_request': self.number,
'message': "%s%r failed on this reviewed PR." % (self.ping(), ci),
})
def _auto_init(self):
super(PullRequests, self)._auto_init()
# incorrect index: unique(number, target, repository).
tools.drop_index(self._cr, 'runbot_merge_unique_pr_per_target', self._table)
# correct index:
tools.create_unique_index(
self._cr, 'runbot_merge_unique_pr_per_repo', self._table, ['repository', 'number'])
self._cr.execute("CREATE INDEX IF NOT EXISTS runbot_merge_pr_head "
"ON runbot_merge_pull_requests "
"USING hash (head)")
@property
def _tagstate(self):
if self.state == 'ready' and self.staging_id.heads:
return 'staged'
return self.state
@api.model
def create(self, vals):
pr = super().create(vals)
c = self.env['runbot_merge.commit'].search([('sha', '=', pr.head)])
pr._validate(json.loads(c.statuses or '{}'))
if pr.state not in ('closed', 'merged'):
self.env['runbot_merge.pull_requests.feedback'].create({
'repository': pr.repository.id,
'pull_request': pr.number,
'message': f"[Pull request status dashboard]({pr.url}).",
})
return pr
def _from_gh(self, description, author=None, branch=None, repo=None):
if repo is None:
repo = self.env['runbot_merge.repository'].search([
('name', '=', description['base']['repo']['full_name']),
])
if branch is None:
branch = self.env['runbot_merge.branch'].with_context(active_test=False).search([
('name', '=', description['base']['ref']),
('project_id', '=', repo.project_id.id),
])
if author is None:
author = self.env['res.partner'].search([
('github_login', '=', description['user']['login']),
], limit=1)
message = description['title'].strip()
body = description['body'] and description['body'].strip()
if body:
message += '\n\n' + body
return self.env['runbot_merge.pull_requests'].create({
'state': 'opened' if description['state'] == 'open' else 'closed',
'number': description['number'],
'label': repo._remap_label(description['head']['label']),
'author': author.id,
'target': branch.id,
'repository': repo.id,
'head': description['head']['sha'],
'squash': description['commits'] == 1,
'message': message,
'draft': description['draft'],
})
def write(self, vals):
if vals.get('squash'):
vals['merge_method'] = False
prev = None
if 'target' in vals or 'message' in vals:
prev = {
pr.id: {'target': pr.target, 'message': pr.message}
for pr in self
}
w = super().write(vals)
newhead = vals.get('head')
if newhead:
c = self.env['runbot_merge.commit'].search([('sha', '=', newhead)])
self._validate(json.loads(c.statuses or '{}'))
if prev:
for pr in self:
old_target = prev[pr.id]['target']
if pr.target != old_target:
pr.unstage(
"target (base) branch was changed from %r to %r",
old_target.display_name, pr.target.display_name,
)
old_message = prev[pr.id]['message']
if pr.merge_method in ('merge', 'rebase-merge') and pr.message != old_message:
pr.unstage("merge message updated")
return w
def _check_linked_prs_statuses(self, commit=False):
""" Looks for linked PRs where at least one of the PRs is in a ready
state and the others are not, notifies the other PRs.
:param bool commit: whether to commit the tnx after each comment
"""
# similar to Branch.try_staging's query as it's a subset of that
# other query's behaviour
self.env.cr.execute("""
SELECT
array_agg(pr.id) AS match
FROM runbot_merge_pull_requests pr
WHERE
-- exclude terminal states (so there's no issue when
-- deleting branches & reusing labels)
pr.state != 'merged'
AND pr.state != 'closed'
GROUP BY
pr.target,
CASE
WHEN pr.label SIMILAR TO '%%:patch-[[:digit:]]+'
THEN pr.id::text
ELSE pr.label
END
HAVING
-- one of the batch's PRs should be ready & not marked
bool_or(pr.state = 'ready' AND NOT pr.link_warned)
-- one of the others should be unready
AND bool_or(pr.state != 'ready')
-- but ignore batches with one of the prs at p0
AND bool_and(pr.priority != 0)
""")
for [ids] in self.env.cr.fetchall():
prs = self.browse(ids)
ready = prs.filtered(lambda p: p.state == 'ready')
unready = (prs - ready).sorted(key=lambda p: (p.repository.name, p.number))
for r in ready:
self.env['runbot_merge.pull_requests.feedback'].create({
'repository': r.repository.id,
'pull_request': r.number,
'message': "{}linked pull request(s) {} not ready. Linked PRs are not staged until all of them are ready.".format(
r.ping(),
', '.join(map('{0.display_name}'.format, unready))
)
})
r.link_warned = True
if commit:
self.env.cr.commit()
# send feedback for multi-commit PRs without a merge_method (which
# we've not warned yet)
methods = ''.join(
'* `%s` to %s\n' % pair
for pair in type(self).merge_method.selection
if pair[0] != 'squash'
)
for r in self.search([
('state', '=', 'ready'),
('squash', '=', False),
('merge_method', '=', False),
('method_warned', '=', False),
]):
self.env['runbot_merge.pull_requests.feedback'].create({
'repository': r.repository.id,
'pull_request': r.number,
'message': "%sbecause this PR has multiple commits, I need to know how to merge it:\n\n%s" % (
r.ping(),
methods,
)
})
r.method_warned = True
if commit:
self.env.cr.commit()
def _parse_commit_message(self, message):
""" Parses a commit message to split out the pseudo-headers (which
should be at the end) from the body, and serialises back with a
predefined pseudo-headers ordering.
"""
return Message.from_message(message)
def _is_mentioned(self, message, *, full_reference=False):
"""Returns whether ``self`` is mentioned in ``message```
:param str | PullRequest message:
:param bool full_reference: whether the repository name must be present
:rtype: bool
"""
if full_reference:
pattern = fr'\b{re.escape(self.display_name)}\b'
else:
repository = self.repository.name # .replace('/', '\\/')
pattern = fr'( |\b{repository})#{self.number}\b'
return bool(re.search(pattern, message if isinstance(message, str) else message.message))
def _build_merge_message(self, message, related_prs=()):
# handle co-authored commits (https://help.github.com/articles/creating-a-commit-with-multiple-authors/)
m = self._parse_commit_message(message)
if not self._is_mentioned(message):
m.body += '\n\ncloses {pr.display_name}'.format(pr=self)
for r in related_prs:
if not r._is_mentioned(message, full_reference=True):
m.headers.add('Related', r.display_name)
if self.reviewed_by:
m.headers.add('signed-off-by', self.reviewed_by.formatted_email)
return m
def _add_self_references(self, commits):
"""Adds a footer reference to ``self`` to all ``commits`` if they don't
already refer to the PR.
"""
for c in (c['commit'] for c in commits):
if not self._is_mentioned(c['message']):
m = self._parse_commit_message(c['message'])
m.headers.pop('Part-Of', None)
m.headers.add('Part-Of', self.display_name)
c['message'] = str(m)
def _stage(self, gh, target, related_prs=()):
# nb: pr_commits is oldest to newest so pr.head is pr_commits[-1]
_, prdict = gh.pr(self.number)
commits = prdict['commits']
method = self.merge_method or ('rebase-ff' if commits == 1 else None)
if commits > 50 and method.startswith('rebase'):
raise exceptions.Unmergeable(self, "Rebasing 50 commits is too much.")
if commits > 250:
raise exceptions.Unmergeable(
self, "Merging PRs of 250 or more commits is not supported "
"(https://developer.github.com/v3/pulls/#list-commits-on-a-pull-request)"
)
pr_commits = gh.commits(self.number)
for c in pr_commits:
if not (c['commit']['author']['email'] and c['commit']['committer']['email']):
raise exceptions.Unmergeable(
self,
f"All commits must have author and committer email, "
f"missing email on {c['sha']} indicates the authorship is "
f"most likely incorrect."
)
pr_head = pr_commits[-1]['sha']
if pr_head != self.head:
raise exceptions.Mismatch(self.head, pr_head, commits == 1)
if self.reviewed_by and self.reviewed_by.name == self.reviewed_by.github_login:
# XXX: find other trigger(s) to sync github name?
gh_name = gh.user(self.reviewed_by.github_login)['name']
if gh_name:
self.reviewed_by.name = gh_name
# update pr message in case an update was missed
msg = f'{prdict["title"]}\n\n{prdict.get("body") or ""}'.strip()
if self.message != msg:
self.message = msg
# NOTE: lost merge v merge/copy distinction (head being
# a merge commit reused instead of being re-merged)
return method, getattr(self, '_stage_' + method.replace('-', '_'))(
gh, target, pr_commits, related_prs=related_prs)
def _stage_squash(self, gh, target, commits, related_prs=()):
msg = self._build_merge_message(self, related_prs=related_prs)
authorship = {}
authors = {
(c['commit']['author']['name'], c['commit']['author']['email'])
for c in commits
}
if len(authors) == 1:
name, email = authors.pop()
authorship['author'] = {'name': name, 'email': email}
else:
msg.headers.extend(sorted(
('Co-Authored-By', "%s <%s>" % author)
for author in authors
))
committers = {
(c['commit']['committer']['name'], c['commit']['committer']['email'])
for c in commits
}
if len(committers) == 1:
name, email = committers.pop()
authorship['committer'] = {'name': name, 'email': email}
# should committers also be added to co-authors?
original_head = gh.head(target)
merge_tree = gh.merge(self.head, target, 'temp merge')['tree']['sha']
head = gh('post', 'git/commits', json={
**authorship,
'message': str(msg),
'tree': merge_tree,
'parents': [original_head],
}).json()['sha']
gh.set_ref(target, head)
commits_map = {c['sha']: head for c in commits}
commits_map[''] = head
self.commits_map = json.dumps(commits_map)
return head
def _stage_rebase_ff(self, gh, target, commits, related_prs=()):
# updates head commit with PR number (if necessary) then rebases
# on top of target
msg = self._build_merge_message(commits[-1]['commit']['message'], related_prs=related_prs)
commits[-1]['commit']['message'] = str(msg)
self._add_self_references(commits[:-1])
head, mapping = gh.rebase(self.number, target, commits=commits)
self.commits_map = json.dumps({**mapping, '': head})
return head
def _stage_rebase_merge(self, gh, target, commits, related_prs=()):
self._add_self_references(commits)
h, mapping = gh.rebase(self.number, target, reset=True, commits=commits)
msg = self._build_merge_message(self, related_prs=related_prs)
merge_head = gh.merge(h, target, str(msg))['sha']
self.commits_map = json.dumps({**mapping, '': merge_head})
return merge_head
def _stage_merge(self, gh, target, commits, related_prs=()):
pr_head = commits[-1] # oldest to newest
base_commit = None
head_parents = {p['sha'] for p in pr_head['parents']}
if len(head_parents) > 1:
# look for parent(s?) of pr_head not in PR, means it's
# from target (so we merged target in pr)
merge = head_parents - {c['sha'] for c in commits}
external_parents = len(merge)
if external_parents > 1:
raise exceptions.Unmergeable(
"The PR head can only have one parent from the base branch "
"(not part of the PR itself), found %d: %s" % (
external_parents,
', '.join(merge)
))
if external_parents == 1:
[base_commit] = merge
commits_map = {c['sha']: c['sha'] for c in commits}
if base_commit:
# replicate pr_head with base_commit replaced by
# the current head
original_head = gh.head(target)
merge_tree = gh.merge(pr_head['sha'], target, 'temp merge')['tree']['sha']
new_parents = [original_head] + list(head_parents - {base_commit})
msg = self._build_merge_message(pr_head['commit']['message'], related_prs=related_prs)
copy = gh('post', 'git/commits', json={
'message': str(msg),
'tree': merge_tree,
'author': pr_head['commit']['author'],
'committer': pr_head['commit']['committer'],
'parents': new_parents,
}).json()
gh.set_ref(target, copy['sha'])
# merge commit *and old PR head* map to the pr head replica
commits_map[''] = commits_map[pr_head['sha']] = copy['sha']
self.commits_map = json.dumps(commits_map)
return copy['sha']
else:
# otherwise do a regular merge
msg = self._build_merge_message(self)
merge_head = gh.merge(self.head, target, str(msg))['sha']
# and the merge commit is the normal merge head
commits_map[''] = merge_head
self.commits_map = json.dumps(commits_map)
return merge_head
def unstage(self, reason, *args):
""" If the PR is staged, cancel the staging. If the PR is split and
waiting, remove it from the split (possibly delete the split entirely)
"""
split_batches = self.with_context(active_test=False).mapped('batch_ids').filtered('split_id')
if len(split_batches) > 1:
_logger.warning("Found a PR linked with more than one split batch: %s (%s)", self, split_batches)
for b in split_batches:
if len(b.split_id.batch_ids) == 1:
# only the batch of this PR -> delete split
b.split_id.unlink()
else:
# else remove this batch from the split
b.split_id = False
self.staging_id.cancel('%s ' + reason, self.display_name, *args)
def _try_closing(self, by):
# ignore if the PR is already being updated in a separate transaction
# (most likely being merged?)
self.env.cr.execute('''
SELECT id, state FROM runbot_merge_pull_requests
WHERE id = %s AND state != 'merged'
FOR UPDATE SKIP LOCKED;
''', [self.id])
if not self.env.cr.fetchone():
return False
self.env.cr.execute('''
UPDATE runbot_merge_pull_requests
SET state = 'closed'
WHERE id = %s
''', [self.id])
self.env.cr.commit()
self.modified(['state'])
self.unstage("closed by %s", by)
return True
# state changes on reviews
RPLUS = {
'opened': 'approved',
'validated': 'ready',
}
RMINUS = {
'approved': 'opened',
'ready': 'validated',
'error': 'validated',
}
_TAGS = {
False: set(),
'opened': {'seen 🙂'},
}
_TAGS['validated'] = _TAGS['opened'] | {'CI 🤖'}
_TAGS['approved'] = _TAGS['opened'] | {'r+ 👌'}
_TAGS['ready'] = _TAGS['validated'] | _TAGS['approved']
_TAGS['staged'] = _TAGS['ready'] | {'merging 👷'}
_TAGS['merged'] = _TAGS['ready'] | {'merged 🎉'}
_TAGS['error'] = _TAGS['opened'] | {'error 🙅'}
_TAGS['closed'] = _TAGS['opened'] | {'closed 💔'}
ALL_TAGS = set.union(*_TAGS.values())
class Tagging(models.Model):
"""
Queue of tag changes to make on PRs.
Several PR state changes are driven by webhooks, webhooks should return
quickly, performing calls to the Github API would *probably* get in the
way of that. Instead, queue tagging changes into this table whose
execution can be cron-driven.
"""
_name = _description = 'runbot_merge.pull_requests.tagging'
repository = fields.Many2one('runbot_merge.repository', required=True)
# store the PR number (not id) as we need a Tagging for PR objects
# being deleted (retargeted to non-managed branches)
pull_request = fields.Integer(group_operator=None)
tags_remove = fields.Char(required=True, default='[]')
tags_add = fields.Char(required=True, default='[]')
def create(self, values):
if values.pop('state_from', None):
values['tags_remove'] = ALL_TAGS
if 'state_to' in values:
values['tags_add'] = _TAGS[values.pop('state_to')]
if not isinstance(values.get('tags_remove', ''), str):
values['tags_remove'] = json.dumps(list(values['tags_remove']))
if not isinstance(values.get('tags_add', ''), str):
values['tags_add'] = json.dumps(list(values['tags_add']))
return super().create(values)
def _send(self):
# noinspection SqlResolve
self.env.cr.execute("""
SELECT
t.repository as repo_id,
t.pull_request as pr_number,
array_agg(t.id) as ids,
array_agg(t.tags_remove::json) as to_remove,
array_agg(t.tags_add::json) as to_add
FROM runbot_merge_pull_requests_tagging t
GROUP BY t.repository, t.pull_request
""")
Repos = self.env['runbot_merge.repository']
ghs = {}
to_remove = []
for repo_id, pr, ids, remove, add in self.env.cr.fetchall():
repo = Repos.browse(repo_id)
gh = ghs.get(repo)
if not gh:
gh = ghs[repo] = repo.github()
# fold all grouped PRs'
tags_remove, tags_add = set(), set()
for minus, plus in zip(remove, add):
tags_remove.update(minus)
# need to remove minuses from to_add in case we get e.g.
# -foo +bar; -bar +baz, if we don't remove the minus, we'll end
# up with -foo +bar +baz instead of -foo +baz
tags_add.difference_update(minus)
tags_add.update(plus)
try:
gh.change_tags(pr, tags_remove, tags_add)
except Exception:
_logger.exception(
"Error while trying to change the tags of %s#%s from %s to %s",
repo.name, pr, remove, add,
)
else:
to_remove.extend(ids)
self.browse(to_remove).unlink()
class Feedback(models.Model):
""" Queue of feedback comments to send to PR users
"""
_name = _description = 'runbot_merge.pull_requests.feedback'
repository = fields.Many2one('runbot_merge.repository', required=True)
# store the PR number (not id) as we may want to send feedback to PR
# objects on non-handled branches
pull_request = fields.Integer(group_operator=None)
message = fields.Char()
close = fields.Boolean()
token_field = fields.Selection(
[('github_token', "Mergebot")],
default='github_token',
string="Bot User",
help="Token field (from repo's project) to use to post messages"
)
def _send(self):
ghs = {}
to_remove = []
for f in self.search([]):
repo = f.repository
gh = ghs.get((repo, f.token_field))
if not gh:
gh = ghs[(repo, f.token_field)] = repo.github(f.token_field)
try:
message = f.message
with contextlib.suppress(json.JSONDecodeError):
data = json.loads(message or '')
message = data.get('message')
if data.get('base'):
gh('PATCH', f'pulls/{f.pull_request}', json={'base': data['base']})
if f.close:
pr_to_notify = self.env['runbot_merge.pull_requests'].search([
('repository', '=', repo.id),
('number', '=', f.pull_request),
])
if pr_to_notify:
pr_to_notify._notify_merged(gh, data)
if f.close:
gh.close(f.pull_request)
if message:
gh.comment(f.pull_request, message)
except Exception:
_logger.exception(
"Error while trying to %s %s#%s (%s)",
'close' if f.close else 'send a comment to',
repo.name, f.pull_request,
utils.shorten(f.message, 200)
)
else:
to_remove.append(f.id)
self.browse(to_remove).unlink()
class Commit(models.Model):
"""Represents a commit onto which statuses might be posted,
independent of everything else as commits can be created by
statuses only, by PR pushes, by branch updates, ...
"""
_name = _description = 'runbot_merge.commit'
sha = fields.Char(required=True)
statuses = fields.Char(help="json-encoded mapping of status contexts to states", default="{}")
to_check = fields.Boolean(default=False)
def create(self, values):
values['to_check'] = True
r = super(Commit, self).create(values)
return r
def write(self, values):
values.setdefault('to_check', True)
r = super(Commit, self).write(values)
return r
def _notify(self):
Stagings = self.env['runbot_merge.stagings']
PRs = self.env['runbot_merge.pull_requests']
# chances are low that we'll have more than one commit
for c in self.search([('to_check', '=', True)]):
try:
c.to_check = False
st = json.loads(c.statuses)
pr = PRs.search([('head', '=', c.sha)])
if pr:
pr._validate(st)
stagings = Stagings.search([('heads', 'ilike', c.sha)]).filtered(
lambda s, h=c.sha: any(
head == h
for repo, head in json.loads(s.heads).items()
if not repo.endswith('^')
)
)
if stagings:
stagings._validate()
except Exception:
_logger.exception("Failed to apply commit %s (%s)", c, c.sha)
self.env.cr.rollback()
else:
self.env.cr.commit()
_sql_constraints = [
('unique_sha', 'unique (sha)', 'no duplicated commit'),
]
def _auto_init(self):
res = super(Commit, self)._auto_init()
self._cr.execute("""
CREATE INDEX IF NOT EXISTS runbot_merge_unique_statuses
ON runbot_merge_commit
USING hash (sha)
""")
self._cr.execute("""
CREATE INDEX IF NOT EXISTS runbot_merge_to_process
ON runbot_merge_commit ((1)) WHERE to_check
""")
return res
class Stagings(models.Model):
_name = _description = 'runbot_merge.stagings'
target = fields.Many2one('runbot_merge.branch', required=True)
batch_ids = fields.One2many(
'runbot_merge.batch', 'staging_id',
context={'active_test': False},
)
state = fields.Selection([
('success', 'Success'),
('failure', 'Failure'),
('pending', 'Pending'),
('cancelled', "Cancelled"),
('ff_failed', "Fast forward failed")
], default='pending')
active = fields.Boolean(default=True)
staged_at = fields.Datetime(default=fields.Datetime.now)
timeout_limit = fields.Datetime(store=True, compute='_compute_timeout_limit')
reason = fields.Text("Reason for final state (if any)")
# seems simpler than adding yet another indirection through a model
heads = fields.Char(required=True, help="JSON-encoded map of heads, one per repo in the project")
head_ids = fields.Many2many('runbot_merge.commit', compute='_compute_statuses')
statuses = fields.Binary(compute='_compute_statuses')
statuses_cache = fields.Text()
def write(self, vals):
# don't allow updating the statuses_cache
vals.pop('statuses_cache', None)
if 'state' not in vals:
return super().write(vals)
previously_pending = self.filtered(lambda s: s.state == 'pending')
super(Stagings, self).write(vals)
for staging in previously_pending:
if staging.state != 'pending':
super(Stagings, staging).write({
'statuses_cache': json.dumps(staging.statuses)
})
return True
def name_get(self):
return [
(staging.id, "%d (%s, %s%s)" % (
staging.id,
staging.target.name,
staging.state,
(', ' + staging.reason) if staging.reason else '',
))
for staging in self
]
@api.depends('heads')
def _compute_statuses(self):
""" Fetches statuses associated with the various heads, returned as
(repo, context, state, url)
"""
Commits = self.env['runbot_merge.commit']
for st in self:
heads = {
head: repo for repo, head in json.loads(st.heads).items()
if not repo.endswith('^')
}
commits = st.head_ids = Commits.search([('sha', 'in', list(heads.keys()))])
if st.statuses_cache:
st.statuses = json.loads(st.statuses_cache)
continue
st.statuses = [
(
heads[commit.sha],
context,
status.get('state') or 'pending',
status.get('target_url') or ''
)
for commit in commits
for context, st in json.loads(commit.statuses).items()
for status in [to_status(st)]
]
# only depend on staged_at as it should not get modified, but we might
# update the CI timeout after the staging have been created and we
# *do not* want to update the staging timeouts in that case
@api.depends('staged_at')
def _compute_timeout_limit(self):
for st in self:
st.timeout_limit = fields.Datetime.to_string(
fields.Datetime.from_string(st.staged_at)
+ datetime.timedelta(minutes=st.target.project_id.ci_timeout)
)
def _validate(self):
Commits = self.env['runbot_merge.commit']
for s in self:
if s.state != 'pending':
continue
repos = {
repo.name: repo
for repo in self.env['runbot_merge.repository'].search([])
.having_branch(s.target)
}
# maps commits to the statuses they need
required_statuses = [
(head, repos[repo].status_ids._for_staging(s).mapped('context'))
for repo, head in json.loads(s.heads).items()
if not repo.endswith('^')
]
# maps commits to their statuses
cmap = {
c.sha: json.loads(c.statuses)
for c in Commits.search([('sha', 'in', [h for h, _ in required_statuses])])
}
update_timeout_limit = False
st = 'success'
for head, reqs in required_statuses:
statuses = cmap.get(head) or {}
for v in map(lambda n: state_(statuses, n), reqs):
if st == 'failure' or v in ('error', 'failure'):
st = 'failure'
elif v is None:
st = 'pending'
elif v == 'pending':
st = 'pending'
update_timeout_limit = True
else:
assert v == 'success'
vals = {'state': st}
if update_timeout_limit:
vals['timeout_limit'] = fields.Datetime.to_string(datetime.datetime.now() + datetime.timedelta(minutes=s.target.project_id.ci_timeout))
_logger.debug("%s got pending status, bumping timeout to %s (%s)", self, vals['timeout_limit'], cmap)
s.write(vals)
def action_cancel(self):
w = self.env['runbot_merge.stagings.cancel'].create({
'staging_id': self.id,
})
return {
'type': 'ir.actions.act_window',
'target': 'new',
'name': f'Cancel staging {self.id} ({self.target.name})',
'view_mode': 'form',
'res_model': w._name,
'res_id': w.id,
}
def cancel(self, reason, *args):
self = self.filtered('active')
if not self:
return
_logger.info("Cancelling staging %s: " + reason, self, *args)
self.mapped('batch_ids').write({'active': False})
self.write({
'active': False,
'state': 'cancelled',
'reason': reason % args,
})
def fail(self, message, prs=None):
_logger.info("Staging %s failed: %s", self, message)
prs = prs or self.batch_ids.prs
prs.write({'state': 'error'})
for pr in prs:
self.env['runbot_merge.pull_requests.feedback'].create({
'repository': pr.repository.id,
'pull_request': pr.number,
'message': "%sstaging failed: %s" % (pr.ping(), message),
})
self.batch_ids.write({'active': False})
self.write({
'active': False,
'state': 'failure',
'reason': message,
})
def try_splitting(self):
batches = len(self.batch_ids)
if batches > 1:
midpoint = batches // 2
h, t = self.batch_ids[:midpoint], self.batch_ids[midpoint:]
# NB: batches remain attached to their original staging
sh = self.env['runbot_merge.split'].create({
'target': self.target.id,
'batch_ids': [(4, batch.id, 0) for batch in h],
})
st = self.env['runbot_merge.split'].create({
'target': self.target.id,
'batch_ids': [(4, batch.id, 0) for batch in t],
})
_logger.info("Split %s to %s (%s) and %s (%s)",
self, h, sh, t, st)
self.batch_ids.write({'active': False})
self.write({
'active': False,
'state': 'failure',
'reason': self.reason if self.state == 'failure' else 'timed out'
})
return True
# single batch => the staging is an unredeemable failure
if self.state != 'failure':
# timed out, just mark all PRs (wheee)
self.fail('timed out (>{} minutes)'.format(self.target.project_id.ci_timeout))
return False
# try inferring which PR failed and only mark that one
for repo, head in json.loads(self.heads).items():
if repo.endswith('^'):
continue
required_statuses = set(
self.env['runbot_merge.repository']
.search([('name', '=', repo)])
.status_ids
._for_staging(self)
.mapped('context'))
commit = self.env['runbot_merge.commit'].search([('sha', '=', head)])
statuses = json.loads(commit.statuses or '{}')
reason = next((
ctx for ctx, result in statuses.items()
if ctx in required_statuses
if to_status(result).get('state') in ('error', 'failure')
), None)
if not reason:
continue
pr = next((
pr for pr in self.batch_ids.prs
if pr.repository.name == repo
), None)
status = to_status(statuses[reason])
viewmore = ''
if status.get('target_url'):
viewmore = ' (view more at %(target_url)s)' % status
if pr:
self.fail("%s%s" % (reason, viewmore), pr)
else:
self.fail('%s on %s%s' % (reason, head, viewmore))
return False
# the staging failed but we don't have a specific culprit, fail
# everything
self.fail("unknown reason")
return False
def check_status(self):
"""
Checks the status of an active staging:
* merges it if successful
* splits it if failed (or timed out) and more than 1 batch
* marks the PRs as failed otherwise
* ignores if pending (or cancelled or ff_failed but those should also
be disabled)
"""
logger = _logger.getChild('cron')
if not self.active:
logger.info("Staging %s is not active, ignoring status check", self)
return
logger.info("Checking active staging %s (state=%s)", self, self.state)
project = self.target.project_id
if self.state == 'success':
gh = {repo.name: repo.github() for repo in project.repo_ids.having_branch(self.target)}
staging_heads = json.loads(self.heads)
self.env.cr.execute('''
SELECT 1 FROM runbot_merge_pull_requests
WHERE id in %s
FOR UPDATE
''', [tuple(self.mapped('batch_ids.prs.id'))])
try:
self._safety_dance(gh, staging_heads)
except exceptions.FastForwardError as e:
logger.warning(
"Could not fast-forward successful staging on %s:%s",
e.args[0], self.target.name,
exc_info=True
)
self.write({
'state': 'ff_failed',
'reason': str(e.__cause__ or e.__context__ or e)
})
else:
prs = self.mapped('batch_ids.prs')
logger.info(
"%s FF successful, marking %s as merged",
self, prs
)
prs.write({'state': 'merged'})
pseudobranch = None
if self.target == project.branch_ids[:1]:
pseudobranch = project._next_freeze()
for pr in prs:
self.env['runbot_merge.pull_requests.feedback'].create({
'repository': pr.repository.id,
'pull_request': pr.number,
'message': json.dumps({
'sha': json.loads(pr.commits_map)[''],
}),
'close': True,
})
if pseudobranch:
self.env['runbot_merge.pull_requests.tagging'].create({
'repository': pr.repository.id,
'pull_request': pr.number,
'tags_add': json.dumps([pseudobranch]),
})
finally:
self.batch_ids.write({'active': False})
self.write({'active': False})
elif self.state == 'failure' or self.is_timed_out():
self.try_splitting()
def is_timed_out(self):
return fields.Datetime.from_string(self.timeout_limit) < datetime.datetime.now()
def _safety_dance(self, gh, staging_heads):
""" Reverting updates doesn't work if the branches are protected
(because a revert is basically a force push). So we can update
REPO_A, then fail to update REPO_B for some reason, and we're hosed.
To try and make this issue less likely, do the safety dance:
* First, perform a dry run using the tmp branches (which can be
force-pushed and sacrificed), that way if somebody pushed directly
to REPO_B during the staging we catch it. If we're really unlucky
they could still push after the dry run but...
* An other issue then is that the github call sometimes fails for no
noticeable reason (e.g. network failure or whatnot), if it fails
on REPO_B when REPO_A has already been updated things get pretty
bad. In that case, wait a bit and retry for now. A more complex
strategy (including disabling the branch entirely until somebody
has looked at and fixed the issue) might be necessary.
:returns: the last repo it tried to update (probably the one on which
it failed, if it failed)
"""
# FIXME: would make sense for FFE to be richer, and contain the repo name
repo_name = None
tmp_target = 'tmp.' + self.target.name
# first force-push the current targets to all tmps
for repo_name in staging_heads.keys():
if repo_name.endswith('^'):
continue
g = gh[repo_name]
g.set_ref(tmp_target, g.head(self.target.name))
# then attempt to FF the tmp to the staging
for repo_name, head in staging_heads.items():
if repo_name.endswith('^'):
continue
gh[repo_name].fast_forward(tmp_target, staging_heads.get(repo_name + '^') or head)
# there is still a race condition here, but it's way
# lower than "the entire staging duration"...
first = True
for repo_name, head in staging_heads.items():
if repo_name.endswith('^'):
continue
for pause in [0.1, 0.3, 0.5, 0.9, 0]: # last one must be 0/falsy of we lose the exception
try:
# if the staging has a $repo^ head, merge that,
# otherwise merge the regular (CI'd) head
gh[repo_name].fast_forward(
self.target.name,
staging_heads.get(repo_name + '^') or head
)
except exceptions.FastForwardError:
# The GH API regularly fails us. If the failure does not
# occur on the first repository, retry a few times with a
# little pause.
if not first and pause:
time.sleep(pause)
continue
raise
else:
break
first = False
return repo_name
class Split(models.Model):
_name = _description = 'runbot_merge.split'
target = fields.Many2one('runbot_merge.branch', required=True)
batch_ids = fields.One2many('runbot_merge.batch', 'split_id', context={'active_test': False})
class Batch(models.Model):
""" A batch is a "horizontal" grouping of *codependent* PRs: PRs with
the same label & target but for different repositories. These are
assumed to be part of the same "change" smeared over multiple
repositories e.g. change an API in repo1, this breaks use of that API
in repo2 which now needs to be updated.
"""
_name = _description = 'runbot_merge.batch'
target = fields.Many2one('runbot_merge.branch', required=True)
staging_id = fields.Many2one('runbot_merge.stagings')
split_id = fields.Many2one('runbot_merge.split')
prs = fields.Many2many('runbot_merge.pull_requests')
active = fields.Boolean(default=True)
@api.constrains('target', 'prs')
def _check_prs(self):
for batch in self:
repos = self.env['runbot_merge.repository']
for pr in batch.prs:
if pr.target != batch.target:
raise ValidationError("A batch and its PRs must have the same branch, got %s and %s" % (batch.target, pr.target))
if pr.repository in repos:
raise ValidationError("All prs of a batch must have different target repositories, got a duplicate %s on %s" % (pr.repository, pr))
repos |= pr.repository
def stage(self, meta, prs):
"""
Updates meta[*][head] on success
:return: () or Batch object (if all prs successfully staged)
"""
new_heads = {}
for pr in prs:
gh = meta[pr.repository]['gh']
_logger.info(
"Staging pr %s for target %s; method=%s",
pr.display_name, pr.target.name,
pr.merge_method or (pr.squash and 'single') or None
)
target = 'tmp.{}'.format(pr.target.name)
original_head = gh.head(target)
try:
try:
method, new_heads[pr] = pr._stage(gh, target, related_prs=(prs - pr))
_logger.info(
"Staged pr %s to %s by %s: %s -> %s",
pr.display_name, pr.target.name, method,
original_head, new_heads[pr]
)
except Exception:
# reset the head which failed, as rebase() may have partially
# updated it (despite later steps failing)
gh.set_ref(target, original_head)
# then reset every previous update
for to_revert in new_heads.keys():
it = meta[to_revert.repository]
it['gh'].set_ref('tmp.{}'.format(to_revert.target.name), it['head'])
raise
except github.MergeError:
raise exceptions.MergeError(pr)
except exceptions.Mismatch as e:
old_head, new_head, to_squash = e.args
pr.write({
'state': 'opened',
'squash': to_squash,
'head': new_head,
})
_logger.warning(
"head mismatch on %s: had %s but found %s",
pr.display_name, old_head, new_head
)
self.env['runbot_merge.pull_requests.feedback'].create({
'repository': pr.repository.id,
'pull_request': pr.number,
'message': "%swe apparently missed an update to this PR "
"and tried to stage it in a state which "
"might not have been approved. PR has been "
"updated to %s, please check and approve or "
"re-approve." % (pr.ping(), new_head)
})
return self.env['runbot_merge.batch']
# update meta to new heads
for pr, head in new_heads.items():
meta[pr.repository]['head'] = head
return self.create({
'target': prs[0].target.id,
'prs': [(4, pr.id, 0) for pr in prs],
})
class FetchJob(models.Model):
_name = _description = 'runbot_merge.fetch_job'
active = fields.Boolean(default=True)
repository = fields.Many2one('runbot_merge.repository', required=True)
number = fields.Integer(required=True, group_operator=None)
def _check(self, commit=False):
"""
:param bool commit: commit after each fetch has been executed
"""
while True:
f = self.search([], limit=1)
if not f:
return
self.env.cr.execute("SAVEPOINT runbot_merge_before_fetch")
try:
f.repository._load_pr(f.number)
except Exception:
self.env.cr.execute("ROLLBACK TO SAVEPOINT runbot_merge_before_fetch")
_logger.exception("Failed to load pr %s, skipping it", f.number)
finally:
self.env.cr.execute("RELEASE SAVEPOINT runbot_merge_before_fetch")
f.active = False
if commit:
self.env.cr.commit()
# The commit (and PR) statuses was originally a map of ``{context:state}``
# however it turns out to clarify error messages it'd be useful to have
# a bit more information e.g. a link to the CI's build info on failure and
# all that. So the db-stored statuses are now becoming a map of
# ``{ context: {state, target_url, description } }``. The issue here is
# there's already statuses stored in the db so we need to handle both
# formats, hence these utility functions)
def state_(statuses, name):
""" Fetches the status state """
name = name.strip()
v = statuses.get(name)
if isinstance(v, dict):
return v.get('state')
return v
def to_status(v):
""" Converts old-style status values (just a state string) to new-style
(``{state, target_url, description}``)
:type v: str | dict
:rtype: dict
"""
if isinstance(v, dict):
return v
return {'state': v, 'target_url': None, 'description': None}
refline = re.compile(rb'([\da-f]{40}) ([^\0\n]+)(\0.*)?\n?$')
ZERO_REF = b'0'*40
def parse_refs_smart(read):
""" yields pkt-line data (bytes), or None for flush lines """
def read_line():
length = int(read(4), 16)
if length == 0:
return None
return read(length - 4)
header = read_line()
assert header.rstrip() == b'# service=git-upload-pack', header
assert read_line() is None, "failed to find first flush line"
# read lines until second delimiter
for line in iter(read_line, None):
if line.startswith(ZERO_REF):
break # empty list (no refs)
m = refline.match(line)
yield m[1].decode(), m[2].decode()
BREAK = re.compile(r'''
^
[ ]{0,3} # 0-3 spaces of indentation
# followed by a sequence of three or more matching -, _, or * characters,
# each followed optionally by any number of spaces or tabs
# so needs to start with a _, - or *, then have at least 2 more such
# interspersed with any number of spaces or tabs
([*_-])
([ \t]*\1){2,}
[ \t]*
$
''', flags=re.VERBOSE)
SETEX_UNDERLINE = re.compile(r'''
^
[ ]{0,3} # no more than 3 spaces indentation
[-=]+ # a sequence of = characters or a sequence of - characters
[ ]* # any number of trailing spaces
$
# we don't care about "a line containing a single -" because we want to
# disambiguate SETEX headings from thematic breaks, and thematic breaks have
# 3+ -. Doesn't look like GH interprets `- - -` as a line so yay...
''', flags=re.VERBOSE)
HEADER = re.compile('^([A-Za-z-]+): (.*)$')
class Message:
@classmethod
def from_message(cls, msg):
in_headers = True
maybe_setex = None
# creating from PR message -> remove content following break
msg, handle_break = (msg, False) if isinstance(msg, str) else (msg.message, True)
headers = []
body = []
# don't process the title (first line) of the commit message
msg = msg.splitlines()
for line in reversed(msg[1:]):
if maybe_setex:
# NOTE: actually slightly more complicated: it's a SETEX heading
# only if preceding line(s) can be interpreted as a
# paragraph so e.g. a title followed by a line of dashes
# would indeed be a break, but this should be good enough
# for now, if we need more we'll need a full-blown
# markdown parser probably
if line: # actually a SETEX title -> add underline to body then process current
body.append(maybe_setex)
else: # actually break, remove body then process current
body = []
maybe_setex = None
if not line:
if not in_headers and body and body[-1]:
body.append(line)
continue
if handle_break and BREAK.match(line):
if SETEX_UNDERLINE.match(line):
maybe_setex = line
else:
body = []
continue
h = HEADER.match(line)
if h:
# c-a-b = special case from an existing test, not sure if actually useful?
if in_headers or h.group(1).lower() == 'co-authored-by':
headers.append(h.groups())
continue
body.append(line)
in_headers = False
# if there are non-title body lines, add a separation after the title
if body and body[-1]:
body.append('')
body.append(msg[0])
return cls('\n'.join(reversed(body)), Headers(reversed(headers)))
def __init__(self, body, headers=None):
self.body = body
self.headers = headers or Headers()
def __setattr__(self, name, value):
# make sure stored body is always stripped
if name == 'body':
value = value and value.strip()
super().__setattr__(name, value)
def __str__(self):
if not self.headers:
return self.body + '\n'
with io.StringIO(self.body) as msg:
msg.write(self.body)
msg.write('\n\n')
# https://git.wiki.kernel.org/index.php/CommitMessageConventions
# seems to mostly use capitalised names (rather than title-cased)
keys = list(OrderedSet(k.capitalize() for k in self.headers.keys()))
# c-a-b must be at the very end otherwise github doesn't see it
keys.sort(key=lambda k: k == 'Co-authored-by')
for k in keys:
for v in self.headers.getlist(k):
msg.write(k)
msg.write(': ')
msg.write(v)
msg.write('\n')
return msg.getvalue()
def sub(self, pattern, repl, *, flags):
""" Performs in-place replacements on the body
"""
self.body = re.sub(pattern, repl, self.body, flags=flags)