from __future__ import annotations import ast import builtins import collections import contextlib import datetime import itertools import json import logging import re import time from enum import IntEnum from functools import reduce from operator import itemgetter from typing import Optional, Union, List, Iterator, Tuple import psycopg2.errors import sentry_sdk import werkzeug from markupsafe import Markup from odoo import api, fields, models, tools, Command from odoo.exceptions import AccessError, UserError from odoo.osv import expression from odoo.tools import html_escape, Reverse, mute_logger from . import commands from .utils import enum, readonly, dfm from .. import github, exceptions, controllers, utils _logger = logging.getLogger(__name__) FOOTER = '\nMore info at https://github.com/odoo/odoo/wiki/Mergebot#forward-port\n' 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' id: int sequence = fields.Integer(default=50, group_operator=None) name = fields.Char(required=True) project_id = fields.Many2one('runbot_merge.project', required=True, index=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_create_multi def create(self, vals_list): for vals in vals_list: if 'status_ids' in vals: continue 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_list) 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') -> github.GH: return github.GH(self.project_id[token_field], self.name) def _auto_init(self): res = super()._auto_init() tools.create_unique_index( self.env.cr, 'runbot_merge_unique_repo', self._table, ['name']) return res def _load_pr( self, number: int, *, closing: bool = False, squash: bool = False, ping: str | None = None, ): gh = self.github() # fetch PR object and handle as *opened* issue, pr = gh.pr(number) repo_name = pr['base']['repo']['full_name'] if not self.project_id._has_branch(pr['base']['ref']): _logger.info("Tasked with loading %s PR %s#%d for un-managed branch %s:%s, ignoring", pr['state'], repo_name, number, self.name, pr['base']['ref']) if not closing: self.env.ref('runbot_merge.pr.load.unmanaged')._send( repository=self, pull_request=number, format_args = { 'pr': pr, 'repository': self, }, ) return # if the PR is already loaded, force sync a few attributes pr_id = self.env['runbot_merge.pull_requests'].search([ ('repository.name', '=', repo_name), ('number', '=', number), ]) if pr_id: if squash: pr_id.squash = pr['commits'] == 1 return sync = controllers.handle_pr(self.env, { 'action': 'synchronize', 'pull_request': pr, 'sender': {'login': self.project_id.github_prefix} }) edit = controllers.handle_pr(self.env, { 'action': 'edited', 'pull_request': pr, 'changes': { 'base': {'ref': {'from': pr_id.target.name}}, 'title': {'from': pr_id.message.splitlines()[0]}, 'body': {'from', ''.join(pr_id.message.splitlines(keepends=True)[2:])}, }, 'sender': {'login': self.project_id.github_prefix}, }) edit2 = '' if pr_id.draft != pr['draft']: edit2 = controllers.handle_pr(self.env, { 'action': 'converted_to_draft' if pr['draft'] else 'ready_for_review', 'pull_request': pr, 'sender': {'login': self.project_id.github_prefix} }) + '. ' if pr_id.state != 'closed' and 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 pr_id.state = 'closed' self.env['runbot_merge.pull_requests.feedback'].create({ 'repository': pr_id.repository.id, 'pull_request': number, 'message': f"{edit}. {edit2}{sync}.", }) return # special case for closed PRs, just ignore all events and skip feedback if closing: self.env['runbot_merge.pull_requests']._from_gh(pr) return 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, }) pr_id = self.env['runbot_merge.pull_requests'].search([ ('repository.name', '=', repo_name), ('number', '=', number), ]) 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 pr_id.closed = True self.env.ref('runbot_merge.pr.load.fetched')._send( repository=self, pull_request=number, format_args={ 'pr': pr_id, 'ping': ping or pr_id.ping, }, ) 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' id: int name = fields.Char(required=True) project_id = fields.Many2one('runbot_merge.project', required=True, index=True) active_staging_id = fields.Many2one( 'runbot_merge.stagings', compute='_compute_active_staging', store=True, index=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) staging_enabled = fields.Boolean(default=True) def _auto_init(self): res = super()._auto_init() tools.create_unique_index( self.env.cr, 'runbot_merge_unique_branch_per_repo', self._table, ['name', 'project_id']) return res @api.depends('name', 'active', 'project_id.name') def _compute_display_name(self): for b in self: b.display_name = f"{b.project_id.name}:{b.name}" + ('' if b.active else ' (inactive)') def write(self, vals): if vals.get('active') is False and (actives := self.filtered('active')): actives.active_staging_id.cancel( "Target branch deactivated by %r.", self.env.user.login, ) tmpl = self.env.ref('runbot_merge.pr.branch.disabled') self.env['runbot_merge.pull_requests.feedback'].create([{ 'repository': pr.repository.id, 'pull_request': pr.number, 'message': tmpl._format(pr=pr), } for pr in actives.prs]) self.env.ref('runbot_merge.branch_cleanup')._trigger() if ( (vals.get('staging_enabled') is True and not all(self.mapped('staging_enabled'))) or (vals.get('active') is True and not all(self.mapped('active'))) ): self.env.ref('runbot_merge.staging_cron')._trigger() super().write(vals) 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 class SplitOffWizard(models.TransientModel): _name = "runbot_merge.pull_requests.split_off" _description = "wizard to split a PR off of its current batch and into a different one" pr_id = fields.Many2one("runbot_merge.pull_requests", required=True) new_label = fields.Char(string="New Label") def button_apply(self): self.pr_id._split_off(self.new_label) self.unlink() return {'type': 'ir.actions.act_window_close'} ACL = collections.namedtuple('ACL', 'is_admin is_reviewer is_author') class PullRequests(models.Model): _name = 'runbot_merge.pull_requests' _description = "Pull Request" _inherit = ['mail.thread'] _order = 'number desc' _rec_name = 'number' id: int display_name: str target = fields.Many2one('runbot_merge.branch', required=True, index=True, tracking=True) target_sequence = fields.Integer(related='target.sequence') repository = fields.Many2one('runbot_merge.repository', required=True) project = fields.Many2one(related='repository.project_id') # NB: check that target & repo have same project & provide project related? closed = fields.Boolean(default=False, tracking=True) error = fields.Boolean(string="in error", default=False, tracking=True) skipchecks = fields.Boolean(related='batch_id.skipchecks', inverse='_inverse_skipchecks') cancel_staging = fields.Boolean(related='batch_id.cancel_staging') merge_date = fields.Datetime( related='batch_id.merge_date', inverse=readonly, readonly=True, tracking=True, store=True, ) state = fields.Selection([ ('opened', 'Opened'), ('closed', 'Closed'), ('validated', 'Validated'), ('approved', 'Approved'), ('ready', 'Ready'), # staged? ('merged', 'Merged'), ('error', 'Error'), ], compute='_compute_state', inverse=readonly, readonly=True, store=True, index=True, tracking=True, column_type=enum(_name, 'state'), ) number = fields.Integer(required=True, index=True, group_operator=None) author = fields.Many2one('res.partner', index=True) head = fields.Char(required=True, tracking=True) label = fields.Char( required=True, index=True, tracking=True, help="Label of the source branch (owner:branchname), used for " "cross-repository branch-matching" ) refname = fields.Char(compute='_compute_refname') message = fields.Text(required=True) message_html = fields.Html(compute='_compute_message_html', sanitize=False) draft = fields.Boolean( default=False, required=True, tracking=True, help="A draft PR can not be merged", ) squash = fields.Boolean(default=False, tracking=True) 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, tracking=True, column_type=enum(_name, 'merge_method')) method_warned = fields.Boolean(default=False) reviewed_by = fields.Many2one('res.partner', index=True, tracking=True) delegates = fields.Many2many('res.partner', help="Delegate reviewers, not intrinsically reviewers but can review this PR") priority = fields.Selection(related="batch_id.priority", inverse=readonly, readonly=True) overrides = fields.Char(required=True, default='{}', tracking=True) statuses = fields.Text(help="Copy of the statuses from the HEAD commit, as a Python literal", default="{}") statuses_full = fields.Text( compute='_compute_statuses', help="Compilation of the full status of the PR (commit statuses + overrides), as JSON", store=True, ) status = fields.Selection([ ('pending', 'Pending'), ('failure', 'Failure'), ('success', 'Success'), ], compute='_compute_statuses', store=True, inverse=readonly, readonly=True, column_type=enum(_name, 'status')) previous_failure = fields.Char(default='{}') batch_id = fields.Many2one('runbot_merge.batch', index=True) staging_id = fields.Many2one('runbot_merge.stagings', compute='_compute_staging', inverse=readonly, readonly=True, store=True) staging_ids = fields.Many2many('runbot_merge.stagings', string="Stagings", compute='_compute_stagings', inverse=readonly, readonly=True, context={"active_test": False}) @api.depends( 'closed', 'batch_id.batch_staging_ids.runbot_merge_stagings_id.active', ) def _compute_staging(self): for p in self: if p.closed: p.staging_id = False else: p.staging_id = p.batch_id.staging_ids.filtered('active') @api.depends('batch_id.batch_staging_ids.runbot_merge_stagings_id') def _compute_stagings(self): for p in self: p.staging_ids = p.batch_id.staging_ids 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', store=True, 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') ping = fields.Char(compute='_compute_ping', recursive=True) source_id = fields.Many2one('runbot_merge.pull_requests', index=True, help="the original source of this FP even if parents were detached along the way") parent_id = fields.Many2one( 'runbot_merge.pull_requests', index=True, help="a PR with a parent is an automatic forward port", tracking=True, ) root_id = fields.Many2one('runbot_merge.pull_requests', compute='_compute_root', recursive=True) forwardport_ids = fields.One2many('runbot_merge.pull_requests', 'source_id') limit_id = fields.Many2one('runbot_merge.branch', help="Up to which branch should this PR be forward-ported", tracking=True) detach_reason = fields.Char() _sql_constraints = [( 'fw_constraint', 'check(source_id is null or num_nonnulls(parent_id, detach_reason) = 1)', "fw PRs must either be attached or have a reason for being detached", )] @api.depends('label') def _compute_refname(self): for pr in self: pr.refname = pr.label.split(':', 1)[-1] @api.depends( 'author.github_login', 'reviewed_by.github_login', 'source_id.author.github_login', 'source_id.reviewed_by.github_login', ) def _compute_ping(self): for pr in self: if source := pr.source_id: contacts = source.author | source.reviewed_by | pr.reviewed_by else: contacts = pr.author | pr.reviewed_by s = ' '.join(f'@{p.github_login}' for p in contacts) pr.ping = s and (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('parent_id.root_id') def _compute_root(self): for p in self: p.root_id = reduce(lambda _, p: p, self._iter_ancestors()) @api.depends('message') def _compute_message_title(self): for pr in self: pr.message_title = next(iter(pr.message.splitlines()), '') @api.depends("message") def _compute_message_html(self): for pr in self: match pr.message.split('\n\n', 1): case [title]: pr.message_html = Markup('

%s

') % title case [title, description]: pr.message_html = Markup('

%s

\n%s') % ( title, dfm(pr.repository.name, description), ) case _: pr.message_html = "" @api.depends('repository.name', 'number', 'message') def _compute_display_name(self): name_template = '%(repo_name)s#%(number)d' if self.env.context.get('pr_include_title'): name_template += ' (%(message_title)s)' for p in self: p.display_name = name_template % p def _inverse_skipchecks(self): for p in self: p.batch_id.skipchecks = p.skipchecks if p.skipchecks: p.reviewed_by = self.env.user.partner_id @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().mapped(lambda r: (r.id, r.display_name)) @property def _approved(self): return self.state in ('approved', 'ready') @property def _ready(self): return (self.squash or self.merge_method) and self._approved and self.status == 'success' @property def _linked_prs(self): return self.batch_id.prs - self @property def limit_pretty(self): if self.limit_id: return self.limit_id.name branches = self.repository.project_id.branch_ids if ((bf := self.repository.branch_filter) or '[]') != '[]': branches = branches.filtered_domain(ast.literal_eval(bf)) return branches[:1].name @api.depends( 'batch_id.prs.draft', 'batch_id.prs.state', 'batch_id.skipchecks', ) def _compute_is_blocked(self): self.blocked = False requirements = ( lambda p: not p.draft, lambda p: p.state == 'ready' \ or p.batch_id.skipchecks \ and all(pr.state != 'error' for pr in p.batch_id.prs) ) messages = ('is in draft', 'is not ready') for pr in self: if pr.state in ('merged', 'closed'): continue blocking, message = next(( (blocking, message) for blocking in pr.batch_id.prs for requirement, message in zip(requirements, messages) if not requirement(blocking) ), (None, None)) if blocking == pr: pr.blocked = message elif blocking: pr.blocked = f"linked PR {blocking.display_name} {message}" def _get_overrides(self) -> dict[str, dict[str, str]]: if self.parent_id: return self.parent_id._get_overrides() | json.loads(self.overrides) if self: return json.loads(self.overrides) return {} def _get_or_schedule( self, repo_name: str, number: int, *, target: str | None = None, closing: bool = False, commenter: str | None = None, ) -> PullRequests | None: repo = self.env['runbot_merge.repository'].search([('name', '=', repo_name)]) if not repo: source = self.env['runbot_merge.events_sources'].search([('repository', '=', repo_name)]) _logger.warning( "Got a PR notification for unknown repository %s (source %s)", repo_name, source, ) return if target: b = self.env['runbot_merge.branch'].with_context(active_test=False).search([ ('project_id', '=', repo.project_id.id), ('name', '=', target), ]) tmpl = None if b.active \ else 'runbot_merge.handle.branch.inactive' if b\ else 'runbot_merge.pr.fetch.unmanaged' else: tmpl = None pr = self.search([('repository', '=', repo.id), ('number', '=', number)]) if pr and not pr.target.active: tmpl = 'runbot_merge.handle.branch.inactive' target = pr.target.name if tmpl and not closing: self.env.ref(tmpl)._send( repository=repo, pull_request=number, format_args={'repository': repo_name, 'branch': target, 'number': number}, ) if pr: return pr # if the branch is unknown or inactive, no need to fetch the PR if tmpl: return Fetch = self.env['runbot_merge.fetch_job'] if Fetch.search([('repository', '=', repo.id), ('number', '=', number)]): return Fetch.create({ 'repository': repo.id, 'number': number, 'closing': closing, 'commenter': commenter, }) def _iter_ancestors(self) -> Iterator[PullRequests]: while self: yield self self = self.parent_id def _iter_descendants(self) -> Iterator[PullRequests]: pr = self while pr := self.search([('parent_id', '=', pr.id)]): yield pr def _parse_commands(self, author, comment, login): 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') commandlines = self.repository.project_id._find_commands(comment['body'] or '') if not commandlines: _logger.info("found no commands in comment of %s (%s) (%s)", login, name, utils.shorten(comment['body'] or '', 50) ) return 'ok' def feedback(message: Optional[str] = None, close: bool = False): self.env['runbot_merge.pull_requests.feedback'].create({ 'repository': self.repository.id, 'pull_request': self.number, 'message': message, 'close': close, }) is_admin, is_reviewer, is_author = self._pr_acl(author) source_author = self.source_id._pr_acl(author).is_author # nota: 15.0 `has_group` completely doesn't work if the recordset is empty super_admin = is_admin and author.user_ids and author.user_ids.has_group('runbot_merge.group_admin') help_list: list[type(commands.Command)] = list(filter(None, [ commands.Help, (is_reviewer or (self.source_id and source_author)) and not self.reviewed_by and commands.Approve, (is_author or source_author) and self.reviewed_by and commands.Reject, (is_author or source_author) and self.error and commands.Retry, is_author and not self.source_id and commands.FW, is_author and commands.Limit, source_author and self.source_id and commands.Close, is_reviewer and commands.MergeMethod, is_reviewer and commands.Delegate, is_admin and commands.Priority, super_admin and commands.SkipChecks, is_admin and commands.CancelStaging, author.override_rights and commands.Override, is_author and commands.Check, ])) def format_help(warn_ignore: bool, address: bool = True) -> str: s = [ 'Currently available commands{}:'.format( f" for @{login}" if address else "" ), '', '|command||', '|-|-|', ] for command_type in help_list: for cmd, text in command_type.help(is_reviewer): s.append(f"|`{cmd}`|{text}|") s.extend(['', 'Note: this help text is dynamic and will change with the state of the PR.']) if warn_ignore: s.extend(["", "Warning: in invoking help, every other command has been ignored."]) return "\n".join(s) try: cmds: List[commands.Command] = [ ps for line in commandlines for ps in commands.Parser(line.rstrip()) ] except Exception as e: _logger.info( "error %s while parsing comment of %s (%s): %s", e, login, name, utils.shorten(comment['body'] or '', 50), ) feedback(message=f"""@{login} {e.args[0]}. For your own safety I've ignored *everything in your entire comment*. {format_help(False, address=False)} """) return 'error' if any(isinstance(cmd, commands.Help) for cmd in cmds): self.env['runbot_merge.pull_requests.feedback'].create({ 'repository': self.repository.id, 'pull_request': self.number, 'message': format_help(len(cmds) != 1), }) return "help" if not (is_author or self.source_id or (any(isinstance(cmd, commands.Override) for cmd in cmds) and author.override_rights)): # no point even parsing commands _logger.info("ignoring comment of %s (%s): no ACL to %s", login, name, self.display_name) self.env.ref('runbot_merge.command.access.no')._send( repository=self.repository, pull_request=self.number, format_args={'user': login, 'pr': self} ) return 'ignored' rejections = [] for command in cmds: msg = None match command: case commands.Approve() if self.draft: msg = "draft PRs can not be approved." case commands.Approve() if self.source_id and (source_author or is_reviewer): if selected := [p for p in self._iter_ancestors() if p.number in command]: for pr in selected: # ignore already reviewed PRs, unless it's the one # being r+'d, this means ancestors in error will not # be warned about if pr == self or not pr.reviewed_by: pr._approve(author, login) else: msg = f"tried to approve PRs {command.fmt()} but no such PR is an ancestors of {self.number}" case commands.Approve() if is_reviewer: if command.ids is None or command.ids == [self.number]: msg = self._approve(author, login) else: msg = f"tried to approve PRs {command.fmt()} but the current PR is {self.number}" case commands.Reject() if is_author or source_author: if self.batch_id.skipchecks or self.reviewed_by: if self.error: self.error = False if self.reviewed_by: self.reviewed_by = False if self.batch_id.skipchecks: self.batch_id.skipchecks = False self.env.ref("runbot_merge.command.unapprove.p0")._send( repository=self.repository, pull_request=self.number, format_args={'user': login, 'pr': self}, ) if self.source_id.forwardport_ids.filtered( lambda p: p.reviewed_by and p.state not in ('merged', 'closed') ): feedback("Note that only this forward-port has been" " unapproved, sibling forward ports may" " have to be unapproved individually.") self.unstage("unreviewed (r-) by %s", login) else: msg = "r- makes no sense in the current PR state." case commands.MergeMethod() if is_reviewer: self.merge_method = command.value explanation = next(label for value, label in type(self).merge_method.selection if value == command.value) self.env.ref("runbot_merge.command.method")._send( repository=self.repository, pull_request=self.number, format_args={'new_method': explanation, 'pr': self, 'user': login}, ) # if the merge method is the only thing preventing (but not # *blocking*) staging, trigger a staging if self.state == 'ready': self.env.ref("runbot_merge.staging_cron")._trigger() case commands.Retry() if is_author or source_author: if self.error: self.error = False else: msg = "retry makes no sense when the PR is not in error." case commands.Check() if is_author: self.env['runbot_merge.fetch_job'].create({ 'repository': self.repository.id, 'number': self.number, }) case commands.Delegate(users) if is_reviewer: if not users: delegates = self.author else: delegates = self.env['res.partner'] for login in users: delegates |= delegates.search([('github_login', '=', login)]) or delegates.create({ 'name': login, 'github_login': login, }) delegates.write({'delegate_reviewer': [(4, self.id, 0)]}) case commands.Priority() if is_admin: self.batch_id.priority = str(command) case commands.SkipChecks() if super_admin: self.batch_id.skipchecks = True self.reviewed_by = author if not (self.squash or self.merge_method): self.env.ref('runbot_merge.check_linked_prs_status')._trigger() for p in self.batch_id.prs - self: if not p.reviewed_by: p.reviewed_by = author case commands.CancelStaging() if is_admin: self.batch_id.cancel_staging = True if not self.batch_id.blocked: if splits := self.target.split_ids: splits.unlink() self.target.active_staging_id.cancel( "Unstaged by %s on %s", author.github_login, self.display_name, ) case commands.Override(statuses): for status in statuses: overridable = author.override_rights\ .filtered(lambda r: not r.repository_id or (r.repository_id == self.repository))\ .mapped('context') if status in overridable: self.overrides = json.dumps({ **json.loads(self.overrides), status: { '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': '{}'}) else: msg = f"you are not allowed to override {status!r}." # FW case commands.Close() if source_author: feedback(close=True) case commands.FW(): message = None match command: case commands.FW.NO if is_author or source_author: message = "Disabled forward-porting." case commands.FW.DEFAULT if is_author or source_author: message = "Waiting for CI to create followup forward-ports." case commands.FW.SKIPCI if is_reviewer: message = "Not waiting for CI to create followup forward-ports." # case commands.FW.SKIPMERGE if is_reviewer: # message = "Not waiting for merge to create followup forward-ports." case _: msg = f"you don't have the right to {command}." if message: # TODO: feedback? if self.source_id: "if the pr is not a source, ignore (maybe?)" elif not self.merge_date: "if the PR is not merged, it'll be fw'd normally" elif self.batch_id.fw_policy != 'no' or command == commands.FW.NO: "if the policy is not being flipped from no to something else, nothing to do" elif branch_key(self.limit_id) <= branch_key(self.target): "if the limit is lower than current (old style ignore) there's nothing to do" else: message = f"Starting forward-port. {message}" self.env['forwardport.batches'].create({ 'batch_id': self.batch_id.id, 'source': 'merge', }) (self.source_id or self).batch_id.fw_policy = command.name.lower() feedback(message=message) case commands.Limit(branch) if is_author: if branch is None: feedback(message="'ignore' is deprecated, use 'fw=no' to disable forward porting.") limit = branch or self.target.name for p in self.batch_id.prs: ping, m = p._maybe_update_limit(limit) if ping is Ping.ERROR and p == self: msg = m else: if ping: m = f"@{login} {m}" self.env['runbot_merge.pull_requests.feedback'].create({ 'repository': p.repository.id, 'pull_request': p.number, 'message': m, }) case commands.Limit(): msg = "you can't set a forward-port limit." # NO! case _: msg = f"you can't {command}." if msg is not None: rejections.append(msg) cmdstr = ', '.join(map(str, cmds)) if not rejections: _logger.info("%s (%s) applied %s", login, name, cmdstr) self._track_set_author(author, fallback=True) return 'applied ' + cmdstr self.env.cr.rollback() rejections_list = ''.join(f'\n- {r}' for r in rejections) _logger.info("%s (%s) tried to apply %s%s", login, name, cmdstr, rejections_list) footer = '' if len(cmds) == len(rejections) else "\n\nFor your own safety I've ignored everything in your comment." if rejections_list: rejections = ' ' + rejections_list.removeprefix("\n- ") if rejections_list.count('\n- ') == 1 else rejections_list feedback(message=f"@{login}{rejections}{footer}") return 'rejected' def _maybe_update_limit(self, limit: str) -> Tuple[Ping, str]: limit_id = self.env['runbot_merge.branch'].with_context(active_test=False).search([ ('project_id', '=', self.repository.project_id.id), ('name', '=', limit), ]) if not limit_id: return Ping.ERROR, f"there is no branch {limit!r}, it can't be used as a forward port target." if limit_id != self.target and not limit_id.active: return Ping.ERROR, f"branch {limit_id.name!r} is disabled, it can't be used as a forward port target." # not forward ported yet, just acknowledge the request if not self.source_id and self.state != 'merged': self.limit_id = limit_id if branch_key(limit_id) <= branch_key(self.target): return Ping.NO, "Forward-port disabled (via limit)." else: suffix = '' if self.batch_id.fw_policy == 'no': self.batch_id.fw_policy = 'default' suffix = " Re-enabled forward-porting (you should use "\ "`fw=default` to re-enable forward porting "\ "after disabling)." return Ping.NO, f"Forward-porting to {limit_id.name!r}.{suffix}" # if the PR has been forwardported prs = (self | self.forwardport_ids | self.source_id | self.source_id.forwardport_ids) tip = max(prs, key=pr_key) # if the fp tip was closed it's fine if tip.state == 'closed': return Ping.ERROR, f"{tip.display_name} is closed, no forward porting is going on" prs.limit_id = limit_id real_limit = max(limit_id, tip.target, key=branch_key) addendum = '' # check if tip was queued for forward porting, try to cancel if we're # supposed to stop here if real_limit == tip.target and (task := self.env['forwardport.batches'].search([('batch_id', '=', tip.batch_id.id)])): try: with self.env.cr.savepoint(): self.env.cr.execute( "SELECT FROM forwardport_batches " "WHERE id = %s FOR UPDATE NOWAIT", [task.id]) except psycopg2.errors.LockNotAvailable: # row locked = port occurring and probably going to succeed, # so next(real_limit) likely a done deal already return Ping.ERROR, ( f"Forward port of {tip.display_name} likely already " f"ongoing, unable to cancel, close next forward port " f"when it completes.") else: self.env.cr.execute("DELETE FROM forwardport_batches WHERE id = %s", [task.id]) if real_limit != tip.target: # forward porting was previously stopped at tip, and we want it to # resume if tip.state == 'merged': if tip.batch_id.source.fw_policy == 'no': # hack to ping the user but not rollback the transaction return Ping.YES, f"can not forward-port, policy is 'no' on {(tip.source_id or tip).display_name}" self.env['forwardport.batches'].create({ 'batch_id': tip.batch_id.id, 'source': 'fp' if tip.parent_id else 'merge', }) resumed = tip else: resumed = tip.batch_id._schedule_fp_followup() if resumed: addendum += f', resuming forward-port stopped at {tip.display_name}' if real_limit != limit_id: addendum += f' (instead of the requested {limit_id.name!r} because {tip.display_name} already exists)' # get a "stable" root rather than self's to avoid divertences between # PRs across a root divide (where one post-root would point to the root, # and one pre-root would point to the source, or a previous root) root = tip.root_id # reference the root being forward ported unless we are the root root_ref = '' if root == self else f' {root.display_name}' msg = f"Forward-porting{root_ref} to {real_limit.name!r}{addendum}." # send a message to the source & root except for self, if they exist root_msg = f'Forward-porting to {real_limit.name!r} (from {self.display_name}).' self.env['runbot_merge.pull_requests.feedback'].create([ { 'repository': p.repository.id, 'pull_request': p.number, 'message': root_msg, 'token_field': 'fp_github_token', } # send messages to source and root unless root is self (as it # already gets the normal message) for p in (self.source_id | root) - self ]) return Ping.NO, msg def _find_next_target(self) -> Optional[Branch]: """ Finds the branch between target and limit_id which follows reference """ root = (self.source_id or self) if self.target == root.limit_id: return None branches = root.target.project_id.with_context(active_test=False)._forward_port_ordered() if (branch_filter := self.repository.branch_filter) and branch_filter != '[]': branches = branches.filtered_domain(ast.literal_eval(branch_filter)) branches = list(branches) from_ = branches.index(self.target) + 1 to_ = branches.index(root.limit_id) + 1 if root.limit_id else None # return the first active branch in the set return next(( branch for branch in branches[from_:to_] if branch.active ), None) def _approve(self, author, login): oldstate = self.state newstate = RPLUS.get(oldstate) if not author.email: return "I must know your email before you can review PRs. Please contact an administrator." if not newstate: # Don't fail the entire command if someone tries to approve an # already-approved PR. if self.error: msg = "This PR is already reviewed, it's in error, you might want to `retry` it instead " \ "(if you have already confirmed the error is not legitimate)." else: msg = "This PR is already reviewed, reviewing it again is useless." self.env['runbot_merge.pull_requests.feedback'].create({ 'repository': self.repository.id, 'pull_request': self.number, 'message': msg, }) return None self.reviewed_by = author _logger.debug( "r+ on %s by %s (%s->%s) status=%s message? %s", self.display_name, author.github_login, oldstate, newstate, self.status, self.status == 'failure' ) if self.status == 'failure': # the normal infrastructure is for failure and # prefixes messages with "I'm sorry" self.env.ref("runbot_merge.command.approve.failure")._send( repository=self.repository, pull_request=self.number, format_args={'user': login, 'pr': self}, ) if not (self.squash or self.merge_method): self.env.ref('runbot_merge.check_linked_prs_status')._trigger() return None def _pr_acl(self, user) -> ACL: 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 if is_admin: return ACL(True, True, True) # delegate on source = delegate on PR if self.source_id and self.source_id in user.delegate_reviewer: return ACL(False, True, True) # delegate on any ancestors ~ delegate on PR (maybe should be any descendant of source?) if any(p in user.delegate_reviewer for p in self._iter_ancestors()): return ACL(False, True, True) # user is probably always False on a forward port return ACL(False, False, self.author == user) 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 updateable = self.filtered(lambda p: not p.merge_date) updateable.statuses = statuses or '{}' for pr in updateable: if pr.status == "failure": statuses = json.loads(pr.statuses_full) for ci in pr.repository.status_ids._for_pr(pr).mapped('context'): status = statuses.get(ci) or {'state': 'pending'} if status['state'] in ('error', 'failure'): pr._notify_ci_new_failure(ci, status) self.batch_id._schedule_fp_followup() def modified(self, fnames, create=False, before=False): """ By default, Odoo can't express recursive *dependencies* which is exactly what we need for statuses: they depend on the current PR's overrides, and the parent's overrides, and *its* parent's overrides, ... One option would be to create a stored computed field which accumulates the overrides as *fields* can be recursive, but... """ if 'overrides' in fnames: descendants_or_self = self.concat(*self._iter_descendants()) self.env.add_to_compute(self._fields['status'], descendants_or_self) self.env.add_to_compute(self._fields['statuses_full'], descendants_or_self) self.env.add_to_compute(self._fields['state'], descendants_or_self) super().modified(fnames, create, before) @api.depends( 'statuses', 'overrides', 'target', 'parent_id', 'skipchecks', 'repository.status_ids.context', 'repository.status_ids.branch_filter', 'repository.status_ids.prs', ) def _compute_statuses(self): for pr in self: statuses = {**json.loads(pr.statuses), **pr._get_overrides()} pr.statuses_full = json.dumps(statuses, indent=4) if pr.skipchecks: pr.status = 'success' continue st = 'success' for ci in pr.repository.status_ids._for_pr(pr): v = (statuses.get(ci.context) or {'state': 'pending'})['state'] if v in ('error', 'failure'): st = 'failure' break if v == 'pending': st = 'pending' if pr.status != 'failure' and st == 'failure': pr.unstage("had CI failure after staging") pr.status = st @api.depends( "status", "reviewed_by", "closed", "error" , "batch_id.merge_date", "batch_id.skipchecks", ) def _compute_state(self): for pr in self: if pr.closed: pr.state = "closed" elif pr.batch_id.merge_date: pr.state = 'merged' elif pr.error: pr.state = "error" elif pr.batch_id.skipchecks: # skipchecks behaves as both approval and status override pr.state = "ready" else: states = ("opened", "approved", "validated", "ready") pr.state = states[bool(pr.reviewed_by) | ((pr.status == "success") << 1)] def _notify_ci_new_failure(self, ci, st): prev = json.loads(self.previous_failure) 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.ref("runbot_merge.failure.approved")._send( repository=self.repository, pull_request=self.number, format_args={'pr': self, 'status': ci} ) elif self.state == 'opened' and self.parent_id: # only care about FP PRs which are not approved / staged / merged yet self.env.ref('runbot_merge.forwardport.ci.failed')._send( repository=self.repository, pull_request=self.number, token_field='fp_github_token', format_args={'pr': self, 'ci': ci}, ) def _auto_init(self): for field in self._fields.values(): if not isinstance(field, fields.Selection) or field.column_type[0] == 'varchar': continue t = field.column_type[1] self.env.cr.execute("SELECT 1 FROM pg_type WHERE typname = %s", [t]) if not self.env.cr.rowcount: self.env.cr.execute( f"CREATE TYPE {t} AS ENUM %s", [tuple(s for s, _ in field.selection)] ) super()._auto_init() # incorrect index: unique(number, target, repository). tools.drop_index(self.env.cr, 'runbot_merge_unique_pr_per_target', self._table) # correct index: tools.create_unique_index( self.env.cr, 'runbot_merge_unique_pr_per_repo', self._table, ['repository', 'number']) self.env.cr.execute("CREATE INDEX IF NOT EXISTS runbot_merge_pr_head " "ON runbot_merge_pull_requests " "USING hash (head)") def _get_batch(self, *, target, label): batch = self.env['runbot_merge.batch'] if not re.search(r':patch-\d+$', label): batch = batch.search([ ('merge_date', '=', False), ('prs.target', '=', target), ('prs.label', '=', label), ]) return batch or batch.create({}) @api.model_create_multi def create(self, vals_list): batches = {} for vals in vals_list: batch_key = vals['target'], vals['label'] batch = batches.get(batch_key) if batch is None: batch = batches[batch_key] = self._get_batch(target=vals['target'], label=vals['label']) vals['batch_id'] = batch.id if 'limit_id' not in vals: limits = {p.limit_id for p in batch.prs} if len(limits) == 1: vals['limit_id'] = limits.pop().id elif limits: repo = self.env['runbot_merge.repository'].browse(vals['repository']) _logger.warning( "Unable to set limit on %s#%s: found multiple limits in batch (%s)", repo.name, vals['number'], ', '.join( f'{p.display_name} => {p.limit_id.name}' for p in batch.prs ) ) prs = super().create(vals_list) for pr in prs: c = self.env['runbot_merge.commit'].search([('sha', '=', pr.head)]) pr._validate(c.statuses) if pr.state not in ('closed', 'merged'): self.env.ref('runbot_merge.pr.created')._send( repository=pr.repository, pull_request=pr.number, format_args={'pr': pr}, ) return prs 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) return self.env['runbot_merge.pull_requests'].create({ 'closed': description['state'] != 'open', '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': utils.make_message(description), 'draft': description['draft'], }) def write(self, vals): if vals.get('squash'): vals['merge_method'] = False # when explicitly marking a PR as ready if vals.get('state') == 'ready': # skip validation self.batch_id.skipchecks = True # mark current user as reviewer vals.setdefault('reviewed_by', self.env.user.partner_id.id) for p in self.batch_id.prs - self: if not p.reviewed_by: p.reviewed_by = self.env.user.partner_id.id for pr in self: if (t := vals.get('target')) is not None and pr.target.id != t: pr.unstage( "target (base) branch was changed from %r to %r", pr.target.display_name, self.env['runbot_merge.branch'].browse(t).display_name, ) if 'message' in vals: merge_method = vals['merge_method'] if 'merge_method' in vals else pr.merge_method if merge_method not in (False, 'rebase-ff') and pr.message != vals['message']: pr.unstage("merge message updated") match vals.get('closed'): case True if not self.closed: vals['reviewed_by'] = False case False if self.closed and not self.batch_id: vals['batch_id'] = self._get_batch( target=vals.get('target') or self.target.id, label=vals.get('label') or self.label, ) w = super().write(vals) newhead = vals.get('head') if newhead: authors = self.env.cr.precommit.data.get(f'mail.tracking.author.{self._name}', {}) for p in self: if not (writer := authors.get(p.id)): writer = self.env.user.partner_id p.unstage("updated by %s", writer.github_login or writer.name) # this can be batched c = self.env['runbot_merge.commit'].search([('sha', '=', newhead)]) self._validate(c.statuses) 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') """) 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.ref('runbot_merge.pr.linked.not_ready')._send( repository=r.repository, pull_request=r.number, format_args={ 'pr': r, 'siblings': ', '.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', 'in', ("approved", "ready")), ('staging_id', '=', False), ('squash', '=', False), ('merge_method', '=', False), ('method_warned', '=', False), ]): self.env.ref('runbot_merge.pr.merge_method')._send( repository=r.repository, pull_request=r.number, format_args={'pr': r, 'methods':methods}, ) r.method_warned = True if commit: self.env.cr.commit() def _build_message(self, message: Union['PullRequests', str], related_prs: 'PullRequests' = (), merge: bool = True) -> 'Message': # handle co-authored commits (https://help.github.com/articles/creating-a-commit-with-multiple-authors/) m = Message.from_message(message) if not is_mentioned(message, self): if merge: m.body += f'\n\ncloses {self.display_name}' else: m.headers.pop('Part-Of', None) m.headers.add('Part-Of', self.display_name) for r in related_prs: if not is_mentioned(message, r, full_reference=True): m.headers.add('Related', r.display_name) # ensures all reviewers in the review path are on the PR in order: # original reviewer, then last conflict reviewer, then current PR reviewers = (self | self.root_id | self.source_id)\ .mapped('reviewed_by.formatted_email') sobs = m.headers.getlist('signed-off-by') m.headers.remove('signed-off-by') m.headers.extend( ('signed-off-by', signer) for signer in sobs if signer not in reviewers ) m.headers.extend( ('signed-off-by', reviewer) for reviewer in reversed(reviewers) ) return m 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: Split = self.batch_id.split_id if split: if split.source_id.likely_false_positive: split.source_id.likely_false_positive = False split.source_id.message_post( body=f"Assuming failure is a true positive due to {self.display_name} being removed from split.", ) if len(split.batch_ids) == 1: # only the batch of this PR -> delete split split.unlink() else: # else remove this batch from the split self.batch_id.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.flush_recordset(['state', 'batch_id']) self.env.cr.execute(''' SELECT batch_id FROM runbot_merge_pull_requests WHERE id = %s AND state != 'merged' AND state != 'closed' FOR UPDATE SKIP LOCKED; ''', [self.id]) if not self.env.cr.rowcount: return False self.unstage("closed by %s", by) self.with_context(forwardport_detach_warn=False).write({ 'closed': True, 'reviewed_by': False, 'parent_id': False, 'detach_reason': f"Closed by {by}", }) self.search([('parent_id', '=', self.id)]).write({ 'parent_id': False, 'detach_reason': f"{by} closed parent PR {self.display_name}", }) return True def _fp_conflict_feedback(self, previous_pr, conflicts): (h, out, err, hh) = conflicts.get(previous_pr) or (None, None, None, None) if h: sout = serr = '' if out.strip(): sout = f"\nstdout:\n```\n{out}\n```\n" if err.strip(): serr = f"\nstderr:\n```\n{err}\n```\n" lines = '' if len(hh) > 1: lines = '\n' + ''.join( '* %s%s\n' % (sha, ' <- on this commit' if sha == h else '') for sha in hh ) template = 'runbot_merge.forwardport.failure' format_args = { 'pr': self, 'commits': lines, 'stdout': sout, 'stderr': serr, 'footer': FOOTER, } elif any(conflicts.values()): template = 'runbot_merge.forwardport.linked' format_args = { 'pr': self, 'siblings': ', '.join(p.display_name for p in (self.batch_id.prs - self)), 'footer': FOOTER, } elif not self._find_next_target(): ancestors = "".join( f"* {p.display_name}\n" for p in previous_pr._iter_ancestors() if p.parent_id if p.state not in ('closed', 'merged') if p.target.active ) template = 'runbot_merge.forwardport.final' format_args = { 'pr': self, 'containing': ' containing:' if ancestors else '.', 'ancestors': ancestors, 'footer': FOOTER, } else: template = 'runbot_merge.forwardport.intermediate' format_args = { 'pr': self, 'footer': FOOTER, } self.env.ref(template)._send( repository=self.repository, pull_request=self.number, token_field='fp_github_token', format_args=format_args, ) def button_split(self): if len(self.batch_id.prs) == 1: raise UserError("Splitting a batch with a single PR is dumb") w = self.env['runbot_merge.pull_requests.split_off'].create({ 'pr_id': self.id, 'new_label': self.label, }) return { 'type': 'ir.actions.act_window', 'res_model': w._name, 'res_id': w.id, 'target': 'new', 'views': [(False, 'form')], } def _split_off(self, new_label): # should not be usable to move a PR between batches (maybe later) batch = self.env['runbot_merge.batch'] if not re.search(r':patch-\d+$', new_label): if batch.search([ ('merge_date', '=', False), ('prs.label', '=', new_label), ]): raise UserError("Can not split off to an existing batch") self.write({ 'label': new_label, 'batch_id': batch.create({}).id, }) class Ping(IntEnum): NO = 0 YES = 1 ERROR = 2 # ordering is a bit unintuitive because the lowest sequence (and name) # is the last link of the fp chain, reasoning is a bit more natural the # other way around (highest object is the last), especially with Python # not really having lazy sorts in the stdlib def branch_key(b: Branch, /, _key=itemgetter('sequence', 'name')): return Reverse(_key(b)) def pr_key(p: PullRequests, /): return branch_key(p.target) # state changes on reviews RPLUS = { 'opened': 'approved', 'validated': 'ready', } _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='[]') @api.model_create_multi def create(self, vals_list): for values in vals_list: 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'])) if any(vals_list): self.env.ref('runbot_merge.labels_cron')._trigger() return super().create(vals_list) 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.info( "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, index=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, index=True) 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" ) @api.model_create_multi def create(self, vals_list): # any time a feedback is created, it can be sent self.env.ref('runbot_merge.feedback_cron')._trigger() return super().create(vals_list) 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 FeedbackTemplate(models.Model): _name = 'runbot_merge.pull_requests.feedback.template' _description = "str.format templates for feedback messages, no integration," \ "but that's their purpose" _inherit = ['mail.thread'] template = fields.Text(tracking=True) help = fields.Text(readonly=True) def _format(self, **args): return self.template.format_map(args) def _send(self, *, repository: Repository, pull_request: int, format_args: dict, token_field: Optional[str] = None) -> Optional[Feedback]: try: feedback = { 'repository': repository.id, 'pull_request': pull_request, 'message': self.template.format_map(format_args), } if token_field: feedback['token_field'] = token_field return self.env['runbot_merge.pull_requests.feedback'].create(feedback) except Exception: _logger.exception("Failed to render template %s", self.get_external_id()) raise class StagingCommits(models.Model): _name = 'runbot_merge.stagings.commits' _description = "Mergeable commits for stagings, always the actually merged " \ "commit, never a uniquifier" _log_access = False staging_id = fields.Many2one('runbot_merge.stagings', required=True) commit_id = fields.Many2one('runbot_merge.commit', index=True, required=True) repository_id = fields.Many2one('runbot_merge.repository', required=True) def _auto_init(self): super()._auto_init() # the same commit can be both head and tip (?) tools.create_unique_index( self.env.cr, self._table + "_unique", self._table, ['staging_id', 'commit_id'] ) # there should be one head per staging per repository, unless one is a # real head and one is a uniquifier head tools.create_unique_index( self.env.cr, self._table + "_unique_per_repo", self._table, ['staging_id', 'repository_id'], ) class StagingHeads(models.Model): _name = 'runbot_merge.stagings.heads' _description = "Staging heads, may be the staging's commit or may be a " \ "uniquifier (discarded on success)" _log_access = False staging_id = fields.Many2one('runbot_merge.stagings', required=True) commit_id = fields.Many2one('runbot_merge.commit', index=True, required=True) repository_id = fields.Many2one('runbot_merge.repository', required=True) def _auto_init(self): super()._auto_init() # the same commit can be both head and tip (?) tools.create_unique_index( self.env.cr, self._table + "_unique", self._table, ['staging_id', 'commit_id'] ) # there should be one head per staging per repository, unless one is a # real head and one is a uniquifier head tools.create_unique_index( self.env.cr, self._table + "_unique_per_repo", self._table, ['staging_id', 'repository_id'], ) 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' _rec_name = 'sha' sha = fields.Char(required=True) statuses = fields.Char(help="json-encoded mapping of status contexts to states", default="{}") to_check = fields.Boolean(default=False) head_ids = fields.Many2many('runbot_merge.stagings', relation='runbot_merge_stagings_heads', column2='staging_id', column1='commit_id') commit_ids = fields.Many2many('runbot_merge.stagings', relation='runbot_merge_stagings_commits', column2='staging_id', column1='commit_id') pull_requests = fields.One2many('runbot_merge.pull_requests', compute='_compute_prs') @api.model_create_multi def create(self, vals_list): for values in vals_list: values['to_check'] = True r = super().create(vals_list) self.env.ref("runbot_merge.process_updated_commits")._trigger() return r def write(self, values): values.setdefault('to_check', True) r = super().write(values) if values['to_check']: self.env.ref("runbot_merge.process_updated_commits")._trigger() return r @mute_logger('odoo.sql_db') def _notify(self): Stagings = self.env['runbot_merge.stagings'] PRs = self.env['runbot_merge.pull_requests'] serialization_failures = False # chances are low that we'll have more than one commit for c in self.search([('to_check', '=', True)]): sha = c.sha pr = PRs.search([('head', '=', sha)]) stagings = Stagings.search([ ('head_ids.sha', '=', sha), ('state', '=', 'pending'), ('target.project_id.staging_statuses', '=', True), ]) try: c.to_check = False c.flush_recordset(['to_check']) if pr: pr._validate(c.statuses) pr._track_set_log_message(html_escape(f"statuses changed on {sha}")) if stagings: stagings._notify(c) except psycopg2.errors.SerializationFailure: serialization_failures = True _logger.info("Failed to apply commit %s: serialization failure", sha) self.env.cr.rollback() except Exception: _logger.exception("Failed to apply commit %s", sha) self.env.cr.rollback() else: self.env.cr.commit() if serialization_failures: self.env.ref("runbot_merge.process_updated_commits")._trigger() _sql_constraints = [ ('unique_sha', 'unique (sha)', 'no duplicated commit'), ] def _auto_init(self): res = super()._auto_init() self.env.cr.execute(""" CREATE INDEX IF NOT EXISTS runbot_merge_unique_statuses ON runbot_merge_commit USING hash (sha) """) self.env.cr.execute(""" CREATE INDEX IF NOT EXISTS runbot_merge_to_process ON runbot_merge_commit ((1)) WHERE to_check """) return res def _compute_prs(self): for c in self: c.pull_requests = self.env['runbot_merge.pull_requests'].search([ ('head', '=', c.sha), ]) class Stagings(models.Model): _name = 'runbot_merge.stagings' _description = "A set of batches being tested for integration" _inherit = ['mail.thread'] target = fields.Many2one('runbot_merge.branch', required=True, index=True) parent_id = fields.Many2one('runbot_merge.stagings') child_ids = fields.One2many('runbot_merge.stagings', 'parent_id') likely_false_positive = fields.Boolean(default=False) staging_batch_ids = fields.One2many('runbot_merge.staging.batch', 'runbot_merge_stagings_id') batch_ids = fields.Many2many( 'runbot_merge.batch', context={'active_test': False}, compute="_compute_batch_ids", search="_search_batch_ids", ) pr_ids = fields.One2many('runbot_merge.pull_requests', compute='_compute_prs') state = fields.Selection([ ('success', 'Success'), ('failure', 'Failure'), ('pending', 'Pending'), ('cancelled', "Cancelled"), ('ff_failed', "Fast forward failed") ], default='pending', index=True, store=True, compute='_compute_state') active = fields.Boolean(default=True) staged_at = fields.Datetime(default=fields.Datetime.now, index=True) staging_end = fields.Datetime(store=True) staging_duration = fields.Float(compute='_compute_duration') timeout_limit = fields.Datetime(store=True, compute='_compute_timeout_limit') reason = fields.Text("Reason for final state (if any)") head_ids = fields.Many2many('runbot_merge.commit', relation='runbot_merge_stagings_heads', column1='staging_id', column2='commit_id') heads = fields.One2many('runbot_merge.stagings.heads', 'staging_id') commit_ids = fields.Many2many('runbot_merge.commit', relation='runbot_merge_stagings_commits', column1='staging_id', column2='commit_id') commits = fields.One2many('runbot_merge.stagings.commits', 'staging_id') statuses = fields.Binary(compute='_compute_statuses') statuses_cache = fields.Text(default='{}', required=True) issues_to_close = fields.Json(default=lambda _: [], help="list of tasks to close if this staging succeeds") @api.depends('staged_at', 'staging_end') def _compute_duration(self): for s in self: s.staging_duration = ((s.staging_end or fields.Datetime.now()) - s.staged_at).total_seconds() @api.depends('target.name', 'state', 'reason') def _compute_display_name(self): for staging in self: staging.display_name = "%d (%s, %s%s)" % ( staging.id, staging.target.name, staging.state, (', ' + staging.reason) if staging.reason else '', ) @api.depends('staging_batch_ids.runbot_merge_batch_id') def _compute_batch_ids(self): for staging in self: staging.batch_ids = staging.staging_batch_ids.runbot_merge_batch_id def _search_batch_ids(self, operator, value): return [('staging_batch_ids.runbot_merge_batch_id', operator, value)] @api.depends('heads', 'statuses_cache') def _compute_statuses(self): """ Fetches statuses associated with the various heads, returned as (repo, context, state, url) """ heads = {h.commit_id: h.repository_id for h in self.mapped('heads')} all_heads = self.mapped('head_ids') for st in self: statuses = json.loads(st.statuses_cache) commits = st.head_ids.with_prefetch(all_heads._prefetch_ids) st.statuses = [ ( heads[commit].name, context, status.get('state') or 'pending', status.get('target_url') or '' ) for commit in commits for context, status in statuses.get(commit.sha, {}).items() ] def write(self, vals): if timeout := vals.get('timeout_limit'): self.env.ref("runbot_merge.merge_cron")\ ._trigger(fields.Datetime.to_datetime(timeout)) if vals.get('active') is False: vals['staging_end'] = fields.Datetime.now() self.env.ref("runbot_merge.staging_cron")._trigger() return super().write(vals) # 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): timeouts = set() for st in self: t = st.timeout_limit = st.staged_at + datetime.timedelta(minutes=st.target.project_id.ci_timeout) timeouts.add(t) if timeouts: # we might have very different limits for each staging so need to schedule them all self.env.ref("runbot_merge.merge_cron")._trigger_list(timeouts) @api.depends('batch_ids.prs') def _compute_prs(self): for staging in self: staging.pr_ids = staging.batch_ids.prs def _notify(self, c: Commit) -> None: self.env.cr.execute(""" UPDATE runbot_merge_stagings SET statuses_cache = CASE WHEN statuses_cache::jsonb->%(sha)s IS NULL THEN jsonb_insert(statuses_cache::jsonb, ARRAY[%(sha)s], %(statuses)s::jsonb) ELSE statuses_cache::jsonb || jsonb_build_object(%(sha)s, %(statuses)s::jsonb) END::text WHERE id = any(%(ids)s) """, {'sha': c.sha, 'statuses': c.statuses, 'ids': self.ids}) self.modified(['statuses_cache']) def post_status(self, sha, context, status, *, target_url=None, description=None): if not self.env.user.has_group('runbot_merge.status'): raise AccessError("You are not allowed to post a status.") now = datetime.datetime.now().isoformat(timespec='seconds') for s in self: if not s.target.project_id.staging_rpc: continue if not any(c.commit_id.sha == sha for c in s.commits): raise ValueError(f"Staging {s.id} does not have the commit {sha}") st = json.loads(s.statuses_cache) st.setdefault(sha, {})[context] = { 'state': status, 'target_url': target_url, 'description': description, 'updated_at': now, } s.statuses_cache = json.dumps(st) return True @api.depends( "statuses_cache", "target", "heads.commit_id.sha", "heads.repository_id.status_ids.branch_filter", "heads.repository_id.status_ids.context", ) def _compute_state(self): for staging in self: if staging.state != 'pending': continue # maps commits to the statuses they need required_statuses = [ (h.commit_id.sha, h.repository_id.status_ids._for_staging(staging).mapped('context')) for h in staging.heads ] cmap = json.loads(staging.statuses_cache) last_pending = "" state = 'success' for head, reqs in required_statuses: statuses = cmap.get(head) or {} for status in (statuses.get(n, {}) for n in reqs): v = status.get('state') if state == 'failure' or v in ('error', 'failure'): state = 'failure' elif v is None: state = 'pending' elif v == 'pending': state = 'pending' last_pending = max(last_pending, status.get('updated_at', '')) else: assert v == 'success' staging.state = state if staging.state != 'pending': self.env.ref("runbot_merge.merge_cron")._trigger() if last_pending: timeout = datetime.datetime.fromisoformat(last_pending) \ + datetime.timedelta(minutes=staging.target.project_id.ci_timeout) if timeout > staging.timeout_limit: staging.timeout_limit = timeout self.env.ref("runbot_merge.merge_cron")._trigger(timeout) _logger.debug("%s got pending status, bumping timeout to %s", staging, timeout) 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 False _logger.info("Cancelling staging %s: " + reason, self, *args) self.parent_id.likely_false_positive = False self.write({ 'active': False, 'state': 'cancelled', 'reason': reason % args, }) return True def fail(self, message, prs=None): _logger.info("Staging %s failed: %s", self, message) prs = prs or self.batch_ids.prs prs._track_set_log_message(f'staging {self.id} failed: {message}') prs.error = True for pr in prs: self.env.ref('runbot_merge.pr.staging.fail')._send( repository=pr.repository, pull_request=pr.number, format_args={'pr': pr, 'message': message}, ) if not self.target.split_ids: self.parent_id.likely_false_positive = False self.write({ 'active': False, 'state': 'failure', 'reason': message, }) return True 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, st = self.env['runbot_merge.split'].create([{ 'target': self.target.id, 'source_id': (self.parent_id or self).id, 'batch_ids': [Command.link(batch.id) for batch in h], }, { 'target': self.target.id, 'source_id': (self.parent_id or self).id, 'batch_ids': [Command.link(batch.id) for batch in t], }]) _logger.info("Split %s to %s (%s) and %s (%s)", self, h, sh, t, st) self.write({ 'active': False, 'state': 'failure', 'reason': self.reason if self.state == 'failure' else 'timed out', 'likely_false_positive': not self.parent_id, }) return True # single batch => the staging is an irredeemable 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 staging_statuses = json.loads(self.statuses_cache) # try inferring which PR failed and only mark that one for head in self.heads: required_statuses = set(head.repository_id.status_ids._for_staging(self).mapped('context')) statuses = staging_statuses.get(head.commit_id.sha, {}) reason = next(( ctx for ctx, result in statuses.items() if ctx in required_statuses if result.get('state') in ('error', 'failure') ), None) if not reason: continue pr = next((pr for pr in self.batch_ids.prs if pr.repository == head.repository_id), None) 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.commit_id.sha, 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)} 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: with sentry_sdk.start_span(description="merge staging") as span: span.set_tag("staging", self.id) span.set_tag("branch", self.target.name) self._safety_dance(gh, self.commits) except exceptions.FastForwardError as e: logger.warning( "Could not fast-forward successful staging on %s:%s: %s", e.args[0], self.target.name, e, ) self.write({ 'state': 'ff_failed', 'reason': str(e.__cause__ or e.__context__ or e) }) else: prs = self.mapped('batch_ids.prs') prs._track_set_log_message(f'staging {self.id} succeeded') logger.info( "%s FF successful, marking %s as merged", self, prs.mapped('display_name'), ) self.batch_ids.merge_date = fields.Datetime.now() 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]), }) if self.issues_to_close: self.env['runbot_merge.issues_closer'].create(self.issues_to_close) self.env.ref('runbot_merge.issues_closer_cron')._trigger() finally: 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_commits: StagingCommits): """ 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. """ tmp_target = 'tmp.' + self.target.name # first force-push the current targets to all tmps for repo_name in staging_commits.mapped('repository_id.name'): g = gh[repo_name] g.set_ref(tmp_target, g.head(self.target.name)) # then attempt to FF the tmp to the staging commits for c in staging_commits: gh[c.repository_id.name].fast_forward(tmp_target, c.commit_id.sha) # there is still a race condition here, but it's way # lower than "the entire staging duration"... for i, c in enumerate(staging_commits): for pause in [0.1, 0.3, 0.5, 0.9, 0]: # last one must be 0/falsy of we lose the exception try: gh[c.repository_id.name].fast_forward( self.target.name, c.commit_id.sha ) except exceptions.FastForwardError: if i and pause: time.sleep(pause) continue raise else: break @api.returns('runbot_merge.stagings') def for_heads(self, *heads): """Returns the staging(s) with all the specified heads. Heads should be unique git oids. """ if not heads: return self.browse(()) joins = ''.join( f'\nJOIN runbot_merge_stagings_heads h{i} ON h{i}.staging_id = s.id' f'\nJOIN runbot_merge_commit c{i} ON c{i}.id = h{i}.commit_id AND c{i}.sha = %s\n' for i in range(len(heads)) ) self.env.cr.execute(f"SELECT s.id FROM runbot_merge_stagings s {joins}", heads) stagings = self.browse(id for [id] in self.env.cr.fetchall()) stagings.check_access_rights('read') stagings.check_access_rule('read') return stagings @api.returns('runbot_merge.stagings') def for_commits(self, *heads): """Returns the staging(s) with all the specified commits (heads which have actually been merged). Commits should be unique git oids. """ if not heads: return self.browse(()) joins = ''.join( f'\nJOIN runbot_merge_stagings_commits h{i} ON h{i}.staging_id = s.id' f'\nJOIN runbot_merge_commit c{i} ON c{i}.id = h{i}.commit_id AND c{i}.sha = %s\n' for i in range(len(heads)) ) self.env.cr.execute(f"SELECT s.id FROM runbot_merge_stagings s {joins}", heads) stagings = self.browse(id for [id] in self.env.cr.fetchall()) stagings.check_access_rights('read') stagings.check_access_rule('read') return stagings class Split(models.Model): _name = _description = 'runbot_merge.split' target = fields.Many2one('runbot_merge.branch', required=True) source_id = fields.Many2one('runbot_merge.stagings', required=True) batch_ids = fields.One2many('runbot_merge.batch', 'split_id', context={'active_test': False}) def unlink(self): if not self.env.context.get('staging_split'): self.source_id.likely_false_positive = False return super().unlink() 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) closing = fields.Boolean(default=False) commits_at = fields.Datetime(index="btree_not_null") commenter = fields.Char() @api.model_create_multi def create(self, vals_list): now = fields.Datetime.now() self.env.ref('runbot_merge.fetch_prs_cron')._trigger({ fields.Datetime.to_datetime( vs.get('commits_at') or now ) for vs in vals_list }) return super().create(vals_list) def _check(self, commit=False): """ :param bool commit: commit after each fetch has been executed """ now = getattr(builtins, 'current_date', None) or fields.Datetime.to_string(datetime.datetime.now()) while True: f = self.search([ '|', ('commits_at', '=', False), ('commits_at', '<=', now) ], limit=1) if not f: return f.active = False self.env.cr.execute("SAVEPOINT runbot_merge_before_fetch") try: f.repository._load_pr( f.number, closing=f.closing, squash=bool(f.commits_at), ping=f.commenter and f'@{f.commenter} ', ) 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") if commit: self.env.cr.commit() from .stagings_create import is_mentioned, Message