svndumpfilter_tests.py [plain text]
import os
import sys
import tempfile
import svntest
from svntest.verify import SVNExpectedStdout, SVNExpectedStderr
from svnadmin_tests import load_and_verify_dumpstream, test_create
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
Item = svntest.wc.StateItem
def filter_and_return_output(dump, bufsize=0, *varargs):
"""Filter the array of lines passed in 'dump' and return the output
and errput"""
if isinstance(dump, str):
dump = [ dump ]
try:
varargs.index('-q')
expected_errput = None except:
try:
varargs.index('--quiet')
expected_errput = None
except:
expected_errput = svntest.verify.AnyOutput
exit_code, output, errput = svntest.main.run_command_stdin(
svntest.main.svndumpfilter_binary, expected_errput, bufsize, 1,
dump, *varargs)
if sys.platform == 'win32':
errput = map(lambda x : x.replace('\r\n', '\n'), errput)
return output, errput
@Issue(2982)
def reflect_dropped_renumbered_revs(sbox):
"reflect dropped renumbered revs in svn:mergeinfo"
test_create(sbox)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svndumpfilter_tests_data',
'with_merges.dump')
dumpfile = open(dumpfile_location).read()
filtered_out, filtered_err = filter_and_return_output(
dumpfile, 0, "include",
"trunk", "branch1",
"--skip-missing-merge-sources",
"--drop-empty-revs",
"--renumber-revs", "--quiet")
load_and_verify_dumpstream(sbox, [], [], None, filtered_out,
"--ignore-uuid")
url = sbox.repo_url
expected_output = svntest.verify.UnorderedOutput([
url + "/trunk - /branch1:4-5\n",
])
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
test_create(sbox)
filtered_out, filtered_err = filter_and_return_output(
dumpfile, 0, "exclude", "branch1",
"--skip-missing-merge-sources",
"--drop-empty-revs",
"--renumber-revs", "--quiet")
load_and_verify_dumpstream(sbox, [], [], None, filtered_out,
"--ignore-uuid")
expected_output = svntest.verify.UnorderedOutput([
url + "/trunk - \n",
])
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
@Issue(3181)
def svndumpfilter_loses_mergeinfo(sbox):
"svndumpfilter loses mergeinfo"
test_create(sbox)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svndumpfilter_tests_data',
'with_merges.dump')
dumpfile = open(dumpfile_location).read()
filtered_out, filtered_err = filter_and_return_output(dumpfile, 0, "include",
"trunk", "branch1",
"--quiet")
load_and_verify_dumpstream(sbox, [], [], None, filtered_out)
url = sbox.repo_url
expected_output = svntest.verify.UnorderedOutput([
url + "/trunk - /branch1:4-8\n",
])
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
def _simple_dumpfilter_test(sbox, dumpfile, *dumpargs):
wc_dir = sbox.wc_dir
filtered_output, filtered_err = filter_and_return_output(dumpfile, 0,
'--quiet',
*dumpargs)
load_and_verify_dumpstream(sbox, [], [], None, filtered_output,
'--ignore-uuid')
expected_disk = svntest.main.greek_state.copy()
expected_disk.remove('A/B/E/alpha')
expected_disk.remove('A/B/E/beta')
expected_disk.remove('A/B/E')
expected_disk.remove('A/D/H/chi')
expected_disk.remove('A/D/H/psi')
expected_disk.remove('A/D/H/omega')
expected_disk.remove('A/D/H')
expected_disk.remove('A/D/G/pi')
expected_disk.remove('A/D/G/rho')
expected_disk.remove('A/D/G/tau')
expected_disk.remove('A/D/G')
expected_output = svntest.wc.State(wc_dir, {
'A' : Item(status='A '),
'A/B' : Item(status='A '),
'A/B/lambda' : Item(status='A '),
'A/B/F' : Item(status='A '),
'A/mu' : Item(status='A '),
'A/C' : Item(status='A '),
'A/D' : Item(status='A '),
'A/D/gamma' : Item(status='A '),
'iota' : Item(status='A '),
})
expected_status = svntest.actions.get_virginal_state(wc_dir, 1)
expected_status.remove('A/B/E/alpha')
expected_status.remove('A/B/E/beta')
expected_status.remove('A/B/E')
expected_status.remove('A/D/H/chi')
expected_status.remove('A/D/H/psi')
expected_status.remove('A/D/H/omega')
expected_status.remove('A/D/H')
expected_status.remove('A/D/G/pi')
expected_status.remove('A/D/G/rho')
expected_status.remove('A/D/G/tau')
expected_status.remove('A/D/G')
svntest.actions.run_and_verify_update(wc_dir,
expected_output,
expected_disk,
expected_status)
@Issue(2697)
def dumpfilter_with_targets(sbox):
"svndumpfilter --targets blah"
test_create(sbox)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svndumpfilter_tests_data',
'greek_tree.dump')
dumpfile = open(dumpfile_location).read()
(fd, targets_file) = tempfile.mkstemp(dir=svntest.main.temp_dir)
try:
targets = open(targets_file, 'w')
targets.write('/A/D/H\n')
targets.write('/A/D/G\n')
targets.close()
_simple_dumpfilter_test(sbox, dumpfile,
'exclude', '/A/B/E', '--targets', targets_file)
finally:
os.close(fd)
os.remove(targets_file)
def dumpfilter_with_patterns(sbox):
"svndumpfilter --pattern PATH_PREFIX"
test_create(sbox)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svndumpfilter_tests_data',
'greek_tree.dump')
dumpfile = open(dumpfile_location).read()
_simple_dumpfilter_test(sbox, dumpfile,
'exclude', '--pattern', '/A/D/[GH]*', '/A/[B]/E*')
@Issue(3020)
def filter_mergeinfo_revs_outside_of_dump_stream(sbox):
"filter mergeinfo revs outside of dump stream"
test_create(sbox)
partial_dump = os.path.join(os.path.dirname(sys.argv[0]),
'svndumpfilter_tests_data',
'mergeinfo_included_partial.dump')
partial_dump_contents = open(partial_dump).read()
filtered_dumpfile2, filtered_out = filter_and_return_output(
partial_dump_contents,
8192, "include", "trunk", "branches",
"--skip-missing-merge-sources",
"--quiet")
load_and_verify_dumpstream(sbox, [], [], None, filtered_dumpfile2,
'--ignore-uuid')
url = sbox.repo_url + "/branches"
expected_output = svntest.verify.UnorderedOutput([
url + "/B1 - /branches/B2:6-7\n",
"/trunk:4\n",
url + "/B2 - /trunk:4\n",
url + "/B1/B/E - /branches/B2/B/E:6-7\n",
"/trunk/B/E:3-4\n"])
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
test_create(sbox)
skeleton_dumpfile = open(os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'skeleton_repos.dump')).read()
load_and_verify_dumpstream(sbox, [], [], None, skeleton_dumpfile,
'--ignore-uuid')
partial_dump2 = os.path.join(os.path.dirname(sys.argv[0]),
'svndumpfilter_tests_data',
'mergeinfo_included_partial.dump')
partial_dump_contents2 = open(partial_dump2).read()
filtered_dumpfile2, filtered_err = filter_and_return_output(
partial_dump_contents2,
8192, "exclude", "branches/B2",
"--skip-missing-merge-sources",
"--drop-empty-revs",
"--renumber-revs")
expected_err = [
"Excluding (and dropping empty revisions for) prefixes:\n",
" '/branches/B2'\n",
"\n",
"Revision 6 committed as 6.\n",
"Revision 7 skipped.\n", "Revision 8 committed as 7.\n",
"Revision 9 committed as 8.\n",
"Revision 10 committed as 9.\n",
"Revision 11 skipped.\n", "Revision 12 skipped.\n", "Revision 13 committed as 10.\n",
"Revision 14 committed as 11.\n",
"Revision 15 committed as 12.\n",
"\n",
"Dropped 3 revisions.\n",
"\n",
"Revisions renumbered as follows:\n",
" 15 => 12\n",
" 14 => 11\n",
" 13 => 10\n",
" 12 => (dropped)\n", " 11 => (dropped)\n", " 10 => 9\n",
" 9 => 8\n",
" 8 => 7\n",
" 7 => (dropped)\n", " 6 => 6\n",
"\n",
"Dropped 2 nodes:\n",
" '/branches/B2'\n",
" '/branches/B2/D/H/chi'\n",
"\n"]
svntest.verify.verify_outputs(
"Actual svndumpfilter stderr does not agree with expected stderr",
None, filtered_err, None, expected_err)
load_and_verify_dumpstream(sbox, [], [], None, filtered_dumpfile2,
'--parent-dir', '/Projects/Project-X',
'--ignore-uuid')
url = sbox.repo_url + "/Projects/Project-X/branches"
expected_output = svntest.verify.UnorderedOutput([
url + "/B1 - /Projects/Project-X/trunk:9\n",
url + "/B1/B/E - /Projects/Project-X/trunk/B/E:8-9\n"])
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
@Issue(3020)
def dropped_but_not_renumbered_empty_revs(sbox):
"mergeinfo maps correctly when dropping revs"
test_create(sbox)
full_dump = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'mergeinfo_included_full.dump')
full_dump_contents = open(full_dump).read()
filtered_dumpfile, filtered_out = filter_and_return_output(
full_dump_contents,
8192, "exclude", "branches/B2",
"--skip-missing-merge-sources", "--drop-empty-revs")
load_and_verify_dumpstream(sbox, [], [], None, filtered_dumpfile,
'--ignore-uuid')
url = sbox.repo_url + "/branches"
expected_output = svntest.verify.UnorderedOutput([
url + "/B1 - /trunk:6,8\n",
url + "/B1/B/E - /trunk/B/E:5-8\n"])
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
@Issue(4234)
def dumpfilter_targets_expect_leading_slash_prefixes(sbox):
"dumpfilter targets expect leading '/' in prefixes"
test_create(sbox)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svndumpfilter_tests_data',
'greek_tree.dump')
dumpfile = open(dumpfile_location).read()
(fd, targets_file) = tempfile.mkstemp(dir=svntest.main.temp_dir)
try:
targets = open(targets_file, 'w')
targets.write('A/D/H\n')
targets.write('A/D/G\n')
targets.close()
_simple_dumpfilter_test(sbox, dumpfile,
'exclude', '/A/B/E', '--targets', targets_file)
finally:
os.close(fd)
os.remove(targets_file)
test_list = [ None,
reflect_dropped_renumbered_revs,
svndumpfilter_loses_mergeinfo,
dumpfilter_with_targets,
dumpfilter_with_patterns,
filter_mergeinfo_revs_outside_of_dump_stream,
dropped_but_not_renumbered_empty_revs,
dumpfilter_targets_expect_leading_slash_prefixes,
]
if __name__ == '__main__':
svntest.main.run_tests(test_list)