/brz/remove-bazaar

# Copyright (C) 2006-2011 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Tests for repository implementations - tests a repository format."""

from io import BytesIO

import re

from ... import (

    branch as _mod_branch,

    commit,

    controldir,

    delta as _mod_delta,

    errors,

    gpg,

    info,

    repository,

    revision as _mod_revision,

    tests,

    transport,

    upgrade,

    workingtree,

    )

from ...bzr import (

    branch as _mod_bzrbranch,

    inventory,

    remote,

    repository as bzrrepository,

    )

from ...bzr import (

    knitpack_repo,

    )

from .. import (

    per_repository,

    test_server,

    )

from ..matchers import *

class TestRepositoryMakeBranchAndTree(per_repository.TestCaseWithRepository):

    def test_repository_format(self):

        # make sure the repository on tree.branch is of the desired format,

        # because developers use this api to setup the tree, branch and

        # repository for their tests: having it now give the right repository

        # type would invalidate the tests.

        tree = self.make_branch_and_tree('repo')

        self.assertIsInstance(tree.branch.repository._format,

                              self.repository_format.__class__)

class TestRepository(per_repository.TestCaseWithRepository):

    def assertFormatAttribute(self, attribute, allowed_values):

        """Assert that the format has an attribute 'attribute'."""

        repo = self.make_repository('repo')

        self.assertIn(getattr(repo._format, attribute), allowed_values)

    def assertRepositoryAttribute(self, attribute, allowed_values):

        """Assert that the repo has an attribute 'attribute'."""

        repo = self.make_repository('repo')

        self.assertIn(getattr(repo, attribute), allowed_values)

    def test_attribute_fast_deltas(self):

        """Test the format.fast_deltas attribute."""

        self.assertFormatAttribute('fast_deltas', (True, False))

    def test_attribute_supports_nesting_repositories(self):

        """Test the format.supports_nesting_repositories."""

        self.assertFormatAttribute('supports_nesting_repositories',

                                   (True, False))

    def test_attribute_supports_multiple_authors(self):

        """Test the format.supports_multiple_authors."""

        self.assertFormatAttribute('supports_multiple_authors',

                                   (True, False))

    def test_attribute_supports_unreferenced_revisions(self):

        """Test the format.supports_unreferenced_revisions."""

        self.assertFormatAttribute('supports_unreferenced_revisions',

                                   (True, False))

    def test_attribute__fetch_reconcile(self):

        """Test the _fetch_reconcile attribute."""

        self.assertFormatAttribute('_fetch_reconcile', (True, False))

    def test_attribute_format_experimental(self):

        self.assertFormatAttribute('experimental', (True, False))

    def test_attribute_format_pack_compresses(self):

        self.assertFormatAttribute('pack_compresses', (True, False))

    def test_attribute_format_supports_full_versioned_files(self):

        self.assertFormatAttribute('supports_full_versioned_files',

                                   (True, False))

    def test_attribute_format_supports_funky_characters(self):

        self.assertFormatAttribute('supports_funky_characters',

                                   (True, False))

    def test_attribute_format_supports_leaving_lock(self):

        self.assertFormatAttribute('supports_leaving_lock',

                                   (True, False))

    def test_attribute_format_versioned_directories(self):

        self.assertFormatAttribute(

            'supports_versioned_directories', (True, False))

    def test_attribute_format_revision_graph_can_have_wrong_parents(self):

        self.assertFormatAttribute('revision_graph_can_have_wrong_parents',

                                   (True, False))

    def test_attribute_format_supports_random_access(self):

        self.assertRepositoryAttribute('supports_random_access', (True, False))

    def test_attribute_format_supports_setting_revision_ids(self):

        self.assertFormatAttribute('supports_setting_revision_ids',

                                   (True, False))

    def test_attribute_format_supports_storing_branch_nick(self):

        self.assertFormatAttribute('supports_storing_branch_nick',

                                   (True, False))

    def test_attribute_format_supports_custom_revision_properties(self):

        self.assertFormatAttribute(

            'supports_custom_revision_properties',

            (True, False))

    def test_attribute_format_supports_overriding_transport(self):

        repo = self.make_repository('repo')

        self.assertIn(

            repo._format.supports_overriding_transport, (True, False))

        repo.control_transport.copy_tree('.', '../repository.backup')

        backup_transport = repo.control_transport.clone('../repository.backup')

        if repo._format.supports_overriding_transport:

            backup = repo._format.open(

                repo.controldir,

                _override_transport=backup_transport)

            self.assertIs(backup_transport, backup.control_transport)

        else:

            self.assertRaises(TypeError, repo._format.open,

                              repo.controldir, _override_transport=backup_transport)

    def test_format_is_deprecated(self):

        repo = self.make_repository('repo')

        self.assertIn(repo._format.is_deprecated(), (True, False))

    def test_format_is_supported(self):

        repo = self.make_repository('repo')

        self.assertIn(repo._format.is_supported(), (True, False))

    def test_attribute_format_records_per_file_revision(self):

        self.assertFormatAttribute('records_per_file_revision',

                                   (True, False))

    def test_clone_to_default_format(self):

        # TODO: Test that cloning a repository preserves all the information

        # such as signatures[not tested yet] etc etc.

        # when changing to the current default format.

        tree_a = self.make_branch_and_tree('a')

        self.build_tree(['a/foo'])

        tree_a.add('foo')

        file_id = tree_a.path2id('foo')

        rev1 = tree_a.commit('rev1')

        bzrdirb = self.make_controldir('b')

        repo_b = tree_a.branch.repository.clone(bzrdirb)

        tree_b = repo_b.revision_tree(rev1)

        tree_b.lock_read()

        self.addCleanup(tree_b.unlock)

        tree_b.get_file_text('foo')

        repo_b.get_revision(rev1)

    def test_supports_rich_root(self):

        tree = self.make_branch_and_tree('a')

        tree.commit('')

        second_revision = tree.commit('')

        rev_tree = tree.branch.repository.revision_tree(second_revision)

        rev_tree.lock_read()

        self.addCleanup(rev_tree.unlock)

        root_revision = rev_tree.get_file_revision(u'')

        rich_root = (root_revision != second_revision)

        self.assertEqual(rich_root,

                         tree.branch.repository.supports_rich_root())

    def test_clone_specific_format(self):

        """todo"""

    def test_format_initialize_find_open(self):

        # loopback test to check the current format initializes to itself.

        if not self.repository_format.is_supported():

            # unsupported formats are not loopback testable

            # because the default open will not open them and

            # they may not be initializable.

            return

        # supported formats must be able to init and open

        t = self.get_transport()

        readonly_t = self.get_readonly_transport()

        made_control = self.bzrdir_format.initialize(t.base)

        made_repo = self.repository_format.initialize(made_control)

        self.assertEqual(made_control, made_repo.controldir)

        # find it via controldir opening:

        opened_control = controldir.ControlDir.open(readonly_t.base)

        direct_opened_repo = opened_control.open_repository()

        self.assertEqual(direct_opened_repo.__class__, made_repo.__class__)

        self.assertEqual(opened_control, direct_opened_repo.controldir)

        self.assertIsInstance(direct_opened_repo._format,

                              self.repository_format.__class__)

        # find it via Repository.open

        opened_repo = repository.Repository.open(readonly_t.base)

        self.assertIsInstance(opened_repo, made_repo.__class__)

        self.assertEqual(made_repo._format.__class__,

                         opened_repo._format.__class__)

        # if it has a unique id string, can we probe for it ?

        try:

            self.repository_format.get_format_string()

        except NotImplementedError:

            return

        self.assertEqual(self.repository_format,

                         bzrrepository.RepositoryFormatMetaDir.find_format(opened_control))

    def test_format_matchingcontroldir(self):

        self.assertEqual(self.repository_format,

                         self.repository_format._matchingcontroldir.repository_format)

        self.assertEqual(self.repository_format,

                         self.bzrdir_format.repository_format)

    def test_format_network_name(self):

        repo = self.make_repository('r')

        format = repo._format

        network_name = format.network_name()

        self.assertIsInstance(network_name, bytes)

        # We want to test that the network_name matches the actual format on

        # disk.  For local repositories, that means that using network_name as

        # a key in the registry gives back the same format.  For remote

        # repositories, that means that the network_name of the

        # RemoteRepositoryFormat we have locally matches the actual format

        # present on the remote side.

        if isinstance(format, remote.RemoteRepositoryFormat):

            repo._ensure_real()

            real_repo = repo._real_repository

            self.assertEqual(real_repo._format.network_name(), network_name)

        else:

            registry = repository.network_format_registry

            looked_up_format = registry.get(network_name)

            self.assertEqual(format.__class__, looked_up_format.__class__)

    def test_create_repository(self):

        # bzrdir can construct a repository for itself.

        if not self.bzrdir_format.is_supported():

            # unsupported formats are not loopback testable

            # because the default open will not open them and

            # they may not be initializable.

            return

        t = self.get_transport()

        made_control = self.bzrdir_format.initialize(t.base)

        made_repo = made_control.create_repository()

        # Check that we have a repository object.

        made_repo.has_revision(b'foo')

        self.assertEqual(made_control, made_repo.controldir)

    def test_create_repository_shared(self):

        # bzrdir can construct a shared repository.

        if not self.bzrdir_format.is_supported():

            # unsupported formats are not loopback testable

            # because the default open will not open them and

            # they may not be initializable.

            return

        t = self.get_transport()

        made_control = self.bzrdir_format.initialize(t.base)

        try:

            made_repo = made_control.create_repository(shared=True)

        except errors.IncompatibleFormat:

            # not all repository formats understand being shared, or

            # may only be shared in some circumstances.

            return

        # Check that we have a repository object.

        made_repo.has_revision(b'foo')

        self.assertEqual(made_control, made_repo.controldir)

        self.assertTrue(made_repo.is_shared())

    def test_revision_tree(self):

        wt = self.make_branch_and_tree('.')

        rev1 = wt.commit('lala!', allow_pointless=True)

        root_id = wt.path2id('')

        tree = wt.branch.repository.revision_tree(rev1)

        with tree.lock_read():

            self.assertEqual(rev1, tree.get_file_revision(u''))

            expected = inventory.InventoryDirectory(root_id, '', None)

            expected.revision = rev1

            self.assertEqual([('', 'V', 'directory', expected)],

                             list(tree.list_files(include_root=True)))

        self.assertRaises(ValueError, wt.branch.repository.revision_tree, None)

        tree = wt.branch.repository.revision_tree(_mod_revision.NULL_REVISION)

        with tree.lock_read():

            self.assertEqual([], list(tree.list_files(include_root=True)))

    def test_get_revision_delta(self):

        tree_a = self.make_branch_and_tree('a')

        self.build_tree(['a/foo'])

        tree_a.add('foo')

        rev1 = tree_a.commit('rev1')

        self.build_tree(['a/vla'])

        tree_a.add('vla')

        rev2 = tree_a.commit('rev2')

        delta = tree_a.branch.repository.get_revision_delta(rev1)

        self.assertIsInstance(delta, _mod_delta.TreeDelta)

        self.assertEqual([('foo', 'file')], [(c.path[1], c.kind[1]) for c in delta.added])

        delta = tree_a.branch.repository.get_revision_delta(rev2)

        self.assertIsInstance(delta, _mod_delta.TreeDelta)

        self.assertEqual([('vla', 'file')], [(c.path[1], c.kind[1]) for c in delta.added])

    def test_clone_bzrdir_repository_revision(self):

        # make a repository with some revisions,

        # and clone it, this should not have unreferenced revisions.

        # also: test cloning with a revision id of NULL_REVISION -> empty repo.

        raise tests.TestSkipped('revision limiting is not implemented yet.')

    def test_clone_repository_basis_revision(self):

        raise tests.TestSkipped(

            'the use of a basis should not add noise data to the result.')

    def test_clone_shared_no_tree(self):

        # cloning a shared repository keeps it shared

        # and preserves the make_working_tree setting.

        made_control = self.make_controldir('source')

        try:

            made_repo = made_control.create_repository(shared=True)

        except errors.IncompatibleFormat:

            # not all repository formats understand being shared, or

            # may only be shared in some circumstances.

            return

        try:

            made_repo.set_make_working_trees(False)

        except errors.UnsupportedOperation:

            # the repository does not support having its tree-making flag

            # toggled.

            return

        result = made_control.clone(self.get_url('target'))

        # Check that we have a repository object.

        made_repo.has_revision(b'foo')

        self.assertEqual(made_control, made_repo.controldir)

        self.assertTrue(result.open_repository().is_shared())

        self.assertFalse(result.open_repository().make_working_trees())

    def test_upgrade_preserves_signatures(self):

        if not self.repository_format.supports_revision_signatures:

            raise tests.TestNotApplicable(

                "repository does not support signing revisions")

        wt = self.make_branch_and_tree('source')

        a = wt.commit('A', allow_pointless=True)

        repo = wt.branch.repository

        repo.lock_write()

        repo.start_write_group()

        try:

            repo.sign_revision(a, gpg.LoopbackGPGStrategy(None))

        except errors.UnsupportedOperation:

            self.assertFalse(repo._format.supports_revision_signatures)

            raise tests.TestNotApplicable(

                "signatures not supported by repository format")

        repo.commit_write_group()

        repo.unlock()

        old_signature = repo.get_signature_text(a)

        try:

            old_format = controldir.ControlDirFormat.get_default_format()

            # This gives metadir branches something they can convert to.

            # it would be nice to have a 'latest' vs 'default' concept.

            format = controldir.format_registry.make_controldir(

                'development-subtree')

            upgrade.upgrade(repo.controldir.root_transport.base, format=format)

        except errors.UpToDateFormat:

            # this is in the most current format already.

            return

        except errors.BadConversionTarget as e:

            raise tests.TestSkipped(str(e))

        wt = workingtree.WorkingTree.open(wt.basedir)

        new_signature = wt.branch.repository.get_signature_text(a)

        self.assertEqual(old_signature, new_signature)

    def test_format_description(self):

        repo = self.make_repository('.')

        text = repo._format.get_format_description()

        self.assertTrue(len(text))

    def test_format_supports_external_lookups(self):

        repo = self.make_repository('.')

        self.assertIn(repo._format.supports_external_lookups, (True, False))

    def assertMessageRoundtrips(self, message):

        """Assert that message roundtrips to a repository and back intact."""

        tree = self.make_branch_and_tree('.')

        a = tree.commit(message, allow_pointless=True)

        rev = tree.branch.repository.get_revision(a)

        serializer = getattr(tree.branch.repository, "_serializer", None)

        if serializer is not None and serializer.squashes_xml_invalid_characters:

            # we have to manually escape this as we dont try to

            # roundtrip xml invalid characters in the xml-based serializers.

            escaped_message, escape_count = re.subn(

                u'[^\x09\x0A\x0D\u0020-\uD7FF\uE000-\uFFFD]+',

                lambda match: match.group(0).encode(

                    'unicode_escape').decode('ascii'),

                message)

            self.assertEqual(rev.message, escaped_message)

        else:

            self.assertEqual(rev.message, message)

        # insist the class is unicode no matter what came in for

        # consistency.

        self.assertIsInstance(rev.message, str)

    def test_commit_unicode_message(self):

        # a siple unicode message should be preserved

        self.assertMessageRoundtrips(u'foo bar gamm\xae plop')

    def test_commit_unicode_control_characters(self):

        # a unicode message with control characters should roundtrip too.

        unichars = [chr(x) for x in range(256)]

        # '\r' is not directly allowed anymore, as it used to be translated

        # into '\n' anyway

        unichars[ord('\r')] = u'\n'

        self.assertMessageRoundtrips(

            u"All 8-bit chars: " + ''.join(unichars))

    def test_check_repository(self):

        """Check a fairly simple repository's history"""

        tree = self.make_branch_and_tree('.')

        a_rev = tree.commit('initial empty commit', allow_pointless=True)

        result = tree.branch.repository.check()

        # writes to log; should accept both verbose or non-verbose

        result.report_results(verbose=True)

        result.report_results(verbose=False)

    def test_get_revisions(self):

        tree = self.make_branch_and_tree('.')

        a_rev = tree.commit('initial empty commit', allow_pointless=True)

        b_rev = tree.commit('second empty commit', allow_pointless=True)

        c_rev = tree.commit('third empty commit', allow_pointless=True)

        repo = tree.branch.repository

        revision_ids = [a_rev, b_rev, c_rev]

        revisions = repo.get_revisions(revision_ids)

        self.assertEqual(len(revisions), 3)

        zipped = list(zip(revisions, revision_ids))

        self.assertEqual(len(zipped), 3)

        for revision, revision_id in zipped:

            self.assertEqual(revision.revision_id, revision_id)

            self.assertEqual(revision, repo.get_revision(revision_id))

    def test_iter_revisions(self):

        tree = self.make_branch_and_tree('.')

        a_rev = tree.commit('initial empty commit', allow_pointless=True)

        b_rev = tree.commit('second empty commit', allow_pointless=True)

        c_rev = tree.commit('third empty commit', allow_pointless=True)

        d_rev = b'd-rev'

        repo = tree.branch.repository

        revision_ids = [a_rev, c_rev, b_rev, d_rev]

        revid_with_rev = repo.iter_revisions(revision_ids)

        self.assertEqual(

            set((revid, rev.revision_id if rev is not None else None)

                for (revid, rev) in revid_with_rev),

            {(a_rev, a_rev),

             (b_rev, b_rev),

             (c_rev, c_rev),

             (d_rev, None)})

    def test_root_entry_has_revision(self):

        tree = self.make_branch_and_tree('.')

        revid = tree.commit('message')

        rev_tree = tree.branch.repository.revision_tree(tree.last_revision())

        rev_tree.lock_read()

        self.addCleanup(rev_tree.unlock)

        self.assertEqual(revid, rev_tree.get_file_revision(u''))

    def test_pointless_commit(self):

        tree = self.make_branch_and_tree('.')

        self.assertRaises(commit.PointlessCommit, tree.commit, 'pointless',

                          allow_pointless=False)

        tree.commit('pointless', allow_pointless=True)

    def test_format_attributes(self):

        """All repository formats should have some basic attributes."""

        # create a repository to get a real format instance, not the

        # template from the test suite parameterization.

        repo = self.make_repository('.')

        repo._format.rich_root_data

        repo._format.supports_tree_reference

    def test_iter_files_bytes(self):

        tree = self.make_branch_and_tree('tree')

        self.build_tree_contents([('tree/file1', b'foo'),

                                  ('tree/file2', b'bar')])

        tree.add(['file1', 'file2'])

        file1_id = tree.path2id('file1')

        file2_id = tree.path2id('file2')

        rev1 = tree.commit('rev1')

        self.build_tree_contents([('tree/file1', b'baz')])

        rev2 = tree.commit('rev2')

        repository = tree.branch.repository

        repository.lock_read()

        self.addCleanup(repository.unlock)

        extracted = dict((i, b''.join(b)) for i, b in

                         repository.iter_files_bytes(

                         [(file1_id, rev1, 'file1-old'),

                          (file1_id, rev2, 'file1-new'),

                          (file2_id, rev1, 'file2'),

                          ]))

        self.assertEqual(b'foo', extracted['file1-old'])

        self.assertEqual(b'bar', extracted['file2'])

        self.assertEqual(b'baz', extracted['file1-new'])

        self.assertRaises(errors.RevisionNotPresent, list,

                          repository.iter_files_bytes(

                              [(file1_id, b'rev3', 'file1-notpresent')]))

        self.assertRaises((errors.RevisionNotPresent, errors.NoSuchId), list,

                          repository.iter_files_bytes(

                          [(b'file3-id', b'rev3', 'file1-notpresent')]))

    def test_get_graph(self):

        """Bare-bones smoketest that all repositories implement get_graph."""

        repo = self.make_repository('repo')

        repo.lock_read()

        self.addCleanup(repo.unlock)

        repo.get_graph()

    def test_graph_ghost_handling(self):

        if not self.repository_format.supports_ghosts:

            raise tests.TestNotApplicable('format does not support ghosts')

        tree = self.make_branch_and_tree('here')

        tree.lock_write()

        self.addCleanup(tree.unlock)

        rev1 = tree.commit('initial commit')

        tree.add_parent_tree_id(b'ghost')

        rev2 = tree.commit('commit-with-ghost')

        graph = tree.branch.repository.get_graph()

        parents = graph.get_parent_map([b'ghost', rev2])

        self.assertTrue(b'ghost' not in parents)

        self.assertEqual(parents[rev2], (rev1, b'ghost'))

    def test_get_known_graph_ancestry(self):

        tree = self.make_branch_and_tree('here')

        tree.lock_write()

        self.addCleanup(tree.unlock)