tools/dist/backport/merger.py - subversion - Git at Google

 #!/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

   # TODO(interactive mode): exclude STATUS from reverts
   # TODO(interactive mode): save local mods to disk, as backport.pl does
   run_revert()

   run_svn_quiet(['update'])
   # 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()
	#!/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

	# TODO(interactive mode): exclude STATUS from reverts
	# TODO(interactive mode): save local mods to disk, as backport.pl does
	run_revert()

	run_svn_quiet(['update'])
	# 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()