def test_branchmap_rebuildmeta(self):
        '''test rebuildmeta on a branchmapped clone'''
        repo_path = self.load_svndump('branchmap.svndump')
        branchmap = open(self.branchmap, 'w')
        branchmap.write("badname = dit\n")
        branchmap.write("feature = dah\n")
        branchmap.close()
        ui = self.ui()
        ui.setconfig('hgsubversion', 'branchmap', self.branchmap)
        commands.clone(ui, test_util.fileurl(repo_path),
                       self.wc_path, branchmap=self.branchmap)
        originfo = self.repo.svnmeta().branches

        # clone & rebuild
        ui = self.ui()
        src, dest = test_util.hgclone(ui, self.wc_path, self.wc_path + '_clone',
                                      update=False)
        src = test_util.getlocalpeer(src)
        dest = test_util.getlocalpeer(dest)
        svncommands.rebuildmeta(ui, dest,
                                args=[test_util.fileurl(repo_path)])

        # just check the keys; assume the contents are unaffected by the branch
        # map and thus properly tested by other tests
        self.assertEquals(sorted(src.svnmeta().branches),
                          sorted(dest.svnmeta().branches))

    def test_corruption(self):
        SUCCESS = 0
        FAILURE = 1

        repo, repo_path = self.load_and_fetch('correct.svndump', layout='single',
                                              subdir='')

        ui = self.ui()

        self.assertEqual(SUCCESS, verify.verify(ui, self.repo, rev='tip'))

        corrupt_source = test_util.fileurl(self.load_svndump('corrupt.svndump'))

        repo.ui.setconfig('paths', 'default', corrupt_source)

        ui.pushbuffer()
        code = verify.verify(ui, repo, rev='tip')
        actual = ui.popbuffer()

        actual = actual.replace(corrupt_source, '$REPO')
        actual = set(actual.splitlines())

        expected = set([
            'verifying 78e965230a13 against [email protected]',
            'missing file: missing-file',
            'wrong flags for: executable-file',
            'wrong flags for: symlink',
            'wrong flags for: regular-file',
            'difference in: another-regular-file',
            'difference in: regular-file',
            'unexpected file: empty-file',
        ])

        self.assertEqual((FAILURE, expected), (code, actual))

 def test_list_authors_map(self):
     repo_path = self.load_svndump('replace_trunk_with_branch.svndump')
     author_path = os.path.join(repo_path, 'authors')
     svncommands.listauthors(self.ui(),
                             args=[test_util.fileurl(repo_path)],
                             authors=author_path)
     self.assertMultiLineEqual(open(author_path).read(), 'Augie=\nevil=\n')

    def test_push_to_non_tip(self):
        self.test_push_to_branch(push=False)
        wc2path = self.wc_path + '_clone'
        u = self.repo.ui
        hg.clone(self.repo.ui, self.wc_path, wc2path, update=False)
        res = self.pushrevisions()
        self.assertEqual(0, res)
        oldf = open(os.path.join(self.wc_path, '.hg', 'hgrc'))
        hgrc = oldf.read()
        oldf.close()
        shutil.rmtree(self.wc_path)
        hg.clone(u, wc2path, self.wc_path, update=False)
        oldf = open(os.path.join(self.wc_path, '.hg', 'hgrc'), 'w')
        oldf.write(hgrc)
        oldf.close()

        # do a commit here
        self.commitchanges([('foobaz', 'foobaz', 'This file is added on default.', ),
                            ],
                           parent='default',
                           message='commit to default')
        from hgsubversion import svncommands
        svncommands.rebuildmeta(u,
                                self.repo,
                                args=[test_util.fileurl(self.repo_path)])


        hg.update(self.repo, self.repo['tip'].node())
        oldnode = self.repo['tip'].hex()
        self.pushrevisions(expected_extra_back=1)
        self.assertNotEqual(oldnode, self.repo['tip'].hex(), 'Revision was not pushed.')

    def test_push_single_dir_renamed_branch(self, stupid=False):
        # Tests pulling and pushing with a renamed branch
        # Based on test_push_single_dir
        test_util.load_svndump_fixture(self.repo_path,
                                       'branch_from_tag.svndump')
        cmd = ['clone', '--layout=single', '--branch=flaf']
        if stupid:
            cmd.append('--stupid')
        cmd += [test_util.fileurl(self.repo_path), self.wc_path]
        dispatch.dispatch(cmd)

        def file_callback(repo, memctx, path):
            if path == 'adding_file':
                return context.memfilectx(path=path,
                                          data='foo',
                                          islink=False,
                                          isexec=False,
                                          copied=False)
            raise IOError(errno.EINVAL, 'Invalid operation: ' + path)
        ctx = context.memctx(self.repo,
                             (self.repo['tip'].node(), node.nullid),
                             'automated test',
                             ['adding_file'],
                             file_callback,
                             'an_author',
                             '2009-10-19 18:49:30 -0500',
                             {'branch': 'default',})
        self.repo.commitctx(ctx)
        hg.update(self.repo, self.repo['tip'].node())
        self.pushrevisions()
        self.assertTrue('adding_file' in self.svnls(''))

        self.assertEquals(set(['flaf']),
                          set(self.repo[i].branch() for i in self.repo))

 def test_list_authors(self):
     repo_path = self.load_svndump('replace_trunk_with_branch.svndump')
     u = self.ui()
     u.pushbuffer()
     svncommands.listauthors(u,
                                  args=[test_util.fileurl(repo_path)],
                                  authors=None)
     actual = u.popbuffer()
     self.assertMultiLineEqual(actual, 'Augie\nevil\n')

def _do_case(self, name, stupid, single):
    subdir = test_util.subdir.get(name, '')
    layout = 'auto'
    if single:
        layout = 'single'
    self._load_fixture_and_fetch(name, subdir=subdir, stupid=stupid, layout=layout)
    assert len(self.repo) > 0
    wc2_path = self.wc_path + '_clone'
    u = ui.ui()
    src, dest = hg.clone(u, self.wc_path, wc2_path, update=False)

    # insert a wrapper that prevents calling changectx.children()
    def failfn(orig, ctx):
        self.fail('calling %s is forbidden; it can cause massive slowdowns '
                  'when rebuilding large repositories' % orig)

    origchildren = getattr(context.changectx, 'children')
    extensions.wrapfunction(context.changectx, 'children', failfn)

    try:
        svncommands.rebuildmeta(u, dest,
                                args=[test_util.fileurl(self.repo_path +
                                                        subdir), ])
    finally:
        # remove the wrapper
        context.changectx.children = origchildren

    self.assertTrue(os.path.isdir(os.path.join(src.path, 'svn')),
                    'no .hg/svn directory in the source!')
    self.assertTrue(os.path.isdir(os.path.join(src.path, 'svn')),
                    'no .hg/svn directory in the destination!')
    dest = hg.repository(u, os.path.dirname(dest.path))
    for tf in ('rev_map', 'uuid', 'tagmap', 'layout', 'subdir', ):
        stf = os.path.join(src.path, 'svn', tf)
        self.assertTrue(os.path.isfile(stf), '%r is missing!' % stf)
        dtf = os.path.join(dest.path, 'svn', tf)
        self.assertTrue(os.path.isfile(dtf), '%r is missing!' % tf)
        old, new = open(stf).read(), open(dtf).read()
        self.assertMultiLineEqual(old, new)
        self.assertEqual(src.branchtags(), dest.branchtags())
    srcbi = pickle.load(open(os.path.join(src.path, 'svn', 'branch_info')))
    destbi = pickle.load(open(os.path.join(dest.path, 'svn', 'branch_info')))
    self.assertEqual(sorted(srcbi.keys()), sorted(destbi.keys()))
    revkeys = svnmeta.SVNMeta(dest).revmap.keys()
    for branch in destbi:
        srcinfo = srcbi[branch]
        destinfo = destbi[branch]
        if srcinfo[:2] == (None, 0) or destinfo[:2] == (None, 0):
            self.assertTrue(srcinfo[2] <= destinfo[2],
                            'Latest revision for %s decreased from %d to %d!'
                            % (branch or 'default', srcinfo[2], destinfo[2]))
            self.assertEqual(srcinfo[0], destinfo[0])
        else:
            pr = sorted(filter(lambda x: x[1] == srcinfo[0] and x[0] <= srcinfo[1],
                        revkeys), reverse=True)[0][0]
            self.assertEqual(pr, destinfo[1])
            self.assertEqual(srcinfo[2], destinfo[2])

    def test_quoting(self):
        ui = self.ui()
        repo_path = self.load_svndump('non_ascii_path_1.svndump')

        repo_url = test_util.fileurl(repo_path)
        subdir = '/b\xC3\xB8b'
        quoted_subdir = urllib.quote(subdir)

        repo1 = svnrepo.svnremoterepo(ui, repo_url + subdir)
        repo2 = svnrepo.svnremoterepo(ui, repo_url + quoted_subdir)
        self.assertEqual(repo1.svnurl, repo2.svnurl)

 def test_file_map_exclude(self, stupid=False):
     test_util.load_svndump_fixture(self.repo_path, 'replace_trunk_with_branch.svndump')
     filemap = open(self.filemap, 'w')
     filemap.write("exclude alpha\n")
     filemap.close()
     ui = self.ui(stupid)
     ui.setconfig('hgsubversion', 'filemap', self.filemap)
     commands.clone(ui, test_util.fileurl(self.repo_path),
                    self.wc_path, filemap=self.filemap)
     self.assertEqual(node.hex(self.repo[0].node()), '2c48f3525926ab6c8b8424bcf5eb34b149b61841')
     self.assertEqual(node.hex(self.repo['default'].node()), 'b37a3c0297b71f989064d9b545b5a478bbed7cc1')

 def test_file_map(self, stupid=False):
     test_util.load_svndump_fixture(self.repo_path, 'replace_trunk_with_branch.svndump')
     filemap = open(self.filemap, 'w')
     filemap.write("include alpha\n")
     filemap.close()
     ui = self.ui(stupid)
     ui.setconfig('hgsubversion', 'filemap', self.filemap)
     commands.clone(ui, test_util.fileurl(self.repo_path),
                    self.wc_path, filemap=self.filemap)
     self.assertEqual(node.hex(self.repo[0].node()), '88e2c7492d83e4bf30fbb2dcbf6aa24d60ac688d')
     self.assertEqual(node.hex(self.repo['default'].node()), 'e524296152246b3837fe9503c83b727075835155')

 def test_branchmap_tagging(self):
     '''test tagging a renamed branch, which used to raise an exception'''
     repo_path = self.load_svndump('commit-to-tag.svndump')
     branchmap = open(self.branchmap, 'w')
     branchmap.write("magic = art\n")
     branchmap.close()
     ui = self.ui()
     ui.setconfig('hgsubversion', 'branchmap', self.branchmap)
     commands.clone(ui, test_util.fileurl(repo_path),
                    self.wc_path, branchmap=self.branchmap)
     branches = set(self.repo[i].branch() for i in self.repo)
     self.assertEquals(sorted(branches), ['art', 'closeme'])

 def test_branchmap_empty_commit(self):
     '''test mapping an empty commit on a renamed branch'''
     repo_path = self.load_svndump('propset-branch.svndump')
     branchmap = open(self.branchmap, 'w')
     branchmap.write("the-branch = bob\n")
     branchmap.close()
     ui = self.ui()
     ui.setconfig('hgsubversion', 'branchmap', self.branchmap)
     commands.clone(ui, test_util.fileurl(repo_path),
                    self.wc_path, branchmap=self.branchmap)
     branches = set(self.repo[i].branch() for i in self.repo)
     self.assertEquals(sorted(branches), ['bob', 'default'])

    def test_empty_log_message(self):
        repo, repo_path = self.load_and_fetch('empty-log-message.svndump')

        self.assertEqual(repo['tip'].description(), '')

        test_util.rmtree(self.wc_path)

        ui = self.ui()
        ui.setconfig('hgsubversion', 'defaultmessage', 'blyf')
        commands.clone(ui, test_util.fileurl(repo_path), self.wc_path)

        self.assertEqual(self.repo['tip'].description(), 'blyf')

 def test_author_map_closing_author(self):
     repo_path = self.load_svndump('replace_trunk_with_branch.svndump')
     authormap = open(self.authors, 'w')
     authormap.write("evil=Testy <[email protected]>")
     authormap.close()
     ui = self.ui()
     ui.setconfig('hgsubversion', 'authormap', self.authors)
     commands.clone(ui, test_util.fileurl(repo_path),
                    self.wc_path, authors=self.authors)
     self.assertEqual(self.repo[0].user(),
                      '[email protected]')
     self.assertEqual(self.repo['tip'].user(),
                     'Testy <[email protected]>')

    def test_tagren_changed(self):
        repo_path = self.load_svndump('commit-to-tag.svndump')
        tagmap = open(self.tagmap, 'w')
        tagmap.write("edit-at-create = edit-past\n")
        tagmap.write("also-edit = \n")
        tagmap.write("will-edit = edit-future\n")
        tagmap.close()

        ui = self.ui()
        ui.setconfig('hgsubversion', 'tagmap', self.tagmap)
        commands.clone(ui, test_util.fileurl(repo_path),
                       self.wc_path, tagmap=self.tagmap)
        tags = self.repo.tags()

 def test_svnrebuildmeta(self):
     otherpath = self.load_svndump('binaryfiles-broken.svndump')
     otherurl = test_util.fileurl(otherpath)
     self.load_and_fetch('replace_trunk_with_branch.svndump')
     # rebuildmeta with original repo
     svncommands.rebuildmeta(self.ui(), repo=self.repo, args=[])
     # rebuildmeta with unrelated repo
     self.assertRaises(hgutil.Abort,
                       svncommands.rebuildmeta,
                       self.ui(), repo=self.repo, args=[otherurl])
     # rebuildmeta --unsafe-skip-uuid-check with unrelated repo
     svncommands.rebuildmeta(self.ui(), repo=self.repo, args=[otherurl],
                             unsafe_skip_uuid_check=True)

 def _loadwithfilemap(self, svndump, filemapcontent,
         failonmissing=True):
     repo_path = self.load_svndump(svndump)
     filemap = open(self.filemap, 'w')
     filemap.write(filemapcontent)
     filemap.close()
     ui = self.ui()
     ui.setconfig('hgsubversion', 'filemap', self.filemap)
     ui.setconfig('hgsubversion', 'failoninvalidreplayfile', 'true')
     ui.setconfig('hgsubversion', 'failonmissing', failonmissing)
     commands.clone(ui, test_util.fileurl(repo_path),
                    self.wc_path, filemap=self.filemap)
     return self.repo

 def setUp(self):
     self.tmpdir = tempfile.mkdtemp('svnwrap_test')
     self.repo_path = '%s/testrepo' % self.tmpdir
     subprocess.call(['svnadmin', 'create', self.repo_path, ])
     inp = open(os.path.join(os.path.dirname(__file__), 'fixtures',
                             'project_root_at_repo_root.svndump'))
     proc = subprocess.call(['svnadmin', 'load', self.repo_path, ],
                             stdin=inp,
                             close_fds=test_util.canCloseFds,
                             stdout=subprocess.PIPE,
                             stderr=subprocess.STDOUT)
     assert proc == 0
     self.repo = svnwrap.SubversionRepo(test_util.fileurl(self.repo_path))

 def test_author_map(self):
     repo_path = self.load_svndump('replace_trunk_with_branch.svndump')
     authormap = open(self.authors, 'w')
     authormap.write('Augie=Augie Fackler <[email protected]> # stuffy\n')
     authormap.write("Augie Fackler <[email protected]>\n")
     authormap.close()
     ui = self.ui()
     ui.setconfig('hgsubversion', 'authormap', self.authors)
     commands.clone(ui, test_util.fileurl(repo_path),
                    self.wc_path, authors=self.authors)
     self.assertEqual(self.repo[0].user(),
                      'Augie Fackler <[email protected]>')
     self.assertEqual(self.repo['tip'].user(),
                     '[email protected]')

 def test_branchmap(self):
     repo_path = self.load_svndump('branchmap.svndump')
     branchmap = open(self.branchmap, 'w')
     branchmap.write("badname = good-name # stuffy\n")
     branchmap.write("feature = default\n")
     branchmap.close()
     ui = self.ui()
     ui.setconfig('hgsubversion', 'branchmap', self.branchmap)
     commands.clone(ui, test_util.fileurl(repo_path),
                    self.wc_path, branchmap=self.branchmap)
     branches = set(self.repo[i].branch() for i in self.repo)
     self.assert_('badname' not in branches)
     self.assert_('good-name' in branches)
     self.assertEquals(self.repo[2].branch(), 'default')