import os
import shutil
import sys
import svntest
from svntest.verify import SVNExpectedStdout, SVNExpectedStderr
from svntest.verify import SVNUnexpectedStderr
Skip = svntest.testcase.Skip
SkipUnless = svntest.testcase.SkipUnless
XFail = svntest.testcase.XFail
Item = svntest.wc.StateItem
def get_txns(repo_dir):
"Get the txn names using 'svnadmin lstxns'."
exit_code, output_lines, error_lines = svntest.main.run_svnadmin('lstxns',
repo_dir)
txns = sorted([output_lines.strip(x) for x in output_lines])
return txns
def load_and_verify_dumpstream(sbox, expected_stdout, expected_stderr,
revs, dump, *varargs):
"""Load the array of lines passed in 'dump' into the
current tests' repository and verify the repository content
using the array of wc.States passed in revs. VARARGS are optional
arguments passed to the 'load' command"""
if type(dump) is type(""):
dump = [ dump ]
exit_code, output, errput = svntest.main.run_command_stdin(
svntest.main.svnadmin_binary, expected_stderr, 1, dump,
'load', '--quiet', sbox.repo_dir, *varargs)
if expected_stdout:
if expected_stdout == svntest.verify.AnyOutput:
if len(output) == 0:
raise SVNExpectedStdout
else:
svntest.verify.compare_and_display_lines(
"Standard output", "STDOUT:", expected_stdout, output)
if expected_stderr:
if expected_stderr == svntest.verify.AnyOutput:
if len(errput) == 0:
raise SVNExpectedStderr
else:
svntest.verify.compare_and_display_lines(
"Standard error output", "STDERR:", expected_stderr, errput)
return
if revs:
for rev in range(len(revs)):
svntest.actions.run_and_verify_svn("Updating to r%s" % (rev+1),
svntest.verify.AnyOutput, [],
"update", "-r%s" % (rev+1),
sbox.wc_dir)
wc_tree = svntest.tree.build_tree_from_wc(sbox.wc_dir)
rev_tree = revs[rev].old_tree()
try:
svntest.tree.compare_trees ("rev/disk", rev_tree, wc_tree)
except svntest.tree.SVNTreeError:
svntest.verify.display_trees(None, 'WC TREE', wc_tree, rev_tree)
raise
def test_create(sbox):
"'svnadmin create'"
repo_dir = sbox.repo_dir
wc_dir = sbox.wc_dir
svntest.main.safe_rmtree(repo_dir, 1)
svntest.main.safe_rmtree(wc_dir)
svntest.main.create_repos(repo_dir)
svntest.actions.run_and_verify_svn("Creating rev 0 checkout",
["Checked out revision 0.\n"], [],
"checkout",
sbox.repo_url, wc_dir)
svntest.actions.run_and_verify_svn(
"Running status",
[], [],
"status", wc_dir)
svntest.actions.run_and_verify_svn(
"Running verbose status",
[" 0 0 ? %s\n" % wc_dir], [],
"status", "--verbose", wc_dir)
def clean_dumpfile():
return \
[ "SVN-fs-dump-format-version: 2\n\n",
"UUID: 668cc64a-31ed-0310-8ccb-b75d75bb44e3\n\n",
"Revision-number: 0\n",
"Prop-content-length: 56\n",
"Content-length: 56\n\n",
"K 8\nsvn:date\nV 27\n2005-01-08T21:48:13.838745Z\nPROPS-END\n\n\n",
"Revision-number: 1\n",
"Prop-content-length: 98\n",
"Content-length: 98\n\n",
"K 7\nsvn:log\nV 0\n\nK 10\nsvn:author\nV 4\nerik\n",
"K 8\nsvn:date\nV 27\n2005-01-08T21:51:16.313791Z\nPROPS-END\n\n\n",
"Node-path: A\n",
"Node-kind: file\n",
"Node-action: add\n",
"Prop-content-length: 35\n",
"Text-content-length: 5\n",
"Text-content-md5: e1cbb0c3879af8347246f12c559a86b5\n",
"Content-length: 40\n\n",
"K 12\nsvn:keywords\nV 2\nId\nPROPS-END\ntext\n\n\n"]
dumpfile_revisions = \
[ svntest.wc.State('', { 'A' : svntest.wc.StateItem(contents="text\n") }) ]
def extra_headers(sbox):
"loading of dumpstream with extra headers"
test_create(sbox)
dumpfile = clean_dumpfile()
dumpfile[3:3] = \
[ "X-Comment-Header: Ignored header normally not in dump stream\n" ]
load_and_verify_dumpstream(sbox,[],[], dumpfile_revisions, dumpfile,
'--ignore-uuid')
def extra_blockcontent(sbox):
"load success on oversized Content-length"
test_create(sbox)
dumpfile = clean_dumpfile()
dumpfile[8:9] = \
[ "Extra-content-length: 10\n",
"Content-length: 108\n\n" ]
dumpfile[11] = dumpfile[11][:-2] + "extra text\n\n\n"
load_and_verify_dumpstream(sbox,[],[], dumpfile_revisions, dumpfile,
'--ignore-uuid')
def inconsistent_headers(sbox):
"load failure on undersized Content-length"
test_create(sbox)
dumpfile = clean_dumpfile()
dumpfile[-2] = "Content-length: 30\n\n"
load_and_verify_dumpstream(sbox, [], svntest.verify.AnyOutput,
dumpfile_revisions, dumpfile)
def empty_date(sbox):
"preserve date-less revisions in load (issue #2729)"
test_create(sbox)
dumpfile = clean_dumpfile()
dumpfile[7:11] = \
[ "Prop-content-length: 52\n",
"Content-length: 52\n\n",
"K 7\nsvn:log\nV 0\n\nK 10\nsvn:author\nV 4\nerik\nPROPS-END\n\n\n"
]
load_and_verify_dumpstream(sbox,[],[], dumpfile_revisions, dumpfile,
'--ignore-uuid')
svntest.actions.run_and_verify_svn(None, [], [], "propget",
"--revprop", "-r1", "svn:date",
sbox.wc_dir)
def dump_copied_dir(sbox):
"'svnadmin dump' on copied directory"
sbox.build()
wc_dir = sbox.wc_dir
repo_dir = sbox.repo_dir
old_C_path = os.path.join(wc_dir, 'A', 'C')
new_C_path = os.path.join(wc_dir, 'A', 'B', 'C')
svntest.main.run_svn(None, 'cp', old_C_path, new_C_path)
svntest.main.run_svn(None, 'ci', wc_dir, '--quiet',
'-m', 'log msg')
exit_code, output, errput = svntest.main.run_svnadmin("dump", repo_dir)
if svntest.verify.compare_and_display_lines(
"Output of 'svnadmin dump' is unexpected.",
'STDERR', ["* Dumped revision 0.\n",
"* Dumped revision 1.\n",
"* Dumped revision 2.\n"], errput):
raise svntest.Failure
def dump_move_dir_modify_child(sbox):
"'svnadmin dump' on modified child of copied dir"
sbox.build()
wc_dir = sbox.wc_dir
repo_dir = sbox.repo_dir
B_path = os.path.join(wc_dir, 'A', 'B')
Q_path = os.path.join(wc_dir, 'A', 'Q')
svntest.main.run_svn(None, 'cp', B_path, Q_path)
svntest.main.file_append(os.path.join(Q_path, 'lambda'), 'hello')
svntest.main.run_svn(None, 'ci', wc_dir, '--quiet',
'-m', 'log msg')
exit_code, output, errput = svntest.main.run_svnadmin("dump", repo_dir)
svntest.verify.compare_and_display_lines(
"Output of 'svnadmin dump' is unexpected.",
'STDERR', ["* Dumped revision 0.\n",
"* Dumped revision 1.\n",
"* Dumped revision 2.\n"], errput)
exit_code, output, errput = svntest.main.run_svnadmin("dump", "-r",
"0:HEAD", repo_dir)
svntest.verify.compare_and_display_lines(
"Output of 'svnadmin dump' is unexpected.",
'STDERR', ["* Dumped revision 0.\n",
"* Dumped revision 1.\n",
"* Dumped revision 2.\n"], errput)
def dump_quiet(sbox):
"'svnadmin dump --quiet'"
sbox.build(create_wc = False)
exit_code, output, errput = svntest.main.run_svnadmin("dump", sbox.repo_dir,
'--quiet')
svntest.verify.compare_and_display_lines(
"Output of 'svnadmin dump --quiet' is unexpected.",
'STDERR', [], errput)
def hotcopy_dot(sbox):
"'svnadmin hotcopy PATH .'"
sbox.build()
backup_dir, backup_url = sbox.add_repo_path('backup')
os.mkdir(backup_dir)
cwd = os.getcwd()
os.chdir(backup_dir)
svntest.actions.run_and_verify_svnadmin(
None, None, [],
"hotcopy", os.path.join(cwd, sbox.repo_dir), '.')
os.chdir(cwd)
exit_code, origout, origerr = svntest.main.run_svnadmin("dump",
sbox.repo_dir,
'--quiet')
exit_code, backout, backerr = svntest.main.run_svnadmin("dump",
backup_dir,
'--quiet')
if origerr or backerr or origout != backout:
raise svntest.Failure
def hotcopy_format(sbox):
"'svnadmin hotcopy' checking db/format file"
sbox.build()
backup_dir, backup_url = sbox.add_repo_path('backup')
exit_code, output, errput = svntest.main.run_svnadmin("hotcopy",
sbox.repo_dir,
backup_dir)
if errput:
print("Error: hotcopy failed")
raise svntest.Failure
fp = open(os.path.join(sbox.repo_dir, "db", "format"))
contents1 = fp.read()
fp.close()
fp2 = open(os.path.join(backup_dir, "db", "format"))
contents2 = fp2.read()
fp2.close()
if contents1 != contents2:
print("Error: db/format file contents do not match after hotcopy")
raise svntest.Failure
def setrevprop(sbox):
"'setlog' and 'setrevprop', bypassing hooks'"
sbox.build()
iota_path = os.path.join(sbox.wc_dir, "iota")
exit_code, output, errput = svntest.main.run_svnadmin("setlog",
sbox.repo_dir,
"-r0",
"--bypass-hooks",
iota_path)
if errput:
print("Error: 'setlog' failed")
raise svntest.Failure
svntest.actions.run_and_verify_svn(None,
[ "This is the file 'iota'.\n", "\n" ],
[], "propget", "--revprop", "-r0",
"svn:log", sbox.wc_dir)
foo_path = os.path.join(sbox.wc_dir, "foo")
svntest.main.file_write(foo_path, "foo")
exit_code, output, errput = svntest.main.run_svnadmin("setrevprop",
sbox.repo_dir,
"-r0", "svn:author",
foo_path)
if errput:
print("Error: 'setrevprop' failed")
raise svntest.Failure
svntest.actions.run_and_verify_svn(None, [ "foo\n" ], [], "propget",
"--revprop", "-r0", "svn:author",
sbox.wc_dir)
def verify_windows_paths_in_repos(sbox):
"verify a repository containing paths like 'c:hi'"
sbox.build(create_wc = False)
repo_url = sbox.repo_url
chi_url = sbox.repo_url + '/c:hi'
svntest.actions.run_and_verify_svn(None, None, [],
'mkdir', '-m', 'log_msg',
chi_url)
exit_code, output, errput = svntest.main.run_svnadmin("verify",
sbox.repo_dir)
svntest.verify.compare_and_display_lines(
"Error while running 'svnadmin verify'.",
'STDERR', ["* Verified revision 0.\n",
"* Verified revision 1.\n",
"* Verified revision 2.\n"], errput)
def fsfs_file(repo_dir, kind, rev):
if svntest.main.server_minor_version >= 5:
if svntest.main.fsfs_sharding is None:
return os.path.join(repo_dir, 'db', kind, '0', rev)
else:
shard = int(rev) // svntest.main.fsfs_sharding
path = os.path.join(repo_dir, 'db', kind, str(shard), rev)
if svntest.main.fsfs_packing is None or kind == 'revprops':
return path
elif os.path.exists(path):
return path
else:
return os.path.join(repo_dir, 'db', kind, ('%d.pack' % shard), 'pack')
else:
return os.path.join(repo_dir, 'db', kind, rev)
def verify_incremental_fsfs(sbox):
"""svnadmin verify detects corruption dump can't"""
sbox.build(create_wc = False)
repo_url = sbox.repo_url
E_url = sbox.repo_url + '/A/B/E'
svntest.actions.run_and_verify_svn(None, None, [],
'mkdir', '-m', 'log_msg',
E_url + '/bravo')
r2 = fsfs_file(sbox.repo_dir, 'revs', '2')
if r2.endswith('pack'):
raise svntest.Skip
fp = open(r2, 'wb')
fp.write("""id: 0-2.0.r2/0
type: dir
count: 0
cpath: /A/B/E/bravo
copyroot: 0 /
PLAIN
K 5
alpha
V 17
file 3-1.0.r1/719
K 4
beta
V 17
file 4-1.0.r1/840
K 5
bravo
V 14
dir 0-2.0.r2/0
END
ENDREP
id: 2-1.0.r2/181
type: dir
pred: 2-1.0.r1/1043
count: 1
text: 2 69 99 99 f63001f7fddd1842d8891474d0982111
cpath: /A/B/E
copyroot: 0 /
PLAIN
K 1
E
V 16
dir 2-1.0.r2/181
K 1
F
V 17
dir 5-1.0.r1/1160
K 6
lambda
V 17
file 6-1.0.r1/597
END
ENDREP
id: 1-1.0.r2/424
type: dir
pred: 1-1.0.r1/1335
count: 1
text: 2 316 95 95 bccb66379b4f825dac12b50d80211bae
cpath: /A/B
copyroot: 0 /
PLAIN
K 1
B
V 16
dir 1-1.0.r2/424
K 1
C
V 17
dir 7-1.0.r1/1569
K 1
D
V 17
dir 8-1.0.r1/3061
K 2
mu
V 18
file i-1.0.r1/1451
END
ENDREP
id: 0-1.0.r2/692
type: dir
pred: 0-1.0.r1/3312
count: 1
text: 2 558 121 121 c9b5a2d26473a4e28088673dda9df804
cpath: /A
copyroot: 0 /
PLAIN
K 1
A
V 16
dir 0-1.0.r2/692
K 4
iota
V 18
file j-1.0.r1/3428
END
ENDREP
id: 0.0.r2/904
type: dir
pred: 0.0.r1/3624
count: 2
text: 2 826 65 65 e44e4151d0d124533338619f082c8c9a
cpath: /
copyroot: 0 /
_0.0.t1-1 add false false /A/B/E/bravo
904 1031
""")
fp.close()
exit_code, output, errput = svntest.main.run_svnadmin("verify", "-r2",
sbox.repo_dir)
svntest.verify.verify_outputs(
message=None, actual_stdout=output, actual_stderr=errput,
expected_stdout=None,
expected_stderr=".*Found malformed header in revision file")
def recover_fsfs(sbox):
"recover a repository (FSFS only)"
sbox.build()
current_path = os.path.join(sbox.repo_dir, 'db', 'current')
svntest.main.file_append(os.path.join(sbox.wc_dir, 'iota'), 'newer line\n')
svntest.main.run_svn(None, 'ci', sbox.wc_dir, '--quiet', '-m', 'log msg')
svntest.main.file_append(os.path.join(sbox.wc_dir, 'iota'), 'newest line\n')
svntest.main.run_svn(None, 'ci', sbox.wc_dir, '--quiet', '-m', 'log msg')
expected_current_contents = svntest.main.file_read(current_path)
os.rename(os.path.join(sbox.repo_dir, 'db','current'),
os.path.join(sbox.repo_dir, 'db','was_current'));
exit_code, output, errput = svntest.main.run_svnadmin("recover",
sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
actual_current_contents = svntest.main.file_read(current_path)
svntest.verify.compare_and_display_lines(
"Contents of db/current is unexpected.",
'db/current', expected_current_contents, actual_current_contents)
svntest.main.file_write(current_path, '2\n')
exit_code, output, errput = svntest.main.run_svnadmin("recover",
sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
actual_current_contents = svntest.main.file_read(current_path)
svntest.verify.compare_and_display_lines(
"Contents of db/current is unexpected.",
'db/current', expected_current_contents, actual_current_contents)
svntest.main.file_write(current_path, '1\n')
exit_code, output, errput = svntest.main.run_svnadmin("recover",
sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
actual_current_contents = svntest.main.file_read(current_path)
svntest.verify.compare_and_display_lines(
"Contents of db/current is unexpected.",
'db/current', expected_current_contents, actual_current_contents)
svntest.main.file_write(current_path, 'fish\n')
exit_code, output, errput = svntest.main.run_svnadmin("recover",
sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
actual_current_contents = svntest.main.file_read(current_path)
svntest.verify.compare_and_display_lines(
"Contents of db/current is unexpected.",
'db/current', expected_current_contents, actual_current_contents)
def load_with_parent_dir(sbox):
"'svnadmin load --parent-dir' reparents mergeinfo"
test_create(sbox)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'mergeinfo_included.dump')
dumpfile = svntest.main.file_read(dumpfile_location)
svntest.actions.run_and_verify_svn(None,
['\n', 'Committed revision 1.\n'],
[], "mkdir", sbox.repo_url + "/sample",
"-m", "Create sample dir")
load_and_verify_dumpstream(sbox,[],[], None, dumpfile, '--parent-dir',
'/sample')
svntest.actions.run_and_verify_svn(None,
[sbox.repo_url +
"/sample/branch - /sample/trunk:5-7\n"],
[], 'propget', 'svn:mergeinfo', '-R',
sbox.repo_url + '/sample/branch')
svntest.actions.run_and_verify_svn(None,
[sbox.repo_url +
"/sample/branch1 - " +
"/sample/branch:6-9\n"],
[], 'propget', 'svn:mergeinfo', '-R',
sbox.repo_url + '/sample/branch1')
def set_uuid(sbox):
"test 'svnadmin setuuid'"
sbox.build(create_wc=False)
exit_code, output, errput = svntest.main.run_svnlook('uuid', sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
orig_uuid = output[0].rstrip()
svntest.actions.run_and_verify_svnadmin(None, None, '^.*Malformed UUID.*$',
'setuuid', sbox.repo_dir, 'abcdef')
svntest.actions.run_and_verify_svnadmin(None, [], None,
'setuuid', sbox.repo_dir)
exit_code, output, errput = svntest.main.run_svnlook('uuid', sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
new_uuid = output[0].rstrip()
if new_uuid == orig_uuid:
print("Error: new UUID matches the original one")
raise svntest.Failure
svntest.actions.run_and_verify_svnadmin(None, [], None,
'setuuid', sbox.repo_dir, orig_uuid)
exit_code, output, errput = svntest.main.run_svnlook('uuid', sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
new_uuid = output[0].rstrip()
if new_uuid != orig_uuid:
print("Error: new UUID doesn't match the original one")
raise svntest.Failure
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 = svntest.main.file_read(dumpfile_location)
svntest.actions.run_and_verify_svn(None, ['\n', 'Committed revision 1.\n'],
[], "mkdir", sbox.repo_url + "/toplevel",
"-m", "Create toplevel dir")
load_and_verify_dumpstream(sbox,[],[], None, dumpfile)
load_and_verify_dumpstream(sbox,[],[], None, dumpfile, '--parent-dir',
'/toplevel')
svntest.actions.run_and_verify_svn(None, ["/trunk:1-4\n"],
[], 'propget', 'svn:mergeinfo',
sbox.repo_url + '/branch2')
svntest.actions.run_and_verify_svn(None, ["/branch1:5-9\n"],
[], 'propget', 'svn:mergeinfo',
sbox.repo_url + '/trunk')
svntest.actions.run_and_verify_svn(None, ["/toplevel/trunk:1-13\n"],
[], 'propget', 'svn:mergeinfo',
sbox.repo_url + '/toplevel/branch2')
svntest.actions.run_and_verify_svn(None, ["/toplevel/branch1:14-18\n"],
[], 'propget', 'svn:mergeinfo',
sbox.repo_url + '/toplevel/trunk')
svntest.actions.run_and_verify_svn(None, ["/toplevel/trunk:1-12\n"],
[], 'propget', 'svn:mergeinfo',
sbox.repo_url + '/toplevel/branch1')
svntest.actions.run_and_verify_svn(None, ["/trunk:1-3\n"],
[], 'propget', 'svn:mergeinfo',
sbox.repo_url + '/branch1')
def fsfs_recover_handle_missing_revs_or_revprops_file(sbox):
"""fsfs recovery checks missing revs / revprops files"""
sbox.build()
svntest.main.file_append(os.path.join(sbox.wc_dir, 'iota'), 'newer line\n')
svntest.main.run_svn(None, 'ci', sbox.wc_dir, '--quiet', '-m', 'log msg')
svntest.main.file_append(os.path.join(sbox.wc_dir, 'iota'), 'newest line\n')
svntest.main.run_svn(None, 'ci', sbox.wc_dir, '--quiet', '-m', 'log msg')
rev_3 = fsfs_file(sbox.repo_dir, 'revs', '3')
rev_was_3 = rev_3 + '.was'
os.rename(rev_3, rev_was_3)
exit_code, output, errput = svntest.main.run_svnadmin("recover",
sbox.repo_dir)
if svntest.verify.verify_outputs(
"Output of 'svnadmin recover' is unexpected.", None, errput, None,
".*Expected current rev to be <= %s but found 3"
% (rev_3.endswith('pack') and '[012]' or '2')):
raise svntest.Failure
os.rename(rev_was_3, rev_3)
revprop_3 = fsfs_file(sbox.repo_dir, 'revprops', '3')
revprop_was_3 = revprop_3 + '.was'
os.rename(revprop_3, revprop_was_3)
exit_code, output, errput = svntest.main.run_svnadmin("recover",
sbox.repo_dir)
if svntest.verify.verify_outputs(
"Output of 'svnadmin recover' is unexpected.", None, errput, None,
".*Revision 3 has a revs file but no revprops file"):
raise svntest.Failure
os.rename(revprop_was_3, revprop_3)
os.rename(revprop_3, revprop_was_3)
os.mkdir(revprop_3)
exit_code, output, errput = svntest.main.run_svnadmin("recover",
sbox.repo_dir)
if svntest.verify.verify_outputs(
"Output of 'svnadmin recover' is unexpected.", None, errput, None,
".*Revision 3 has a non-file where its revprops file should be.*"):
raise svntest.Failure
def create_in_repo_subdir(sbox):
"'svnadmin create /path/to/repo/subdir'"
repo_dir = sbox.repo_dir
wc_dir = sbox.wc_dir
svntest.main.safe_rmtree(repo_dir, 1)
svntest.main.safe_rmtree(wc_dir)
svntest.main.create_repos(repo_dir)
try:
subdir = os.path.join(repo_dir, 'Z')
svntest.main.create_repos(subdir)
except svntest.main.SVNRepositoryCreateFailure:
return
raise svntest.Failure
test_list = [ None,
extra_headers,
extra_blockcontent,
inconsistent_headers,
empty_date,
dump_copied_dir,
dump_move_dir_modify_child,
dump_quiet,
hotcopy_dot,
hotcopy_format,
setrevprop,
verify_windows_paths_in_repos,
SkipUnless(verify_incremental_fsfs, svntest.main.is_fs_type_fsfs),
SkipUnless(recover_fsfs, svntest.main.is_fs_type_fsfs),
load_with_parent_dir,
set_uuid,
reflect_dropped_renumbered_revs,
SkipUnless(fsfs_recover_handle_missing_revs_or_revprops_file,
svntest.main.is_fs_type_fsfs),
create_in_repo_subdir,
]
if __name__ == '__main__':
svntest.main.run_tests(test_list)