path: root/tools/dist/backport/merger.py

#!/usr/bin/env python3

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

"""
backport.merger - library for running STATUS merges
"""

import backport.status

import contextlib
import functools
import logging
import os
import re
import subprocess
import sys
import tempfile
import time
import unittest

logger = logging.getLogger(__name__)

# The 'svn' binary
SVN = os.getenv('SVN', 'svn')
# TODO: maybe run 'svn info' to check if it works / fail early?


class UnableToMergeException(Exception):
  pass


def invoke_svn(argv):
  "Run svn with ARGV as argv[1:].  Return (exit_code, stdout, stderr)."
  # TODO(interactive mode): disable --non-interactive
  child_env = os.environ.copy()
  child_env.update({'LC_ALL': 'C'})
  argv = [SVN, '--non-interactive', '--config-option=config:miscellany:log-encoding=UTF-8'] + argv
  child = subprocess.Popen(argv,
                           stdout=subprocess.PIPE,
                           stderr=subprocess.PIPE,
                           env=child_env)
  stdout, stderr = child.communicate()
  return child.returncode, stdout.decode('UTF-8'), stderr.decode('UTF-8')

def run_svn(argv, expected_stderr=None):
  """Run svn with ARGV as argv[1:].  If EXPECTED_STDERR is None, raise if the
  exit code is non-zero or stderr is non-empty.  Else, treat EXPECTED_STDERR as
  a regexp, and ignore an errorful exit or stderr messages if the latter match
  the regexp.  Return exit_code, stdout, stderr."""

  exit_code, stdout, stderr = invoke_svn(argv)
  if exit_code == 0 and not stderr:
    return exit_code, stdout, stderr
  elif expected_stderr and re.compile(expected_stderr).search(stderr):
    return exit_code, stdout, stderr
  else:
    logger.warning("Unexpected stderr: %r", stderr)
    # TODO: pass stdout/stderr to caller?
    raise subprocess.CalledProcessError(returncode=exit_code,
                                        cmd=[SVN] + argv)

def run_svn_quiet(argv, *args, **kwargs):
  "Wrapper for run_svn(-q)."
  return run_svn(['-q'] + argv, *args, **kwargs)

class Test_invoking_cmdline_client(unittest.TestCase):
  def test_run_svn(self):
    _, stdout, _ = run_svn(['--version', '-q'])
    self.assertRegex(stdout, r'^1\.[0-9]+\.[0-9]+')

    run_svn(['--version', '--no-such-option'], "invalid option")

    with self.assertLogs() as cm:
      with self.assertRaises(subprocess.CalledProcessError):
        run_svn(['--version', '--no-such-option'])
      self.assertRegex(cm.output[0], "Unexpected stderr.*")

  def test_svn_version(self):
    self.assertGreaterEqual(svn_version(), (1, 0))


@functools.lru_cache(maxsize=1)
def svn_version():
  "Return the version number of the 'svn' binary as a (major, minor) tuple."
  _, stdout, _ = run_svn(['--version', '-q'])
  match = re.compile(r'(\d+)\.(\d+)').match(stdout)
  assert match
  return tuple(map(int, match.groups()))

def run_revert():
  return run_svn(['revert', '-q', '-R', './'])

def last_changed_revision(path_or_url):
  "Return the 'Last Changed Rev:' of PATH_OR_URL."

  if svn_version() >= (1, 9):
    return int(run_svn(['info', '--show-item=last-changed-revision', '--',
                        path_or_url])[1])
  else:
    _, lines, _ = run_svn(['info', '--', path_or_url]).splitlines()
    for line in lines:
      if line.startswith('Last Changed Rev:'):
        return int(line.split(':', 1)[1])
    else:
      raise Exception("'svn info' did not print last changed revision")

def no_local_mods(path):
  "Check PATH for local mods.  Raise if there are any."
  if run_svn(['status', '-q', '--', path])[1]:
    raise UnableToMergeException("Local mods on {!r}".format(path))

def _includes_only_svn_mergeinfo_changes(status_output):
  """Return TRUE iff there is exactly one local mod, and it is an svn:mergeinfo
  change.  Use the provided `status -q` output."""

  if len(status_output.splitlines()) != 1:
    return False

  _, diff_output, _ = run_svn(['diff'])

  pattern = re.compile(r'^(Added|Modified|Deleted): ')
  targets = (line.split(':', 1)[1].strip()
             for line in diff_output.splitlines()
               if pattern.match(line))
  if set(targets) == {'svn:mergeinfo'}:
    return True

  return False


@contextlib.contextmanager
def log_message_file(logmsg):
  "Context manager that returns a file containing the text LOGMSG."
  with tempfile.NamedTemporaryFile(mode='w+', encoding="UTF-8") as logmsg_file:
    logmsg_file.write(logmsg)
    logmsg_file.flush()
    yield logmsg_file.name

def merge(entry, expected_stderr=None, *, commit=False):
  """Merges ENTRY into the working copy at cwd.

  Do not commit the result, unless COMMIT is true.  When committing,
  remove ENTRY from its STATUS file prior to committing.

  EXPECTED_STDERR will be passed to run_svn() for the actual 'merge' command."""

  assert isinstance(entry, backport.status.StatusEntry)
  assert entry.valid()
  assert entry.status_file

  sf = entry.status_file

  # TODO(interactive mode): catch the exception
  validate_branch_contains_named_revisions(entry)

  # Prepare mergeargs and logmsg.
  logmsg = ""
  if entry.branch:
    branch_url = sf.branch_url(entry.branch)
    if svn_version() >= (1, 8):
      mergeargs = ['--', branch_url]
      logmsg = "Merge {}:\n".format(entry.noun())
      reintegrated_word = "merged"
    else:
      mergeargs = ['--reintegrate', '--', branch_url]
      logmsg = "Reintegrate {}:\n".format(entry.noun())
      reintegrated_word = "reintegrated"
    logmsg += "\n"
  elif entry.revisions:
    mergeargs = []
    if entry.accept:
      mergeargs.append('--accept=%s' % (entry.accept,))
      logmsg += "Merge {} from trunk, with --accept={}:\n".\
                    format(entry.noun(), entry.accept)
    else:
      logmsg += "Merge {} from trunk:\n".format(entry.noun())
    logmsg += "\n"
    mergeargs.extend('-c' + str(revision) for revision in entry.revisions)
    mergeargs.extend(['--', sf.trunk_url()])
  logmsg += entry.raw

  no_local_mods('.')

  # TODO: use select() to restore interweaving of stdout/stderr
  _, stdout, stderr = run_svn_quiet(['merge'] + mergeargs, expected_stderr)
  sys.stdout.write(stdout)
  sys.stderr.write(stderr)

  _, stdout, _ = run_svn(['status', '-q'])
  if _includes_only_svn_mergeinfo_changes(stdout):
    raise UnableToMergeException("Entry %s includes only svn:mergeinfo changes"
                                 % entry)

  if commit:
    sf.remove(entry)
    sf.unparse(open('./STATUS', 'w'))

    # HACK to make backport_tests pass - the tests should be changed!
    s = open('./STATUS').read()
    if s.endswith('\n\n'):
      s = s[:-1]
    open('./STATUS', 'w').write(s)

    # Don't assume we can pass UTF-8 in argv.
    with log_message_file(logmsg) as logmsg_filename:
      run_svn_quiet(['commit', '-F', logmsg_filename])

  # TODO(interactive mode): add the 'svn status' display

  if entry.branch:
    revnum = last_changed_revision('./STATUS')

    if commit:
      # Sleep to avoid out-of-order commit notifications
      if not os.getenv("SVN_BACKPORT_DONT_SLEEP"): # enabled by the test suite
          time.sleep(15)
      second_logmsg = "Remove the {!r} branch, {} in r{}."\
                          .format(entry.branch, reintegrated_word, revnum)
      run_svn(['rm', '-m', second_logmsg, '--', branch_url])
      time.sleep(1)

def validate_branch_contains_named_revisions(entry):
  """Validate that every revision explicitly named in ENTRY has either been
  merged to its backport branch from trunk, or has been committed directly to
  its backport branch.  Entries that declare no backport branches are
  considered valid.  Return on success, raise on failure."""
  if not entry.branch:
    return # valid

  if svn_version() < (1,5): # doesn't have 'svn mergeinfo' subcommand
    return # skip check

  sf = entry.status_file
  branch_url = sf.branch_url(entry.branch)
  present_str = (
    run_svn(['mergeinfo', '--show-revs=merged', '--', sf.trunk_url(), branch_url])[1]
    +
    run_svn(['mergeinfo', '--show-revs=eligible', '--', branch_url])[1]
  )

  present = map(int, re.compile(r'(\d+)').findall(present_str))

  absent = set(entry.revisions) - set(present)

  if absent:
    raise UnableToMergeException("Revisions '{}' nominated but not included "
                                 "in branch".format(
                                   ', '.join('r%d' % revno
                                             for revno in absent)))



def setUpModule():
  "Set-up function, invoked by 'python -m unittest'."
  # Suppress warnings generated by the test data.
  # TODO: some test functions assume .assertLogs is available, they fail with
  # AttributeError if it's absent (e.g., on python < 3.4).
  try:
    unittest.TestCase.assertLogs
  except AttributeError:
    logger.setLevel(logging.ERROR)

if __name__ == '__main__':
  unittest.main()