/brz/remove-bazaar

# Copyright (C) 2008 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, see <http://www.gnu.org/licenses/>.

from __future__ import absolute_import

import time

from .... import (

    tests,

    )

from ..helpers import (

    kind_to_mode,

    )

from . import (

    FastimportFeature,

    )

try:

    from fastimport import commands

except ImportError:

    commands = object()

def load_tests(loader, standard_tests, pattern):

    """Parameterize tests for all versions of groupcompress."""

    scenarios = [

        ('pack-0.92', {'branch_format': 'pack-0.92'}),

        ('1.9-rich-root', {'branch_format': '1.9-rich-root'}),

    ]

    from ....bzr.groupcompress_repo import RepositoryFormat2a

    scenarios.append(('2a', {'branch_format': '2a'}))

    suite = loader.suiteClass()

    result = tests.multiply_tests(standard_tests, scenarios, suite)

    return result

class TestCaseForGenericProcessor(tests.TestCaseWithTransport):

    _test_needs_features = [FastimportFeature]

    branch_format = "pack-0.92"

    def get_handler(self):

        from ..processors import (

            generic_processor,

            )

        branch = self.make_branch('.', format=self.branch_format)

        handler = generic_processor.GenericProcessor(branch.controldir)

        return handler, branch

    # FIXME: [] as a default is bad, as it is mutable, but I want

    # to use None to mean "don't check this".

    def assertChanges(self, branch, revno, expected_added=[],

                      expected_removed=[], expected_modified=[],

                      expected_renamed=[], expected_kind_changed=[]):

        """Check the changes introduced in a revision of a branch.

        This method checks that a revision introduces expected changes.

        The required changes are passed in as a list, where

        each entry contains the needed information about the change.

        If you do not wish to assert anything about a particular

        category then pass None instead.

        branch: The branch.

        revno: revision number of revision to check.

        expected_added: a list of (filename,) tuples that must have

            been added in the delta.

        expected_removed: a list of (filename,) tuples that must have

            been removed in the delta.

        expected_modified: a list of (filename,) tuples that must have

            been modified in the delta.

        expected_renamed: a list of (old_path, new_path) tuples that

            must have been renamed in the delta.

        expected_kind_changed: a list of (path, old_kind, new_kind) tuples

            that must have been changed in the delta.

        :return: revtree1, revtree2

        """

        repo = branch.repository

        revtree1 = repo.revision_tree(branch.get_rev_id(revno - 1))

        revtree2 = repo.revision_tree(branch.get_rev_id(revno))

        changes = revtree2.changes_from(revtree1)

        self._check_changes(changes, expected_added, expected_removed,

                            expected_modified, expected_renamed, expected_kind_changed)

        return revtree1, revtree2

    def _check_changes(self, changes, expected_added=[],

                       expected_removed=[], expected_modified=[],

                       expected_renamed=[], expected_kind_changed=[]):

        """Check the changes in a TreeDelta

        This method checks that the TreeDelta contains the expected

        modifications between the two trees that were used to generate

        it. The required changes are passed in as a list, where

        each entry contains the needed information about the change.

        If you do not wish to assert anything about a particular

        category then pass None instead.

        changes: The TreeDelta to check.

        expected_added: a list of (filename,) tuples that must have

            been added in the delta.

        expected_removed: a list of (filename,) tuples that must have

            been removed in the delta.

        expected_modified: a list of (filename,) tuples that must have

            been modified in the delta.

        expected_renamed: a list of (old_path, new_path) tuples that

            must have been renamed in the delta.

        expected_kind_changed: a list of (path, old_kind, new_kind) tuples

            that must have been changed in the delta.

        """

        renamed = changes.renamed

        added = changes.added + changes.copied

        removed = changes.removed

        modified = changes.modified

        kind_changed = changes.kind_changed

        if expected_renamed is not None:

            self.assertEquals(len(renamed), len(expected_renamed),

                              "%s is renamed, expected %s" % (renamed, expected_renamed))

            renamed_files = [(item.path[0], item.path[1]) for item in renamed]

            for expected_renamed_entry in expected_renamed:

                expected_renamed_entry = (

                    expected_renamed_entry[0].decode('utf-8'),

                    expected_renamed_entry[1].decode('utf-8'))

                self.assertTrue(expected_renamed_entry in renamed_files,

                                "%s is not renamed, %s are" % (expected_renamed_entry,

                                                               renamed_files))

        if expected_added is not None:

            self.assertEquals(len(added), len(expected_added),

                              "%s is added" % str(added))

            added_files = [(item.path[1],) for item in added]

            for expected_added_entry in expected_added:

                expected_added_entry = (

                    expected_added_entry[0].decode('utf-8'), )

                self.assertTrue(expected_added_entry in added_files,

                                "%s is not added, %s are" % (expected_added_entry,

                                                             added_files))

        if expected_removed is not None:

            self.assertEquals(len(removed), len(expected_removed),

                              "%s is removed" % str(removed))

            removed_files = [(item.path[0],) for item in removed]

            for expected_removed_entry in expected_removed:

                expected_removed_entry = (

                    expected_removed_entry[0].decode('utf-8'), )

                self.assertTrue(expected_removed_entry in removed_files,

                                "%s is not removed, %s are" % (expected_removed_entry,

                                                               removed_files))

        if expected_modified is not None:

            self.assertEquals(len(modified), len(expected_modified),

                              "%s is modified" % str(modified))

            modified_files = [(item.path[1],) for item in modified]

            for expected_modified_entry in expected_modified:

                expected_modified_entry = (

                    expected_modified_entry[0].decode('utf-8'), )

                self.assertTrue(expected_modified_entry in modified_files,

                                "%s is not modified, %s are" % (

                                    expected_modified_entry, modified_files))

        if expected_kind_changed is not None:

            self.assertEquals(len(kind_changed), len(expected_kind_changed),

                              "%s is kind-changed, expected %s" % (kind_changed,

                                                                   expected_kind_changed))

            kind_changed_files = [(item.path[1], item.kind[0], item.kind[1])

                                  for item in kind_changed]

            for expected_kind_changed_entry in expected_kind_changed:

                expected_kind_changed_entry = (

                    expected_kind_changed_entry[0].decode('utf-8'), ) + expected_kind_changed_entry[1:]

                self.assertTrue(expected_kind_changed_entry in

                                kind_changed_files, "%s is not kind-changed, %s are" % (

                                    expected_kind_changed_entry, kind_changed_files))

    def assertContent(self, branch, tree, path, content):

        with branch.lock_read():

            self.assertEqual(tree.get_file_text(path.decode('utf-8')), content)

    def assertSymlinkTarget(self, branch, tree, path, target):

        with branch.lock_read():

            self.assertEqual(tree.get_symlink_target(

                path.decode('utf-8')), target)

    def assertExecutable(self, branch, tree, path, executable):

        with branch.lock_read():

            self.assertEqual(tree.is_executable(

                path.decode('utf-8')), executable)

    def assertRevisionRoot(self, revtree, path):

        self.assertEqual(revtree.get_revision_id(),

                         revtree.get_file_revision(path.decode('utf-8')))

class TestImportToPackTag(TestCaseForGenericProcessor):

    def file_command_iter(self, path, kind='file', content=b'aaa',

                          executable=False, to_kind=None, to_content=b'bbb', to_executable=None):

        # Revno 1: create a file or symlink

        # Revno 2: modify it

        if to_kind is None:

            to_kind = kind

        if to_executable is None:

            to_executable = executable

        def command_list():

            author = [b'', b'bugs@a.com', time.time(), time.timezone]

            committer = [b'', b'elmer@a.com', time.time(), time.timezone]

            def files_one():

                yield commands.FileModifyCommand(path,

                                                 kind_to_mode(kind, executable), None, content)

            yield commands.CommitCommand(b'head', b'1', author,

                                         committer, b"commit 1", None, [], files_one)

            def files_two():

                yield commands.FileModifyCommand(path,

                                                 kind_to_mode(to_kind, to_executable), None, to_content)

            # pass "head" for from_ to show that #401249 is worse than I knew

            yield commands.CommitCommand(b'head', b'2', author,

                                         committer, b"commit 2", b"head", [], files_two)

            yield commands.TagCommand(b'tag1', b':1', committer, b"tag 1")

            # pass "head" for from_ to demonstrate #401249

            yield commands.TagCommand(b'tag2', b'head', committer, b"tag 2")

        return command_list

    def test_tag(self):

        handler, branch = self.get_handler()

        path = b'a'

        raise tests.KnownFailure("non-mark committish not yet supported"

                                 "- bug #410249")

        handler.process(self.file_command_iter(path))

class TestImportZeroMarker(TestCaseForGenericProcessor):

    def test_tag(self):

        handler, branch = self.get_handler()

        def command_list():

            committer = [b'', b'elmer@a.com', time.time(), time.timezone]

            yield commands.TagCommand(b'tag1', b':0', committer, b"tag 1")

        handler.process(command_list)

class TestImportToPackModify(TestCaseForGenericProcessor):

    def file_command_iter(self, path, kind='file', content=b'aaa',

                          executable=False, to_kind=None, to_content=b'bbb', to_executable=None):

        # Revno 1: create a file or symlink

        # Revno 2: modify it

        if to_kind is None:

            to_kind = kind

        if to_executable is None:

            to_executable = executable

        mode = kind_to_mode(kind, executable)

        to_mode = kind_to_mode(to_kind, to_executable)

        def command_list():

            author = [b'', b'bugs@a.com', time.time(), time.timezone]

            committer = [b'', b'elmer@a.com', time.time(), time.timezone]

            def files_one():

                yield commands.FileModifyCommand(path, mode, None, content)

            yield commands.CommitCommand(b'head', b'1', author,

                                         committer, b"commit 1", None, [], files_one)

            def files_two():

                yield commands.FileModifyCommand(path, to_mode, None, to_content)

            yield commands.CommitCommand(b'head', b'2', author,

                                         committer, b"commit 2", b":1", [], files_two)

        return command_list

    def test_modify_file_in_root(self):

        handler, branch = self.get_handler()

        path = b'a'

        handler.process(self.file_command_iter(path))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(path,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_modified=[(path,)])

        self.assertContent(branch, revtree1, path, b"aaa")

        self.assertContent(branch, revtree2, path, b"bbb")

        self.assertRevisionRoot(revtree1, path)

        self.assertRevisionRoot(revtree2, path)

    def test_modify_file_in_subdir(self):

        handler, branch = self.get_handler()

        path = b'a/a'

        handler.process(self.file_command_iter(path))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(b'a',), (path,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_modified=[(path,)])

        self.assertContent(branch, revtree1, path, b"aaa")

        self.assertContent(branch, revtree2, path, b"bbb")

    def test_modify_symlink_in_root(self):

        handler, branch = self.get_handler()

        path = b'a'

        handler.process(self.file_command_iter(path, kind='symlink'))

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_modified=[(path,)])

        self.assertSymlinkTarget(branch, revtree1, path, "aaa")

        self.assertSymlinkTarget(branch, revtree2, path, "bbb")

        self.assertRevisionRoot(revtree1, path)

        self.assertRevisionRoot(revtree2, path)

    def test_modify_symlink_in_subdir(self):

        handler, branch = self.get_handler()

        path = b'a/a'

        handler.process(self.file_command_iter(path, kind='symlink'))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(b'a',), (path,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_modified=[(path,)])

        self.assertSymlinkTarget(branch, revtree1, path, "aaa")

        self.assertSymlinkTarget(branch, revtree2, path, "bbb")

    def test_modify_file_becomes_symlink(self):

        handler, branch = self.get_handler()

        path = b'a/a'

        handler.process(self.file_command_iter(path,

                                               kind='file', to_kind='symlink'))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(b'a',), (path,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_kind_changed=[(path, 'file', 'symlink')])

        self.assertContent(branch, revtree1, path, b"aaa")

        self.assertSymlinkTarget(branch, revtree2, path, "bbb")

    def test_modify_symlink_becomes_file(self):

        handler, branch = self.get_handler()

        path = b'a/a'

        handler.process(self.file_command_iter(path,

                                               kind='symlink', to_kind='file'))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(b'a',), (path,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_kind_changed=[(path, 'symlink', 'file')])

        self.assertSymlinkTarget(branch, revtree1, path, "aaa")

        self.assertContent(branch, revtree2, path, b"bbb")

    def test_modify_file_now_executable(self):

        handler, branch = self.get_handler()

        path = b'a/a'

        handler.process(self.file_command_iter(path,

                                               executable=False, to_executable=True, to_content=b'aaa'))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(b'a',), (path,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_modified=[(path,)])

        self.assertExecutable(branch, revtree1, path, False)

        self.assertExecutable(branch, revtree2, path, True)

    def test_modify_file_no_longer_executable(self):

        handler, branch = self.get_handler()

        path = b'a/a'

        handler.process(self.file_command_iter(path,

                                               executable=True, to_executable=False, to_content=b'aaa'))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(b'a',), (path,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_modified=[(path,)])

        self.assertExecutable(branch, revtree1, path, True)

        self.assertExecutable(branch, revtree2, path, False)

class TestImportToPackModifyTwice(TestCaseForGenericProcessor):

    """This tests when the same file is modified twice in the one commit.

    Note: hg-fast-export produces data like this on occasions.

    """

    def file_command_iter(self, path, kind='file', content=b'aaa',

                          executable=False, to_kind=None, to_content=b'bbb', to_executable=None):

        # Revno 1: create a file twice

        if to_kind is None:

            to_kind = kind

        if to_executable is None:

            to_executable = executable

        def command_list():

            author = [b'', b'bugs@a.com', time.time(), time.timezone]

            committer = [b'', b'elmer@a.com', time.time(), time.timezone]

            def files_one():

                yield commands.FileModifyCommand(path, kind_to_mode(kind, executable),

                                                 None, content)

                yield commands.FileModifyCommand(path, kind_to_mode(to_kind, to_executable),

                                                 None, to_content)

            yield commands.CommitCommand(b'head', b'1', author,

                                         committer, b"commit 1", None, [], files_one)

        return command_list

    def test_modify_file_twice_in_root(self):

        handler, branch = self.get_handler()

        path = b'a'

        handler.process(self.file_command_iter(path))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(path,)])

        self.assertContent(branch, revtree1, path, b"aaa")

        self.assertRevisionRoot(revtree1, path)

class TestImportToPackModifyTricky(TestCaseForGenericProcessor):

    def file_command_iter(self, path1, path2, kind='file'):

        # Revno 1: create a file or symlink in a directory

        # Revno 2: create a second file that implicitly deletes the

        # first one because either:

        # * the new file is a in directory with the old file name

        # * the new file has the same name as the directory of the first

        def command_list():

            author = [b'', b'bugs@a.com', time.time(), time.timezone]

            committer = [b'', b'elmer@a.com', time.time(), time.timezone]

            def files_one():

                yield commands.FileModifyCommand(path1, kind_to_mode(kind, False),

                                                 None, b"aaa")

            yield commands.CommitCommand(b'head', b'1', author,

                                         committer, b"commit 1", None, [], files_one)

            def files_two():

                yield commands.FileModifyCommand(path2, kind_to_mode(kind, False),

                                                 None, b"bbb")

            yield commands.CommitCommand(b'head', b'2', author,

                                         committer, b"commit 2", b":1", [], files_two)

        return command_list

    def test_modify_file_becomes_directory(self):

        handler, branch = self.get_handler()

        path1 = b'a/b'

        path2 = b'a/b/c'

        handler.process(self.file_command_iter(path1, path2))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(b'a',), (path1,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_added=[(path2,)],

                                                expected_kind_changed=[(path1, 'file', 'directory')])

        self.assertContent(branch, revtree1, path1, b"aaa")

        self.assertContent(branch, revtree2, path2, b"bbb")

    def test_modify_directory_becomes_file(self):

        handler, branch = self.get_handler()

        path1 = b'a/b/c'

        path2 = b'a/b'

        handler.process(self.file_command_iter(path1, path2))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(b'a',), (b'a/b',), (path1,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_removed=[(path1,), ],

                                                expected_kind_changed=[(path2, 'directory', 'file')])

        self.assertContent(branch, revtree1, path1, b"aaa")

        self.assertContent(branch, revtree2, path2, b"bbb")

    def test_modify_symlink_becomes_directory(self):

        handler, branch = self.get_handler()

        path1 = b'a/b'

        path2 = b'a/b/c'

        handler.process(self.file_command_iter(path1, path2, 'symlink'))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(b'a',), (path1,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_added=[(path2,)],

                                                expected_kind_changed=[(path1, 'symlink', 'directory')])

        self.assertSymlinkTarget(branch, revtree1, path1, "aaa")

        self.assertSymlinkTarget(branch, revtree2, path2, "bbb")

    def test_modify_directory_becomes_symlink(self):

        handler, branch = self.get_handler()

        path1 = b'a/b/c'

        path2 = b'a/b'

        handler.process(self.file_command_iter(path1, path2, 'symlink'))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(b'a',), (b'a/b',), (path1,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_removed=[(path1,), ],

                                                expected_kind_changed=[(path2, 'directory', 'symlink')])

        self.assertSymlinkTarget(branch, revtree1, path1, "aaa")

        self.assertSymlinkTarget(branch, revtree2, path2, "bbb")

class TestImportToPackDelete(TestCaseForGenericProcessor):

    def file_command_iter(self, path, kind='file'):

        # Revno 1: create a file or symlink

        # Revno 2: delete it

        def command_list():

            author = [b'', b'bugs@a.com', time.time(), time.timezone]

            committer = [b'', b'elmer@a.com', time.time(), time.timezone]

            def files_one():

                yield commands.FileModifyCommand(path, kind_to_mode(kind, False),

                                                 None, b"aaa")

            yield commands.CommitCommand(b'head', b'1', author,

                                         committer, b"commit 1", None, [], files_one)

            def files_two():

                yield commands.FileDeleteCommand(path)

            yield commands.CommitCommand(b'head', b'2', author,

                                         committer, b"commit 2", b":1", [], files_two)

        return command_list

    def test_delete_file_in_root(self):

        handler, branch = self.get_handler()

        path = b'a'

        handler.process(self.file_command_iter(path))

        revtree0, revtree1 = self.assertChanges(branch, 1,

                                                expected_added=[(path,)])

        revtree1, revtree2 = self.assertChanges(branch, 2,

                                                expected_removed=[(path,)])

        self.assertContent(branch, revtree1, path, b"aaa")

        self.assertRevisionRoot(revtree1, path)

    def test_delete_file_in_subdir(self):

        handler, branch = self.get_handler()

        path = b'a/a'

        handler.process(self.file_command_iter(path))