#!/usr/bin/env python # # svnsync_tests.py: Tests SVNSync's repository mirroring capabilities. # # 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 sys, os # Test suite-specific modules import re # Our testing module import svntest from svntest.verify import SVNUnexpectedStdout, SVNUnexpectedStderr from svntest.verify import SVNExpectedStderr from svntest.verify import AnyOutput from svntest.main import server_has_partial_replay # (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 Item = svntest.wc.StateItem ###################################################################### # Helper routines def run_sync(url, source_url=None, source_prop_encoding=None, expected_output=AnyOutput, expected_error=[]): "Synchronize the mirror repository with the master" if source_url is not None: args = ["synchronize", url, source_url] else: # Allow testing of old source-URL-less syntax args = ["synchronize", url] if source_prop_encoding: args.append("--source-prop-encoding") args.append(source_prop_encoding) # Normal expected output is of the form: # ['Transmitting file data .......\n', # optional # 'Committed revision 1.\n', # 'Copied properties for revision 1.\n', ...] svntest.actions.run_and_verify_svnsync(expected_output, expected_error, *args) def run_copy_revprops(url, source_url, source_prop_encoding=None, expected_output=AnyOutput, expected_error=[]): "Copy revprops to the mirror repository from the master" args = ["copy-revprops", url, source_url] if source_prop_encoding: args.append("--source-prop-encoding") args.append(source_prop_encoding) # Normal expected output is of the form: # ['Copied properties for revision 1.\n', ...] svntest.actions.run_and_verify_svnsync(expected_output, expected_error, *args) def run_init(dst_url, src_url, source_prop_encoding=None): "Initialize the mirror repository from the master" args = ["initialize", dst_url, src_url] if source_prop_encoding: args.append("--source-prop-encoding") args.append(source_prop_encoding) expected_output = ['Copied properties for revision 0.\n'] svntest.actions.run_and_verify_svnsync(expected_output, [], *args) def run_info(url, expected_output=AnyOutput, expected_error=[]): "Print synchronization information of the repository" # Normal expected output is of the form: # ['From URL: http://....\n', # 'From UUID: XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX\n', # 'Last Merged Revision: XXX\n'] svntest.actions.run_and_verify_svnsync(expected_output, expected_error, "info", url) def setup_and_sync(sbox, dump_file_contents, subdir=None, bypass_prop_validation=False, source_prop_encoding=None, is_src_ra_local=None, is_dest_ra_local=None): """Create a repository for SBOX, load it with DUMP_FILE_CONTENTS, then create a mirror repository and sync it with SBOX. If is_src_ra_local or is_dest_ra_local is True, then run_init, run_sync, and run_copy_revprops will use the file:// scheme for the source and destination URLs. Return the mirror sandbox.""" # Create the empty master repository. sbox.build(create_wc=False, empty=True) # Load the repository from DUMP_FILE_PATH. svntest.actions.run_and_verify_load(sbox.repo_dir, dump_file_contents, bypass_prop_validation) # Create the empty destination repository. dest_sbox = sbox.clone_dependent() dest_sbox.build(create_wc=False, empty=True) # Setup the mirror repository. Feed it the UUID of the source repository. exit_code, output, errput = svntest.main.run_svnlook("uuid", sbox.repo_dir) svntest.actions.run_and_verify_svnadmin2(None, None, 0, 'setuuid', dest_sbox.repo_dir, output[0][:-1]) # Create the revprop-change hook for this test svntest.actions.enable_revprop_changes(dest_sbox.repo_dir) repo_url = sbox.repo_url cwd = os.getcwd() if is_src_ra_local: repo_url = sbox.file_protocol_url() if subdir: repo_url = repo_url + subdir dest_repo_url = dest_sbox.repo_url if is_dest_ra_local: dest_repo_url = dest_sbox.file_protocol_url() run_init(dest_repo_url, repo_url, source_prop_encoding) run_sync(dest_repo_url, repo_url, source_prop_encoding=source_prop_encoding) run_copy_revprops(dest_repo_url, repo_url, source_prop_encoding=source_prop_encoding) return dest_sbox def verify_mirror(dest_sbox, exp_dump_file_contents): """Compare the contents of the mirror repository in DEST_SBOX with EXP_DUMP_FILE_CONTENTS, by comparing the parsed dump stream content. First remove svnsync rev-props from the DEST_SBOX repository. """ # Remove some SVNSync-specific housekeeping properties from the # mirror repository in preparation for the comparison dump. for prop_name in ("svn:sync-from-url", "svn:sync-from-uuid", "svn:sync-last-merged-rev"): svntest.actions.run_and_verify_svn( None, [], "propdel", "--revprop", "-r", "0", prop_name, dest_sbox.repo_url) # Create a dump file from the mirror repository. dest_dump = svntest.actions.run_and_verify_dump(dest_sbox.repo_dir) svntest.verify.compare_dump_files( "Dump files", "DUMP", exp_dump_file_contents, dest_dump) def run_test(sbox, dump_file_name, subdir=None, exp_dump_file_name=None, bypass_prop_validation=False, source_prop_encoding=None, is_src_ra_local=None, is_dest_ra_local=None): """Load a dump file, sync repositories, and compare contents with the original or another dump file.""" # This directory contains all the dump files svnsync_tests_dir = os.path.join(os.path.dirname(sys.argv[0]), 'svnsync_tests_data') # Load the specified dump file into the master repository. master_dumpfile_contents = open(os.path.join(svnsync_tests_dir, dump_file_name), 'rb').readlines() dest_sbox = setup_and_sync(sbox, master_dumpfile_contents, subdir, bypass_prop_validation, source_prop_encoding, is_src_ra_local, is_dest_ra_local) # Compare the dump produced by the mirror repository with either the original # dump file (used to create the master repository) or another specified dump # file. if exp_dump_file_name: exp_dump_file_contents = open(os.path.join(svnsync_tests_dir, exp_dump_file_name), 'rb').readlines() else: exp_dump_file_contents = master_dumpfile_contents verify_mirror(dest_sbox, exp_dump_file_contents) ###################################################################### # Tests #---------------------------------------------------------------------- def copy_and_modify(sbox): "copy and modify" run_test(sbox, "copy-and-modify.dump") #---------------------------------------------------------------------- def copy_from_previous_version_and_modify(sbox): "copy from previous version and modify" run_test(sbox, "copy-from-previous-version-and-modify.dump") #---------------------------------------------------------------------- def copy_from_previous_version(sbox): "copy from previous version" run_test(sbox, "copy-from-previous-version.dump") #---------------------------------------------------------------------- def modified_in_place(sbox): "modified in place" run_test(sbox, "modified-in-place.dump") #---------------------------------------------------------------------- def tag_empty_trunk(sbox): "tag empty trunk" run_test(sbox, "tag-empty-trunk.dump") #---------------------------------------------------------------------- def tag_trunk_with_dir(sbox): "tag trunk containing a sub-directory" run_test(sbox, "tag-trunk-with-dir.dump") #---------------------------------------------------------------------- def tag_trunk_with_file(sbox): "tag trunk containing a file" run_test(sbox, "tag-trunk-with-file.dump") #---------------------------------------------------------------------- def tag_trunk_with_file2(sbox): "tag trunk containing a file (#2)" run_test(sbox, "tag-trunk-with-file2.dump") #---------------------------------------------------------------------- def tag_with_modified_file(sbox): "tag with a modified file" run_test(sbox, "tag-with-modified-file.dump") #---------------------------------------------------------------------- def dir_prop_change(sbox): "directory property changes" run_test(sbox, "dir-prop-change.dump") #---------------------------------------------------------------------- def file_dir_file(sbox): "files and dirs mixed together" run_test(sbox, "file-dir-file.dump") #---------------------------------------------------------------------- def copy_parent_modify_prop(sbox): "copy parent and modify prop" run_test(sbox, "copy-parent-modify-prop.dump") #---------------------------------------------------------------------- def detect_meddling(sbox): "detect non-svnsync commits in destination" sbox.build("svnsync-meddling") dest_sbox = sbox.clone_dependent() dest_sbox.build(create_wc=False, empty=True) # Make our own destination checkout (have to do it ourself because # it is not greek). svntest.main.safe_rmtree(dest_sbox.wc_dir) svntest.actions.run_and_verify_svn(None, [], 'co', dest_sbox.repo_url, dest_sbox.wc_dir) svntest.actions.enable_revprop_changes(dest_sbox.repo_dir) run_init(dest_sbox.repo_url, sbox.repo_url) run_sync(dest_sbox.repo_url) svntest.actions.run_and_verify_svn(None, [], 'up', dest_sbox.wc_dir) # Commit some change to the destination, which should be detected by svnsync svntest.main.file_append(os.path.join(dest_sbox.wc_dir, 'A', 'B', 'lambda'), 'new lambda text') svntest.actions.run_and_verify_svn(None, [], 'ci', '-m', 'msg', dest_sbox.wc_dir) expected_error = r".*Destination HEAD \(2\) is not the last merged revision \(1\).*" run_sync(dest_sbox.repo_url, None, expected_output=[], expected_error=expected_error) def url_encoding(sbox): "test url encoding issues" run_test(sbox, "url-encoding-bug.dump") # A test for copying revisions that lack a property that already exists # on the destination rev as part of the commit (i.e. svn:author in this # case, but svn:date would also work). def no_author(sbox): "test copying revs with no svn:author revprops" run_test(sbox, "no-author.dump") def copy_revprops(sbox): "test copying revprops other than svn:*" run_test(sbox, "revprops.dump") @SkipUnless(server_has_partial_replay) def only_trunk(sbox): "test syncing subdirectories" run_test(sbox, "svnsync-trunk-only.dump", "/trunk", "svnsync-trunk-only.expected.dump") @SkipUnless(server_has_partial_replay) def only_trunk_A_with_changes(sbox): "test syncing subdirectories with changes on root" run_test(sbox, "svnsync-trunk-A-changes.dump", "/trunk/A", "svnsync-trunk-A-changes.expected.dump") # test for issue #2904 @Issue(2904) def move_and_modify_in_the_same_revision(sbox): "test move parent and modify child file in same rev" run_test(sbox, "svnsync-move-and-modify.dump") def info_synchronized(sbox): "test info cmd on a synchronized repo" sbox.build("svnsync-info-syncd", False) # Get the UUID of the source repository. exit_code, output, errput = svntest.main.run_svnlook("uuid", sbox.repo_dir) src_uuid = output[0].strip() dest_sbox = sbox.clone_dependent() dest_sbox.build(create_wc=False, empty=True) svntest.actions.enable_revprop_changes(dest_sbox.repo_dir) run_init(dest_sbox.repo_url, sbox.repo_url) run_sync(dest_sbox.repo_url) expected_out = ['Source URL: %s\n' % sbox.repo_url, 'Source Repository UUID: %s\n' % src_uuid, 'Last Merged Revision: 1\n', ] svntest.actions.run_and_verify_svnsync(expected_out, [], "info", dest_sbox.repo_url) def info_not_synchronized(sbox): "test info cmd on an un-synchronized repo" sbox.build("svnsync-info-not-syncd", False) run_info(sbox.repo_url, [], ".*Repository '%s' is not initialized.*" % sbox.repo_url) #---------------------------------------------------------------------- def copy_bad_line_endings(sbox): "copy with inconsistent line endings in svn:* props" run_test(sbox, "copy-bad-line-endings.dump", exp_dump_file_name="copy-bad-line-endings.expected.dump", bypass_prop_validation=True) def copy_bad_line_endings2(sbox): "copy with non-LF line endings in svn:* props" run_test(sbox, "copy-bad-line-endings2.dump", exp_dump_file_name="copy-bad-line-endings2.expected.dump", bypass_prop_validation=True) def copy_bad_encoding(sbox): "copy and reencode non-UTF-8 svn:* props" run_test(sbox, "copy-bad-encoding.dump", exp_dump_file_name="copy-bad-encoding.expected.dump", bypass_prop_validation=True, source_prop_encoding="ISO-8859-3") #---------------------------------------------------------------------- def delete_svn_props(sbox): "copy with svn:* prop deletions" run_test(sbox, "delete-svn-props.dump") @Issue(3438) def commit_a_copy_of_root(sbox): "commit a copy of root causes sync to fail" #Testcase for issue 3438. run_test(sbox, "repo-with-copy-of-root-dir.dump") # issue #3641 'svnsync fails to partially copy a repository'. # This currently fails because while replacements with history # within copies are handled, replacements without history inside # copies cause the sync to fail: # # >svnsync synchronize %TEST_REPOS_ROOT_URL%/svnsync_tests-29-1 # %TEST_REPOS_ROOT_URL%/svnsync_tests-29/trunk/H # Transmitting file data ...\..\..\subversion\svnsync\main.c:1444: (apr_err=160013) # ..\..\..\subversion\svnsync\main.c:1391: (apr_err=160013) # ..\..\..\subversion\libsvn_ra\ra_loader.c:1168: (apr_err=160013) # ..\..\..\subversion\libsvn_delta\path_driver.c:254: (apr_err=160013) # ..\..\..\subversion\libsvn_repos\replay.c:480: (apr_err=160013) # ..\..\..\subversion\libsvn_repos\replay.c:276: (apr_err=160013) # ..\..\..\subversion\libsvn_repos\replay.c:290: (apr_err=160013) # ..\..\..\subversion\libsvn_fs_base\tree.c:1258: (apr_err=160013) # ..\..\..\subversion\libsvn_fs_base\tree.c:1258: (apr_err=160013) # ..\..\..\subversion\libsvn_fs_base\tree.c:1236: (apr_err=160013) # ..\..\..\subversion\libsvn_fs_base\tree.c:931: (apr_err=160013) # ..\..\..\subversion\libsvn_fs_base\tree.c:742: (apr_err=160013) # svnsync: File not found: revision 4, path '/trunk/H/Z/B/lambda' # # See also http://svn.haxx.se/dev/archive-2010-11/0411.shtml and # # # Note: For those who may poke around this test in the future, r3 of # delete-revprops.dump was created with the following svnmucc command: # # svnmucc.exe -mm cp head %ROOT_URL%/trunk/A %ROOT_URL%/trunk/H # rm %ROOT_URL%/trunk/H/B # cp head %ROOT_URL%/trunk/X %ROOT_URL%/trunk/B # # r4 was created with this svnmucc command: # # svnmucc.exe -mm cp head %ROOT_URL%/trunk/A %ROOT_URL%/trunk/H/Z # rm %ROOT_URL%/trunk/H/Z/B # mkdir %ROOT_URL%/trunk/H/Z/B @Issue(3641) def descend_into_replace(sbox): "descending into replaced dir looks in src" run_test(sbox, "descend-into-replace.dump", subdir='/trunk/H', exp_dump_file_name = "descend-into-replace.expected.dump") # issue #3728 @Issue(3728) def delete_revprops(sbox): "copy-revprops with removals" svnsync_tests_dir = os.path.join(os.path.dirname(sys.argv[0]), 'svnsync_tests_data') initial_contents = open(os.path.join(svnsync_tests_dir, "delete-revprops.dump"), 'rb').readlines() expected_contents = open(os.path.join(svnsync_tests_dir, "delete-revprops.expected.dump"), 'rb').readlines() # Create the initial repos and mirror, and sync 'em. dest_sbox = setup_and_sync(sbox, initial_contents) # Now remove a revprop from r1 of the source, and run 'svnsync # copy-revprops' to re-sync 'em. svntest.actions.enable_revprop_changes(sbox.repo_dir) exit_code, out, err = svntest.main.run_svn(None, 'pdel', '-r', '1', '--revprop', 'issue-id', sbox.repo_url) if err: raise SVNUnexpectedStderr(err) run_copy_revprops(dest_sbox.repo_url, sbox.repo_url) # Does the result look as we expected? verify_mirror(dest_sbox, expected_contents) @Issue(3870) @SkipUnless(svntest.main.is_posix_os) def fd_leak_sync_from_serf_to_local(sbox): "fd leak during sync from serf to local" import resource resource.setrlimit(resource.RLIMIT_NOFILE, (128, 128)) run_test(sbox, "largemods.dump", is_src_ra_local=None, is_dest_ra_local=True) #---------------------------------------------------------------------- @Issue(4476) def mergeinfo_contains_r0(sbox): "mergeinfo contains r0" def make_node_record(node_name, mi): """Return a dumpfile node-record for adding a (directory) node named NODE_NAME with mergeinfo MI. Return it as a list of newline-terminated lines. """ headers_tmpl = """\ Node-path: %s Node-kind: dir Node-action: add Prop-content-length: %d Content-length: %d """ content_tmpl = """\ K 13 svn:mergeinfo V %d %s PROPS-END """ content = content_tmpl % (len(mi), mi) headers = headers_tmpl % (node_name, len(content), len(content)) record = headers + '\n' + content + '\n\n' return record.splitlines(True) # The test case mergeinfo (before, after) syncing, separated here with # spaces instead of newlines test_mi = [ ("", ""), # unchanged ("/a:1", "/a:1"), ("/a:1 /b:1*,2","/a:1 /b:1*,2"), ("/:0:1", "/:0:1"), # unchanged; colon-zero in filename ("/a:0", ""), # dropped entirely ("/a:0*", ""), ("/a:0 /b:0*", ""), ("/a:1 /b:0", "/a:1"), # one kept, one dropped ("/a:0 /b:1", "/b:1"), ("/a:0,1 /b:1", "/a:1 /b:1"), # one kept, one changed ("/a:1 /b:0,1", "/a:1 /b:1"), ("/a:0,1 /b:0*,1 /c:0,2 /d:0-1 /e:0-1,3 /f:0-2 /g:0-3", "/a:1 /b:1 /c:2 /d:1 /e:1,3 /f:1-2 /g:1-3"), # all changed ("/a:0:0-1", "/a:0:1"), # changed; colon-zero in filename ] # Get the constant prefix for each dumpfile dump_file_name = "mergeinfo-contains-r0.dump" svnsync_tests_dir = os.path.join(os.path.dirname(sys.argv[0]), 'svnsync_tests_data') dump_in = open(os.path.join(svnsync_tests_dir, dump_file_name), 'rb').readlines() dump_out = list(dump_in) # duplicate the list # Add dumpfile node records containing the test mergeinfo for n, mi in enumerate(test_mi): node_name = "D" + str(n) mi_in = mi[0].replace(' ', '\n') mi_out = mi[1].replace(' ', '\n') dump_in.extend(make_node_record(node_name, mi_in)) dump_out.extend(make_node_record(node_name, mi_out)) # Run the sync dest_sbox = setup_and_sync(sbox, dump_in, bypass_prop_validation=True) # Compare the dump produced by the mirror repository with expected verify_mirror(dest_sbox, dump_out) ######################################################################## # Run the tests # list all tests here, starting with None: test_list = [ None, copy_and_modify, copy_from_previous_version_and_modify, copy_from_previous_version, modified_in_place, tag_empty_trunk, tag_trunk_with_dir, tag_trunk_with_file, tag_trunk_with_file2, tag_with_modified_file, dir_prop_change, file_dir_file, copy_parent_modify_prop, detect_meddling, url_encoding, no_author, copy_revprops, only_trunk, only_trunk_A_with_changes, move_and_modify_in_the_same_revision, info_synchronized, info_not_synchronized, copy_bad_line_endings, copy_bad_line_endings2, copy_bad_encoding, delete_svn_props, commit_a_copy_of_root, descend_into_replace, delete_revprops, fd_leak_sync_from_serf_to_local, # calls setrlimit mergeinfo_contains_r0, ] if __name__ == '__main__': svntest.main.run_tests(test_list) # NOTREACHED ### End of file.