def test_log3_url(self):
        """Test svn_client_log3 on a file:// URL"""
        directory = urljoin(self.repos_uri + "/", "trunk/dir1")

        start = core.svn_opt_revision_t()
        end = core.svn_opt_revision_t()
        core.svn_opt_parse_revision(start, end, "4:0")
        client.log3((directory,), start, start, end, 1, True, False, self.log_receiver, self.client_ctx)
        self.assertEqual(self.change_author, "john")
        self.assertEqual(self.log_message, "More directories.")
        self.assertEqual(len(self.changed_paths), 3)
        for dir in ("/trunk/dir1", "/trunk/dir2", "/trunk/dir3"):
            self.assert_(dir in self.changed_paths)
            self.assertEqual(self.changed_paths[dir].action, "A")

  def test_log3_url(self):
    """Test svn_client_log3 on a file:// URL"""
    dir = urljoin(REPOS_URL+"/", "trunk/dir1")

    start = core.svn_opt_revision_t()
    end = core.svn_opt_revision_t()
    core.svn_opt_parse_revision(start, end, "4:0")
    client.log3((dir,), start, start, end, 1, True, False, self.log_receiver,
        self.client_ctx)
    self.assertEqual(self.change_author, "john")
    self.assertEqual(self.log_message, "More directories.")
    self.assertEqual(len(self.changed_paths), 3)
    for dir in ('/trunk/dir1', '/trunk/dir2', '/trunk/dir3'):
      self.assert_(self.changed_paths.has_key(dir))
      self.assertEqual(self.changed_paths[dir].action, 'A')

def _svn_rev(number):
    value = core.svn_opt_revision_value_t()
    value.number = number
    revision = core.svn_opt_revision_t()
    revision.kind = core.svn_opt_revision_number
    revision.value = value
    return revision

  def test_inherited_props(self):
    """Test inherited props"""

    trunk_url = self.repos_uri + '/trunk'
    client.propset_remote('svn:global-ignores', '*.q', trunk_url,
                          False, 12, {}, None, self.client_ctx)

    head = core.svn_opt_revision_t()
    head.kind = core.svn_opt_revision_head
    props, iprops, rev = client.propget5('svn:global-ignores', trunk_url,
                                         head, head, core.svn_depth_infinity,
                                         None, self.client_ctx)
    self.assertEquals(props[trunk_url], '*.q\n')

    dir1_url = trunk_url + '/dir1'
    props, iprops, rev = client.propget5('svn:global-ignores', dir1_url,
                                         head, head, core.svn_depth_infinity,
                                         None, self.client_ctx)
    self.assertEquals(iprops[trunk_url], {'svn:global-ignores':'*.q\n'})

    self.proplist_receiver_trunk_calls = 0
    client.proplist4(trunk_url, head, head, core.svn_depth_empty, None, True,
                     self.proplist_receiver_trunk, self.client_ctx)
    self.assertEquals(self.proplist_receiver_trunk_calls, 1)

    self.proplist_receiver_dir1_calls = 0
    self.proplist_receiver_dir1_key = trunk_url
    client.proplist4(dir1_url, head, head, core.svn_depth_empty, None, True,
                     self.proplist_receiver_dir1, self.client_ctx)
    self.assertEquals(self.proplist_receiver_dir1_calls, 1)

    def __init__(self, session_id=None):
        
        # Setup parameters expected by the base class 
        if session_id is None:
            session_id = ADMIN_SESSION_ID
        self.mParentSession = getSession(session_id)
        self.mChangeset = self.mParentSession.changeset

        # Call base constructor
        ccs_revision.__init__(self, self.mParentSession, self.mChangeset, \
                False)
        
        # Setup a working directory for this revision
        self.rDir = mkdtemp("", "ccsd")

        # Checkout the current configuration HEAD to this directory
        try:
            rev = core.svn_opt_revision_t()
            rev.kind = core.svn_opt_revision_head
            client.svn_client_checkout("%s/ca" % self.svnroot, self.rDir, \
                    rev, True, self.ctx, self.pool)
            self.mCurRev = rev
        except core.SubversionException:
            # CA not initialised
            raise ccs_ca_error("infrastructure not found in repository!")
        
        # Check basic repository structure
        if self.mParentSession is not None and self.mChangeset is not None:
            self.checkRepoStructure()
        
        # Start with no errors
        self.mErrors = {}

    def test_merge_peg3(self):
        """Test svn_client_merge_peg3."""
        head = core.svn_opt_revision_t()
        head.kind = core.svn_opt_revision_head
        wc_path = self.temper.alloc_empty_dir("-merge_peg3")

        client.checkout3(self.repos_uri, wc_path, head, head, core.svn_depth_infinity, True, False, self.client_ctx)

        # Let's try to backport a change from the v1x branch
        trunk_path = core.svn_dirent_join(wc_path, "trunk")
        v1x_path = core.svn_dirent_join(wc_path, "branches/v1x")

        start = core.svn_opt_revision_t()
        start.kind = core.svn_opt_revision_number
        start.value.number = 8

        end = core.svn_opt_revision_t()
        end.kind = core.svn_opt_revision_number
        end.value.number = 9

        rrange = core.svn_opt_revision_range_t()
        rrange.start = start
        rrange.end = end

        client.merge_peg3(
            v1x_path,
            (rrange,),
            end,
            trunk_path,
            core.svn_depth_infinity,
            False,
            False,
            False,
            False,
            None,
            self.client_ctx,
        )

        # Did it take effect?
        readme_path_native = core.svn_dirent_local_style(core.svn_dirent_join(trunk_path, "README.txt"))

        readme = open(readme_path_native, "r")
        readme_text = readme.read()
        readme.close()

        self.assertEqual(readme_text, "This is a test.\n")

  def test_log5(self):
    """Test svn_client_log5."""
    start = core.svn_opt_revision_t()
    start.kind = core.svn_opt_revision_number
    start.value.number = 0

    end = core.svn_opt_revision_t()
    end.kind = core.svn_opt_revision_number
    end.value.number = 4

    rev_range = core.svn_opt_revision_range_t()
    rev_range.start = start
    rev_range.end = end

    self.received_revisions = []

    client.log5((self.repos_uri,), end, (rev_range,), 0, False, True, False, (),
        self.log_entry_receiver, self.client_ctx)

    self.assertEqual(self.received_revisions, range(0, 5))

    def test_checkout(self):
        """Test svn_client_checkout2."""

        rev = core.svn_opt_revision_t()
        rev.kind = core.svn_opt_revision_head

        path = self.temper.alloc_empty_dir("-checkout")

        self.assertRaises(ValueError, client.checkout2, self.repos_uri, path, None, None, True, True, self.client_ctx)

        client.checkout2(self.repos_uri, path, rev, rev, True, True, self.client_ctx)

  def test_info_file(self):
    """Test svn_client_info on working copy file and remote files."""

    # This test requires a file /trunk/README.txt of size 8 bytes
    # in the repository.
    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_head
    wc_path = core.svn_path_canonicalize(tempfile.mktemp())

    client.checkout2(REPOS_URL, wc_path, rev, rev, True, True,
                     self.client_ctx)
    adm_access = wc.adm_open3(None, wc_path, True, -1, None)

    try:
      # Test 1: Run info -r BASE. We expect the size value to be filled in.
      rev.kind = core.svn_opt_revision_base
      readme_path = '%s/trunk/README.txt' % wc_path
      readme_url = '%s/trunk/README.txt' % REPOS_URL
      client.info(readme_path, rev, rev, self.info_receiver,
                  False, self.client_ctx)

      self.assertEqual(self.path, os.path.basename(readme_path))
      self.info.assert_valid()
      self.assertEqual(self.info.working_size, client.SWIG_SVN_INFO_SIZE_UNKNOWN)
      self.assertEqual(self.info.size, 8)

      # Test 2: Run info (revision unspecified). We expect the working_size value
      # to be filled in.
      rev.kind = core.svn_opt_revision_unspecified
      client.info(readme_path, rev, rev, self.info_receiver,
                  False, self.client_ctx)

      self.assertEqual(self.path, readme_path)
      self.info.assert_valid()
      self.assertEqual(self.info.size, client.SWIG_SVN_INFO_SIZE_UNKNOWN)
      # README.txt contains one EOL char, so on Windows it will be expanded from
      # LF to CRLF hence the working_size will be 9 instead of 8.
      if os.name == 'nt':
        self.assertEqual(self.info.working_size, 9)
      else:
        self.assertEqual(self.info.working_size, 8)

      # Test 3: Run info on the repository URL of README.txt. We expect the size
      # value to be filled in.
      rev.kind = core.svn_opt_revision_head
      client.info(readme_url, rev, rev, self.info_receiver,
                  False, self.client_ctx)
      self.info.assert_valid()
      self.assertEqual(self.info.working_size, client.SWIG_SVN_INFO_SIZE_UNKNOWN)
      self.assertEqual(self.info.size, 8)
    finally:
      wc.adm_close(adm_access)
      core.svn_io_remove_dir(wc_path)

    def test_url_from_path(self):
        """Test svn_client_url_from_path for a file:// URL"""
        self.assertEquals(client.url_from_path(self.repos_uri), self.repos_uri)

        rev = core.svn_opt_revision_t()
        rev.kind = core.svn_opt_revision_head

        path = self.temper.alloc_empty_dir("-url_from_path")

        client.checkout2(self.repos_uri, path, rev, rev, True, True, self.client_ctx)

        self.assertEquals(client.url_from_path(path), self.repos_uri)

  def test_url_from_path(self):
    """Test svn_client_url_from_path for a file:// URL"""
    self.assertEquals(client.url_from_path(REPOS_URL), REPOS_URL)

    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_head

    path = tempfile.mktemp('-url_from_path')

    client.checkout2(REPOS_URL, path, rev, rev, True, True,
                     self.client_ctx)

    self.assertEquals(client.url_from_path(path), REPOS_URL)

    def test_update4(self):
        """Test update and the notify function callbacks"""

        rev = core.svn_opt_revision_t()
        rev.kind = core.svn_opt_revision_number
        rev.value.number = 0

        path = self.temper.alloc_empty_dir("-update")

        self.assertRaises(ValueError, client.checkout2, self.repos_uri, path, None, None, True, True, self.client_ctx)

        client.checkout2(self.repos_uri, path, rev, rev, True, True, self.client_ctx)

        def notify_func(path, action, kind, mime_type, content_state, prop_state, rev):
            self.notified_paths.append(path)

        self.client_ctx.notify_func = client.svn_swig_py_notify_func
        self.client_ctx.notify_baton = notify_func
        rev.value.number = 1
        self.notified_paths = []
        client.update4((path,), rev, core.svn_depth_unknown, True, False, False, False, False, self.client_ctx)
        expected_paths = [
            path,
            os.path.join(path, "branches"),
            os.path.join(path, "tags"),
            os.path.join(path, "trunk"),
            path,
            path,
        ]
        # All normal subversion apis process paths in Subversion's canonical format,
        # which isn't the platform specific format
        expected_paths = [x.replace(os.path.sep, "/") for x in expected_paths]
        self.notified_paths.sort()
        expected_paths.sort()

        self.assertEquals(self.notified_paths, expected_paths)

        def notify_func2(notify, pool):
            self.notified_paths.append(notify.path)

        self.client_ctx.notify_func2 = client.svn_swig_py_notify_func2
        self.client_ctx.notify_baton2 = notify_func2
        rev.value.number = 2
        self.notified_paths = []
        expected_paths = [path, os.path.join(path, "trunk", "README.txt"), os.path.join(path, "trunk"), path, path]
        expected_paths = [x.replace(os.path.sep, "/") for x in expected_paths]
        client.update4((path,), rev, core.svn_depth_unknown, True, False, False, False, False, self.client_ctx)
        self.notified_paths.sort()
        expected_paths.sort()
        self.assertEquals(self.notified_paths, expected_paths)

    def test_info(self):
        """Test svn_client_info on an empty repository"""

        # Run info
        revt = core.svn_opt_revision_t()
        revt.kind = core.svn_opt_revision_head
        client.info(self.repos_uri, revt, revt, self.info_receiver, False, self.client_ctx)

        # Check output from running info. This also serves to verify that
        # the internal 'info' object is still valid
        self.assertEqual(self.path, os.path.basename(self.repos_path))
        self.info.assert_valid()
        self.assertEqual(self.info.URL, self.repos_uri)
        self.assertEqual(self.info.repos_root_URL, self.repos_uri)

  def test_checkout(self):
    """Test svn_client_checkout2."""

    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_head

    path = tempfile.mktemp('-checkout')

    self.assertRaises(ValueError, client.checkout2,
                      REPOS_URL, path, None, None, True, True,
                      self.client_ctx)

    client.checkout2(REPOS_URL, path, rev, rev, True, True,
            self.client_ctx)

    def test_uuid_from_path(self):
        """Test svn_client_uuid_from_path."""
        rev = core.svn_opt_revision_t()
        rev.kind = core.svn_opt_revision_head

        path = self.temper.alloc_empty_dir("-uuid_from_path")

        client.checkout2(self.repos_uri, path, rev, rev, True, True, self.client_ctx)

        wc_adm = wc.adm_open3(None, path, False, 0, None)

        self.assertEquals(
            client.uuid_from_path(path, wc_adm, self.client_ctx), client.uuid_from_url(self.repos_uri, self.client_ctx)
        )

        self.assert_(isinstance(client.uuid_from_path(path, wc_adm, self.client_ctx), basestring))

def main():
  try:
    opts, args = my_getopt(sys.argv[1:], "h?f", ["help", "fix"])
  except Exception as e:
    sys.stderr.write(""" Improperly used """)
    sys.exit(1)

  if len(args) == 1:
   wcpath = args[0]
   wcpath = os.path.abspath(wcpath)
  else:
    usage()
    sys.exit(1)

  fix = 0
  current_path = os.getcwd()
  hash_file = os.path.join(current_path, ".hashfile")
  newmergeinfo_file = os.path.join(current_path, ".newmergeinfo")

  temp_pool = core.svn_pool_create()
  ctx = client.svn_client_create_context(temp_pool)
  depth = core.svn_depth_infinity
  revision = core.svn_opt_revision_t()
  revision.kind = core.svn_opt_revision_unspecified

  for opt, values in opts:
    if opt == "--help" or opt in ("-h", "-?"):
      usage()
    elif opt == "--fix" or opt == "-f":
      fix = 1

  # Check for any local modifications in the working copy
  check_local_modifications(wcpath, temp_pool)

  parsed_original_mergeinfo = get_original_mergeinfo(wcpath, revision,
                                                     depth, ctx, temp_pool)

  repo_root = client.svn_client_root_url_from_path(wcpath, ctx, temp_pool)

  core.svn_config_ensure(None)

  if fix == 0:
    sanitize_mergeinfo(parsed_original_mergeinfo, repo_root, wcpath, ctx,
                       hash_file, newmergeinfo_file, temp_pool)
  if fix == 1:
    fix_sanitized_mergeinfo(parsed_original_mergeinfo, repo_root, wcpath,
                            ctx, hash_file, newmergeinfo_file, temp_pool)

  def test_uuid_from_path(self):
    """Test svn_client_uuid_from_path."""
    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_head

    path = tempfile.mktemp('uuid_from_path')

    client.checkout2(REPOS_URL, path, rev, rev, True, True,
                     self.client_ctx)

    wc_adm = wc.adm_open3(None, path, False, 0, None)

    self.assertEquals(client.uuid_from_path(path, wc_adm, self.client_ctx),
                      client.uuid_from_url(REPOS_URL, self.client_ctx))

    self.assert_(isinstance(client.uuid_from_path(path, wc_adm,
                            self.client_ctx), types.StringTypes))

def make_optrev(rev):
    """Return an svn_opt_revision_t build from revision specifier REV.
    REV maybe be an integer value or one of the following string
    specifiers: 'HEAD', 'BASE'."""

    try:
        revnum = int(rev)
        kind = core.svn_opt_revision_number
    except ValueError:
        revnum = -1
        if rev == 'HEAD':
            kind = core.svn_opt_revision_head
        elif rev == 'BASE':
            kind = core.svn_opt_revision_base
        else:
            raise Exception("Unsupported revision specified '%s'" % str(rev))
    optrev = core.svn_opt_revision_t()
    optrev.kind = kind
    optrev.value.number = revnum
    return optrev

  def setUp(self):
    """Load a Subversion repository"""

    self.temper = utils.Temper()

    # Isolate each test from the others with a fresh repository.
    (self.repos, _, self.repos_uri) = self.temper.alloc_known_repo(
      'trac/versioncontrol/tests/svnrepos.dump', suffix='-wc-repo')
    self.fs = repos.fs(self.repos)

    self.path = self.temper.alloc_empty_dir(suffix='-wc-wc')

    client_ctx = client.create_context()

    rev = core.svn_opt_revision_t()
    rev.kind = core.svn_opt_revision_head

    client.checkout2(self.repos_uri, self.path, rev, rev, True, True,
            client_ctx)

    self.wc = wc.adm_open3(None, self.path, True, -1, None)

    def setUp(self):
        """Load a Subversion repository"""

        # Isolate each test from the others with a fresh repository.
        # Eventually, we should move this into a shared TestCase base
        # class that all test cases in this directory can use.
        SubversionRepositoryTestSetup().setUp()

        # Open repository directly for cross-checking
        self.repos = repos.open(REPOS_PATH)
        self.fs = repos.fs(self.repos)

        self.path = core.svn_path_canonicalize(tempfile.mktemp())

        client_ctx = client.create_context()

        rev = core.svn_opt_revision_t()
        rev.kind = core.svn_opt_revision_head

        client.checkout2(REPOS_URL, self.path, rev, rev, True, True, client_ctx)

        self.wc = wc.adm_open3(None, self.path, True, -1, None)