Google Git
Sign in
apache / subversion / refs/tags/1.14.1 / . / subversion / tests / cmdline / svnadmin_tests.py
blob: e96eee50d087c77586bebe1fef031ee76082fff7 [file] [log] [blame]
#!/usr/bin/env python
#
# svnadmin_tests.py: testing the 'svnadmin' tool.
#
# Subversion is a tool for revision control.
# See http://subversion.apache.org for more information.
#
# ====================================================================
# 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.
######################################################################
# General modules
import os
import logging
import re
import shutil
import sys
import threading
import time
import gzip
logger = logging.getLogger()
# Our testing module
import svntest
from svntest.verify import SVNExpectedStdout, SVNExpectedStderr
from svntest.verify import SVNUnexpectedStderr
from svntest.verify import UnorderedOutput
from svntest.main import SVN_PROP_MERGEINFO
# (abbreviation)
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
SkipDumpLoadCrossCheck = svntest.testcase.SkipDumpLoadCrossCheck_deco
Item = svntest.wc.StateItem
def read_rep_cache(repo_dir):
"""Return the rep-cache contents as a dict {hash: (rev, index, ...)}.
"""
db_path = os.path.join(repo_dir, 'db', 'rep-cache.db')
db1 = svntest.sqlite3.connect(db_path)
schema1 = db1.execute("pragma user_version").fetchone()[0]
# Can't test newer rep-cache schemas with an old built-in SQLite; see the
# documentation of STMT_CREATE_SCHEMA_V2 in ../../libsvn_fs_fs/rep-cache-db.sql
if schema1 >= 2 and svntest.sqlite3.sqlite_version_info < (3, 8, 2):
raise svntest.Failure("Can't read rep-cache schema %d using old "
"Python-SQLite version %s < (3,8,2)" %
(schema1,
svntest.sqlite3.sqlite_version_info))
content = { row[0]: row[1:] for row in
db1.execute("select * from rep_cache") }
return content
def check_hotcopy_bdb(src, dst):
"Verify that the SRC BDB repository has been correctly copied to DST."
### TODO: This function should be extended to verify all hotcopied files,
### not just compare the output of 'svnadmin dump'. See check_hotcopy_fsfs().
exit_code, origout, origerr = svntest.main.run_svnadmin("dump", src,
'--quiet')
exit_code, backout, backerr = svntest.main.run_svnadmin("dump", dst,
'--quiet')
if origerr or backerr or origout != backout:
raise svntest.Failure
def check_hotcopy_fsfs_fsx(src, dst):
# Walk the source and compare all files to the destination
for src_dirpath, src_dirs, src_files in os.walk(src):
# Verify that the current directory exists in the destination
dst_dirpath = src_dirpath.replace(src, dst)
if not os.path.isdir(dst_dirpath):
raise svntest.Failure("%s does not exist in hotcopy "
"destination" % dst_dirpath)
# Verify that all dirents in the current directory also exist in source
for dst_dirent in os.listdir(dst_dirpath):
# Ignore auto-created empty lock files as they may or may not
# be present and are neither required by nor do they harm to
# the destination repository.
if dst_dirent == 'pack-lock':
continue
if dst_dirent == 'write-lock':
continue
# Ignore auto-created rep-cache.db-journal file
if dst_dirent == 'rep-cache.db-journal':
continue
src_dirent = os.path.join(src_dirpath, dst_dirent)
if not os.path.exists(src_dirent):
raise svntest.Failure("%s does not exist in hotcopy "
"source" % src_dirent)
# Compare all files in this directory
for src_file in src_files:
# Ignore auto-created empty lock files as they may or may not
# be present and are neither required by nor do they harm to
# the destination repository.
if src_file == 'pack-lock':
continue
if src_file == 'write-lock':
continue
# Ignore auto-created rep-cache.db-journal file
if src_file == 'rep-cache.db-journal':
continue
src_path = os.path.join(src_dirpath, src_file)
dst_path = os.path.join(dst_dirpath, src_file)
if not os.path.isfile(dst_path):
raise svntest.Failure("%s does not exist in hotcopy "
"destination" % dst_path)
# Special case for db/uuid: Only the UUID in the first line needs
# to match. Source and target must have the same number of lines
# (due to having the same format).
if src_path == os.path.join(src, 'db', 'uuid'):
lines1 = open(src_path, 'rb').read().split(b"\n")
lines2 = open(dst_path, 'rb').read().split(b"\n")
if len(lines1) != len(lines2):
raise svntest.Failure("%s differs in number of lines"
% dst_path)
if lines1[0] != lines2[0]:
raise svntest.Failure("%s contains different uuid: '%s' vs. '%s'"
% (dst_path, lines1[0], lines2[0]))
continue
# Special case for rep-cache: It will always differ in a byte-by-byte
# comparison, so compare db tables instead.
if src_file == 'rep-cache.db':
db1 = svntest.sqlite3.connect(src_path)
db2 = svntest.sqlite3.connect(dst_path)
schema1 = db1.execute("pragma user_version").fetchone()[0]
schema2 = db2.execute("pragma user_version").fetchone()[0]
if schema1 != schema2:
raise svntest.Failure("rep-cache schema differs: '%s' vs. '%s'"
% (schema1, schema2))
# Can't test newer rep-cache schemas with an old built-in SQLite.
if schema1 >= 2 and svntest.sqlite3.sqlite_version_info < (3, 8, 2):
continue
rows1 = []
rows2 = []
for row in db1.execute("select * from rep_cache order by hash"):
rows1.append(row)
for row in db2.execute("select * from rep_cache order by hash"):
rows2.append(row)
if len(rows1) != len(rows2):
raise svntest.Failure("number of rows in rep-cache differs")
for i in range(len(rows1)):
if rows1[i] != rows2[i]:
raise svntest.Failure("rep-cache row %i differs: '%s' vs. '%s'"
% (i, rows1[i], rows2[i]))
continue
# Special case for revprop-generation: It will always be zero in
# the hotcopy destination (i.e. a fresh cache generation)
if src_file == 'revprop-generation':
f2 = open(dst_path, 'r')
revprop_gen = int(f2.read().strip())
if revprop_gen != 0:
raise svntest.Failure("Hotcopy destination has non-zero " +
"revprop generation")
continue
f1 = open(src_path, 'rb')
f2 = open(dst_path, 'rb')
while True:
offset = 0
BUFSIZE = 1024
buf1 = f1.read(BUFSIZE)
buf2 = f2.read(BUFSIZE)
if not buf1 or not buf2:
if not buf1 and not buf2:
# both at EOF
break
elif buf1:
raise svntest.Failure("%s differs at offset %i" %
(dst_path, offset))
elif buf2:
raise svntest.Failure("%s differs at offset %i" %
(dst_path, offset))
if len(buf1) != len(buf2):
raise svntest.Failure("%s differs in length" % dst_path)
for i in range(len(buf1)):
if buf1[i] != buf2[i]:
raise svntest.Failure("%s differs at offset %i"
% (dst_path, offset))
offset += 1
f1.close()
f2.close()
def check_hotcopy_fsfs(src, dst):
"Verify that the SRC FSFS repository has been correctly copied to DST."
check_hotcopy_fsfs_fsx(src, dst)
def check_hotcopy_fsx(src, dst):
"Verify that the SRC FSX repository has been correctly copied to DST."
check_hotcopy_fsfs_fsx(src, dst)
#----------------------------------------------------------------------
# How we currently test 'svnadmin' --
#
# 'svnadmin create': Create an empty repository, test that the
# root node has a proper created-revision,
# because there was once a bug where it
# didn't.
#
# Note also that "svnadmin create" is tested
# implicitly every time we run a python test
# script. (An empty repository is always
# created and then imported into; if this
# subcommand failed catastrophically, every
# test would fail and we would know instantly.)
#
# 'svnadmin createtxn'
# 'svnadmin rmtxn': See below.
#
# 'svnadmin lstxns': We don't care about the contents of transactions;
# we only care that they exist or not.
# Therefore, we can simply parse transaction headers.
#
# 'svnadmin dump': A couple regression tests that ensure dump doesn't
# error out, and one to check that the --quiet option
# really does what it's meant to do. The actual
# contents of the dump aren't verified at all.
#
# ### TODO: someday maybe we could parse the contents of trees too.
#
######################################################################
# Helper routines
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 patch_format(repo_dir, shard_size):
"""Rewrite the format of the FSFS or FSX repository REPO_DIR so
that it would use sharding with SHARDS revisions per shard."""
format_path = os.path.join(repo_dir, "db", "format")
contents = open(format_path, 'rb').read()
processed_lines = []
for line in contents.split(b"\n"):
if line.startswith(b"layout "):
processed_lines.append(("layout sharded %d" % shard_size).encode())
else:
processed_lines.append(line)
new_contents = b"\n".join(processed_lines)
os.chmod(format_path, svntest.main.S_ALL_RW)
with open(format_path, 'wb') as f:
f.write(new_contents)
def is_sharded(repo_dir):
"""Return whether the FSFS repository REPO_DIR is sharded."""
format_path = os.path.join(repo_dir, "db", "format")
contents = open(format_path, 'rb').read()
for line in contents.split(b"\n"):
if line.startswith(b"layout sharded"):
return True
return False
def load_and_verify_dumpstream(sbox, expected_stdout, expected_stderr,
revs, check_props, 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. If CHECK_PROPS is True, check properties
of each rev's items. VARARGS are optional arguments passed to the
'load' command."""
dump = svntest.main.ensure_list(dump)
exit_code, output, errput = svntest.main.run_command_stdin(
svntest.main.svnadmin_binary, expected_stderr, 0, True, dump,
'load', '--quiet', sbox.repo_dir, *varargs)
if expected_stdout:
if expected_stdout is 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 is svntest.verify.AnyOutput:
if len(errput) == 0:
raise SVNExpectedStderr
else:
svntest.verify.compare_and_display_lines(
"Standard error output", "STDERR:", expected_stderr, errput)
# The expected error occurred, so don't try to verify the result
return
if revs:
# verify revs as wc states
for rev in range(len(revs)):
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [],
"update", "-r%s" % (rev+1),
sbox.wc_dir)
rev_tree = revs[rev]
svntest.actions.verify_disk(sbox.wc_dir, rev_tree, check_props)
def load_dumpstream(sbox, dump, *varargs):
"Load dump text without verification."
return load_and_verify_dumpstream(sbox, None, None, None, False, dump,
*varargs)
class FSFS_Index:
"""Manages indexes of a rev file in a FSFS format 7 repository.
The interface returns P2L information and allows for item offsets
and lengths to be modified. """
def __init__(self, sbox, revision):
self.by_item = { }
self.revision = revision
self.repo_dir = sbox.repo_dir
self._read()
def _read(self):
""" Read P2L index using svnfsfs. """
exit_code, output, errput = svntest.main.run_svnfsfs('dump-index',
'-r' + str(self.revision),
self.repo_dir)
svntest.verify.verify_outputs("Error while dumping index",
[], errput, [], [])
svntest.verify.verify_exit_code(None, exit_code, 0)
self.by_item.clear()
for line in output:
values = line.split()
if len(values) >= 4 and values[0] != 'Start':
item = int(values[4])
self.by_item[item] = values
def _write(self):
""" Rewrite indexes using svnfsfs. """
by_offset = {}
for key in self.by_item:
values = self.by_item[key]
by_offset[int(values[0], 16)] = values
lines = []
for (offset, values) in sorted(by_offset.items()):
values = by_offset[offset]
line = values[0] + ' ' + values[1] + ' ' + values[2] + ' ' + \
values[3] + ' ' + values[4] + '\n';
lines.append(line.encode())
exit_code, output, errput = svntest.main.run_command_stdin(
svntest.main.svnfsfs_binary, 0, 0, False, lines,
'load-index', self.repo_dir)
svntest.verify.verify_outputs("Error while rewriting index",
output, errput, [], [])
svntest.verify.verify_exit_code(None, exit_code, 0)
def get_item(self, item):
""" Return offset, length and type of ITEM. """
values = self.by_item[item]
offset = int(values[0], 16)
len = int(values[1], 16)
type = values[2]
return (offset, len, type)
def modify_item(self, item, offset, len):
""" Modify offset and length of ITEM. """
values = self.by_item[item]
values[0] = '%x' % offset
values[1] = '%x' % len
self._write()
def repo_format(sbox):
""" Return the repository format number for SBOX."""
format_file = open(os.path.join(sbox.repo_dir, "db", "format"))
format = int(format_file.read()[:1])
format_file.close()
return format
def set_changed_path_list(sbox, revision, changes):
""" Replace the changed paths list in the revision file REVISION in SBOX
with the text CHANGES."""
idx = None
# read full file
fp = open(fsfs_file(sbox.repo_dir, 'revs', str(revision)), 'r+b')
contents = fp.read()
length = len(contents)
if repo_format(sbox) < 7:
# replace the changed paths list
header = contents[contents.rfind(b'\n', length - 64, length - 1):]
body_len = int(header.split(b' ')[1])
else:
# read & parse revision file footer
footer_length = contents[length-1];
if isinstance(footer_length, str):
footer_length = ord(footer_length)
footer = contents[length - footer_length - 1:length-1]
l2p_offset = int(footer.split(b' ')[0])
l2p_checksum = footer.split(b' ')[1]
p2l_offset = int(footer.split(b' ')[2])
p2l_checksum = footer.split(b' ')[3]
idx = FSFS_Index(sbox, revision)
(offset, item_len, item_type) = idx.get_item(1)
# split file contents
body_len = offset
indexes = contents[l2p_offset:length - footer_length - 1]
# construct new footer, include indexes as are
file_len = body_len + len(changes) + 1
p2l_offset += file_len - l2p_offset
header = str(file_len).encode() + b' ' + l2p_checksum + b' ' \
+ str(p2l_offset).encode() + b' ' + p2l_checksum
header += bytes([len(header)])
header = b'\n' + indexes + header
contents = contents[:body_len] + changes + header
# set new contents
fp.seek(0)
fp.write(contents)
fp.truncate()
fp.close()
if repo_format(sbox) >= 7:
idx.modify_item(1, offset, len(changes) + 1)
######################################################################
# Tests
#----------------------------------------------------------------------
# dump stream tests need a dump file
def clean_dumpfile():
return \
[ b"SVN-fs-dump-format-version: 2\n\n",
b"UUID: 668cc64a-31ed-0310-8ccb-b75d75bb44e3\n\n",
b"Revision-number: 0\n",
b"Prop-content-length: 56\n",
b"Content-length: 56\n\n",
b"K 8\nsvn:date\nV 27\n2005-01-08T21:48:13.838745Z\nPROPS-END\n\n\n",
b"Revision-number: 1\n",
b"Prop-content-length: 98\n",
b"Content-length: 98\n\n",
b"K 7\nsvn:log\nV 0\n\nK 10\nsvn:author\nV 4\nerik\n",
b"K 8\nsvn:date\nV 27\n2005-01-08T21:51:16.313791Z\nPROPS-END\n\n\n",
b"Node-path: A\n",
b"Node-kind: file\n",
b"Node-action: add\n",
b"Prop-content-length: 35\n",
b"Text-content-length: 5\n",
b"Text-content-md5: e1cbb0c3879af8347246f12c559a86b5\n",
b"Content-length: 40\n\n",
b"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"
sbox.build(empty=True)
dumpfile = clean_dumpfile()
dumpfile[3:3] = \
[ b"X-Comment-Header: Ignored header normally not in dump stream\n" ]
load_and_verify_dumpstream(sbox,[],[], dumpfile_revisions, False, dumpfile,
'--ignore-uuid')
#----------------------------------------------------------------------
# Ensure loading continues after skipping a bit of unknown extra content.
def extra_blockcontent(sbox):
"load success on oversized Content-length"
sbox.build(empty=True)
dumpfile = clean_dumpfile()
# Replace "Content-length" line with two lines
dumpfile[8:9] = \
[ b"Extra-content-length: 10\n",
b"Content-length: 108\n\n" ]
# Insert the extra content after "PROPS-END\n"
dumpfile[11] = dumpfile[11][:-2] + b"extra text\n\n\n"
load_and_verify_dumpstream(sbox,[],[], dumpfile_revisions, False, dumpfile,
'--ignore-uuid')
#----------------------------------------------------------------------
def inconsistent_headers(sbox):
"load failure on undersized Content-length"
sbox.build(empty=True)
dumpfile = clean_dumpfile()
dumpfile[-2] = b"Content-length: 30\n\n"
load_and_verify_dumpstream(sbox, [], svntest.verify.AnyOutput,
dumpfile_revisions, False, dumpfile)
#----------------------------------------------------------------------
# Test for issue #2729: Datestamp-less revisions in dump streams do
# not remain so after load
@Issue(2729)
def empty_date(sbox):
"preserve date-less revisions in load"
sbox.build(empty=True)
dumpfile = clean_dumpfile()
# Replace portions of the revision data to drop the svn:date revprop.
dumpfile[7:11] = \
[ b"Prop-content-length: 52\n",
b"Content-length: 52\n\n",
b"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, False, dumpfile,
'--ignore-uuid')
# Verify that the revision still lacks the svn:date property.
svntest.actions.run_and_verify_svn([], '.*(E195011|E200017).*svn:date',
"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)
sbox.simple_commit(message='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')
sbox.simple_commit(message='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, dump, 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, [],
"hotcopy", os.path.join(cwd, sbox.repo_dir), '.')
os.chdir(cwd)
if svntest.main.is_fs_type_fsfs():
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
if svntest.main.is_fs_type_bdb():
check_hotcopy_bdb(sbox.repo_dir, backup_dir)
if svntest.main.is_fs_type_fsx():
check_hotcopy_fsx(sbox.repo_dir, backup_dir)
#----------------------------------------------------------------------
# This test is redundant for FSFS. The hotcopy_dot and hotcopy_incremental
# tests cover this check for FSFS already.
@SkipUnless(svntest.main.is_fs_type_bdb)
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:
logger.warn("Error: hotcopy failed")
raise svntest.Failure
# verify that the db/format files are the same
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:
logger.warn("Error: db/format file contents do not match after hotcopy")
raise svntest.Failure
#----------------------------------------------------------------------
def setrevprop(sbox):
"setlog, setrevprop, delrevprop; bypass hooks"
sbox.build()
# Try a simple log property modification.
iota_path = os.path.join(sbox.wc_dir, "iota")
mu_path = sbox.ospath('A/mu')
svntest.actions.run_and_verify_svnadmin([], [],
"setlog", sbox.repo_dir, "-r0",
"--bypass-hooks",
iota_path)
# Make sure it fails without --bypass-hooks. (We haven't called
# svntest.actions.enable_revprop_changes().)
#
# Note that we attempt to set the log message to a different value than the
# successful call.
svntest.actions.run_and_verify_svnadmin([], svntest.verify.AnyOutput,
"setlog", sbox.repo_dir, "-r0",
mu_path)
# Verify that the revprop value matches what we set when retrieved
# through the client.
svntest.actions.run_and_verify_svn([ "This is the file 'iota'.\n", "\n" ],
[], "propget", "--revprop", "-r0",
"svn:log", sbox.wc_dir)
# Try an author property modification.
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:
logger.warn("Error: 'setrevprop' failed")
raise svntest.Failure
# Verify that the revprop value matches what we set when retrieved
# through the client.
svntest.actions.run_and_verify_svn([ "foo\n" ], [], "propget",
"--revprop", "-r0", "svn:author",
sbox.wc_dir)
# Delete the property.
svntest.actions.run_and_verify_svnadmin([], [],
"delrevprop", "-r0", sbox.repo_dir,
"svn:author")
svntest.actions.run_and_verify_svnlook([], ".*E200017.*svn:author.*",
"propget", "--revprop", "-r0",
sbox.repo_dir, "svn:author")
def verify_windows_paths_in_repos(sbox):
"verify a repository containing paths like 'c:hi'"
# setup a repo with a directory '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, [],
'mkdir', '-m', 'log_msg',
chi_url)
exit_code, output, errput = svntest.main.run_svnadmin("verify",
sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
# unfortunately, some backends needs to do more checks than other
# resulting in different progress output
if svntest.main.is_fs_log_addressing():
svntest.verify.compare_and_display_lines(
"Error while running 'svnadmin verify'.",
'STDOUT', ["* Verifying metadata at revision 0 ...\n",
"* Verifying repository metadata ...\n",
"* Verified revision 0.\n",
"* Verified revision 1.\n",
"* Verified revision 2.\n"], output)
elif svntest.main.fs_has_rep_sharing() and not svntest.main.is_fs_type_bdb():
svntest.verify.compare_and_display_lines(
"Error while running 'svnadmin verify'.",
'STDOUT', ["* Verifying repository metadata ...\n",
"* Verified revision 0.\n",
"* Verified revision 1.\n",
"* Verified revision 2.\n"], output)
else:
svntest.verify.compare_and_display_lines(
"Error while running 'svnadmin verify'.",
'STDOUT', ["* Verified revision 0.\n",
"* Verified revision 1.\n",
"* Verified revision 2.\n"], output)
#----------------------------------------------------------------------
# Returns the filename of the rev or revprop file (according to KIND)
# numbered REV in REPO_DIR, which must be in the first shard if we're
# using a sharded repository.
def fsfs_file(repo_dir, kind, rev):
if svntest.main.options.server_minor_version >= 5:
if svntest.main.options.fsfs_sharding is None:
if svntest.main.is_fs_type_fsx():
rev = 'r' + rev
return os.path.join(repo_dir, 'db', kind, '0', rev)
else:
shard = int(rev) // svntest.main.options.fsfs_sharding
if svntest.main.is_fs_type_fsx():
rev = 'r' + rev
path = os.path.join(repo_dir, 'db', kind, str(shard), rev)
if svntest.main.options.fsfs_packing is None or kind == 'revprops':
# we don't pack revprops
return path
elif os.path.exists(path):
# rev exists outside a pack file.
return path
else:
# didn't find the plain file; assume it's in a pack file
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"""
if svntest.main.options.fsfs_version is not None and \
svntest.main.options.fsfs_version not in [4, 6]:
raise svntest.Skip("Unsupported prepackaged repository version")
# setup a repo with a directory 'c:hi'
# use physical addressing as this is hard to provoke with logical addressing
sbox.build(create_wc = False,
minor_version = min(svntest.main.options.server_minor_version,8))
repo_url = sbox.repo_url
E_url = sbox.repo_url + '/A/B/E'
# Create A/B/E/bravo in r2.
svntest.actions.run_and_verify_svn(None, [],
'mkdir', '-m', 'log_msg',
E_url + '/bravo')
# Corrupt r2's reference to A/C by replacing "dir 7-1.0.r1/1568" with
# "dir 7-1.0.r1/1569" (increment offset) and updating the checksum for
# this directory listing to "c9b5a2d26473a4e28088673dda9df804" so that
# the listing itself is valid.
r2 = fsfs_file(sbox.repo_dir, 'revs', '2')
if r2.endswith('pack'):
raise svntest.Skip("Test doesn't handle packed revisions")
fp = open(r2, 'wb')
fp.write(b"""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"
"|.*Missing id field in node-rev.*")
#----------------------------------------------------------------------
# Helper for two test functions.
def corrupt_and_recover_db_current(sbox, minor_version=None):
"""Build up a MINOR_VERSION sandbox and test different recovery scenarios
with missing, out-of-date or even corrupt db/current files. Recovery should
behave the same way with all values of MINOR_VERSION, hence this helper
containing the common code that allows us to check it."""
sbox.build(minor_version=minor_version)
current_path = os.path.join(sbox.repo_dir, 'db', 'current')
# Commit up to r3, so we can test various recovery scenarios.
svntest.main.file_append(os.path.join(sbox.wc_dir, 'iota'), 'newer line\n')
sbox.simple_commit(message='log msg')
svntest.main.file_append(os.path.join(sbox.wc_dir, 'iota'), 'newest line\n')
sbox.simple_commit(message='log msg')
# Remember the contents of the db/current file.
expected_current_contents = open(current_path).read()
# Move aside the current file for r3.
os.rename(os.path.join(sbox.repo_dir, 'db','current'),
os.path.join(sbox.repo_dir, 'db','was_current'))
# Run 'svnadmin recover' and check that the current file is recreated.
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)
# Now try writing db/current to be one rev lower than it should be.
svntest.main.file_write(current_path, '2\n')
# Run 'svnadmin recover' and check that the current file is fixed.
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)
# Now try writing db/current to be *two* revs lower than it should be.
svntest.main.file_write(current_path, '1\n')
# Run 'svnadmin recover' and check that the current file is fixed.
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)
# Now try writing db/current to be fish revs lower than it should be.
#
# Note: I'm not actually sure it's wise to recover from this, but
# detecting it would require rewriting fs_fs.c:get_youngest() to
# check the actual contents of its buffer, since atol() will happily
# convert "fish" to 0.
svntest.main.file_write(current_path, 'fish\n')
# Run 'svnadmin recover' and check that the current file is fixed.
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)
@SkipUnless(svntest.main.is_fs_type_fsfs)
def fsfs_recover_db_current(sbox):
"fsfs recover db/current"
corrupt_and_recover_db_current(sbox)
@SkipUnless(svntest.main.is_fs_type_fsfs)
def fsfs_recover_old_db_current(sbox):
"fsfs recover db/current --compatible-version=1.3"
# Around trunk@1573728, 'svnadmin recover' wrongly errored out
# for the --compatible-version=1.3 repositories with missing or
# invalid db/current file:
# svnadmin: E160006: No such revision 1
corrupt_and_recover_db_current(sbox, minor_version=3)
#----------------------------------------------------------------------
@Issue(2983)
def load_with_parent_dir(sbox):
"'svnadmin load --parent-dir' reparents mergeinfo"
## See https://issues.apache.org/jira/browse/SVN-2983. ##
sbox.build(empty=True)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'mergeinfo_included.dump')
dumpfile = svntest.actions.load_dumpfile(dumpfile_location)
# Create 'sample' dir in sbox.repo_url, and load the dump stream there.
svntest.actions.run_and_verify_svn(['Committing transaction...\n',
'Committed revision 1.\n'],
[], "mkdir", sbox.repo_url + "/sample",
"-m", "Create sample dir")
load_dumpstream(sbox, dumpfile, '--parent-dir', '/sample')
# Verify the svn:mergeinfo properties for '--parent-dir'
svntest.actions.run_and_verify_svn([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([sbox.repo_url +
"/sample/branch1 - " +
"/sample/branch:6-9\n"],
[], 'propget', 'svn:mergeinfo', '-R',
sbox.repo_url + '/sample/branch1')
# Create 'sample-2' dir in sbox.repo_url, and load the dump stream again.
# This time, don't include a leading slash on the --parent-dir argument.
# See issue #3547.
svntest.actions.run_and_verify_svn(['Committing transaction...\n',
'Committed revision 11.\n'],
[], "mkdir", sbox.repo_url + "/sample-2",
"-m", "Create sample-2 dir")
load_dumpstream(sbox, dumpfile, '--parent-dir', 'sample-2')
# Verify the svn:mergeinfo properties for '--parent-dir'.
svntest.actions.run_and_verify_svn([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([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)
# Squirrel away the original repository UUID.
exit_code, output, errput = svntest.main.run_svnlook('uuid', sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
orig_uuid = output[0].rstrip()
# Try setting a new, bogus UUID.
svntest.actions.run_and_verify_svnadmin(None, '^.*Malformed UUID.*$',
'setuuid', sbox.repo_dir, 'abcdef')
# Try generating a brand new UUID.
svntest.actions.run_and_verify_svnadmin([], 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:
logger.warn("Error: new UUID matches the original one")
raise svntest.Failure
# Now, try setting the UUID back to the original value.
svntest.actions.run_and_verify_svnadmin([], 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:
logger.warn("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"
## See https://issues.apache.org/jira/browse/SVN-3020. ##
sbox.build(empty=True)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svndumpfilter_tests_data',
'with_merges.dump')
dumpfile = svntest.actions.load_dumpfile(dumpfile_location)
# Create 'toplevel' dir in sbox.repo_url
svntest.actions.run_and_verify_svn(['Committing transaction...\n',
'Committed revision 1.\n'],
[], "mkdir", sbox.repo_url + "/toplevel",
"-m", "Create toplevel dir")
# Load the dump stream in sbox.repo_url
load_dumpstream(sbox, dumpfile)
# Load the dump stream in toplevel dir
load_dumpstream(sbox, dumpfile, '--parent-dir', '/toplevel')
# Verify the svn:mergeinfo properties
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(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"""
# Set up a repository containing the greek tree.
sbox.build()
# Commit up to r3, so we can test various recovery scenarios.
svntest.main.file_append(os.path.join(sbox.wc_dir, 'iota'), 'newer line\n')
sbox.simple_commit(message='log msg')
svntest.main.file_append(os.path.join(sbox.wc_dir, 'iota'), 'newest line\n')
sbox.simple_commit(message='log msg')
rev_3 = fsfs_file(sbox.repo_dir, 'revs', '3')
rev_was_3 = rev_3 + '.was'
# Move aside the revs file for r3.
os.rename(rev_3, rev_was_3)
# Verify 'svnadmin recover' fails when youngest has a revprops
# file but no revs file.
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"
# For example, if svntest.main.fsfs_sharding == 2, then rev_3 would
# be the pack file for r2:r3, and the error message would report "<= 1".
% (rev_3.endswith('pack') and '[012]' or '2')):
raise svntest.Failure
# Restore the r3 revs file, thus repairing the repository.
os.rename(rev_was_3, rev_3)
revprop_3 = fsfs_file(sbox.repo_dir, 'revprops', '3')
revprop_was_3 = revprop_3 + '.was'
# Move aside the revprops file for r3.
os.rename(revprop_3, revprop_was_3)
# Verify 'svnadmin recover' fails when youngest has a revs file
# but no revprops file (issue #2992).
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
# Restore the r3 revprops file, thus repairing the repository.
os.rename(revprop_was_3, revprop_3)
# Change revprops file to a directory for revision 3
os.rename(revprop_3, revprop_was_3)
os.mkdir(revprop_3)
# Verify 'svnadmin recover' fails when youngest has a revs file
# but revprops file is not a file (another aspect of issue #2992).
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
# Restore the r3 revprops file, thus repairing the repository.
os.rmdir(revprop_3)
os.rename(revprop_was_3, revprop_3)
#----------------------------------------------------------------------
@Skip(svntest.main.tests_use_prepackaged_repository)
def create_in_repo_subdir(sbox):
"'svnadmin create /path/to/repo/subdir'"
sbox.build(create_wc=False, empty=True)
repo_dir = sbox.repo_dir
success = False
try:
# This should fail
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:
# This should fail, too
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)
@SkipDumpLoadCrossCheck()
def verify_with_invalid_revprops(sbox):
"svnadmin verify detects invalid revprops file"
sbox.build(create_wc=False, empty=True)
repo_dir = sbox.repo_dir
# Run a test verify
exit_code, output, errput = svntest.main.run_svnadmin("verify",
sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
if svntest.verify.verify_outputs(
"Output of 'svnadmin verify' is unexpected.", None, output, None,
".*Verified revision 0*"):
raise svntest.Failure
# Empty the revprops file
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
#----------------------------------------------------------------------
# Even *more* testing for issue #3020 'Reflect dropped/renumbered
# revisions in svn:mergeinfo data during svnadmin load'
#
# Full or incremental dump-load cycles should result in the same
# mergeinfo in the loaded repository.
#
# Given a repository 'SOURCE-REPOS' with mergeinfo, and a repository
# 'TARGET-REPOS' (which may or may not be empty), either of the following
# methods to move 'SOURCE-REPOS' to 'TARGET-REPOS' should result in
# the same mergeinfo on 'TARGET-REPOS':
#
# 1) Dump -r1:HEAD from 'SOURCE-REPOS' and load it in one shot to
# 'TARGET-REPOS'.
#
# 2) Dump 'SOURCE-REPOS' in a series of incremental dumps and load
# each of them to 'TARGET-REPOS'.
#
# See https://issues.apache.org/jira/browse/SVN-3020#desc13
@Issue(3020)
def dont_drop_valid_mergeinfo_during_incremental_loads(sbox):
"don't filter mergeinfo revs from incremental dump"
# Create an empty repos.
sbox.build(empty=True)
# PART 1: Load a full dump to an empty repository.
#
# The test repository used here, 'mergeinfo_included_full.dump', is
# this repos:
# __________________________________________
# | |
# | ____________________________|_____
# | | | |
# trunk---r2---r3-----r5---r6-------r8---r9---------------> | |
# r1 | | | | | |
# initial | | | |______ | |
# import copy | copy | merge merge
# | | | merge (r5) (r8)
# | | | (r9) | |
# | | | | | |
# | | V V | |
# | | branches/B2-------r11---r12----> | |
# | | r7 |____| | |
# | | | | |
# | merge |___ | |
# | (r6) | | |
# | |_________________ | | |
# | | merge | |
# | | (r11-12) | |
# | | | | |
# V V V | |
# branches/B1-------------------r10--------r13--> | |
# r4 | |
# | V V
# branches/B1/B/E------------------------------r14---r15->
#
#
# The mergeinfo on this repos@15 is:
#
# Properties on 'branches/B1':
# svn:mergeinfo
# /branches/B2:11-12
# /trunk:6,9
# Properties on 'branches/B1/B/E':
# svn:mergeinfo
# /branches/B2/B/E:11-12
# /trunk/B/E:5-6,8-9
# Properties on 'branches/B2':
# svn:mergeinfo
# /trunk:9
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'mergeinfo_included_full.dump')
dumpfile_full = svntest.actions.load_dumpfile(dumpfile_location)
load_dumpstream(sbox, dumpfile_full, '--ignore-uuid')
# Check that the mergeinfo is as expected.
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(expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
# PART 2: Load a a series of incremental dumps to an empty repository.
#
# Incrementally dump the repository into three dump files:
dump_file_r1_10 = sbox.get_tempname("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 = sbox.get_tempname("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 = sbox.get_tempname("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()
# Blow away the current repos and create an empty one in its place.
sbox.build(empty=True)
# Load the three incremental dump files in sequence.
load_dumpstream(sbox, svntest.actions.load_dumpfile(dump_file_r1_10),
'--ignore-uuid')
load_dumpstream(sbox, svntest.actions.load_dumpfile(dump_file_r11_13),
'--ignore-uuid')
load_dumpstream(sbox, svntest.actions.load_dumpfile(dump_file_r14_15),
'--ignore-uuid')
# Check the mergeinfo, we use the same expected output as before,
# as it (duh!) should be exactly the same as when we loaded the
# repos in one shot.
svntest.actions.run_and_verify_svn(expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
# Now repeat the above two scenarios, but with an initially non-empty target
# repository. First, try the full dump-load in one shot.
#
# PART 3: Load a full dump to an non-empty repository.
#
# Reset our sandbox.
sbox.build(empty=True)
# Load this skeleton repos into the empty target:
#
# Projects/ (Added r1)
# README (Added r2)
# Project-X (Added r3)
# Project-Y (Added r4)
# Project-Z (Added r5)
# docs/ (Added r6)
# README (Added r6)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'skeleton_repos.dump')
dumpfile_skeleton = svntest.actions.load_dumpfile(dumpfile_location)
load_dumpstream(sbox, dumpfile_skeleton, '--ignore-uuid')
# Load 'svnadmin_tests_data/mergeinfo_included_full.dump' in one shot:
load_dumpstream(sbox, dumpfile_full, '--parent-dir', 'Projects/Project-X',
'--ignore-uuid')
# Check that the mergeinfo is as expected. This is exactly the
# same expected mergeinfo we previously checked, except that the
# revisions are all offset +6 to reflect the revions already in
# the skeleton target before we began loading and the leading source
# paths are adjusted by the --parent-dir:
#
# Properties on 'branches/B1':
# svn:mergeinfo
# /Projects/Project-X/branches/B2:17-18
# /Projects/Project-X/trunk:12,15
# Properties on 'branches/B1/B/E':
# svn:mergeinfo
# /Projects/Project-X/branches/B2/B/E:17-18
# /Projects/Project-X/trunk/B/E:11-12,14-15
# Properties on 'branches/B2':
# svn:mergeinfo
# /Projects/Project-X/trunk:15
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(expected_output, [],
'propget', 'svn:mergeinfo', '-R',
sbox.repo_url)
# PART 4: Load a a series of incremental dumps to an non-empty repository.
#
# Reset our sandbox.
sbox.build(empty=True)
# Load this skeleton repos into the empty target:
load_dumpstream(sbox, dumpfile_skeleton, '--ignore-uuid')
# Load the three incremental dump files in sequence.
load_dumpstream(sbox, svntest.actions.load_dumpfile(dump_file_r1_10),
'--parent-dir', 'Projects/Project-X', '--ignore-uuid')
load_dumpstream(sbox, svntest.actions.load_dumpfile(dump_file_r11_13),
'--parent-dir', 'Projects/Project-X', '--ignore-uuid')
load_dumpstream(sbox, svntest.actions.load_dumpfile(dump_file_r14_15),
'--parent-dir', 'Projects/Project-X', '--ignore-uuid')
# Check the resulting mergeinfo. We expect the exact same results
# as Part 3.
# See https://issues.apache.org/jira/browse/SVN-3020#desc16.
svntest.actions.run_and_verify_svn(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"
## See https://issues.apache.org/jira/browse/SVN-2591. ##
# Create a repository.
sbox.build(create_wc=False, empty=True)
original_repo = sbox.repo_dir
hotcopy_repo, hotcopy_url = sbox.add_repo_path('hotcopy')
# Create a file, a dir and a missing path outside the repoitory.
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")
# Symlink definitions: base name -> target relpath.
# Check both existing and nonexistent targets.
# Check targets both within and outside the source repository.
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)),
]
# Create symlinks within the repository directory.
for name, target_relpath in symlinks:
target_path = os.path.join(original_repo, target_relpath)
target_abspath = os.path.abspath(target_path)
# Create two symlinks to each target - one relative, one absolute.
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, [],
"hotcopy", original_repo, hotcopy_repo)
# Check if the symlinks were copied correctly.
for name, target_relpath in symlinks:
target_path = os.path.join(original_repo, target_relpath)
target_abspath = os.path.abspath(target_path)
# Check two symlinks to each target - one relative, one absolute.
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 = b"""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
"""
sbox.build(empty=True)
# Try to load the dumpstream, expecting a failure (because of mixed EOLs).
exp_err = svntest.verify.RegexListOutput(['svnadmin: E125005:.*',
'svnadmin: E125005:.*',
'svnadmin: E125017:.*'],
match_all=False)
load_and_verify_dumpstream(sbox, [], exp_err, dumpfile_revisions,
False, dump_str, '--ignore-uuid')
# Now try it again bypassing prop validation. (This interface takes
# care of the removal and recreation of the original repository.)
svntest.actions.load_repo(sbox, dump_str=dump_str,
bypass_prop_validation=True)
# Getting the property should fail.
svntest.actions.run_and_verify_svn(None, 'svn: E135000: ',
'pg', 'svn:log', '--revprop', '-r1',
sbox.repo_url)
# Now try it again with prop normalization.
svntest.actions.load_repo(sbox, dump_str=dump_str,
bypass_prop_validation=False,
normalize_props=True)
# We should get the expected property value.
exit_code, output, _ = svntest.main.run_svn(None, 'pg', 'svn:log',
'--revprop', '-r1',
'--no-newline',
sbox.repo_url)
svntest.verify.verify_exit_code(None, exit_code, 0)
if output != ['\n', '\n']:
raise svntest.Failure("Unexpected property value %s" % output)
# This test intentionally corrupts a revision and assumes an FSFS
# repository. If you can make it work with BDB please do so.
# However, the verification triggered by this test is in the repos layer
# so it will trigger with either backend anyway.
@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"
if svntest.main.options.fsfs_version is not None and \
svntest.main.options.fsfs_version not in [4, 6]:
raise svntest.Skip("Unsupported prepackaged repository version")
dumpfile = clean_dumpfile()
# Corruption only possible in physically addressed revisions created
# with pre-1.6 servers.
sbox.build(empty=True,
minor_version=min(svntest.main.options.server_minor_version,8))
# Load the dumpstream
load_and_verify_dumpstream(sbox, [], [], dumpfile_revisions, False,
dumpfile, '--ignore-uuid')
# Replace the path 'A' in revision 1 with a non-UTF-8 sequence.
# This has been observed in repositories in the wild, though Subversion
# 1.6 and greater should prevent such filenames from entering the repository.
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 == b"A\n":
# replace 'A' with a latin1 character -- the new path is not valid UTF-8
fp_new.write(b"\xE6\n")
elif line == b"text: 1 340 32 32 a6be7b4cf075fd39e6a99eb69a31232b\n":
# phys, PLAIN directories: fix up the representation checksum
fp_new.write(b"text: 1 340 32 32 f2e93e73272cac0f18fccf16f224eb93\n")
elif line == b"text: 1 340 44 32 a6be7b4cf075fd39e6a99eb69a31232b\n":
# phys, deltified directories: fix up the representation checksum
fp_new.write(b"text: 1 340 44 32 f2e93e73272cac0f18fccf16f224eb93\n")
elif line == b"cpath: /A\n":
# also fix up the 'created path' field
fp_new.write(b"cpath: /\xE6\n")
elif line == b"_0.0.t0-0 add-file true true /A\n":
# and another occurrance
fp_new.write(b"_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)
# Verify the repository, expecting failure
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.*")
# Make sure the repository can still be dumped so that the
# encoding problem can be fixed in a dump/edit/load cycle.
expected_stderr = [
"* Dumped revision 0.\n",
"WARNING 0x0002: E160005: "
"While validating fspath '?\\E6': "
"Path '?\\E6' 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
def test_lslocks_and_rmlocks(sbox):
"test 'svnadmin lslocks' and 'svnadmin rmlocks'"
sbox.build(create_wc=False)
iota_url = sbox.repo_url + '/iota'
lambda_url = sbox.repo_url + '/A/B/lambda'
exit_code, output, errput = svntest.main.run_svnadmin("lslocks",
sbox.repo_dir)
if exit_code or errput or output:
raise svntest.Failure("Error: 'lslocks' failed")
expected_output = svntest.verify.UnorderedRegexListOutput(
["'.*lambda' locked by user 'jrandom'.\n",
"'.*iota' locked by user 'jrandom'.\n"])
# Lock iota and A/B/lambda using svn client
svntest.actions.run_and_verify_svn(expected_output,
[], "lock", "-m", "Locking files",
iota_url, lambda_url)
def expected_output_list(path):
return [
"Path: " + path,
"UUID Token: opaquelocktoken:.*",
"Owner: jrandom",
"Created:.*",
"Expires:.*",
"Comment \(1 line\):",
"Locking files",
"\n", # empty line
]
# List all locks
exit_code, output, errput = svntest.main.run_svnadmin("lslocks",
sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
svntest.verify.verify_exit_code(None, exit_code, 0)
expected_output = svntest.verify.UnorderedRegexListOutput(
expected_output_list('/A/B/lambda') +
expected_output_list('/iota'))
svntest.verify.compare_and_display_lines('lslocks output mismatch',
'output',
expected_output, output)
# List lock in path /A
exit_code, output, errput = svntest.main.run_svnadmin("lslocks",
sbox.repo_dir,
"A")
if errput:
raise SVNUnexpectedStderr(errput)
expected_output = svntest.verify.RegexListOutput(
expected_output_list('/A/B/lambda'))
svntest.verify.compare_and_display_lines('lslocks output mismatch',
'output',
expected_output, output)
svntest.verify.verify_exit_code(None, exit_code, 0)
# Remove locks
exit_code, output, errput = svntest.main.run_svnadmin("rmlocks",
sbox.repo_dir,
"iota",
"A/B/lambda")
expected_output = UnorderedOutput(["Removed lock on '/iota'.\n",
"Removed lock on '/A/B/lambda'.\n"])
svntest.verify.verify_outputs(
"Unexpected output while running 'svnadmin rmlocks'.",
output, [], expected_output, None)
#----------------------------------------------------------------------
@Issue(3734)
def load_ranges(sbox):
"'svnadmin load --revision X:Y'"
## See https://issues.apache.org/jira/browse/SVN-3734. ##
sbox.build(empty=True)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'skeleton_repos.dump')
dumplines = svntest.actions.load_dumpfile(dumpfile_location)
# Load our dumpfile, 2 revisions at a time, verifying that we have
# the correct youngest revision after each load.
load_dumpstream(sbox, dumplines, '-r0:2')
svntest.actions.run_and_verify_svnlook(['2\n'],
None, 'youngest', sbox.repo_dir)
load_dumpstream(sbox, dumplines, '-r3:4')
svntest.actions.run_and_verify_svnlook(['4\n'],
None, 'youngest', sbox.repo_dir)
load_dumpstream(sbox, dumplines, '-r5:6')
svntest.actions.run_and_verify_svnlook(['6\n'],
None, 'youngest', sbox.repo_dir)
# There are ordering differences in the property blocks.
if (svntest.main.options.server_minor_version < 6):
temp = []
for line in dumplines:
if not "Text-content-sha1:" in line:
temp.append(line)
expected_dump = UnorderedOutput(temp)
else:
expected_dump = UnorderedOutput(dumplines)
new_dumpdata = svntest.actions.run_and_verify_dump(sbox.repo_dir)
svntest.verify.compare_and_display_lines("Dump files", "DUMP",
expected_dump, new_dumpdata)
@SkipUnless(svntest.main.is_fs_type_fsfs)
def hotcopy_incremental(sbox):
"'svnadmin hotcopy --incremental PATH .'"
sbox.build()
backup_dir, backup_url = sbox.add_repo_path('backup')
os.mkdir(backup_dir)
cwd = os.getcwd()
for i in [1, 2, 3]:
os.chdir(backup_dir)
svntest.actions.run_and_verify_svnadmin(
None, [],
"hotcopy", "--incremental", os.path.join(cwd, sbox.repo_dir), '.')
os.chdir(cwd)
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
if i < 3:
sbox.simple_mkdir("newdir-%i" % i)
sbox.simple_commit()
@SkipUnless(svntest.main.is_fs_type_fsfs)
@SkipUnless(svntest.main.fs_has_pack)
def hotcopy_incremental_packed(sbox):
"'svnadmin hotcopy --incremental' with packing"
# Configure two files per shard to trigger packing.
sbox.build()
patch_format(sbox.repo_dir, shard_size=2)
backup_dir, backup_url = sbox.add_repo_path('backup')
os.mkdir(backup_dir)
cwd = os.getcwd()
# Pack revisions 0 and 1 if not already packed.
if not (svntest.main.is_fs_type_fsfs and svntest.main.options.fsfs_packing
and svntest.main.options.fsfs_sharding == 2):
svntest.actions.run_and_verify_svnadmin(
['Packing revisions in shard 0...done.\n'], [], "pack",
os.path.join(cwd, sbox.repo_dir))
# Commit 5 more revs, hotcopy and pack after each commit.
for i in [1, 2, 3, 4, 5]:
os.chdir(backup_dir)
svntest.actions.run_and_verify_svnadmin(
None, [],
"hotcopy", "--incremental", os.path.join(cwd, sbox.repo_dir), '.')
os.chdir(cwd)
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
if i < 5:
sbox.simple_mkdir("newdir-%i" % i)
sbox.simple_commit()
if (svntest.main.is_fs_type_fsfs and not svntest.main.options.fsfs_packing
and not i % 2):
expected_output = ['Packing revisions in shard %d...done.\n' % (i/2)]
else:
expected_output = []
svntest.actions.run_and_verify_svnadmin(
expected_output, [], "pack", os.path.join(cwd, sbox.repo_dir))
def locking(sbox):
"svnadmin lock tests"
sbox.build(create_wc=False)
comment_path = os.path.join(svntest.main.temp_dir, "comment")
svntest.main.file_write(comment_path, "dummy comment")
invalid_comment_path = os.path.join(svntest.main.temp_dir, "invalid_comment")
svntest.main.file_write(invalid_comment_path, "character is invalid")
# Test illegal character in comment file.
expected_error = ".*svnadmin: E130004:.*"
svntest.actions.run_and_verify_svnadmin(None,
expected_error, "lock",
sbox.repo_dir,
"iota", "jrandom",
invalid_comment_path)
# Test locking path with --bypass-hooks
expected_output = "'/iota' locked by user 'jrandom'."
svntest.actions.run_and_verify_svnadmin(expected_output,
None, "lock",
sbox.repo_dir,
"iota", "jrandom",
comment_path,
"--bypass-hooks")
# Remove lock
svntest.actions.run_and_verify_svnadmin(None,
None, "rmlocks",
sbox.repo_dir, "iota")
# Test locking path without --bypass-hooks
expected_output = "'/iota' locked by user 'jrandom'."
svntest.actions.run_and_verify_svnadmin(expected_output,
None, "lock",
sbox.repo_dir,
"iota", "jrandom",
comment_path)
# Test locking already locked path.
expected_error = ".*svnadmin: E160035:.*"
svntest.actions.run_and_verify_svnadmin(None,
expected_error, "lock",
sbox.repo_dir,
"iota", "jrandom",
comment_path)
# Test locking non-existent path.
expected_error = ".*svnadmin: E160013:.*"
svntest.actions.run_and_verify_svnadmin(None,
expected_error, "lock",
sbox.repo_dir,
"non-existent", "jrandom",
comment_path)
# Test locking a path while specifying a lock token.
expected_output = "'/A/D/G/rho' locked by user 'jrandom'."
lock_token = "opaquelocktoken:01234567-89ab-cdef-89ab-cdef01234567"
svntest.actions.run_and_verify_svnadmin(expected_output,
None, "lock",
sbox.repo_dir,
"A/D/G/rho", "jrandom",
comment_path, lock_token)
# Test unlocking a path, but provide the wrong lock token.
expected_error = ".*svnadmin: E160040:.*"
wrong_lock_token = "opaquelocktoken:12345670-9ab8-defc-9ab8-def01234567c"
svntest.actions.run_and_verify_svnadmin(None,
expected_error, "unlock",
sbox.repo_dir,
"A/D/G/rho", "jrandom",
wrong_lock_token)
# Test unlocking the path again, but this time provide the correct
# lock token.
expected_output = "'/A/D/G/rho' unlocked by user 'jrandom'."
svntest.actions.run_and_verify_svnadmin(expected_output,
None, "unlock",
sbox.repo_dir,
"A/D/G/rho", "jrandom",
lock_token)
# Install lock/unlock prevention hooks.
hook_path = svntest.main.get_pre_lock_hook_path(sbox.repo_dir)
svntest.main.create_python_hook_script(hook_path, 'import sys; sys.exit(1)')
hook_path = svntest.main.get_pre_unlock_hook_path(sbox.repo_dir)
svntest.main.create_python_hook_script(hook_path, 'import sys; sys.exit(1)')
# Test locking a path. Don't use --bypass-hooks, though, as we wish
# to verify that hook script is really getting executed.
expected_error = ".*svnadmin: E165001:.*"
svntest.actions.run_and_verify_svnadmin(None,
expected_error, "lock",
sbox.repo_dir,
"iota", "jrandom",
comment_path)
# Fetch the lock token for our remaining locked path. (We didn't
# explicitly set it, so it will vary from test run to test run.)
exit_code, output, errput = svntest.main.run_svnadmin("lslocks",
sbox.repo_dir,
"iota")
iota_token = None
for line in output:
if line.startswith("UUID Token: opaquelocktoken:"):
iota_token = line[12:].rstrip()
break
if iota_token is None:
raise svntest.Failure("Unable to lookup lock token for 'iota'")
# Try to unlock a path while providing the correct lock token but
# with a preventative hook in place.
expected_error = ".*svnadmin: E165001:.*"
svntest.actions.run_and_verify_svnadmin(None,
expected_error, "unlock",
sbox.repo_dir,
"iota", "jrandom",
iota_token)
# Finally, use --bypass-hooks to unlock the path (again using the
# correct lock token).
expected_output = "'/iota' unlocked by user 'jrandom'."
svntest.actions.run_and_verify_svnadmin(expected_output,
None, "unlock",
"--bypass-hooks",
sbox.repo_dir,
"iota", "jrandom",
iota_token)
@SkipUnless(svntest.main.is_threaded_python)
@Issue(4129)
def mergeinfo_race(sbox):
"concurrent mergeinfo commits invalidate pred-count"
sbox.build()
# This test exercises two commit-time race condition bugs:
#
# (a) metadata corruption when concurrent commits change svn:mergeinfo (issue #4129)
# (b) false positive SVN_ERR_FS_CONFLICT error with httpv1 commits
# https://mail-archives.apache.org/mod_mbox/subversion-dev/201507.mbox/%3C20150731234536.GA5395@tarsus.local2%3E
#
# Both bugs are timing-dependent and might not reproduce 100% of the time.
wc_dir = sbox.wc_dir
wc2_dir = sbox.add_wc_path('2')
## Create wc2.
svntest.main.run_svn(None, 'checkout', '-q', sbox.repo_url, wc2_dir)
## Some random edits.
svntest.main.run_svn(None, 'mkdir', sbox.ospath('d1', wc_dir))
svntest.main.run_svn(None, 'mkdir', sbox.ospath('d2', wc2_dir))
## Set random mergeinfo properties.
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))
# t2 will trigger the issue #4129 sanity check in fs_fs.c
t1.start(); t2.start()
t1.join(); t2.join()
# Crude attempt to make sure everything worked.
# TODO: better way to catch exceptions in the thread
if svntest.actions.run_and_parse_info(sbox.repo_url)[0]['Revision'] != '3':
raise svntest.Failure("one or both commits failed")
@Issue(4213)
@Skip(svntest.main.is_fs_type_fsx)
def recover_old_empty(sbox):
"recover empty --compatible-version=1.3"
sbox.build(create_wc=False, empty=True, minor_version=3)
svntest.actions.run_and_verify_svnadmin(None, [],
"recover", sbox.repo_dir)
@SkipUnless(svntest.main.is_fs_type_fsfs)
def verify_keep_going(sbox):
"svnadmin verify --keep-going test"
# No support for modifying pack files
if svntest.main.options.fsfs_packing:
raise svntest.Skip('fsfs packing set')
sbox.build(create_wc = False)
repo_url = sbox.repo_url
B_url = sbox.repo_url + '/B'
C_url = sbox.repo_url + '/C'
# Create A/B/E/bravo in r2.
svntest.actions.run_and_verify_svn(None, [],
'mkdir', '-m', 'log_msg',
B_url)
svntest.actions.run_and_verify_svn(None, [],
'mkdir', '-m', 'log_msg',
C_url)
r2 = fsfs_file(sbox.repo_dir, 'revs', '2')
fp = open(r2, 'r+b')
fp.write(b"inserting junk to corrupt the rev")
fp.close()
exit_code, output, errput = svntest.main.run_svnadmin("verify",
"--keep-going",
sbox.repo_dir)
exp_out = svntest.verify.RegexListOutput([".*Verified revision 0.",
".*Verified revision 1.",
".*",
".*Summary.*",
".*r2: E160004:.*",
".*r2: E160004:.*",
".*r3: E160004:.*",
".*r3: E160004:.*"])
if (svntest.main.fs_has_rep_sharing()):
exp_out.insert(0, ".*Verifying.*metadata.*")
exp_err = svntest.verify.RegexListOutput([".*Error verifying revision 2.",
"svnadmin: E160004:.*",
"svnadmin: E160004:.*",
".*Error verifying revision 3.",
"svnadmin: E160004:.*",
"svnadmin: E160004:.*",
"svnadmin: E205012:.*"], False)
if (svntest.main.is_fs_log_addressing()):
exp_err.insert(0, ".*Error verifying repository metadata.")
exp_err.insert(1, "svnadmin: E160004:.*")
if svntest.verify.verify_outputs("Unexpected error while running 'svnadmin verify'.",
output, errput, exp_out, exp_err):
raise svntest.Failure
exit_code, output, errput = svntest.main.run_svnadmin("verify",
sbox.repo_dir)
if (svntest.main.is_fs_log_addressing()):
exp_out = svntest.verify.RegexListOutput([".*Verifying metadata at revision 0.*"])
else:
exp_out = svntest.verify.RegexListOutput([".*Verified revision 0.",
".*Verified revision 1."])
if (svntest.main.fs_has_rep_sharing()):
exp_out.insert(0, ".*Verifying repository metadata.*")
if (svntest.main.is_fs_log_addressing()):
exp_err = svntest.verify.RegexListOutput([
".*Error verifying repository metadata.",
"svnadmin: E160004:.*"], False)
else:
exp_err = svntest.verify.RegexListOutput([".*Error verifying revision 2.",
"svnadmin: E160004:.*",
"svnadmin: E160004:.*"], False)
if svntest.verify.verify_outputs("Unexpected error while running 'svnadmin verify'.",
output, errput, exp_out, exp_err):
raise svntest.Failure
exit_code, output, errput = svntest.main.run_svnadmin("verify",
"--quiet",
sbox.repo_dir)
if (svntest.main.is_fs_log_addressing()):
exp_err = svntest.verify.RegexListOutput([
".*Error verifying repository metadata.",
"svnadmin: E160004:.*"], False)
else:
exp_err = svntest.verify.RegexListOutput([".*Error verifying revision 2.",
"svnadmin: E160004:.*",
"svnadmin: E160004:.*"], False)
if svntest.verify.verify_outputs("Output of 'svnadmin verify' is unexpected.",
None, errput, None, exp_err):
raise svntest.Failure
# Don't leave a corrupt repository
svntest.main.safe_rmtree(sbox.repo_dir, True)
@SkipUnless(svntest.main.is_fs_type_fsfs)
def verify_keep_going_quiet(sbox):
"svnadmin verify --keep-going --quiet test"
# No support for modifying pack files
if svntest.main.options.fsfs_packing:
raise svntest.Skip('fsfs packing set')
sbox.build(create_wc = False)
repo_url = sbox.repo_url
B_url = sbox.repo_url + '/B'
C_url = sbox.repo_url + '/C'
# Create A/B/E/bravo in r2.
svntest.actions.run_and_verify_svn(None, [],
'mkdir', '-m', 'log_msg',
B_url)
svntest.actions.run_and_verify_svn(None, [],
'mkdir', '-m', 'log_msg',
C_url)
r2 = fsfs_file(sbox.repo_dir, 'revs', '2')
fp = open(r2, 'r+b')
fp.write(b"inserting junk to corrupt the rev")
fp.close()
exit_code, output, errput = svntest.main.run_svnadmin("verify",
"--keep-going",
"--quiet",
sbox.repo_dir)
exp_err = svntest.verify.RegexListOutput([".*Error verifying revision 2.",
"svnadmin: E160004:.*",
"svnadmin: E160004:.*",
".*Error verifying revision 3.",
"svnadmin: E160004:.*",
"svnadmin: E160004:.*",
"svnadmin: E205012:.*"], False)
# Insert another expected error from checksum verification
if (svntest.main.is_fs_log_addressing()):
exp_err.insert(0, ".*Error verifying repository metadata.")
exp_err.insert(1, "svnadmin: E160004:.*")
if svntest.verify.verify_outputs(
"Unexpected error while running 'svnadmin verify'.",
output, errput, None, exp_err):
raise svntest.Failure
# Don't leave a corrupt repository
svntest.main.safe_rmtree(sbox.repo_dir, True)
@SkipUnless(svntest.main.is_fs_type_fsfs)
def verify_invalid_path_changes(sbox):
"detect invalid changed path list entries"
# No support for modifying pack files
if svntest.main.options.fsfs_packing:
raise svntest.Skip('fsfs packing set')
sbox.build(create_wc = False)
repo_url = sbox.repo_url
# Create a number of revisions each adding a single path
for r in range(2,20):
svntest.actions.run_and_verify_svn(None, [],
'mkdir', '-m', 'log_msg',
sbox.repo_url + '/B' + str(r))
# modify every other revision to make sure that errors are not simply
# "carried over" but that all corrupts we get detected independently
# add existing node
set_changed_path_list(sbox, 2,
b"_0.0.t1-1 add-dir false false /A\n\n")
# add into non-existent parent
set_changed_path_list(sbox, 4,
b"_0.0.t3-2 add-dir false false /C/X\n\n")
# del non-existent node
set_changed_path_list(sbox, 6,
b"_0.0.t5-2 delete-dir false false /C\n\n")
# del existent node of the wrong kind
#
# THIS WILL NOT BE DETECTED
# since dump mechanism and file don't care about the types of deleted nodes
set_changed_path_list(sbox, 8,
b"_0.0.t7-2 delete-file false false /B3\n\n")
# copy from non-existent node
set_changed_path_list(sbox, 10,
b"_0.0.t9-2 add-dir false false /B10\n6 /B8\n")
# copy from existing node of the wrong kind
set_changed_path_list(sbox, 12,
b"_0.0.t11-2 add-file false false /B12\n9 /B8\n")
# modify non-existent node
set_changed_path_list(sbox, 14,
b"_0.0.t13-2 modify-file false false /A/D/H/foo\n\n")
# modify existent node of the wrong kind
set_changed_path_list(sbox, 16,
b"_0.0.t15-2 modify-file false false /B12\n\n")
# replace non-existent node
set_changed_path_list(sbox, 18,
b"_0.0.t17-2 replace-file false false /A/D/H/foo\n\n")
# find corruptions
exit_code, output, errput = svntest.main.run_svnadmin("verify",
"--keep-going",
sbox.repo_dir)
# Errors generated by FSFS when CHANGED_PATHS is not forced into emulation
exp_out1 = svntest.verify.RegexListOutput([".*Verified revision 0.",
".*Verified revision 1.",
".*Verified revision 3.",
".*Verified revision 5.",
".*Verified revision 7.",
".*Verified revision 8.",
".*Verified revision 9.",
".*Verified revision 11.",
".*Verified revision 13.",
".*Verified revision 15.",
".*Verified revision 17.",
".*Verified revision 19.",
".*",
".*Summary.*",
".*r2: E160020:.*",
".*r2: E160020:.*",
".*r4: E160013:.*",
".*r6: E160013:.*",
".*r6: E160013:.*",
".*r10: E160013:.*",
".*r10: E160013:.*",
".*r12: E145001:.*",
".*r12: E145001:.*",
".*r14: E160013:.*",
".*r14: E160013:.*",
".*r16: E145001:.*",
".*r16: E145001:.*",
".*r18: E160013:.*",
".*r18: E160013:.*"])
exp_err1 = svntest.verify.RegexListOutput([".*Error verifying revision 2.",
"svnadmin: E160020:.*",
"svnadmin: E160020:.*",
".*Error verifying revision 4.",
"svnadmin: E160013:.*",
".*Error verifying revision 6.",
"svnadmin: E160013:.*",
"svnadmin: E160013:.*",
".*Error verifying revision 10.",
"svnadmin: E160013:.*",
"svnadmin: E160013:.*",
".*Error verifying revision 12.",
"svnadmin: E145001:.*",
"svnadmin: E145001:.*",
".*Error verifying revision 14.",
"svnadmin: E160013:.*",
"svnadmin: E160013:.*",
".*Error verifying revision 16.",
"svnadmin: E145001:.*",
"svnadmin: E145001:.*",
".*Error verifying revision 18.",
"svnadmin: E160013:.*",
"svnadmin: E160013:.*",
"svnadmin: E205012:.*"], False)
# If CHANGED_PATHS is emulated, FSFS fails earlier, generating fewer
# of the same messages per revision.
exp_out2 = svntest.verify.RegexListOutput([".*Verified revision 0.",
".*Verified revision 1.",
".*Verified revision 3.",
".*Verified revision 5.",
".*Verified revision 7.",
".*Verified revision 8.",
".*Verified revision 9.",
".*Verified revision 11.",
".*Verified revision 13.",
".*Verified revision 15.",
".*Verified revision 17.",
".*Verified revision 19.",
".*",
".*Summary.*",
".*r2: E160020:.*",
".*r2: E160020:.*",
".*r4: E160013:.*",
".*r6: E160013:.*",
".*r10: E160013:.*",
".*r10: E160013:.*",
".*r12: E145001:.*",
".*r12: E145001:.*",
".*r14: E160013:.*",
".*r16: E145001:.*",
".*r16: E145001:.*",
".*r18: E160013:.*"])
exp_err2 = svntest.verify.RegexListOutput([".*Error verifying revision 2.",
"svnadmin: E160020:.*",
"svnadmin: E160020:.*",
".*Error verifying revision 4.",
"svnadmin: E160013:.*",
".*Error verifying revision 6.",
"svnadmin: E160013:.*",
".*Error verifying revision 10.",
"svnadmin: E160013:.*",
"svnadmin: E160013:.*",
".*Error verifying revision 12.",
"svnadmin: E145001:.*",
"svnadmin: E145001:.*",
".*Error verifying revision 14.",
"svnadmin: E160013:.*",
".*Error verifying revision 16.",
"svnadmin: E145001:.*",
"svnadmin: E145001:.*",
".*Error verifying revision 18.",
"svnadmin: E160013:.*",
"svnadmin: E205012:.*"], False)
# Determine which pattern to use.
# Note that index() will throw an exception if the string can't be found.
try:
rev6_line = errput.index('* Error verifying revision 6.\n');
rev10_line = errput.index('* Error verifying revision 10.\n');
error_count = 0
for line in errput[rev6_line+1:rev10_line]:
if "svnadmin: E" in line:
error_count = error_count + 1
if error_count == 1:
exp_out = exp_out2
exp_err = exp_err2
else:
exp_out = exp_out1
exp_err = exp_err1
except ValueError:
exp_out = exp_out1
exp_err = exp_err1
if (svntest.main.fs_has_rep_sharing()):
exp_out.insert(0, ".*Verifying.*metadata.*")
if svntest.main.options.fsfs_sharding is not None:
for x in range(0, 19 / svntest.main.options.fsfs_sharding):
exp_out.insert(0, ".*Verifying.*metadata.*")
if svntest.main.is_fs_log_addressing():
exp_out.insert(0, ".*Verifying.*metadata.*")
if svntest.verify.verify_outputs("Unexpected error while running 'svnadmin verify'.",
output, errput, exp_out, exp_err):
raise svntest.Failure
exit_code, output, errput = svntest.main.run_svnadmin("verify",
sbox.repo_dir)
exp_out = svntest.verify.RegexListOutput([".*Verified revision 0.",
".*Verified revision 1."])
exp_err = svntest.verify.RegexListOutput([".*Error verifying revision 2.",
"svnadmin: E160020:.*",
"svnadmin: E160020:.*"], False)
if (svntest.main.fs_has_rep_sharing()):
exp_out.insert(0, ".*Verifying.*metadata.*")
if svntest.main.options.fsfs_sharding is not None:
for x in range(0, 19 / svntest.main.options.fsfs_sharding):
exp_out.insert(0, ".*Verifying.*metadata.*")
if svntest.main.is_fs_log_addressing():
exp_out.insert(0, ".*Verifying.*metadata.*")
if svntest.verify.verify_outputs("Unexpected error while running 'svnadmin verify'.",
output, errput, exp_out, exp_err):
raise svntest.Failure
exit_code, output, errput = svntest.main.run_svnadmin("verify",
"--quiet",
sbox.repo_dir)
exp_out = []
exp_err = svntest.verify.RegexListOutput([".*Error verifying revision 2.",
"svnadmin: E160020:.*",
"svnadmin: E160020:.*"], False)
if svntest.verify.verify_outputs("Output of 'svnadmin verify' is unexpected.",
output, errput, exp_out, exp_err):
raise svntest.Failure
# Don't leave a corrupt repository
svntest.main.safe_rmtree(sbox.repo_dir, True)
def verify_denormalized_names(sbox):
"detect denormalized names and name collisions"
sbox.build(create_wc=False, empty=True)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'normalization_check.dump')
load_dumpstream(sbox, svntest.actions.load_dumpfile(dumpfile_location))
exit_code, output, errput = svntest.main.run_svnadmin(
"verify", "--check-normalization", sbox.repo_dir)
expected_output_regex_list = [
".*Verified revision 0.",
".*Verified revision 1.",
".*Verified revision 2.",
".*Verified revision 3.",
# A/{Eacute}/{aring}lpha
"WARNING 0x0003: Duplicate representation of path 'A/.*/.*lpha'",
".*Verified revision 4.",
".*Verified revision 5.",
# Q/{aring}lpha
"WARNING 0x0004: Duplicate representation of path '/Q/.*lpha'"
# A/{Eacute}
" in svn:mergeinfo property of 'A/.*'",
".*Verified revision 6.",
".*Verified revision 7."]
# The BDB backend doesn't do global metadata verification.
if (svntest.main.fs_has_rep_sharing() and not svntest.main.is_fs_type_bdb()):
expected_output_regex_list.insert(0, ".*Verifying repository metadata.*")
if svntest.main.options.fsfs_sharding is not None:
for x in range(0, 7 / svntest.main.options.fsfs_sharding):
expected_output_regex_list.insert(0, ".*Verifying.*metadata.*")
if svntest.main.is_fs_log_addressing():
expected_output_regex_list.insert(0, ".* Verifying metadata at revision 0.*")
exp_out = svntest.verify.RegexListOutput(expected_output_regex_list)
exp_err = svntest.verify.ExpectedOutput([])
svntest.verify.verify_outputs(
"Unexpected error while running 'svnadmin verify'.",
output, errput, exp_out, exp_err)
@SkipUnless(svntest.main.is_fs_type_fsfs)
def fsfs_recover_old_non_empty(sbox):
"fsfs recover non-empty --compatible-version=1.3"
# Around trunk@1560210, 'svnadmin recover' wrongly errored out
# for the --compatible-version=1.3 Greek tree repository:
# svnadmin: E200002: Serialized hash missing terminator
sbox.build(create_wc=False, minor_version=3)
svntest.actions.run_and_verify_svnadmin(None, [], "recover",
sbox.repo_dir)
@SkipUnless(svntest.main.is_fs_type_fsfs)
def fsfs_hotcopy_old_non_empty(sbox):
"fsfs hotcopy non-empty --compatible-version=1.3"
# Around trunk@1560210, 'svnadmin hotcopy' wrongly errored out
# for the --compatible-version=1.3 Greek tree repository:
# svnadmin: E160006: No such revision 1
sbox.build(create_wc=False, minor_version=3)
backup_dir, backup_url = sbox.add_repo_path('backup')
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
sbox.repo_dir, backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
def load_ignore_dates(sbox):
"svnadmin load --ignore-dates"
# All revisions in the loaded repository should come after this time.
start_time = time.localtime()
time.sleep(1)
sbox.build(create_wc=False, empty=True)
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'skeleton_repos.dump')
dumpfile_skeleton = svntest.actions.load_dumpfile(dumpfile_location)
load_dumpstream(sbox, dumpfile_skeleton, '--ignore-dates')
svntest.actions.run_and_verify_svnlook(['6\n'],
None, 'youngest', sbox.repo_dir)
for rev in range(1, 6):
exit_code, output, errput = svntest.main.run_svnlook('date', '-r', rev,
sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
rev_time = time.strptime(output[0].rstrip()[:19], '%Y-%m-%d %H:%M:%S')
if rev_time < start_time:
raise svntest.Failure("Revision time for r%d older than load start time\n"
" rev_time: %s\n"
" start_time: %s"
% (rev, str(rev_time), str(start_time)))
@SkipUnless(svntest.main.is_fs_type_fsfs)
def fsfs_hotcopy_old_with_id_changes(sbox):
"fsfs hotcopy old with node-id and copy-id changes"
# Around trunk@1573728, running 'svnadmin hotcopy' for the
# --compatible-version=1.3 repository with certain node-id and copy-id
# changes ended with mismatching db/current in source and destination:
#
# source: "2 l 1" destination: "2 k 1",
# "3 l 2" "3 4 2"
# (and so on...)
#
# We test this case by creating a --compatible-version=1.3 repository
# and committing things that result in node-id and copy-id changes.
# After every commit, we hotcopy the repository to a new destination
# and check whether the source of the backup and the backup itself are
# identical. We also maintain a separate --incremental backup, which
# is updated and checked after every commit.
sbox.build(create_wc=True, minor_version=3)
inc_backup_dir, inc_backup_url = sbox.add_repo_path('incremental-backup')
# r1 = Initial greek tree sandbox.
backup_dir, backup_url = sbox.add_repo_path('backup-after-r1')
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
"--incremental",
sbox.repo_dir, inc_backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, inc_backup_dir)
# r2 = Add a new property.
sbox.simple_propset('foo', 'bar', 'A/mu')
sbox.simple_commit(message='r2')
backup_dir, backup_url = sbox.add_repo_path('backup-after-r2')
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
"--incremental",
sbox.repo_dir, inc_backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, inc_backup_dir)
# r3 = Copy a file.
sbox.simple_copy('A/B/E', 'A/B/E1')
sbox.simple_commit(message='r3')
backup_dir, backup_url = sbox.add_repo_path('backup-after-r3')
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
"--incremental",
sbox.repo_dir, inc_backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, inc_backup_dir)
# r4 = Remove an existing file ...
sbox.simple_rm('A/D/gamma')
sbox.simple_commit(message='r4')
backup_dir, backup_url = sbox.add_repo_path('backup-after-r4')
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
"--incremental",
sbox.repo_dir, inc_backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, inc_backup_dir)
# r5 = ...and replace it with a new file here.
sbox.simple_add_text("This is the replaced file.\n", 'A/D/gamma')
sbox.simple_commit(message='r5')
backup_dir, backup_url = sbox.add_repo_path('backup-after-r5')
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
"--incremental",
sbox.repo_dir, inc_backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, inc_backup_dir)
# r6 = Add an entirely new file.
sbox.simple_add_text('This is an entirely new file.\n', 'A/C/mu1')
sbox.simple_commit(message='r6')
backup_dir, backup_url = sbox.add_repo_path('backup-after-r6')
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
"--incremental",
sbox.repo_dir, inc_backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, inc_backup_dir)
# r7 = Change the content of the existing file (this changeset does
# not bump the next-id and copy-id counters in the repository).
sbox.simple_append('A/mu', 'This is change in the existing file.\n')
sbox.simple_commit(message='r7')
backup_dir, backup_url = sbox.add_repo_path('backup-after-r7')
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
"--incremental",
sbox.repo_dir, inc_backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, backup_dir)
check_hotcopy_fsfs(sbox.repo_dir, inc_backup_dir)
@SkipUnless(svntest.main.fs_has_pack)
def verify_packed(sbox):
"verify packed with small shards"
# Configure two files per shard to trigger packing.
sbox.build()
patch_format(sbox.repo_dir, shard_size=2)
# Play with our greek tree. These changesets fall into two
# separate shards with r2 and r3 being in shard 1 ...
sbox.simple_append('iota', "Line.\n")
sbox.simple_append('A/D/gamma', "Another line.\n")
sbox.simple_commit(message='r2')
sbox.simple_propset('foo', 'bar', 'iota')
sbox.simple_propset('foo', 'baz', 'A/mu')
sbox.simple_commit(message='r3')
# ...and r4 and r5 being in shard 2.
sbox.simple_rm('A/C')
sbox.simple_copy('A/B/E', 'A/B/E1')
sbox.simple_move('A/mu', 'A/B/mu')
sbox.simple_commit(message='r4')
sbox.simple_propdel('foo', 'A/B/mu')
sbox.simple_commit(message='r5')
if svntest.main.is_fs_type_fsfs and svntest.main.options.fsfs_packing:
# With --fsfs-packing, everything is already packed and we
# can skip this part.
pass
else:
expected_output = ["Packing revisions in shard 0...done.\n",
"Packing revisions in shard 1...done.\n",
"Packing revisions in shard 2...done.\n"]
svntest.actions.run_and_verify_svnadmin(expected_output, [],
"pack", sbox.repo_dir)
if svntest.main.is_fs_log_addressing():
expected_output = ["* Verifying metadata at revision 0 ...\n",
"* Verifying metadata at revision 2 ...\n",
"* Verifying metadata at revision 4 ...\n",
"* Verifying repository metadata ...\n",
"* Verified revision 0.\n",
"* Verified revision 1.\n",
"* Verified revision 2.\n",
"* Verified revision 3.\n",
"* Verified revision 4.\n",
"* Verified revision 5.\n"]
else:
expected_output = ["* Verifying repository metadata ...\n",
"* Verified revision 0.\n",
"* Verified revision 1.\n",
"* Verified revision 2.\n",
"* Verified revision 3.\n",
"* Verified revision 4.\n",
"* Verified revision 5.\n"]
svntest.actions.run_and_verify_svnadmin(expected_output, [],
"verify", sbox.repo_dir)
# Test that 'svnadmin freeze' is nestable. (For example, this ensures it
# won't take system-global locks, only repository-scoped ones.)
#
# This could be useful to easily freeze a small number of repositories at once.
#
# ### We don't actually test that freeze takes a write lock anywhere (not even
# ### in C tests.)
def freeze_freeze(sbox):
"svnadmin freeze svnadmin freeze (some-cmd)"
sbox.build(create_wc=False, read_only=True)
second_repo_dir, _ = sbox.add_repo_path('backup')
svntest.actions.run_and_verify_svnadmin(None, [], "hotcopy",
sbox.repo_dir, second_repo_dir)
if svntest.main.is_fs_type_fsx() or \
(svntest.main.is_fs_type_fsfs() and \
svntest.main.options.server_minor_version < 9):
# FSFS repositories created with --compatible-version=1.8 and less
# erroneously share the filesystem data (locks, shared transaction
# data, ...) between hotcopy source and destination. This is fixed
# for new FS formats, but in order to avoid a deadlock for old formats,
# we have to manually assign a new UUID for the hotcopy destination.
# As of trunk@1618024, the same applies to FSX repositories.
svntest.actions.run_and_verify_svnadmin([], None,
'setuuid', second_repo_dir)
svntest.actions.run_and_verify_svnadmin(None, [],
'freeze', '--', sbox.repo_dir,
svntest.main.svnadmin_binary, 'freeze', '--', second_repo_dir,
sys.executable, '-c', 'True')
arg_file = sbox.get_tempname()
svntest.main.file_write(arg_file,
"%s\n%s\n" % (sbox.repo_dir, second_repo_dir))
svntest.actions.run_and_verify_svnadmin(None, [],
'freeze', '-F', arg_file, '--',
sys.executable, '-c', 'True')
def verify_metadata_only(sbox):
"verify metadata only"
sbox.build(create_wc = False)
exit_code, output, errput = svntest.main.run_svnadmin("verify",
sbox.repo_dir,
"--metadata-only")
if errput:
raise SVNUnexpectedStderr(errput)
# Unfortunately, older formats won't test as thoroughly than newer ones
# resulting in different progress output. BDB will do a full check but
# not produce any output.
if svntest.main.is_fs_log_addressing():
svntest.verify.compare_and_display_lines(
"Unexpected error while running 'svnadmin verify'.",
'STDOUT', ["* Verifying metadata at revision 0 ...\n",
"* Verifying repository metadata ...\n"], output)
elif svntest.main.fs_has_rep_sharing() \
and not svntest.main.is_fs_type_bdb():
svntest.verify.compare_and_display_lines(
"Unexpected error while running 'svnadmin verify'.",
'STDOUT', ["* Verifying repository metadata ...\n"], output)
else:
svntest.verify.compare_and_display_lines(
"Unexpected error while running 'svnadmin verify'.",
'STDOUT', [], output)
@Skip(svntest.main.is_fs_type_bdb)
def verify_quickly(sbox):
"verify quickly using metadata"
sbox.build(create_wc = False)
rev_file = open(fsfs_file(sbox.repo_dir, 'revs', '1'), 'r+b')
# set new contents
rev_file.seek(8)
rev_file.write(b'#')
rev_file.close()
exit_code, output, errput = svntest.main.run_svnadmin("verify",
sbox.repo_dir,
"--metadata-only")
# unfortunately, some backends needs to do more checks than other
# resulting in different progress output
if svntest.main.is_fs_log_addressing():
exp_out = svntest.verify.RegexListOutput([])
exp_err = svntest.verify.RegexListOutput(["svnadmin: E160004:.*"], False)
else:
exp_out = svntest.verify.RegexListOutput([])
exp_err = svntest.verify.RegexListOutput([])
if (svntest.main.fs_has_rep_sharing()):
exp_out.insert(0, ".*Verifying.*metadata.*")
if svntest.verify.verify_outputs("Unexpected error while running 'svnadmin verify'.",
output, errput, exp_out, exp_err):
raise svntest.Failure
# Don't leave a corrupt repository
svntest.main.safe_rmtree(sbox.repo_dir, True)
@SkipUnless(svntest.main.is_fs_type_fsfs)
@SkipUnless(svntest.main.fs_has_pack)
def fsfs_hotcopy_progress(sbox):
"hotcopy progress reporting"
# Check how 'svnadmin hotcopy' reports progress for non-incremental
# and incremental scenarios. The progress output can be affected by
# the --fsfs-packing option, so skip the test if that is the case.
if svntest.main.options.fsfs_packing:
raise svntest.Skip('fsfs packing set')
# Create an empty repository, configure three files per shard.
sbox.build(create_wc=False, empty=True)
patch_format(sbox.repo_dir, shard_size=3)
inc_backup_dir, inc_backup_url = sbox.add_repo_path('incremental-backup')
# Nothing really exciting for the empty repository.
expected_full = [
"* Copied revision 0.\n"
]
expected_incremental = [
"* Copied revision 0.\n",
]
backup_dir, backup_url = sbox.add_repo_path('backup-0')
svntest.actions.run_and_verify_svnadmin(expected_full, [],
'hotcopy',
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(expected_incremental, [],
'hotcopy', '--incremental',
sbox.repo_dir, inc_backup_dir)
# Commit three revisions. After this step we have a full shard
# (r0, r1, r2) and the second shard (r3) with a single revision.
for i in range(3):
svntest.actions.run_and_verify_svn(None, [], 'mkdir',
'-m', svntest.main.make_log_msg(),
sbox.repo_url + '/dir-%i' % i)
expected_full = [
"* Copied revision 0.\n",
"* Copied revision 1.\n",
"* Copied revision 2.\n",
"* Copied revision 3.\n",
]
expected_incremental = [
"* Copied revision 1.\n",
"* Copied revision 2.\n",
"* Copied revision 3.\n",
]
backup_dir, backup_url = sbox.add_repo_path('backup-1')
svntest.actions.run_and_verify_svnadmin(expected_full, [],
'hotcopy',
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(expected_incremental, [],
'hotcopy', '--incremental',
sbox.repo_dir, inc_backup_dir)
# Pack everything (r3 is still unpacked) and hotcopy again. In this case,
# the --incremental output should track the incoming (r0, r1, r2) pack and
# should not mention r3, because it is already a part of the destination
# and is *not* a part of the incoming pack.
svntest.actions.run_and_verify_svnadmin(None, [], 'pack',
sbox.repo_dir)
expected_full = [
"* Copied revisions from 0 to 2.\n",
"* Copied revision 3.\n",
]
expected_incremental = [
"* Copied revisions from 0 to 2.\n",
]
backup_dir, backup_url = sbox.add_repo_path('backup-2')
svntest.actions.run_and_verify_svnadmin(expected_full, [],
'hotcopy',
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(expected_incremental, [],
'hotcopy', '--incremental',
sbox.repo_dir, inc_backup_dir)
# Fill the second shard, pack again, commit several unpacked revisions
# on top of it. Rerun the hotcopy and check the progress output.
for i in range(4, 6):
svntest.actions.run_and_verify_svn(None, [], 'mkdir',
'-m', svntest.main.make_log_msg(),
sbox.repo_url + '/dir-%i' % i)
svntest.actions.run_and_verify_svnadmin(None, [], 'pack',
sbox.repo_dir)
for i in range(6, 8):
svntest.actions.run_and_verify_svn(None, [], 'mkdir',
'-m', svntest.main.make_log_msg(),
sbox.repo_url + '/dir-%i' % i)
expected_full = [
"* Copied revisions from 0 to 2.\n",
"* Copied revisions from 3 to 5.\n",
"* Copied revision 6.\n",
"* Copied revision 7.\n",
]
expected_incremental = [
"* Copied revisions from 3 to 5.\n",
"* Copied revision 6.\n",
"* Copied revision 7.\n",
]
backup_dir, backup_url = sbox.add_repo_path('backup-3')
svntest.actions.run_and_verify_svnadmin(expected_full, [],
'hotcopy',
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(expected_incremental, [],
'hotcopy', '--incremental',
sbox.repo_dir, inc_backup_dir)
@SkipUnless(svntest.main.is_fs_type_fsfs)
def fsfs_hotcopy_progress_with_revprop_changes(sbox):
"incremental hotcopy progress with changed revprops"
# The progress output can be affected by the --fsfs-packing
# option, so skip the test if that is the case.
if svntest.main.options.fsfs_packing:
raise svntest.Skip('fsfs packing set')
# Create an empty repository, commit several revisions and hotcopy it.
sbox.build(create_wc=False, empty=True)
for i in range(6):
svntest.actions.run_and_verify_svn(None, [], 'mkdir',
'-m', svntest.main.make_log_msg(),
sbox.repo_url + '/dir-%i' % i)
expected_output = [
"* Copied revision 0.\n",
"* Copied revision 1.\n",
"* Copied revision 2.\n",
"* Copied revision 3.\n",
"* Copied revision 4.\n",
"* Copied revision 5.\n",
"* Copied revision 6.\n",
]
backup_dir, backup_url = sbox.add_repo_path('backup')
svntest.actions.run_and_verify_svnadmin(expected_output, [],
'hotcopy',
sbox.repo_dir, backup_dir)
# Amend a few log messages in the source, run the --incremental hotcopy.
# The progress output should only mention the corresponding revisions.
revprop_file = sbox.get_tempname()
svntest.main.file_write(revprop_file, "Modified log message.")
for i in [1, 3, 6]:
svntest.actions.run_and_verify_svnadmin(None, [],
'setrevprop',
sbox.repo_dir, '-r', i,
'svn:log', revprop_file)
expected_output = [
"* Copied revision 1.\n",
"* Copied revision 3.\n",
"* Copied revision 6.\n",
]
svntest.actions.run_and_verify_svnadmin(expected_output, [],
'hotcopy', '--incremental',
sbox.repo_dir, backup_dir)
@SkipUnless(svntest.main.is_fs_type_fsfs)
def fsfs_hotcopy_progress_old(sbox):
"hotcopy --compatible-version=1.3 progress"
sbox.build(create_wc=False, empty=True, minor_version=3)
inc_backup_dir, inc_backup_url = sbox.add_repo_path('incremental-backup')
# Nothing really exciting for the empty repository.
expected_full = [
"* Copied revision 0.\n"
]
expected_incremental = [
"* Copied revision 0.\n",
]
backup_dir, backup_url = sbox.add_repo_path('backup-0')
svntest.actions.run_and_verify_svnadmin(expected_full, [],
'hotcopy',
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(expected_incremental, [],
'hotcopy', '--incremental',
sbox.repo_dir, inc_backup_dir)
# Commit three revisions, hotcopy and check the progress output.
for i in range(3):
svntest.actions.run_and_verify_svn(None, [], 'mkdir',
'-m', svntest.main.make_log_msg(),
sbox.repo_url + '/dir-%i' % i)
expected_full = [
"* Copied revision 0.\n",
"* Copied revision 1.\n",
"* Copied revision 2.\n",
"* Copied revision 3.\n",
]
expected_incremental = [
"* Copied revision 1.\n",
"* Copied revision 2.\n",
"* Copied revision 3.\n",
]
backup_dir, backup_url = sbox.add_repo_path('backup-1')
svntest.actions.run_and_verify_svnadmin(expected_full, [],
'hotcopy',
sbox.repo_dir, backup_dir)
svntest.actions.run_and_verify_svnadmin(expected_incremental, [],
'hotcopy', '--incremental',
sbox.repo_dir, inc_backup_dir)
@SkipUnless(svntest.main.fs_has_unique_freeze)
def freeze_same_uuid(sbox):
"freeze multiple repositories with same UUID"
sbox.build(create_wc=False)
first_repo_dir, _ = sbox.add_repo_path('first')
second_repo_dir, _ = sbox.add_repo_path('second')
# Test that 'svnadmin freeze A (svnadmin freeze B)' does not deadlock for
# new FSFS formats, even if 'A' and 'B' share the same UUID. Create two
# repositories by loading the same dump file, ...
svntest.main.create_repos(first_repo_dir)
svntest.main.create_repos(second_repo_dir)
dump_path = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'skeleton_repos.dump')
dump_contents = open(dump_path, 'rb').readlines()
svntest.actions.run_and_verify_load(first_repo_dir, dump_contents)
svntest.actions.run_and_verify_load(second_repo_dir, dump_contents)
# ...and execute the 'svnadmin freeze -F' command.
arg_file = sbox.get_tempname()
svntest.main.file_write(arg_file,
"%s\n%s\n" % (first_repo_dir, second_repo_dir))
svntest.actions.run_and_verify_svnadmin(None, None,
'freeze', '-F', arg_file, '--',
sys.executable, '-c', 'True')
@Skip(svntest.main.is_fs_type_fsx)
def upgrade(sbox):
"upgrade --compatible-version=1.3"
sbox.build(create_wc=False, minor_version=3)
svntest.actions.run_and_verify_svnadmin(None, [], "upgrade",
sbox.repo_dir)
# Does the repository work after upgrade?
svntest.actions.run_and_verify_svn(['Committing transaction...\n',
'Committed revision 2.\n'], [], 'mkdir',
'-m', svntest.main.make_log_msg(),
sbox.repo_url + '/dir')
def load_txdelta(sbox):
"exercising svn_txdelta_target on BDB"
sbox.build(empty=True)
# This dumpfile produced a BDB repository that generated cheksum
# mismatches on read caused by the improper handling of
# svn_txdelta_target ops. The bug was fixed by r1640832.
dumpfile_location = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'load_txdelta.dump.gz')
dumpfile = gzip.open(dumpfile_location, "rb").readlines()
load_dumpstream(sbox, dumpfile)
# Verify would fail with a checksum mismatch:
# * Error verifying revision 14.
# svnadmin: E200014: MD5 checksum mismatch on representation 'r':
# expected: 5182e8876ed894dc7fe28f6ff5b2fee6
# actual: 5121f82875508863ad70daa8244e6947
exit_code, output, errput = svntest.main.run_svnadmin("verify", sbox.repo_dir)
if errput:
raise SVNUnexpectedStderr(errput)
if svntest.verify.verify_outputs(
"Output of 'svnadmin verify' is unexpected.", None, output, None,
".*Verified revision *"):
raise svntest.Failure
@Issues(4563)
def load_no_svndate_r0(sbox):
"load without svn:date on r0"
sbox.build(create_wc=False, empty=True)
# svn:date exits
svntest.actions.run_and_verify_svnlook([' svn:date\n'], [],
'proplist', '--revprop', '-r0',
sbox.repo_dir)
dump_old = [b"SVN-fs-dump-format-version: 2\n", b"\n",
b"UUID: bf52886d-358d-4493-a414-944a6e5ad4f5\n", b"\n",
b"Revision-number: 0\n",
b"Prop-content-length: 10\n",
b"Content-length: 10\n", b"\n",
b"PROPS-END\n", b"\n"]
svntest.actions.run_and_verify_load(sbox.repo_dir, dump_old)
# svn:date should have been removed
svntest.actions.run_and_verify_svnlook([], [],
'proplist', '--revprop', '-r0',
sbox.repo_dir)
# This is only supported for FSFS
# The port to FSX is still pending, BDB won't support it.
@SkipUnless(svntest.main.is_fs_type_fsfs)
def hotcopy_read_only(sbox):
"'svnadmin hotcopy' a read-only source repository"
sbox.build()
svntest.main.chmod_tree(sbox.repo_dir, 0, svntest.main.S_ALL_WRITE)
backup_dir, backup_url = sbox.add_repo_path('backup')
exit_code, output, errput = svntest.main.run_svnadmin("hotcopy",
sbox.repo_dir,
backup_dir)
# r/o repos are hard to clean up. Make it writable again.
svntest.main.chmod_tree(sbox.repo_dir, svntest.main.S_ALL_WRITE,
svntest.main.S_ALL_WRITE)
if errput:
logger.warn("Error: hotcopy failed")
raise SVNUnexpectedStderr(errput)
@SkipUnless(svntest.main.is_fs_type_fsfs)
@SkipUnless(svntest.main.fs_has_pack)
def fsfs_pack_non_sharded(sbox):
"'svnadmin pack' on a non-sharded repository"
# Configure two files per shard to trigger packing.
sbox.build(create_wc = False,
minor_version = min(svntest.main.options.server_minor_version,3))
# Skip for pre-cooked sharded repositories
if is_sharded(sbox.repo_dir):
raise svntest.Skip('sharded pre-cooked repository')
svntest.actions.run_and_verify_svnadmin(
None, [], "upgrade", sbox.repo_dir)
svntest.actions.run_and_verify_svnadmin(
['svnadmin: Warning - this repository is not sharded. Packing has no effect.\n'],
[], "pack", sbox.repo_dir)
def load_revprops(sbox):
"svnadmin load-revprops"
sbox.build(create_wc=False, empty=True)
dump_path = os.path.join(os.path.dirname(sys.argv[0]),
'svnadmin_tests_data',
'skeleton_repos.dump')
dump_contents = open(dump_path, 'rb').readlines()
load_and_verify_dumpstream(sbox, None, [], None, False, dump_contents)
svntest.actions.run_and_verify_svnlook(['Initial setup...\n', '\n'],
[], 'log', '-r1', sbox.repo_dir)
# After loading the dump, amend one of the log message in the repository.
input_file = sbox.get_tempname()
svntest.main.file_write(input_file, 'Modified log message...\n')
svntest.actions.run_and_verify_svnadmin([], [], 'setlog', '--bypass-hooks',
'-r1', sbox.repo_dir, input_file)
svntest.actions.run_and_verify_svnlook(['Modified log message...\n', '\n'],
[], 'log', '-r1', sbox.repo_dir)
# Load the same dump, but with 'svnadmin load-revprops'. Doing so should
# restore the log message to its original state.
svntest.main.run_command_stdin(svntest.main.svnadmin_binary, None, 0,
True, dump_contents, 'load-revprops',
sbox.repo_dir)
svntest.actions.run_and_verify_svnlook(['Initial setup...\n', '\n'],
[], 'log', '-r1', sbox.repo_dir)
def dump_revprops(sbox):
"svnadmin dump-revprops"
sbox.build(create_wc=False)
# Dump revprops only.
exit_code, dump_contents, errput = \
svntest.actions.run_and_verify_svnadmin(None, [], "dump-revprops", "-q",
sbox.repo_dir)
# We expect the dump to contain no path changes
for line in dump_contents:
if line.find(b"Node-path: ") > -1:
logger.warn("Error: path change found in revprops-only dump.")
raise svntest.Failure
# Remember the current log message for r1
exit_code, log_msg, errput = \
svntest.actions.run_and_verify_svnlook(None, [], 'log', '-r1',
sbox.repo_dir)
# Now, change the log message in the repository.
input_file = sbox.get_tempname()
svntest.main.file_write(input_file, 'Modified log message...\n')
svntest.actions.run_and_verify_svnadmin([], [], 'setlog', '--bypass-hooks',
'-r1', sbox.repo_dir, input_file)
svntest.actions.run_and_verify_svnlook(['Modified log message...\n', '\n'],
[], 'log', '-r1', sbox.repo_dir)
# Load the same dump with 'svnadmin load-revprops'. Doing so should
# restore the log message to its original state.
svntest.main.run_command_stdin(svntest.main.svnadmin_binary, None, 0,
True, dump_contents, 'load-revprops',
sbox.repo_dir)
svntest.actions.run_and_verify_svnlook(log_msg, [], 'log', '-r1',
sbox.repo_dir)
@XFail(svntest.main.is_fs_type_fsx)
@Issue(4598)
def dump_no_op_change(sbox):
"svnadmin dump with no-op changes"
sbox.build(create_wc=False, empty=True)
empty_file = sbox.get_tempname()
svntest.main.file_write(empty_file, '')
svntest.actions.run_and_verify_svnmucc(None, [],
'-U', sbox.repo_url,
'-m', svntest.main.make_log_msg(),
'put', empty_file, 'bar')
# Commit a no-op change.
svntest.actions.run_and_verify_svnmucc(None, [],
'-U', sbox.repo_url,
'-m', svntest.main.make_log_msg(),
'put', empty_file, 'bar')
# Dump and load the repository.
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
sbox.repo_dir)
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump)
# We expect svn log -v to yield identical results for both original and
# reconstructed repositories. This used to fail as described in the
# Issue 4598 (https://issues.apache.org/jira/browse/SVN-4598), at least
# around r1706415.
#
# Test svn log -v for r2:
_, expected, _ = svntest.actions.run_and_verify_svn(None, [], 'log', '-v',
'-r2', sbox.repo_url)
found = [True for line in expected if line.find('M /bar\n') != -1]
if not found:
raise svntest.Failure
svntest.actions.run_and_verify_svn(expected, [], 'log', '-v',
'-r2', sbox2.repo_url)
# Test svn log -v for /bar:
_, expected, _ = svntest.actions.run_and_verify_svn(None, [], 'log', '-v',
sbox.repo_url + '/bar')
found = [True for line in expected if line.find('M /bar\n') != -1]
if not found:
raise svntest.Failure
svntest.actions.run_and_verify_svn(expected, [], 'log', '-v',
sbox2.repo_url + '/bar')
@XFail(svntest.main.is_fs_type_bdb)
@XFail(svntest.main.is_fs_type_fsx)
@Issue(4623)
def dump_no_op_prop_change(sbox):
"svnadmin dump with no-op property change"
sbox.build(create_wc=False, empty=True)
empty_file = sbox.get_tempname()
svntest.main.file_write(empty_file, '')
svntest.actions.run_and_verify_svnmucc(None, [],
'-U', sbox.repo_url,
'-m', svntest.main.make_log_msg(),
'put', empty_file, 'bar',
'propset', 'pname', 'pval', 'bar')
# Commit a no-op property change.
svntest.actions.run_and_verify_svnmucc(None, [],
'-U', sbox.repo_url,
'-m', svntest.main.make_log_msg(),
'propset', 'pname', 'pval', 'bar')
# Dump and load the repository.
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
sbox.repo_dir)
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump)
# Test svn log -v for r2:
_, expected, _ = svntest.actions.run_and_verify_svn(None, [], 'log', '-v',
'-r2', sbox.repo_url)
found = [True for line in expected if line.find('M /bar\n') != -1]
if not found:
raise svntest.Failure
svntest.actions.run_and_verify_svn(expected, [], 'log', '-v',
'-r2', sbox2.repo_url)
# Test svn log -v for /bar:
_, expected, _ = svntest.actions.run_and_verify_svn(None, [], 'log', '-v',
sbox.repo_url + '/bar')
found = [True for line in expected if line.find('M /bar\n') != -1]
if not found:
raise svntest.Failure
svntest.actions.run_and_verify_svn(expected, [], 'log', '-v',
sbox2.repo_url + '/bar')
def load_no_flush_to_disk(sbox):
"svnadmin load --no-flush-to-disk"
sbox.build(empty=True)
# Can't test the "not flushing to disk part", but loading the
# dump should work.
dump = clean_dumpfile()
expected = [
svntest.wc.State('', {
'A' : svntest.wc.StateItem(contents="text\n",
props={'svn:keywords': 'Id'})
})
]
load_and_verify_dumpstream(sbox, [], [], expected, True, dump,
'--no-flush-to-disk', '--ignore-uuid')
def dump_to_file(sbox):
"svnadmin dump --file ARG"
sbox.build(create_wc=False, empty=False)
expected_dump = svntest.actions.run_and_verify_dump(sbox.repo_dir)
file = sbox.get_tempname()
svntest.actions.run_and_verify_svnadmin2([],
["* Dumped revision 0.\n",
"* Dumped revision 1.\n"],
0, 'dump', '--file', file,
sbox.repo_dir)
actual_dump = open(file, 'rb').readlines()
svntest.verify.compare_dump_files(None, None, expected_dump, actual_dump)
# Test that svnadmin dump --file overwrites existing files.
file = sbox.get_tempname()
svntest.main.file_write(file, '')
svntest.actions.run_and_verify_svnadmin2([],
["* Dumped revision 0.\n",
"* Dumped revision 1.\n"],
0, 'dump', '--file', file,
sbox.repo_dir)
actual_dump = open(file, 'rb').readlines()
svntest.verify.compare_dump_files(None, None, expected_dump, actual_dump)
def load_from_file(sbox):
"svnadmin load --file ARG"
sbox.build(empty=True)
file = sbox.get_tempname()
with open(file, 'wb') as f:
f.writelines(clean_dumpfile())
svntest.actions.run_and_verify_svnadmin2(None, [],
0, 'load', '--file', file,
'--ignore-uuid', sbox.repo_dir)
expected_tree = \
svntest.wc.State('', {
'A' : svntest.wc.StateItem(contents="text\n",
props={'svn:keywords': 'Id'})
})
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [],
'update', sbox.wc_dir)
svntest.actions.verify_disk(sbox.wc_dir, expected_tree, check_props=True)
def dump_exclude(sbox):
"svnadmin dump with excluded paths"
sbox.build(create_wc=False)
# Dump repository with /A/D/H and /A/B/E paths excluded.
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
'--exclude', '/A/D/H',
'--exclude', '/A/B/E',
sbox.repo_dir)
# Load repository from dump.
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump)
# Check log.
expected_output = svntest.verify.RegexListOutput([
'-+\\n',
'r1\ .*\n',
# '/A/D/H' and '/A/B/E' is not added.
re.escape('Changed paths:\n'),
re.escape(' A /A\n'),
re.escape(' A /A/B\n'),
re.escape(' A /A/B/F\n'),
re.escape(' A /A/B/lambda\n'),
re.escape(' A /A/C\n'),
re.escape(' A /A/D\n'),
re.escape(' A /A/D/G\n'),
re.escape(' A /A/D/G/pi\n'),
re.escape(' A /A/D/G/rho\n'),
re.escape(' A /A/D/G/tau\n'),
re.escape(' A /A/D/gamma\n'),
re.escape(' A /A/mu\n'),
re.escape(' A /iota\n'),
'-+\\n'
])
svntest.actions.run_and_verify_svn(expected_output, [],
'log', '-v', '-q', sbox2.repo_url)
def dump_exclude_copysource(sbox):
"svnadmin dump with excluded copysource"
sbox.build(create_wc=False, empty=True)
# Create default repository structure.
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "mkdir",
sbox.repo_url + '/trunk',
sbox.repo_url + '/branches',
sbox.repo_url + '/tags',
"-m", "Create repository structure.")
# Create a branch.
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "copy",
sbox.repo_url + '/trunk',
sbox.repo_url + '/branches/branch1',
"-m", "Create branch.")
# Dump repository with /trunk excluded.
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
'--exclude', '/trunk',
sbox.repo_dir)
# Load repository from dump.
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump)
# Check log.
expected_output = svntest.verify.RegexListOutput([
'-+\\n',
'r2\ .*\n',
re.escape('Changed paths:\n'),
# Simple add, not copy.
re.escape(' A /branches/branch1\n'),
'-+\\n',
'r1\ .*\n',
# '/trunk' is not added.
re.escape('Changed paths:\n'),
re.escape(' A /branches\n'),
re.escape(' A /tags\n'),
'-+\\n'
])
svntest.actions.run_and_verify_svn(expected_output, [],
'log', '-v', '-q', sbox2.repo_url)
def dump_include(sbox):
"svnadmin dump with included paths"
sbox.build(create_wc=False, empty=True)
# Create a couple of directories.
# Note that we can't use greek tree as it contains only two top-level
# nodes. Including non top-level nodes (e.g. '--include /A/B/E') will
# produce unloadable dump for now.
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "mkdir",
sbox.repo_url + '/A',
sbox.repo_url + '/B',
sbox.repo_url + '/C',
"-m", "Create folder.")
# Dump repository with /A and /C paths included.
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
'--include', '/A',
'--include', '/C',
sbox.repo_dir)
# Load repository from dump.
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump)
# Check log.
expected_output = svntest.verify.RegexListOutput([
'-+\\n',
'r1\ .*\n',
# '/B' is not added.
re.escape('Changed paths:\n'),
re.escape(' A /A\n'),
re.escape(' A /C\n'),
'-+\\n'
])
svntest.actions.run_and_verify_svn(expected_output, [],
'log', '-v', '-q', sbox2.repo_url)
def dump_not_include_copysource(sbox):
"svnadmin dump with not included copysource"
sbox.build(create_wc=False, empty=True)
# Create default repository structure.
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "mkdir",
sbox.repo_url + '/trunk',
sbox.repo_url + '/branches',
sbox.repo_url + '/tags',
"-m", "Create repository structure.")
# Create a branch.
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "copy",
sbox.repo_url + '/trunk',
sbox.repo_url + '/branches/branch1',
"-m", "Create branch.")
# Dump repository with only /branches included.
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
'--include', '/branches',
sbox.repo_dir)
# Load repository from dump.
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump)
# Check log.
expected_output = svntest.verify.RegexListOutput([
'-+\\n',
'r2\ .*\n',
re.escape('Changed paths:\n'),
# Simple add, not copy.
re.escape(' A /branches/branch1\n'),
'-+\\n',
'r1\ .*\n',
# Only '/branches' is added in r1.
re.escape('Changed paths:\n'),
re.escape(' A /branches\n'),
'-+\\n'
])
svntest.actions.run_and_verify_svn(expected_output, [],
'log', '-v', '-q', sbox2.repo_url)
def dump_exclude_by_pattern(sbox):
"svnadmin dump with paths excluded by pattern"
sbox.build(create_wc=False, empty=True)
# Create a couple of directories.
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "mkdir",
sbox.repo_url + '/aaa',
sbox.repo_url + '/aab',
sbox.repo_url + '/aac',
sbox.repo_url + '/bbc',
"-m", "Create repository structure.")
# Dump with paths excluded by pattern.
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
'--exclude', '/aa?',
'--pattern',
sbox.repo_dir)
# Load repository from dump.
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump)
# Check log.
expected_output = svntest.verify.RegexListOutput([
'-+\\n',
'r1\ .*\n',
re.escape('Changed paths:\n'),
# Only '/bbc' is added in r1.
re.escape(' A /bbc\n'),
'-+\\n'
])
svntest.actions.run_and_verify_svn(expected_output, [],
'log', '-v', '-q', sbox2.repo_url)
def dump_include_by_pattern(sbox):
"svnadmin dump with paths included by pattern"
sbox.build(create_wc=False, empty=True)
# Create a couple of directories.
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "mkdir",
sbox.repo_url + '/aaa',
sbox.repo_url + '/aab',
sbox.repo_url + '/aac',
sbox.repo_url + '/bbc',
"-m", "Create repository structure.")
# Dump with paths included by pattern.
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
'--include', '/aa?',
'--pattern',
sbox.repo_dir)
# Load repository from dump.
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump)
# Check log.
expected_output = svntest.verify.RegexListOutput([
'-+\\n',
'r1\ .*\n',
# '/bbc' is not added.
re.escape('Changed paths:\n'),
re.escape(' A /aaa\n'),
re.escape(' A /aab\n'),
re.escape(' A /aac\n'),
'-+\\n'
])
svntest.actions.run_and_verify_svn(expected_output, [],
'log', '-v', '-q', sbox2.repo_url)
def dump_exclude_all_rev_changes(sbox):
"svnadmin dump with all revision changes excluded"
sbox.build(create_wc=False, empty=True)
# Create a couple of directories (r1).
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "mkdir",
sbox.repo_url + '/r1a',
sbox.repo_url + '/r1b',
sbox.repo_url + '/r1c',
"-m", "Revision 1.")
# Create a couple of directories (r2).
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "mkdir",
sbox.repo_url + '/r2a',
sbox.repo_url + '/r2b',
sbox.repo_url + '/r2c',
"-m", "Revision 2.")
# Create a couple of directories (r3).
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "mkdir",
sbox.repo_url + '/r3a',
sbox.repo_url + '/r3b',
sbox.repo_url + '/r3c',
"-m", "Revision 3.")
# Dump with paths excluded by pattern.
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
'--exclude', '/r2?',
'--pattern',
sbox.repo_dir)
# Load repository from dump.
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump)
# Check log. Revision properties ('svn:log' etc.) should be empty for r2.
expected_output = svntest.verify.RegexListOutput([
'-+\\n',
'r3 | jrandom | .* | 1 line\\n',
re.escape('Changed paths:'),
re.escape(' A /r3a'),
re.escape(' A /r3b'),
re.escape(' A /r3c'),
'',
re.escape('Revision 3.'),
'-+\\n',
re.escape('r2 | (no author) | (no date) | 1 line'),
'',
'',
'-+\\n',
'r1 | jrandom | .* | 1 line\\n',
re.escape('Changed paths:'),
re.escape(' A /r1a'),
re.escape(' A /r1b'),
re.escape(' A /r1c'),
'',
re.escape('Revision 1.'),
'-+\\n',
])
svntest.actions.run_and_verify_svn(expected_output, [],
'log', '-v', sbox2.repo_url)
def dump_invalid_filtering_option(sbox):
"dump with --include and --exclude simultaneously"
sbox.build(create_wc=False, empty=False)
# Attempt to dump repository with '--include' and '--exclude' options
# specified simultaneously.
expected_error = ".*: '--exclude' and '--include' options cannot be used " \
"simultaneously"
svntest.actions.run_and_verify_svnadmin(None, expected_error,
'dump', '-q',
'--exclude', '/A/D/H',
'--include', '/A/B/E',
sbox.repo_dir)
@Issue(4725)
def load_issue4725(sbox):
"""load that triggers issue 4725"""
sbox.build(empty=True)
sbox.simple_mkdir('subversion')
sbox.simple_commit()
sbox.simple_mkdir('subversion/trunk')
sbox.simple_mkdir('subversion/branches')
sbox.simple_commit()
sbox.simple_mkdir('subversion/trunk/src')
sbox.simple_commit()
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
sbox.repo_dir)
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump, '-M100')
@Issue(4767)
def dump_no_canonicalize_svndate(sbox):
"svnadmin dump shouldn't canonicalize svn:date"
sbox.build(create_wc=False, empty=True)
svntest.actions.enable_revprop_changes(sbox.repo_dir)
# set svn:date in a non-canonical format (not six decimal places)
propval = "2015-01-01T00:00:00.0Z"
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [],
"propset", "--revprop", "-r0", "svn:date",
propval,
sbox.repo_url)
dump_lines = svntest.actions.run_and_verify_dump(sbox.repo_dir)
assert propval.encode() + b'\n' in dump_lines
def check_recover_prunes_rep_cache(sbox, enable_rep_sharing):
"""Check 'recover' prunes the rep-cache while enable-rep-sharing is
true/false.
"""
# Remember the initial rep cache content.
rep_cache_r1 = read_rep_cache(sbox.repo_dir)
#print '\n'.join([h + ": " + repr(ref) for h, ref in rep_cache_r1.items()])
# Commit one new rep and check the rep-cache is extended.
sbox.simple_append('iota', 'New line.\n')
sbox.simple_commit()
rep_cache_r2 = read_rep_cache(sbox.repo_dir)
if not (len(rep_cache_r2) == len(rep_cache_r1) + 1):
raise svntest.Failure
fsfs_conf = svntest.main.get_fsfs_conf_file_path(sbox.repo_dir)
svntest.main.file_append(fsfs_conf,
# Add a newline in case the existing file doesn't
# end with one.
"\n"
"[rep-sharing]\n"
"enable-rep-sharing = %s\n"
% (('true' if enable_rep_sharing else 'false'),))
# Break r2 in such a way that 'recover' will discard it
head_rev_path = fsfs_file(sbox.repo_dir, 'revs', '2')
os.remove(head_rev_path)
current_path = os.path.join(sbox.repo_dir, 'db', 'current')
svntest.main.file_write(current_path, '1\n')
# Recover back to r1.
svntest.actions.run_and_verify_svnadmin(None, [],
"recover", sbox.repo_dir)
svntest.actions.run_and_verify_svnlook(['1\n'], [], 'youngest',
sbox.repo_dir)
# Check the rep-cache is pruned.
rep_cache_recovered = read_rep_cache(sbox.repo_dir)
if not (rep_cache_recovered == rep_cache_r1):
raise svntest.Failure
@Issue(4077)
@SkipUnless(svntest.main.is_fs_type_fsfs)
@SkipUnless(svntest.main.python_sqlite_can_read_without_rowid)
def recover_prunes_rep_cache_when_enabled(sbox):
"recover prunes rep cache when enabled"
sbox.build()
check_recover_prunes_rep_cache(sbox, enable_rep_sharing=True)
@Issue(4077)
@SkipUnless(svntest.main.is_fs_type_fsfs)
@SkipUnless(svntest.main.python_sqlite_can_read_without_rowid)
def recover_prunes_rep_cache_when_disabled(sbox):
"recover prunes rep cache when disabled"
sbox.build()
check_recover_prunes_rep_cache(sbox, enable_rep_sharing=False)
@Issue(4760)
def dump_include_copied_directory(sbox):
"include copied directory with nested nodes"
sbox.build(create_wc=False)
svntest.actions.run_and_verify_svn(svntest.verify.AnyOutput, [], "copy",
sbox.repo_url + '/A/D',
sbox.repo_url + '/COPY',
"-m", "Create branch.")
# Dump repository with only /COPY path included.
_, dump, _ = svntest.actions.run_and_verify_svnadmin(None, [],
'dump', '-q',
'--include', '/COPY',
sbox.repo_dir)
# Load repository from dump.
sbox2 = sbox.clone_dependent()
sbox2.build(create_wc=False, empty=True)
load_and_verify_dumpstream(sbox2, None, [], None, False, dump)
# Check log.
expected_output = svntest.verify.RegexListOutput([
'-+\\n',
'r2\ .*\n',
# Only '/COPY' is added
re.escape('Changed paths:\n'),
re.escape(' A /COPY'),
re.escape(' A /COPY/G'),
re.escape(' A /COPY/G/pi'),
re.escape(' A /COPY/G/rho'),
re.escape(' A /COPY/G/tau'),
re.escape(' A /COPY/H'),
re.escape(' A /COPY/H/chi'),
re.escape(' A /COPY/H/omega'),
re.escape(' A /COPY/H/psi'),
re.escape(' A /COPY/gamma'),
'-+\\n',
'r1\ .*\n',
'-+\\n'
])
svntest.actions.run_and_verify_svn(expected_output, [],
'log', '-v', '-q', sbox2.repo_url)
def load_normalize_node_props(sbox):
"svnadmin load --normalize node props"
dump_str = b"""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 0
K 10
svn:author
V 2
pl
K 8
svn:date
V 27
2005-05-03T19:10:19.975578Z
PROPS-END
Node-path:\x20
Node-kind: dir
Node-action: change
Prop-content-length: 32
Content-length: 32
K 10
svn:ignore
V 3
\n\r\n
PROPS-END
"""
sbox.build(empty=True)
# Try to load the dumpstream, expecting a failure (because of mixed
# EOLs in the svn:ignore property value).
exp_err = svntest.verify.RegexListOutput(['svnadmin: E125005:.*',
'svnadmin: E125017:.*'],
match_all=False)
load_and_verify_dumpstream(sbox, [], exp_err, dumpfile_revisions,
False, dump_str, '--ignore-uuid')
# Now try it again with prop normalization.
svntest.actions.load_repo(sbox, dump_str=dump_str,
bypass_prop_validation=False,
normalize_props=True)
# We should get the normalized property value.
exit_code, output, _ = svntest.main.run_svn(None, 'pg', 'svn:ignore',
'--no-newline',
sbox.repo_url)
svntest.verify.verify_exit_code(None, exit_code, 0)
if output != ['\n', '\n']:
raise svntest.Failure("Unexpected property value %s" % output)
@SkipUnless(svntest.main.is_fs_type_fsfs)
@SkipUnless(svntest.main.fs_has_rep_sharing)
@SkipUnless(svntest.main.python_sqlite_can_read_without_rowid)
def build_repcache(sbox):
"svnadmin build-repcache"
sbox.build(create_wc = False)
# Remember and remove the existing rep-cache.
rep_cache = read_rep_cache(sbox.repo_dir)
rep_cache_path = os.path.join(sbox.repo_dir, 'db', 'rep-cache.db')
os.remove(rep_cache_path)
# Build a new rep-cache and compare with the original one.
expected_output = ["* Processed revision 1.\n"]
svntest.actions.run_and_verify_svnadmin(expected_output, [],
"build-repcache", sbox.repo_dir)
new_rep_cache = read_rep_cache(sbox.repo_dir)
if new_rep_cache != rep_cache:
raise svntest.Failure
########################################################################
# Run the tests
# list all tests here, starting with None:
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,
fsfs_recover_db_current,
fsfs_recover_old_db_current,
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,
test_lslocks_and_rmlocks,
load_ranges,
hotcopy_incremental,
hotcopy_incremental_packed,
locking,
mergeinfo_race,
recover_old_empty,
verify_keep_going,
verify_keep_going_quiet,
verify_invalid_path_changes,
verify_denormalized_names,
fsfs_recover_old_non_empty,
fsfs_hotcopy_old_non_empty,
load_ignore_dates,
fsfs_hotcopy_old_with_id_changes,
verify_packed,
freeze_freeze,
verify_metadata_only,
verify_quickly,
fsfs_hotcopy_progress,
fsfs_hotcopy_progress_with_revprop_changes,
fsfs_hotcopy_progress_old,
freeze_same_uuid,
upgrade,
load_txdelta,
load_no_svndate_r0,
hotcopy_read_only,
fsfs_pack_non_sharded,
load_revprops,
dump_revprops,
dump_no_op_change,
dump_no_op_prop_change,
load_no_flush_to_disk,
dump_to_file,
load_from_file,
dump_exclude,
dump_exclude_copysource,
dump_include,
dump_not_include_copysource,
dump_exclude_by_pattern,
dump_include_by_pattern,
dump_exclude_all_rev_changes,
dump_invalid_filtering_option,
load_issue4725,
dump_no_canonicalize_svndate,
recover_prunes_rep_cache_when_enabled,
recover_prunes_rep_cache_when_disabled,
dump_include_copied_directory,
load_normalize_node_props,
build_repcache,
]
if __name__ == '__main__':
svntest.main.run_tests(test_list)
# NOTREACHED
### End of file.