lock_tests.py

subversion-1.4.3-1.tar.gz 配置svn的源码

    if line.find("Lock Token:") != -1:
      lock_token = line[12:]
      break
  else:
    print "Error: Lock token not found"
    raise svntest.Failure

  if (repos_lock_token != lock_token):
    print "Error: expected repository lock information not found."
    raise svntest.Failure

#----------------------------------------------------------------------
def unlock_already_unlocked_files(sbox):
  "(un)lock set of files, one already (un)locked"

  sbox.build()
  wc_dir = sbox.wc_dir

  # Deliberately have no direct child of A as a target
  iota_path = os.path.join(wc_dir, 'iota')
  lambda_path = os.path.join(wc_dir, 'A', 'B', 'lambda')
  alpha_path = os.path.join(wc_dir, 'A', 'B', 'E', 'alpha')
  gamma_path = os.path.join(wc_dir, 'A', 'D', 'gamma')

  svntest.actions.run_and_verify_svn(None, ".*locked by user", [], 'lock',
                                     '--username', svntest.main.wc_author2,
                                     '--password', svntest.main.wc_passwd,
                                     '--no-auth-cache',
                                     '-m', 'lock several',
                                     iota_path, lambda_path, alpha_path)
  
  expected_status = svntest.actions.get_virginal_state(wc_dir, 1)
  expected_status.tweak('iota', 'A/B/lambda', 'A/B/E/alpha', writelocked='K')
  svntest.actions.run_and_verify_status(wc_dir, expected_status)

  error_msg = ".*Path '/A/B/E/alpha' is already locked by user '" + \
              svntest.main.wc_author2 + "'.*"
  svntest.actions.run_and_verify_svn(None, None, error_msg,
                                     'lock',
                                     '--username', svntest.main.wc_author2,
                                     '--password', svntest.main.wc_passwd,
                                     '--no-auth-cache',
                                     alpha_path, gamma_path)
  expected_status.tweak('A/D/gamma', writelocked='K')
  svntest.actions.run_and_verify_status(wc_dir, expected_status)

  svntest.actions.run_and_verify_svn(None, ".*unlocked", [], 'unlock',
                                     '--username', svntest.main.wc_author2,
                                     '--password', svntest.main.wc_passwd,
                                     '--no-auth-cache',
                                     lambda_path)

  expected_status.tweak('A/B/lambda', writelocked=None)
  svntest.actions.run_and_verify_status(wc_dir, expected_status)

  error_msg = "(.*No lock on path '/A/B/lambda'.*)" + \
              "|(.*'A/B/lambda' is not locked.*)"
  svntest.actions.run_and_verify_svn(None, None, error_msg,
                                     'unlock',
                                     '--username', svntest.main.wc_author2,
                                     '--password', svntest.main.wc_passwd,
                                     '--no-auth-cache',
                                     '--force',
                                     iota_path, lambda_path, alpha_path)


  expected_status.tweak('iota', 'A/B/E/alpha', writelocked=None)
  svntest.actions.run_and_verify_status(wc_dir, expected_status)

#----------------------------------------------------------------------
def info_moved_path(sbox):
  "show correct lock info on moved path"

  sbox.build()
  wc_dir = sbox.wc_dir
  fname = os.path.join(wc_dir, "iota")
  fname2 = os.path.join(wc_dir, "iota2")

  # Move iota, creating r2.
  svntest.actions.run_and_verify_svn(None, None, [],
                                     "mv", fname, fname2)
  expected_output = svntest.wc.State(wc_dir, {
    'iota2' : Item(verb='Adding'),
    'iota' : Item(verb='Deleting'),
    })
  expected_status = svntest.actions.get_virginal_state(wc_dir, 1)
  expected_status.add({
    "iota2" : Item(status='  ', wc_rev=2)
    })
  expected_status.remove("iota")
  svntest.actions.run_and_verify_commit (wc_dir,
                                         expected_output,
                                         expected_status,
                                         None,
                                         None, None,
                                         None, None,
                                         wc_dir)

  # Create a new, unrelated iota, creating r3.
  svntest.main.file_append(fname, "Another iota")
  svntest.actions.run_and_verify_svn(None, None, [],
                                     "add", fname)
  expected_output = svntest.wc.State(wc_dir, {
    'iota' : Item(verb='Adding'),
    })
  expected_status.add({
    "iota" : Item(status='  ', wc_rev=3)
    })
  svntest.actions.run_and_verify_commit (wc_dir,
                                         expected_output,
                                         expected_status,
                                         None,
                                         None, None,
                                         None, None,
                                         wc_dir)

  # Lock the new iota.
  svntest.actions.run_and_verify_svn(None, ".*locked by user", [],
                                     "lock", fname,
                                     '--username', svntest.main.wc_author,
                                     '--password', svntest.main.wc_passwd)
  expected_status.tweak("iota", writelocked="K")
  svntest.actions.run_and_verify_status(wc_dir, expected_status)

  # Get info for old iota at r1. This shouldn't give us any lock info.
  output, errput = svntest.actions.run_and_verify_svn(None, None, [],
                                                      'info',
                                                      fname2, '-r1')
  # Since we want to make sure that there is *no* lock info, to make this
  # more robust, we also check that the info command actually output some info.
  got_url = 0
  for line in output:
    if line.find("URL:") >= 0:
      got_url = 1
    if line.find("Lock Token:") >= 0:
      print fname2 + " was reported as locked."
      raise svntest.Failure
  if not got_url:
    print "Info didn't output an URL."
    raise svntest.Failure

#----------------------------------------------------------------------
def ls_url_encoded(sbox):
  "ls locked path needing URL encoding"

  sbox.build()
  wc_dir = sbox.wc_dir
  dirname = os.path.join(wc_dir, "space dir")
  fname = os.path.join(dirname, "f")

  # Create a dir with a space in its name and a file therein.
  svntest.actions.run_and_verify_svn(None, None, [],
                                     "mkdir", dirname)
  svntest.main.file_append(fname, "someone was here")
  svntest.actions.run_and_verify_svn(None, None, [],
                                     "add", fname)
  expected_output = svntest.wc.State(wc_dir, {
    'space dir' : Item(verb='Adding'),
    'space dir/f' : Item(verb='Adding'),
    })
  expected_status = svntest.actions.get_virginal_state(wc_dir, 1)
  expected_status.add({
    "space dir" : Item(status='  ', wc_rev=2),
    "space dir/f" : Item(status='  ', wc_rev=2),
    })
  svntest.actions.run_and_verify_commit(wc_dir,
                                        expected_output,
                                        expected_status,
                                        None,
                                        None, None,
                                        None, None,
                                        wc_dir)

  # Lock the file.
  svntest.actions.run_and_verify_svn("Lock space dir/f", ".*locked by user",
                                     [], "lock", fname,
                                     "--username", svntest.main.wc_author,
                                     "--password", svntest.main.wc_passwd)

  # Make sure ls shows it being locked.
  expected_output = " +2 " + re.escape(svntest.main.wc_author) + " +O .+f"
  svntest.actions.run_and_verify_svn("List space dir",
                                     expected_output, [],
                                     "list", "-v", dirname)  

#----------------------------------------------------------------------
# Make sure unlocking a path with the wrong lock token fails.
def unlock_wrong_token(sbox):
  "veriy unlocking with wrong lock token"

  sbox.build()
  wc_dir = sbox.wc_dir

  # lock a file as wc_author
  fname = 'iota'
  file_path = os.path.join(sbox.wc_dir, fname)
  file_url = svntest.main.current_repo_url + "/iota"

  svntest.actions.run_and_verify_svn(None, ".*locked by user", [], 'lock',
                                     '--username', svntest.main.wc_author,
                                     '--password', svntest.main.wc_passwd,
                                     file_path)

  # Steal the lock as the same author, but using an URL to keep the old token
  # in the WC.
  svntest.actions.run_and_verify_svn(None, ".*locked by user", [], 'lock',
                                     '--username', svntest.main.wc_author,
                                     '--password', svntest.main.wc_passwd,
                                     "--force", file_url)

  # Then, unlocking the WC path should fail.
  # ### The error message returned is actually this, but let's worry about that
  # ### another day...
  svntest.actions.run_and_verify_svn(None, None,
                                     ".*((No lock on path)|(400 Bad Request))",
                                     'unlock',
                                     '--username', svntest.main.wc_author,
                                     '--password', svntest.main.wc_passwd,
                                     file_path)

#----------------------------------------------------------------------
# Verify that info shows lock info for locked files with URI-unsafe names
# when run in recursive mode.
def examine_lock_encoded_recurse(sbox):
  "verify recursive info shows lock info"

  sbox.build()
  wc_dir = sbox.wc_dir

  fname = 'A/B/F/one iota'
  comment = 'This is a lock test.'
  file_path = os.path.join(sbox.wc_dir, fname)

  svntest.main.file_append(file_path, "This represents a binary file\n")
  svntest.actions.run_and_verify_svn(None, None, [], "add", file_path)

  expected_output = svntest.wc.State(wc_dir, {
    fname : Item(verb='Adding'),
    })

  expected_status = svntest.actions.get_virginal_state(wc_dir, 1)
  expected_status.add({ fname: Item(wc_rev=2, status='  ') })

  # Commit the file.
  svntest.actions.run_and_verify_commit(wc_dir,
                                        expected_output,
                                        expected_status,
                                        None,
                                        None, None,
                                        None, None,
                                        file_path)

  # lock the file
  svntest.actions.run_and_verify_svn(None, ".*locked by user", [], 'lock',
                                     '--username', svntest.main.wc_author,
                                     '--password', svntest.main.wc_passwd,
                                     '-m', comment, file_path)

  # Run info and check that we get the lock fields.
  output, err = svntest.actions.run_and_verify_svn(None, None, [],
                                                   'info', '-R', 
						   svntest.main.current_repo_url + '/A/B/F')

  lock_info = output[-6:-1]
  if ((len(lock_info) != 5)
      or (lock_info[0][0:28] != 'Lock Token: opaquelocktoken:')
      or (lock_info[1] != 'Lock Owner: ' + svntest.main.wc_author + '\n')
      or (lock_info[2][0:13] != 'Lock Created:')
      or (lock_info[4] != comment + '\n')):
    raise svntest.Failure



########################################################################
# Run the tests

# list all tests here, starting with None:
test_list = [ None,
              lock_file,
              commit_file_keep_lock,
              commit_file_unlock,
              commit_propchange,
              break_lock,
              steal_lock,
              examine_lock,
              handle_defunct_lock,
              enforce_lock,
              defunct_lock,
              deleted_path_lock,
              lock_unlock,
              deleted_dir_lock,
              lock_status,
              stolen_lock_status,
              broken_lock_status,
              lock_non_existent_file,
              out_of_date,
              update_while_needing_lock,
              revert_lock,
              examine_lock_via_url,
              lock_several_files,
              lock_switched_files,
              lock_uri_encoded,
              Skip(lock_and_exebit1, (os.name != 'posix')),
              Skip(lock_and_exebit2, (os.name != 'posix')),
              commit_xml_unsafe_file_unlock,
              repos_lock_with_info,
              unlock_already_unlocked_files,
              info_moved_path,
              ls_url_encoded,
              unlock_wrong_token,
              examine_lock_encoded_recurse,
            ]

if __name__ == '__main__':
  svntest.main.run_tests(test_list)
  # NOTREACHED


### End of file.