import os
import shutil
import sys
import threading
import svntest
from svntest.verify import SVNExpectedStdout, SVNExpectedStderr
from svntest.verify import SVNUnexpectedStderr
from svntest.main import SVN_PROP_MERGEINFO
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 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 isinstance(dump, str):
dump = [ dump ]
exit_code, output, errput = svntest.main.run_command_stdin(
svntest.main.svnadmin_binary, expected_stderr, 0, 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)
@Issue(2729)
def empty_date(sbox):
"preserve date-less revisions in load"
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.options.server_minor_version >= 5:
if svntest.main.options.fsfs_sharding is None:
return os.path.join(repo_dir, 'db', kind, '0', rev)
else:
shard = int(rev) // svntest.main.options.fsfs_sharding
path = os.path.join(repo_dir, 'db', kind, str(shard), rev)
if svntest.main.options.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)
@SkipUnless(svntest.main.is_fs_type_fsfs)
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")
@SkipUnless(svntest.main.is_fs_type_fsfs)
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 = open(current_path).read()
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 = open(current_path).read()
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 = open(current_path).read()
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 = open(current_path).read()
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 = open(current_path).read()
svntest.verify.compare_and_display_lines(
"Contents of db/current is unexpected.",
'db/current', expected_current_contents, actual_current_contents)
@Issue(2983)
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 = open(dumpfile_location).read()
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')
svntest.actions.run_and_verify_svn(None,
['\n', 'Committed revision 11.\n'],
[], "mkdir", sbox.repo_url + "/sample-2",
"-m", "Create sample-2 dir")
load_and_verify_dumpstream(sbox, [], [], None, dumpfile, '--parent-dir',
'sample-2')
svntest.actions.run_and_verify_svn(None,
[sbox.repo_url +
"/sample-2/branch - " +
"/sample-2/trunk:15-17\n"],
[], 'propget', 'svn:mergeinfo', '-R',
sbox.repo_url + '/sample-2/branch')
svntest.actions.run_and_verify_svn(None,
[sbox.repo_url +
"/sample-2/branch1 - " +
"/sample-2/branch:16-19\n"],
[], 'propget', 'svn:mergeinfo', '-R',
sbox.repo_url + '/sample-2/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
@Issue(3020)
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()
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')
url = sbox.repo_url
expected_output = svntest.verify.UnorderedOutput([
url + "/trunk - /branch1:5-9\n",
url + "/toplevel/trunk - /toplevel/branch1:14-18\n",
])
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
@SkipUnless(svntest.main.is_fs_type_fsfs)
@Issue(2992)
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)
success = False
try:
subdir = os.path.join(repo_dir, 'Z')
svntest.main.create_repos(subdir)
except svntest.main.SVNRepositoryCreateFailure:
success = True
if not success:
raise svntest.Failure
cwd = os.getcwd()
success = False
try:
subdir = os.path.join(repo_dir, 'conf')
os.chdir(subdir)
svntest.main.create_repos('Z')
os.chdir(cwd)
except svntest.main.SVNRepositoryCreateFailure:
success = True
os.chdir(cwd)
if not success:
raise svntest.Failure
@SkipUnless(svntest.main.is_fs_type_fsfs)
def verify_with_invalid_revprops(sbox):
"svnadmin verify detects invalid revprops file"
repo_dir = sbox.repo_dir
svntest.main.safe_rmtree(repo_dir, 1)
svntest.main.create_repos(repo_dir)
exit_code, output, errput = svntest.main.run_svnadmin("verify",
sbox.repo_dir)
if svntest.verify.verify_outputs(
"Output of 'svnadmin verify' is unexpected.", None, errput, None,
".*Verified revision 0*"):
raise svntest.Failure
rp_file = open(os.path.join(repo_dir, 'db', 'revprops', '0', '0'), 'w')
rp_file.write('')
rp_file.close()
exit_code, output, errput = svntest.main.run_svnadmin("verify",
sbox.repo_dir)
if svntest.verify.verify_outputs(
"Output of 'svnadmin verify' is unexpected.", None, errput, None,
".*svnadmin: E200002:.*"):
raise svntest.Failure
@Issue(3020)
def dont_drop_valid_mergeinfo_during_incremental_loads(sbox):
"don't filter mergeinfo revs from incremental dump"
test_create(sbox)
dumpfile_full = open(os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'mergeinfo_included_full.dump')).read()
load_and_verify_dumpstream(sbox, [], [], None, dumpfile_full, '--ignore-uuid')
url = sbox.repo_url + '/branches/'
expected_output = svntest.verify.UnorderedOutput([
url + "B1 - /branches/B2:11-12\n",
"/trunk:6,9\n",
url + "B2 - /trunk:9\n",
url + "B1/B/E - /branches/B2/B/E:11-12\n",
"/trunk/B/E:5-6,8-9\n"])
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
dump_file_r1_10 = svntest.main.temp_dir + "-r1-10.dump"
exit_code, output, errput = svntest.main.run_svnadmin(
'dump', sbox.repo_dir, '-r1:10')
dump_fp = open(dump_file_r1_10, 'wb')
dump_fp.writelines(output)
dump_fp.close()
dump_file_r11_13 = svntest.main.temp_dir + "-r11-13.dump"
exit_code, output, errput = svntest.main.run_svnadmin(
'dump', sbox.repo_dir, '--incremental', '-r11:13')
dump_fp = open(dump_file_r11_13, 'wb')
dump_fp.writelines(output)
dump_fp.close()
dump_file_r14_15 = svntest.main.temp_dir + "-r14-15.dump"
exit_code, output, errput = svntest.main.run_svnadmin(
'dump', sbox.repo_dir, '--incremental', '-r14:15')
dump_fp = open(dump_file_r14_15, 'wb')
dump_fp.writelines(output)
dump_fp.close()
test_create(sbox)
load_and_verify_dumpstream(sbox, [], [], None,
open(dump_file_r1_10).read(),
'--ignore-uuid')
load_and_verify_dumpstream(sbox, [], [], None,
open(dump_file_r11_13).read(),
'--ignore-uuid')
load_and_verify_dumpstream(sbox, [], [], None,
open(dump_file_r14_15).read(),
'--ignore-uuid')
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
test_create(sbox)
dumpfile_skeleton = open(os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'skeleton_repos.dump')).read()
load_and_verify_dumpstream(sbox, [], [], None, dumpfile_skeleton,
'--ignore-uuid')
load_and_verify_dumpstream(sbox, [], [], None, dumpfile_full,
'--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/branches/B2:17-18\n",
"/Projects/Project-X/trunk:12,15\n",
url + "B2 - /Projects/Project-X/trunk:15\n",
url + "B1/B/E - /Projects/Project-X/branches/B2/B/E:17-18\n",
"/Projects/Project-X/trunk/B/E:11-12,14-15\n"])
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
test_create(sbox)
load_and_verify_dumpstream(sbox, [], [], None, dumpfile_skeleton,
'--ignore-uuid')
load_and_verify_dumpstream(sbox, [], [], None,
open(dump_file_r1_10).read(),
'--parent-dir', 'Projects/Project-X',
'--ignore-uuid')
load_and_verify_dumpstream(sbox, [], [], None,
open(dump_file_r11_13).read(),
'--parent-dir', 'Projects/Project-X',
'--ignore-uuid')
load_and_verify_dumpstream(sbox, [], [], None,
open(dump_file_r14_15).read(),
'--parent-dir', 'Projects/Project-X',
'--ignore-uuid')
svntest.actions.run_and_verify_svn(None, expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
@SkipUnless(svntest.main.is_posix_os)
@Issue(2591)
def hotcopy_symlink(sbox):
"'svnadmin hotcopy' replicates symlink"
original_repo = sbox.repo_dir
hotcopy_repo, hotcopy_url = sbox.add_repo_path('hotcopy')
svntest.main.safe_rmtree(original_repo, 1)
svntest.main.create_repos(original_repo)
svntest.main.safe_rmtree(sbox.wc_dir, 1)
os.mkdir(sbox.wc_dir)
external_file_path = os.path.join(sbox.wc_dir, "file")
svntest.main.file_write(external_file_path, "An existing file")
external_dir_path = os.path.join(sbox.wc_dir, "dir")
os.mkdir(external_dir_path)
external_missing_path = os.path.join(sbox.wc_dir, "missing")
symlinks = [
('in_repos_file', 'format'),
('in_repos_dir', 'conf'),
('in_repos_missing', 'missing'),
('external_file', os.path.join('..', '..', '..', external_file_path)),
('external_dir', os.path.join('..', '..', '..', external_dir_path)),
('external_missing', os.path.join('..', '..', '..', external_missing_path)),
]
for name, target_relpath in symlinks:
target_path = os.path.join(original_repo, target_relpath)
target_abspath = os.path.abspath(target_path)
symlink_path = os.path.join(original_repo, name)
os.symlink(target_relpath, symlink_path + '_rel')
os.symlink(target_abspath, symlink_path + '_abs')
svntest.actions.run_and_verify_svnadmin(
None, None, [],
"hotcopy", original_repo, hotcopy_repo)
for name, target_relpath in symlinks:
target_path = os.path.join(original_repo, target_relpath)
target_abspath = os.path.abspath(target_path)
symlink_path = os.path.join(hotcopy_repo, name)
if os.readlink(symlink_path + '_rel') != target_relpath:
raise svntest.Failure
if os.readlink(symlink_path + '_abs') != target_abspath:
raise svntest.Failure
def load_bad_props(sbox):
"svnadmin load with invalid svn: props"
dump_str = """SVN-fs-dump-format-version: 2
UUID: dc40867b-38f6-0310-9f5f-f81aa277e06f
Revision-number: 0
Prop-content-length: 56
Content-length: 56
K 8
svn:date
V 27
2005-05-03T19:09:41.129900Z
PROPS-END
Revision-number: 1
Prop-content-length: 99
Content-length: 99
K 7
svn:log
V 3
\n\r\n
K 10
svn:author
V 2
pl
K 8
svn:date
V 27
2005-05-03T19:10:19.975578Z
PROPS-END
Node-path: file
Node-kind: file
Node-action: add
Prop-content-length: 10
Text-content-length: 5
Text-content-md5: e1cbb0c3879af8347246f12c559a86b5
Content-length: 15
PROPS-END
text
"""
test_create(sbox)
load_and_verify_dumpstream(sbox, [], svntest.verify.AnyOutput,
dumpfile_revisions, dump_str,
'--ignore-uuid')
svntest.actions.load_repo(sbox, dump_str=dump_str,
bypass_prop_validation=True)
@SkipUnless(svntest.main.is_fs_type_fsfs)
@SkipUnless(svntest.main.server_enforces_UTF8_fspaths_in_verify)
def verify_non_utf8_paths(sbox):
"svnadmin verify with non-UTF-8 paths"
dumpfile = clean_dumpfile()
test_create(sbox)
load_and_verify_dumpstream(sbox, [], [], dumpfile_revisions, dumpfile,
'--ignore-uuid')
path1 = os.path.join(sbox.repo_dir, "db", "revs", "0", "1")
path_new = os.path.join(sbox.repo_dir, "db", "revs", "0", "1.new")
fp1 = open(path1, 'rb')
fp_new = open(path_new, 'wb')
for line in fp1.readlines():
if line == "A\n":
fp_new.write("\xE6\n")
elif line == "text: 1 279 32 32 d63ecce65d8c428b86f4f8b0920921fe\n":
fp_new.write("text: 1 279 32 32 b50b1d5ed64075b5f632f3b8c30cd6b2\n")
elif line == "cpath: /A\n":
fp_new.write("cpath: /\xE6\n")
elif line == "_0.0.t0-0 add-file true true /A\n":
fp_new.write("_0.0.t0-0 add-file true true /\xE6\n")
else:
fp_new.write(line)
fp1.close()
fp_new.close()
os.remove(path1)
os.rename(path_new, path1)
exit_code, output, errput = svntest.main.run_svnadmin("verify",
sbox.repo_dir)
svntest.verify.verify_outputs(
"Unexpected error while running 'svnadmin verify'.",
[], errput, None, ".*Path '.*' is not in UTF-8.*")
expected_stderr = [
"* Dumped revision 0.\n",
"WARNING 0x0002: E160005: "
"While validating fspath '?\\230': "
"Path '?\\230' is not in UTF-8"
"\n",
"* Dumped revision 1.\n",
]
exit_code, output, errput = svntest.main.run_svnadmin("dump", sbox.repo_dir)
if svntest.verify.compare_and_display_lines(
"Output of 'svnadmin dump' is unexpected.",
'STDERR', expected_stderr, errput):
raise svntest.Failure
@SkipUnless(svntest.main.is_threaded_python)
@Issue(4129)
def mergeinfo_race(sbox):
"concurrent mergeinfo commits invalidate pred-count"
sbox.build()
wc_dir = sbox.wc_dir
wc2_dir = sbox.add_wc_path('2')
svntest.main.run_svn(None, 'checkout', '-q', sbox.repo_url, wc2_dir)
svntest.main.run_svn(None, 'mkdir', sbox.ospath('d1', wc_dir))
svntest.main.run_svn(None, 'mkdir', sbox.ospath('d2', wc2_dir))
svntest.main.run_svn(None, 'ps', 'svn:mergeinfo', '/P:42', sbox.ospath('A', wc_dir))
svntest.main.run_svn(None, 'ps', 'svn:mergeinfo', '/Q:42', sbox.ospath('iota', wc2_dir))
def makethread(some_wc_dir):
def worker():
svntest.main.run_svn(None, 'commit', '-mm', some_wc_dir)
return worker
t1 = threading.Thread(None, makethread(wc_dir))
t2 = threading.Thread(None, makethread(wc2_dir))
t1.start(); t2.start();
t1.join(); t2.join();
if svntest.actions.run_and_parse_info(sbox.repo_url)[0]['Revision'] != '3':
raise svntest.Failure("one or both commits failed")
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,
verify_incremental_fsfs,
recover_fsfs,
load_with_parent_dir,
set_uuid,
reflect_dropped_renumbered_revs,
fsfs_recover_handle_missing_revs_or_revprops_file,
create_in_repo_subdir,
verify_with_invalid_revprops,
dont_drop_valid_mergeinfo_during_incremental_loads,
hotcopy_symlink,
load_bad_props,
verify_non_utf8_paths,
mergeinfo_race,
]
if __name__ == '__main__':
svntest.main.run_tests(test_list)