import contextlib
import functools
import os
import re
import sys
@contextlib.contextmanager
def chdir(dir):
try:
saved_dir = os.getcwd()
os.chdir(dir)
yield
finally:
os.chdir(saved_dir)
sys.path.insert(0, os.path.abspath('../../subversion/tests/cmdline'))
with chdir('../../subversion/tests/cmdline'):
import svntest
Skip = svntest.testcase.Skip_deco
SkipUnless = svntest.testcase.SkipUnless_deco
XFail = svntest.testcase.XFail_deco
Issues = svntest.testcase.Issues_deco
Issue = svntest.testcase.Issue_deco
Wimp = svntest.testcase.Wimp_deco
BACKPORT_PL = os.path.abspath(os.path.join(os.path.dirname(__file__),
'backport.pl'))
STATUS = 'branch/STATUS'
class BackportTest(object):
"""Decorator. See self.__call__()."""
def __init__(self, uuid):
"""The argument is the UUID embedded in the dump file.
If the argument is None, then there is no dump file."""
self.uuid = uuid
def __call__(self, test_func):
"""Return a decorator that: builds TEST_FUNC's sbox, creates
^/subversion/trunk, and calls TEST_FUNC, then compare its output to the
expected dump file named after TEST_FUNC."""
@functools.wraps(test_func)
def wrapped_test_func(sbox):
expected_dump_file = './%s.dump' % (test_func.func_name,)
sbox.build()
sbox.simple_mkdir('subversion', 'subversion/trunk')
sbox.simple_mkdir('subversion/tags', 'subversion/branches')
sbox.simple_move('A', 'subversion/trunk')
sbox.simple_move('iota', 'subversion/trunk')
sbox.simple_commit(message='Create trunk')
sbox.simple_copy('subversion/trunk', 'branch')
sbox.simple_append('branch/STATUS', '')
sbox.simple_add('branch/STATUS')
sbox.simple_commit(message='Create branch, with STATUS file')
sbox.simple_append('subversion/trunk/iota', 'First change\n')
sbox.simple_commit(message='First change')
sbox.simple_append('subversion/trunk/A/mu', 'Second change\n')
sbox.simple_commit(message='Second change')
test_func(sbox)
verify_backport(sbox, expected_dump_file, self.uuid)
return wrapped_test_func
def make_entry(revisions=None, logsummary=None, notes=None, branch=None,
depends=None, votes=None):
assert revisions
if logsummary is None:
logsummary = "default logsummary"
if votes is None:
votes = {+1 : ['jrandom']}
entry = {
'revisions': revisions,
'logsummary': logsummary,
'notes': notes,
'branch': branch,
'depends': depends,
'votes': votes,
}
return entry
def serialize_entry(entry):
return ''.join([
' * %s\n'
% (", ".join("r%ld" % revision for revision in entry['revisions'])),
' %s\n' % (entry['logsummary'],),
' Notes: %s\n' % (entry['notes'],) if entry['notes'] else '',
' Branch: %s\n' % (entry['branch'],) if entry['branch'] else '',
' Depends: %s\n' % (entry['depends'],) if entry['depends'] else '',
' Votes:\n',
''.join(' '
'%s: %s\n' % ({1: '+1', 0: '+0', -1: '-1', -0: '-0'}[vote],
", ".join(entry['votes'][vote]))
for vote in entry['votes']),
'\n', ])
def serialize_STATUS(approveds,
serialize_entry=serialize_entry):
"""Construct and return the contents of a STATUS file.
APPROVEDS is an iterable of ENTRY dicts. The dicts are defined
to have the following keys: 'revisions', a list of revision numbers (ints);
'logsummary'; and 'votes', a dict mapping ±1/±0 (int) to list of voters.
"""
strings = []
strings.append("Status of 1.8.x:\n\n")
strings.append("Candidate changes:\n")
strings.append("==================\n\n")
strings.append("Random new subheading:\n")
strings.append("======================\n\n")
strings.append("Veto-blocked changes:\n")
strings.append("=====================\n\n")
strings.append("Approved changes:\n")
strings.append("=================\n\n")
strings.extend(map(serialize_entry, approveds))
return "".join(strings)
def run_backport(sbox, error_expected=False, extra_env=[]):
"""Run backport.pl. EXTRA_ENV is a list of key=value pairs (str) to set in
the child's environment. ERROR_EXPECTED is propagated to run_command()."""
args = [
'/usr/bin/env',
'SVN=' + svntest.main.svn_binary,
'YES=1', 'MAY_COMMIT=1', 'AVAILID=jrandom',
] + list(extra_env) + [
'perl', BACKPORT_PL,
]
with chdir(sbox.ospath('branch')):
return svntest.main.run_command(args[0], error_expected, False, *(args[1:]))
def verify_backport(sbox, expected_dump_file, uuid):
"""Compare the contents of the SBOX repository with EXPECTED_DUMP_FILE.
Set the UUID of SBOX to UUID beforehand.
Based on svnsync_tests.py:verify_mirror."""
if uuid is None:
return
svntest.actions.enable_revprop_changes(sbox.repo_dir)
for revnum in range(0, 1+int(sbox.youngest())):
svntest.actions.run_and_verify_svnadmin([], [],
"delrevprop", "-r", revnum, sbox.repo_dir, "svn:date")
dest_dump = open(expected_dump_file).readlines()
svntest.actions.run_and_verify_svnadmin(None, [],
'setuuid', '--', sbox.repo_dir, uuid)
src_dump = svntest.actions.run_and_verify_dump(sbox.repo_dir)
svntest.verify.compare_dump_files(
"Dump files", "DUMP", src_dump, dest_dump)
@BackportTest('76cee987-25c9-4d6c-ad40-000000000001')
def backport_indented_entry(sbox):
"parsing of entries with nonstandard indentation"
approved_entries = [
make_entry([4]),
]
def reindenting_serialize_entry(*args, **kwargs):
entry = serialize_entry(*args, **kwargs)
return ('\n' + entry).replace('\n ', '\n')[1:]
sbox.simple_append(STATUS, serialize_STATUS(approved_entries,
serialize_entry=reindenting_serialize_entry))
sbox.simple_commit(message='Nominate r4')
run_backport(sbox)
@BackportTest('76cee987-25c9-4d6c-ad40-000000000002')
def backport_two_approveds(sbox):
"backport with two approveds"
approved_entries = [
make_entry([4]),
make_entry([5]),
]
sbox.simple_append(STATUS, serialize_STATUS(approved_entries))
sbox.simple_commit(message='Nominate r4. Nominate r5.')
run_backport(sbox)
svntest.actions.run_and_verify_svnlook(["8\n"], [],
'youngest', sbox.repo_dir)
sbox.simple_update()
svntest.main.run_svn(None, 'merge', '-r8:6',
'^/branch', sbox.ospath('branch'))
sbox.simple_commit(message='Revert the merges.')
sbox.simple_rm('subversion/trunk/A')
sbox.simple_commit(message='Third change on trunk.')
sbox.simple_append(STATUS, serialize_entry(make_entry([10])))
sbox.simple_commit(message='Nominate r10.')
run_backport(sbox)
@BackportTest('76cee987-25c9-4d6c-ad40-000000000003')
def backport_accept(sbox):
"test --accept parsing"
sbox.simple_append('branch/iota', 'Conflicts with first change\n')
sbox.simple_commit(message="Conflicting change on iota")
approved_entries = [
make_entry([4], notes="Merge with --accept=theirs-conflict."),
]
def reindenting_serialize_entry(*args, **kwargs):
entry = serialize_entry(*args, **kwargs)
return ('\n' + entry).replace('\n ', '\n')[1:]
sbox.simple_append(STATUS, serialize_STATUS(approved_entries,
serialize_entry=reindenting_serialize_entry))
sbox.simple_commit(message='Nominate r4')
run_backport(sbox)
@BackportTest('76cee987-25c9-4d6c-ad40-000000000004')
def backport_branches(sbox):
"test branches"
sbox.simple_append('branch/iota', 'Conflicts with first change')
sbox.simple_commit(message="Conflicting change on iota")
sbox.simple_update()
sbox.simple_copy('branch', 'subversion/branches/r4')
sbox.simple_commit(message='Create a backport branch')
sbox.simple_update()
svntest.main.run_svn(None, 'merge', '--record-only', '-c4',
'^/subversion/trunk', sbox.ospath('subversion/branches/r4'))
sbox.simple_mkdir('subversion/branches/r4/A_resolved')
sbox.simple_append('subversion/branches/r4/iota', "resolved\n", truncate=1)
sbox.simple_commit(message='Conflict resolution via mkdir')
approved_entries = [
make_entry([4], branch="r4")
]
sbox.simple_append(STATUS, serialize_STATUS(approved_entries))
sbox.simple_commit(message='Nominate r4')
run_backport(sbox)
@BackportTest('76cee987-25c9-4d6c-ad40-000000000005')
def backport_multirevisions(sbox):
"test multirevision entries"
approved_entries = [
make_entry([4,5])
]
sbox.simple_append(STATUS, serialize_STATUS(approved_entries))
sbox.simple_commit(message='Nominate a group.')
run_backport(sbox)
@BackportTest(None) def backport_conflicts_detection(sbox):
"test the conflicts detector"
sbox.simple_append('branch/iota', 'Conflicts with first change\n')
sbox.simple_commit(message="Conflicting change on iota")
approved_entries = [
make_entry([4], notes="This will conflict."),
]
sbox.simple_append(STATUS, serialize_STATUS(approved_entries))
sbox.simple_commit(message='Nominate r4')
exit_code, output, errput = run_backport(sbox, True,
["MAY_COMMIT=0"])
expected_output = svntest.verify.RegexOutput(
'Index: iota',
match_all=False,
)
expected_errput = (
r'(?ms)' r'.*Warning summary.*'
r'^r4 [(]default logsummary[)]: Conflicts on iota.*'
)
expected_errput = svntest.verify.RegexListOutput(
[
r'Warning summary',
r'===============',
r'r4 [(]default logsummary[)]: Conflicts on iota',
],
match_all=False)
svntest.verify.verify_outputs(None, output, errput,
expected_output, expected_errput)
svntest.verify.verify_exit_code(None, exit_code, 1)
approved_entries = [
make_entry([4], depends="World peace."),
]
sbox.simple_append(STATUS, serialize_STATUS(approved_entries), truncate=True)
sbox.simple_commit(message='Re-nominate r4')
exit_code, output, errput = run_backport(sbox, extra_env=["MAY_COMMIT=0"])
svntest.verify.verify_outputs(None, output, errput,
"Conflicts found.*, as expected.", [])
@BackportTest(None) def backport_branch_contains(sbox):
"branch must contain the revisions"
sbox.simple_append('branch/iota', 'Conflicts with first change')
sbox.simple_commit(message="Conflicting change on iota")
sbox.simple_update()
sbox.simple_copy('branch', 'subversion/branches/r4')
sbox.simple_commit(message='Create a backport branch')
sbox.simple_update()
svntest.main.run_svn(None, 'merge', '--record-only', '-c4',
'^/subversion/trunk', sbox.ospath('subversion/branches/r4'))
sbox.simple_mkdir('subversion/branches/r4/A_resolved')
sbox.simple_append('subversion/branches/r4/iota', "resolved\n", truncate=1)
sbox.simple_commit(message='Conflict resolution via mkdir')
approved_entries = [
make_entry([4,5], branch="r4")
]
sbox.simple_append(STATUS, serialize_STATUS(approved_entries))
sbox.simple_commit(message='Nominate r4')
exit_code, output, errput = run_backport(sbox, error_expected=True)
expected_errput = svntest.verify.RegexOutput(
".*Revisions 'r5' nominated but not included in branch",
match_all=False,
)
svntest.verify.verify_outputs(None, output, errput,
[], expected_errput)
svntest.verify.verify_exit_code(None, exit_code, 1)
svntest.actions.run_and_verify_svnlook(["9\n"], [],
'youngest', sbox.repo_dir)
svntest.actions.run_and_verify_svn([], [], 'status', '-q',
sbox.repo_dir)
@BackportTest(None) def backport_double_conflict(sbox):
"two-revisioned entry with two conflicts"
sbox.simple_append('branch/iota', 'Conflicts with first change')
sbox.simple_commit(message="Conflicting change on iota")
sbox.simple_update()
sbox.simple_append('subversion/trunk/iota', 'Third line\n')
sbox.simple_commit(message="iota's third line")
approved_entries = [
make_entry([4,7], depends="World peace.")
]
sbox.simple_append(STATUS, serialize_STATUS(approved_entries))
sbox.simple_commit(message='Nominate the r4 group')
exit_code, output, errput = run_backport(sbox, True, ["MAY_COMMIT=0"])
expected_output = 'Conflicts found.*, as expected.'
expected_errput = svntest.verify.RegexOutput(
".*svn: E155015:.*", match_all=False,
)
svntest.verify.verify_outputs(None, output, errput,
expected_output, expected_errput)
svntest.verify.verify_exit_code(None, exit_code, 0)
if any("Warning summary" in line for line in errput):
raise svntest.verify.SVNUnexpectedStderr(errput)
approved_entries = [
make_entry([4,7]) ]
sbox.simple_append(STATUS, serialize_STATUS(approved_entries), truncate=True)
sbox.simple_commit(message='Re-nominate the r4 group')
exit_code, output, errput = run_backport(sbox, True, ["MAY_COMMIT=0"])
expected_errput = r'r4 .*: subshell exited with code (?:[1-9]\d+)'
svntest.verify.verify_exit_code(None, exit_code, 1)
svntest.verify.verify_outputs(None, output, errput,
svntest.verify.AnyOutput, expected_errput)
test_list = [ None,
backport_indented_entry,
backport_two_approveds,
backport_accept,
backport_branches,
backport_multirevisions,
backport_conflicts_detection,
backport_branch_contains,
backport_double_conflict,
]
if __name__ == '__main__':
svntest.main.run_tests(test_list)