apertis-pkg-merge-upstream-to-downstreams

#!/usr/bin/env python3
# SPDX-License-Identifier: MPL-2.0
#
# Copyright © 2019 Collabora Ltd
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import argparse
from sh.contrib import git
from sh import apertis_pkg_merge_updates
import re
from pathlib import Path

def parse_ref(ref: str) -> str:
    return git('rev-parse', '-q', '--verify', ref + '^{commit}', _ok_code=[0, 1]).strip('\n')

def ensure_branch(name: str, fallbacks: list) -> str:
    if parse_ref(f'origin/{name}'):
        return name
    for b in fallbacks:
        commit = parse_ref(f'origin/{b}')
        if commit:
            print(f'Setting branch {name} to point to {b} ({commit:.7})')
            git('update-ref', f'refs/heads/{name}', commit)
            return name
    print(f"Branch {name} not set")
    return None

def existing_upstream_branches(upstream: str):
    branches = set((
        f'{upstream}-proposed-updates',
        f'{upstream}-security',
        f'{upstream}',
    ))
    for branch in list(branches):
        if parse_ref(f"origin/{branch}"):
            git.branch('--track', '-f', branch, f"origin/{branch}", _fg=True)
        if not parse_ref(branch):
            branches.remove(branch)
    return branches

def get_matching_downstream_branch(downstream: str, upstream: str):
    if re.search(r'(dev[0-9]|pre)$', downstream):
        return downstream
    else:
        if upstream.endswith('-security'):
            return ensure_branch(f'{downstream}-security', [f'{downstream}-updates', downstream])
        else:
            return ensure_branch(f'{downstream}-updates', [f'{downstream}-security', downstream])

def push_merge_request(project_url, proposed_branch, upstream, downstream):
  title = f"Update from {upstream} for {downstream}"
  print(f"Pushing: {title}")
  git.push("--force",
    "-o", "merge_request.create",
    "-o", "merge_request.remove_source_branch",
    "-o", f"merge_request.target={downstream}",
    "-o", f"merge_request.title={title}",
    project_url,
    f"{proposed_branch}:{proposed_branch}", _fg=True)

def main():
  parser = argparse.ArgumentParser(description='Merge updates from the upstream repositories to the derivative branches')
  parser.add_argument('--upstream', dest='upstream', type=str,
                      required=True, help='the upstream branch (e.g. debian/buster)')
  parser.add_argument('--downstreams', dest='downstreams', type=str,
                      required=True, help='downstream branches, colon seperate (e.g.  apertis/v2020dev0:apertis/v2019)')
  parser.add_argument('--local-version-suffix', dest="local_suffix", type=str, default="co", help='the local version suffix to be used in new changelog entries')
  parser.add_argument('project_url', type=str, help='git project url to push updates to')
  args = parser.parse_args()

  upstream = args.upstream
  downstreams = args.downstreams.split(":")
  local_suffix = args.local_suffix

  for upstream_branch in existing_upstream_branches(upstream):
   # Mapping of downstream reference to proposed updates
   known_refs = {}
   for downstream in downstreams:
    downstream_branch = get_matching_downstream_branch(downstream, upstream_branch)

    # Skip branch not set
    if not downstream_branch:
        continue

    print(f"Looking at {downstream_branch} <- {upstream_branch}")
    git.checkout(downstream_branch)
    ref = git("rev-parse", "HEAD").rstrip()
    known = known_refs.get(ref)

    if known == None:
      print("New target, doing merge")
      proposed_branch = f"proposed-updates/{upstream_branch}/{ref[0:8]}"
      git.checkout("-B", proposed_branch)
      apertis_pkg_merge_updates(f"--downstream={downstream_branch}", f"--upstream={upstream_branch}", "--local-version-suffix={local_suffix}", _fg=True)
      o = git('diff', '--quiet', f"HEAD..{downstream_branch}", _ok_code=[0,1])
      if o.exit_code == 1:
        print("Merge done, pushing")
        push_merge_request(args.project_url, proposed_branch, upstream_branch, downstream_branch)
        known_refs[ref] = proposed_branch
      else:
        print("No merge required, skipping")
        known_refs[ref] = ""
    else:
      if known == "":
        print("Known target ref, Nothing required, skipping")
        continue

      # Check out known pre-pushed merge, and push that as an MR to the target
      print("Known target, pushing merge request")
      push_merge_request(args.project_url, known, upstream_branch, downstream_branch)

if __name__ == '__main__':
  main()