#!/usr/bin/env python # # svnsync_tests.py: Tests SVNSync's repository mirroring capabilities. # # Subversion is a tool for revision control. # See http://subversion.tigris.org for more information. # # ==================================================================== # Copyright (c) 2005-2007 CollabNet. All rights reserved. # # This software is licensed as described in the file COPYING, which # you should have received as part of this distribution. The terms # are also available at http://subversion.tigris.org/license-1.html. # If newer versions of this license are posted there, you may use a # newer version instead, at your option. # ###################################################################### # General modules import sys, os # Our testing module import svntest from svntest.verify import SVNUnexpectedStdout, SVNUnexpectedStderr from svntest.verify import SVNExpectedStderr from svntest.main import write_restrictive_svnserve_conf from svntest.main import server_has_partial_replay # (abbreviation) Skip = svntest.testcase.Skip SkipUnless = svntest.testcase.SkipUnless XFail = svntest.testcase.XFail Item = svntest.wc.StateItem ###################################################################### # Helper routines def build_repos(sbox): """Avoid the use sbox.build() because we're working with a repos other than the Greek tree.""" # Cleanup after the last run by removing any left-over repository. svntest.main.safe_rmtree(sbox.repo_dir) # Create an empty repository. svntest.main.create_repos(sbox.repo_dir) def run_sync(url, expected_error=None): "Synchronize the mirror repository with the master" exit_code, output, errput = svntest.main.run_svnsync( "synchronize", url, "--username", svntest.main.wc_author, "--password", svntest.main.wc_passwd) if errput: if expected_error is None: raise SVNUnexpectedStderr(errput) else: expected_error = svntest.verify.RegexOutput(expected_error, match_all=False) svntest.verify.compare_and_display_lines(None, "STDERR", expected_error, errput) elif expected_error is not None: raise SVNExpectedStderr if not output and not expected_error: # should be: ['Committed revision 1.\n', 'Committed revision 2.\n'] raise SVNUnexpectedStdout("Missing stdout") def run_copy_revprops(url, expected_error=None): "Copy revprops to the mirror repository from the master" exit_code, output, errput = svntest.main.run_svnsync( "copy-revprops", url, "--username", svntest.main.wc_author, "--password", svntest.main.wc_passwd) if errput: if expected_error is None: raise SVNUnexpectedStderr(errput) else: expected_error = svntest.verify.RegexOutput(expected_error, match_all=False) svntest.verify.compare_and_display_lines(None, "STDERR", expected_error, errput) elif expected_error is not None: raise SVNExpectedStderr if not output and not expected_error: # should be: ['Copied properties for revision 1.\n', # 'Copied properties for revision 2.\n'] raise SVNUnexpectedStdout("Missing stdout") def run_init(dst_url, src_url): "Initialize the mirror repository from the master" exit_code, output, errput = svntest.main.run_svnsync( "initialize", dst_url, src_url, "--username", svntest.main.wc_author, "--password", svntest.main.wc_passwd) if errput: raise SVNUnexpectedStderr(errput) if output != ['Copied properties for revision 0.\n']: raise SVNUnexpectedStdout(output) def run_info(url, expected_error=None): "Print synchronization information of the repository" exit_code, output, errput = svntest.main.run_svnsync( "info", url, "--username", svntest.main.wc_author, "--password", svntest.main.wc_passwd) if errput: if expected_error is None: raise SVNUnexpectedStderr(errput) else: expected_error = svntest.verify.RegexOutput(expected_error, match_all=False) svntest.verify.compare_and_display_lines(None, "STDERR", expected_error, errput) elif expected_error is not None: raise SVNExpectedStderr if not output and not expected_error: # should be: ['From URL: http://....\n', # 'From UUID: XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX\n', # 'Last Merged Revision: XXX\n'] raise SVNUnexpectedStdout("Missing stdout") def run_test(sbox, dump_file_name, subdir = None, exp_dump_file_name = None): """Load a dump file, sync repositories, and compare contents with the original or another dump file.""" # Create the empty master repository. build_repos(sbox) # 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 = file(os.path.join(svnsync_tests_dir, dump_file_name), 'rb').readlines() svntest.actions.run_and_verify_load(sbox.repo_dir, master_dumpfile_contents) # Create the empty destination repository. dest_sbox = sbox.clone_dependent() build_repos(dest_sbox) # 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("Setting UUID", 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 if subdir: repo_url = repo_url + subdir run_init(dest_sbox.repo_url, repo_url) run_sync(dest_sbox.repo_url) run_copy_revprops(dest_sbox.repo_url) # 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, 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) # 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_master_dumpfile_contents = file(os.path.join(svnsync_tests_dir, exp_dump_file_name)).readlines() else: exp_master_dumpfile_contents = master_dumpfile_contents svntest.verify.compare_and_display_lines( "Dump files", "DUMP", exp_master_dumpfile_contents, dest_dump) ###################################################################### # 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() build_repos(dest_sbox) # 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, 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, 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, None, [], 'ci', '-m', 'msg', dest_sbox.wc_dir) run_sync(dest_sbox.repo_url, ".*Destination HEAD \\(2\\) is not the last merged revision \\(1\\).*") #---------------------------------------------------------------------- def basic_authz(sbox): "verify that unreadable content is not synced" sbox.build("svnsync-basic-authz") write_restrictive_svnserve_conf(sbox.repo_dir) dest_sbox = sbox.clone_dependent() build_repos(dest_sbox) svntest.actions.enable_revprop_changes(dest_sbox.repo_dir) run_init(dest_sbox.repo_url, sbox.repo_url) svntest.main.file_write(sbox.authz_file, "[svnsync-basic-authz:/]\n" "* = r\n" "\n" "[svnsync-basic-authz:/A/B]\n" "* = \n" "\n" "[svnsync-basic-authz-1:/]\n" "* = rw\n") run_sync(dest_sbox.repo_url) lambda_url = dest_sbox.repo_url + '/A/B/lambda' # this file should have been blocked by authz svntest.actions.run_and_verify_svn(None, [], svntest.verify.AnyOutput, 'cat', lambda_url) #---------------------------------------------------------------------- def copy_from_unreadable_dir(sbox): "verify that copies from unreadable dirs work" sbox.build("svnsync-copy-from-unreadable-dir") B_url = sbox.repo_url + '/A/B' P_url = sbox.repo_url + '/A/P' # Set a property on the directory we're going to copy, and a file in it, to # confirm that they're transmitted when we later sync the copied directory svntest.actions.run_and_verify_svn(None, None, [], 'pset', 'foo', 'bar', sbox.wc_dir + '/A/B/lambda') svntest.actions.run_and_verify_svn(None, None, [], 'pset', 'baz', 'zot', sbox.wc_dir + '/A/B') svntest.actions.run_and_verify_svn(None, None, [], 'ci', sbox.wc_dir + '/A/B', '-m', 'log_msg') # Now copy that directory so we'll see it in our synced copy svntest.actions.run_and_verify_svn(None, None, [], 'cp', B_url, P_url, '-m', 'Copy B to P') write_restrictive_svnserve_conf(sbox.repo_dir) dest_sbox = sbox.clone_dependent() build_repos(dest_sbox) svntest.actions.enable_revprop_changes(dest_sbox.repo_dir) fp = open(sbox.authz_file, 'w') # For mod_dav_svn's parent path setup we need per-repos permissions in # the authz file... if sbox.repo_url.startswith('http'): fp.write("[svnsync-copy-from-unreadable-dir:/]\n" + "* = r\n" + "\n" + "[svnsync-copy-from-unreadable-dir:/A/B]\n" + "* = \n" + "\n" + "[svnsync-copy-from-unreadable-dir-1:/]\n" + "* = rw") # Otherwise we can just go with the permissions needed for the source # repository. else: fp.write("[/]\n" + "* = r\n" + "\n" + "[/A/B]\n" + "* =\n") fp.close() run_init(dest_sbox.repo_url, sbox.repo_url) run_sync(dest_sbox.repo_url) expected_out = [ 'Changed paths:\n', ' A /A/P\n', ' A /A/P/E\n', ' A /A/P/E/alpha\n', ' A /A/P/E/beta\n', ' A /A/P/F\n', ' A /A/P/lambda\n', '\n', '\n', # log message is stripped ] exit_code, out, err = svntest.main.run_svn(None, 'log', '-r', '3', '-v', dest_sbox.repo_url) if err: raise SVNUnexpectedStderr(err) svntest.verify.compare_and_display_lines(None, 'LOG', expected_out, out[2:11]) svntest.actions.run_and_verify_svn(None, ['bar\n'], [], 'pget', 'foo', dest_sbox.repo_url + '/A/P/lambda') svntest.actions.run_and_verify_svn(None, ['zot\n'], [], 'pget', 'baz', dest_sbox.repo_url + '/A/P') # Issue 2705. def copy_with_mod_from_unreadable_dir(sbox): "verify copies with mods from unreadable dirs" sbox.build("svnsync-copy-with-mod-from-unreadable-dir") # Make a copy of the B directory. svntest.actions.run_and_verify_svn(None, None, [], 'cp', sbox.wc_dir + '/A/B', sbox.wc_dir + '/A/P') # Set a property inside the copied directory. svntest.actions.run_and_verify_svn(None, None, [], 'pset', 'foo', 'bar', sbox.wc_dir + '/A/P/lambda') # Add a new directory and file inside the copied directory. svntest.actions.run_and_verify_svn(None, None, [], 'mkdir', sbox.wc_dir + '/A/P/NEW-DIR') svntest.main.file_append(sbox.wc_dir + '/A/P/E/new-file', "bla bla") svntest.main.run_svn(None, 'add', sbox.wc_dir + '/A/P/E/new-file') # Delete a file inside the copied directory. svntest.actions.run_and_verify_svn(None, None, [], 'rm', sbox.wc_dir + '/A/P/E/beta') # Commit the copy-with-modification. svntest.actions.run_and_verify_svn(None, None, [], 'ci', sbox.wc_dir, '-m', 'log_msg') # Lock down the source repository. write_restrictive_svnserve_conf(sbox.repo_dir) dest_sbox = sbox.clone_dependent() build_repos(dest_sbox) svntest.actions.enable_revprop_changes(dest_sbox.repo_dir) fp = open(sbox.authz_file, 'w') # For mod_dav_svn's parent path setup we need per-repos permissions in # the authz file... if sbox.repo_url.startswith('http'): fp.write("[svnsync-copy-with-mod-from-unreadable-dir:/]\n" + "* = r\n" + "\n" + "[svnsync-copy-with-mod-from-unreadable-dir:/A/B]\n" + "* = \n" + "\n" + "[svnsync-copy-with-mod-from-unreadable-dir-1:/]\n" + "* = rw") # Otherwise we can just go with the permissions needed for the source # repository. else: fp.write("[/]\n" + "* = r\n" + "\n" + "[/A/B]\n" + "* =\n") fp.close() run_init(dest_sbox.repo_url, sbox.repo_url) run_sync(dest_sbox.repo_url) expected_out = [ 'Changed paths:\n', ' A /A/P\n', ' A /A/P/E\n', ' A /A/P/E/alpha\n', ' A /A/P/E/new-file\n', ' A /A/P/F\n', ' A /A/P/NEW-DIR\n', ' A /A/P/lambda\n', '\n', '\n', # log message is stripped ] exit_code, out, err = svntest.main.run_svn(None, 'log', '-r', '2', '-v', dest_sbox.repo_url) if err: raise SVNUnexpectedStderr(err) svntest.verify.compare_and_display_lines(None, 'LOG', expected_out, out[2:12]) svntest.actions.run_and_verify_svn(None, ['bar\n'], [], 'pget', 'foo', dest_sbox.repo_url + '/A/P/lambda') # Issue 2705. def copy_with_mod_from_unreadable_dir_and_copy(sbox): "verify copies with mods from unreadable dirs +copy" sbox.build("svnsync-copy-with-mod-from-unreadable-dir-and-copy") # Make a copy of the B directory. svntest.actions.run_and_verify_svn(None, None, [], 'cp', sbox.wc_dir + '/A/B', sbox.wc_dir + '/A/P') # Copy a (readable) file into the copied directory. svntest.actions.run_and_verify_svn(None, None, [], 'cp', sbox.wc_dir + '/A/D/gamma', sbox.wc_dir + '/A/P/E') # Commit the copy-with-modification. svntest.actions.run_and_verify_svn(None, None, [], 'ci', sbox.wc_dir, '-m', 'log_msg') # Lock down the source repository. write_restrictive_svnserve_conf(sbox.repo_dir) dest_sbox = sbox.clone_dependent() build_repos(dest_sbox) svntest.actions.enable_revprop_changes(dest_sbox.repo_dir) fp = open(sbox.authz_file, 'w') # For mod_dav_svn's parent path setup we need per-repos permissions in # the authz file... if sbox.repo_url.startswith('http'): fp.write("[svnsync-copy-with-mod-from-unreadable-dir-and-copy:/]\n" + "* = r\n" + "\n" + "[svnsync-copy-with-mod-from-unreadable-dir-and-copy:/A/B]\n" + "* = \n" + "\n" + "[svnsync-copy-with-mod-from-unreadable-dir-and-copy-1:/]\n" + "* = rw") # Otherwise we can just go with the permissions needed for the source # repository. else: fp.write("[/]\n" + "* = r\n" + "\n" + "[/A/B]\n" + "* =\n") fp.close() run_init(dest_sbox.repo_url, sbox.repo_url) run_sync(dest_sbox.repo_url) expected_out = [ 'Changed paths:\n', ' A /A/P\n', ' A /A/P/E\n', ' A /A/P/E/alpha\n', ' A /A/P/E/beta\n', ' A /A/P/E/gamma (from /A/D/gamma:1)\n', ' A /A/P/F\n', ' A /A/P/lambda\n', '\n', '\n', # log message is stripped ] exit_code, out, err = svntest.main.run_svn(None, 'log', '-r', '2', '-v', dest_sbox.repo_url) if err: raise SVNUnexpectedStderr(err) svntest.verify.compare_and_display_lines(None, 'LOG', expected_out, out[2:12]) 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") def only_trunk(sbox): "test syncing subdirectories" run_test(sbox, "svnsync-trunk-only.dump", "/trunk", "svnsync-trunk-only.expected.dump") 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 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() build_repos(dest_sbox) svntest.actions.enable_revprop_changes(dest_sbox.repo_dir) run_init(dest_sbox.repo_url, sbox.repo_url) run_sync(dest_sbox.repo_url) exit_code, output, errput = svntest.main.run_svnsync( "info", dest_sbox.repo_url, "--username", svntest.main.wc_author, "--password", svntest.main.wc_passwd) if errput: raise SVNUnexpectedStderr(errput) expected_out = ['Source URL: %s\n' % sbox.repo_url, 'Source Repository UUID: %s\n' % src_uuid, 'Last Merged Revision: 1\n', ] svntest.verify.compare_and_display_lines(None, 'INFO', expected_out, output) 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 lineendings in svn:props" run_test(sbox, "copy-bad-line-endings.dump", exp_dump_file_name="copy-bad-line-endings.expected.dump") #---------------------------------------------------------------------- def delete_svn_props(sbox): "copy with svn:prop deletions" run_test(sbox, "delete-svn-props.dump") 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") ######################################################################## # 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_file2, tag_trunk_with_file, tag_with_modified_file, dir_prop_change, file_dir_file, copy_parent_modify_prop, detect_meddling, Skip(basic_authz, svntest.main.is_ra_type_file), Skip(copy_from_unreadable_dir, svntest.main.is_ra_type_file), Skip(copy_with_mod_from_unreadable_dir, svntest.main.is_ra_type_file), Skip(copy_with_mod_from_unreadable_dir_and_copy, svntest.main.is_ra_type_file), url_encoding, no_author, copy_revprops, SkipUnless(only_trunk, server_has_partial_replay), SkipUnless(only_trunk_A_with_changes, server_has_partial_replay), move_and_modify_in_the_same_revision, info_synchronized, info_not_synchronized, copy_bad_line_endings, delete_svn_props, commit_a_copy_of_root, ] if __name__ == '__main__': svntest.main.run_tests(test_list, serial_only = True) # NOTREACHED ### End of file.