# -*- coding: utf-8 -*-## Copyright (C) 2005 Edgewall Software# Copyright (C) 2005 Christopher Lenz <cmlenz@gmx.de># 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://trac.edgewall.org/wiki/TracLicense.## This software consists of voluntary contributions made by many# individuals. For the exact contribution history, see the revision# history and logs, available at http://trac.edgewall.org/log/.## Author: Christopher Lenz <cmlenz@gmx.de>from datetime import datetimeimport os.pathimport statimport shutilimport sysimport tempfileimport unittestfrom StringIO import StringIOtry:    from svn import core, repos    has_svn = Trueexcept:    has_svn = Falsefrom trac.log import logger_factoryfrom trac.test import TestSetupfrom trac.core import TracErrorfrom trac.util.datefmt import utcfrom trac.versioncontrol import Changeset, Node, NoSuchChangesetfrom trac.versioncontrol.svn_fs import SubversionRepositoryfrom trac.versioncontrol import svn_fsREPOS_PATH = os.path.join(tempfile.gettempdir(), 'trac-svnrepos')HEAD = 21class SubversionRepositoryTestSetup(TestSetup):    def setUp(self):        dumpfile = open(os.path.join(os.path.split(__file__)[0],                                     'svnrepos.dump'))        svn_fs._import_svn()        core.apr_initialize()        pool = core.svn_pool_create(None)        dumpstream = None        try:            r = repos.svn_repos_create(REPOS_PATH, '', '', None, None, pool)            if hasattr(repos, 'svn_repos_load_fs2'):                repos.svn_repos_load_fs2(r, dumpfile, StringIO(),                                        repos.svn_repos_load_uuid_default, '',                                        0, 0, None, pool)            else:                dumpstream = core.svn_stream_from_aprfile(dumpfile, pool)                repos.svn_repos_load_fs(r, dumpstream, None,                                        repos.svn_repos_load_uuid_default, '',                                        None, None, pool)        finally:            if dumpstream:                core.svn_stream_close(dumpstream)            core.svn_pool_destroy(pool)            core.apr_terminate()    def tearDown(self):        if os.name == 'nt':            # The Windows version of 'shutil.rmtree' doesn't override the            # permissions of read-only files, so we have to do it ourselves:            format_file = os.path.join(REPOS_PATH, 'db', 'format')            if os.path.isfile(format_file):                os.chmod(format_file, stat.S_IRWXU)            os.chmod(os.path.join(REPOS_PATH, 'format'), stat.S_IRWXU)        shutil.rmtree(REPOS_PATH)class SubversionRepositoryTestCase(unittest.TestCase):    def setUp(self):        self.repos = SubversionRepository(REPOS_PATH, None,                                          logger_factory('test'))    def tearDown(self):        self.repos = None    def test_repos_normalize_path(self):        self.assertEqual('/', self.repos.normalize_path('/'))        self.assertEqual('/', self.repos.normalize_path(''))        self.assertEqual('/', self.repos.normalize_path(None))        self.assertEqual('trunk', self.repos.normalize_path('trunk'))        self.assertEqual('trunk', self.repos.normalize_path('/trunk'))        self.assertEqual('trunk', self.repos.normalize_path('trunk/'))        self.assertEqual('trunk', self.repos.normalize_path('/trunk/'))    def test_repos_normalize_rev(self):        self.assertEqual(HEAD, self.repos.normalize_rev('latest'))        self.assertEqual(HEAD, self.repos.normalize_rev('head'))        self.assertEqual(HEAD, self.repos.normalize_rev(''))        self.assertRaises(NoSuchChangeset,                          self.repos.normalize_rev, 'something else')        self.assertEqual(HEAD, self.repos.normalize_rev(None))        self.assertEqual(11, self.repos.normalize_rev('11'))        self.assertEqual(11, self.repos.normalize_rev(11))    def test_rev_navigation(self):        self.assertEqual(1, self.repos.oldest_rev)        self.assertEqual(None, self.repos.previous_rev(0))        self.assertEqual(None, self.repos.previous_rev(1))        self.assertEqual(HEAD, self.repos.youngest_rev)        self.assertEqual(6, self.repos.next_rev(5))        self.assertEqual(7, self.repos.next_rev(6))        # ...        self.assertEqual(None, self.repos.next_rev(HEAD))        self.assertRaises(NoSuchChangeset, self.repos.normalize_rev, HEAD + 1)    def test_rev_path_navigation(self):        self.assertEqual(1, self.repos.oldest_rev)        self.assertEqual(None, self.repos.previous_rev(0, 'trunk'))        self.assertEqual(None, self.repos.previous_rev(1, 'trunk'))        self.assertEqual(HEAD, self.repos.youngest_rev)        self.assertEqual(6, self.repos.next_rev(5, 'trunk'))        self.assertEqual(13, self.repos.next_rev(6, 'trunk'))        # ...        self.assertEqual(None, self.repos.next_rev(HEAD, 'trunk'))        # test accentuated characters        self.assertEqual(None, self.repos.previous_rev(17, u'trunk/R\xe9sum\xe9.txt'))        self.assertEqual(17, self.repos.next_rev(16, u'trunk/R\xe9sum\xe9.txt'))    def test_has_node(self):        self.assertEqual(False, self.repos.has_node('/trunk/dir1', 3))        self.assertEqual(True, self.repos.has_node('/trunk/dir1', 4))        self.assertEqual(True, self.repos.has_node('/trunk/dir1'))            def test_get_node(self):        node = self.repos.get_node('/trunk')        self.assertEqual('trunk', node.name)        self.assertEqual('/trunk', node.path)        self.assertEqual(Node.DIRECTORY, node.kind)        self.assertEqual(HEAD, node.rev)        self.assertEqual(datetime(2007,4,30,17,45,26,0,utc),                         node.last_modified)        node = self.repos.get_node('/trunk/README.txt')        self.assertEqual('README.txt', node.name)        self.assertEqual('/trunk/README.txt', node.path)        self.assertEqual(Node.FILE, node.kind)        self.assertEqual(3, node.rev)        self.assertEqual(datetime(2005,4,1,13,24,58,0,utc), node.last_modified)    def test_get_node_specific_rev(self):        node = self.repos.get_node('/trunk', 1)        self.assertEqual('trunk', node.name)        self.assertEqual('/trunk', node.path)        self.assertEqual(Node.DIRECTORY, node.kind)        self.assertEqual(1, node.rev)        self.assertEqual(datetime(2005,4,1,10,0,52,0,utc), node.last_modified)        node = self.repos.get_node('/trunk/README.txt', 2)        self.assertEqual('README.txt', node.name)        self.assertEqual('/trunk/README.txt', node.path)        self.assertEqual(Node.FILE, node.kind)        self.assertEqual(2, node.rev)        self.assertEqual(datetime(2005,4,1,13,12,18,0,utc), node.last_modified)    def test_get_dir_entries(self):        node = self.repos.get_node('/trunk')        entries = node.get_entries()        self.assertEqual('dir1', entries.next().name)        self.assertEqual('mpp_proc', entries.next().name)        self.assertEqual('v2', entries.next().name)        self.assertEqual('README3.txt', entries.next().name)        self.assertEqual(u'R\xe9sum\xe9.txt', entries.next().name)        self.assertEqual('README.txt', entries.next().name)        self.assertRaises(StopIteration, entries.next)    def test_get_file_entries(self):        node = self.repos.get_node('/trunk/README.txt')        entries = node.get_entries()        self.assertRaises(StopIteration, entries.next)    def test_get_dir_content(self):        node = self.repos.get_node('/trunk')        self.assertEqual(None, node.content_length)        self.assertEqual(None, node.content_type)        self.assertEqual(None, node.get_content())    def test_get_file_content(self):        node = self.repos.get_node('/trunk/README.txt')        self.assertEqual(8, node.content_length)        self.assertEqual('text/plain', node.content_type)        self.assertEqual('A test.\n', node.get_content().read())    def test_get_dir_properties(self):        f = self.repos.get_node('/trunk')        props = f.get_properties()        self.assertEqual(1, len(props))    def test_get_file_properties(self):        f = self.repos.get_node('/trunk/README.txt')        props = f.get_properties()        self.assertEqual('native', props['svn:eol-style'])        self.assertEqual('text/plain', props['svn:mime-type'])    def test_created_path_rev(self):        node = self.repos.get_node('/trunk/README3.txt', 15)        self.assertEqual(14, node.rev)        self.assertEqual('/trunk/README3.txt', node.path)        self.assertEqual(14, node.created_rev)        self.assertEqual('trunk/README3.txt', node.created_path)    def test_created_path_rev_parent_copy(self):        node = self.repos.get_node('/tags/v1/README.txt', 15)        self.assertEqual(3, node.rev)        self.assertEqual('/tags/v1/README.txt', node.path)        self.assertEqual(3, node.created_rev)        self.assertEqual('trunk/README.txt', node.created_path)    # Revision Log / node history     def test_get_node_history(self):        node = self.repos.get_node('/trunk/README3.txt')        history = node.get_history()        self.assertEqual(('trunk/README3.txt', 14, 'copy'), history.next())        self.assertEqual(('trunk/README2.txt', 6, 'copy'), history.next())        self.assertEqual(('trunk/README.txt', 3, 'edit'), history.next())        self.assertEqual(('trunk/README.txt', 2, 'add'), history.next())        self.assertRaises(StopIteration, history.next)    def test_get_node_history_follow_copy(self):        node = self.repos.get_node('/tags/v1/README.txt')        history = node.get_history()        self.assertEqual(('tags/v1/README.txt', 7, 'copy'), history.next())        self.assertEqual(('trunk/README.txt', 3, 'edit'), history.next())        self.assertEqual(('trunk/README.txt', 2, 'add'), history.next())        self.assertRaises(StopIteration, history.next)    # Revision Log / path history     def test_get_path_history(self):        history = self.repos.get_path_history('/trunk/README2.txt', None)        self.assertEqual(('trunk/README2.txt', 14, 'delete'), history.next())        self.assertEqual(('trunk/README2.txt', 6, 'copy'), history.next())        self.assertEqual(('trunk/README.txt', 3, 'unknown'), history.next())        self.assertRaises(StopIteration, history.next)    def test_get_path_history_copied_file(self):        history = self.repos.get_path_history('/tags/v1/README.txt', None)        self.assertEqual(('tags/v1/README.txt', 7, 'copy'), history.next())        self.assertEqual(('trunk/README.txt', 3, 'unknown'), history.next())        self.assertRaises(StopIteration, history.next)