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