~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2004, 2005, 2006 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

from cStringIO import StringIO

import os

import sys

import tempfile

from bzrlib import (

    bzrdir,

    errors,

    inventory,

    repository,

    treebuilder,

    )

from bzrlib.builtins import _merge_helper

from bzrlib.bzrdir import BzrDir

from bzrlib.bundle.apply_bundle import install_bundle, merge_bundle

from bzrlib.bundle.bundle_data import BundleTree

from bzrlib.bundle.serializer import write_bundle, read_bundle

from bzrlib.bundle.serializer.v08 import BundleSerializerV08

from bzrlib.bundle.serializer.v09 import BundleSerializerV09

from bzrlib.branch import Branch

from bzrlib.diff import internal_diff

from bzrlib.errors import (BzrError, TestamentMismatch, NotABundle, BadBundle, 

                           NoSuchFile,)

from bzrlib.merge import Merge3Merger

from bzrlib.osutils import has_symlinks, sha_file

from bzrlib.tests import (TestCaseInTempDir, TestCaseWithTransport,

                          TestCase, TestSkipped)

from bzrlib.transform import TreeTransform

from bzrlib.workingtree import WorkingTree

class MockTree(object):

    def __init__(self):

        from bzrlib.inventory import InventoryDirectory, ROOT_ID

        object.__init__(self)

        self.paths = {ROOT_ID: ""}

        self.ids = {"": ROOT_ID}

        self.contents = {}

        self.root = InventoryDirectory(ROOT_ID, '', None)

    inventory = property(lambda x:x)

    def __iter__(self):

        return self.paths.iterkeys()

    def __getitem__(self, file_id):

        if file_id == self.root.file_id:

            return self.root

        else:

            return self.make_entry(file_id, self.paths[file_id])

    def parent_id(self, file_id):

        parent_dir = os.path.dirname(self.paths[file_id])

        if parent_dir == "":

            return None

        return self.ids[parent_dir]

    def iter_entries(self):

        for path, file_id in self.ids.iteritems():

            yield path, self[file_id]

    def get_file_kind(self, file_id):

        if file_id in self.contents:

            kind = 'file'

        else:

            kind = 'directory'

        return kind

    def make_entry(self, file_id, path):

        from bzrlib.inventory import (InventoryEntry, InventoryFile

                                    , InventoryDirectory, InventoryLink)

        name = os.path.basename(path)

        kind = self.get_file_kind(file_id)

        parent_id = self.parent_id(file_id)

        text_sha_1, text_size = self.contents_stats(file_id)

        if kind == 'directory':

            ie = InventoryDirectory(file_id, name, parent_id)

        elif kind == 'file':

            ie = InventoryFile(file_id, name, parent_id)

        elif kind == 'symlink':

            ie = InventoryLink(file_id, name, parent_id)

        else:

            raise BzrError('unknown kind %r' % kind)

        ie.text_sha1 = text_sha_1

        ie.text_size = text_size

        return ie

    def add_dir(self, file_id, path):

        self.paths[file_id] = path

        self.ids[path] = file_id

    def add_file(self, file_id, path, contents):

        self.add_dir(file_id, path)

        self.contents[file_id] = contents

    def path2id(self, path):

        return self.ids.get(path)

    def id2path(self, file_id):

        return self.paths.get(file_id)

    def has_id(self, file_id):

        return self.id2path(file_id) is not None

    def get_file(self, file_id):

        result = StringIO()

        result.write(self.contents[file_id])

        result.seek(0,0)

        return result

    def contents_stats(self, file_id):

        if file_id not in self.contents:

            return None, None

        text_sha1 = sha_file(self.get_file(file_id))

        return text_sha1, len(self.contents[file_id])

class BTreeTester(TestCase):

    """A simple unittest tester for the BundleTree class."""

    def make_tree_1(self):

        mtree = MockTree()

        mtree.add_dir("a", "grandparent")

        mtree.add_dir("b", "grandparent/parent")

        mtree.add_file("c", "grandparent/parent/file", "Hello\n")

        mtree.add_dir("d", "grandparent/alt_parent")

        return BundleTree(mtree, ''), mtree

    def test_renames(self):

        """Ensure that file renames have the proper effect on children"""

        btree = self.make_tree_1()[0]

        self.assertEqual(btree.old_path("grandparent"), "grandparent")

        self.assertEqual(btree.old_path("grandparent/parent"), 

                         "grandparent/parent")

        self.assertEqual(btree.old_path("grandparent/parent/file"),

                         "grandparent/parent/file")

        self.assertEqual(btree.id2path("a"), "grandparent")

        self.assertEqual(btree.id2path("b"), "grandparent/parent")

        self.assertEqual(btree.id2path("c"), "grandparent/parent/file")

        self.assertEqual(btree.path2id("grandparent"), "a")

        self.assertEqual(btree.path2id("grandparent/parent"), "b")

        self.assertEqual(btree.path2id("grandparent/parent/file"), "c")

        assert btree.path2id("grandparent2") is None

        assert btree.path2id("grandparent2/parent") is None

        assert btree.path2id("grandparent2/parent/file") is None

        btree.note_rename("grandparent", "grandparent2")

        assert btree.old_path("grandparent") is None

        assert btree.old_path("grandparent/parent") is None

        assert btree.old_path("grandparent/parent/file") is None

        self.assertEqual(btree.id2path("a"), "grandparent2")

        self.assertEqual(btree.id2path("b"), "grandparent2/parent")

        self.assertEqual(btree.id2path("c"), "grandparent2/parent/file")

        self.assertEqual(btree.path2id("grandparent2"), "a")

        self.assertEqual(btree.path2id("grandparent2/parent"), "b")

        self.assertEqual(btree.path2id("grandparent2/parent/file"), "c")

        assert btree.path2id("grandparent") is None

        assert btree.path2id("grandparent/parent") is None

        assert btree.path2id("grandparent/parent/file") is None

        btree.note_rename("grandparent/parent", "grandparent2/parent2")

        self.assertEqual(btree.id2path("a"), "grandparent2")

        self.assertEqual(btree.id2path("b"), "grandparent2/parent2")

        self.assertEqual(btree.id2path("c"), "grandparent2/parent2/file")

        self.assertEqual(btree.path2id("grandparent2"), "a")

        self.assertEqual(btree.path2id("grandparent2/parent2"), "b")

        self.assertEqual(btree.path2id("grandparent2/parent2/file"), "c")

        assert btree.path2id("grandparent2/parent") is None

        assert btree.path2id("grandparent2/parent/file") is None

        btree.note_rename("grandparent/parent/file", 

                          "grandparent2/parent2/file2")

        self.assertEqual(btree.id2path("a"), "grandparent2")

        self.assertEqual(btree.id2path("b"), "grandparent2/parent2")

        self.assertEqual(btree.id2path("c"), "grandparent2/parent2/file2")

        self.assertEqual(btree.path2id("grandparent2"), "a")

        self.assertEqual(btree.path2id("grandparent2/parent2"), "b")

        self.assertEqual(btree.path2id("grandparent2/parent2/file2"), "c")

        assert btree.path2id("grandparent2/parent2/file") is None

    def test_moves(self):

        """Ensure that file moves have the proper effect on children"""

        btree = self.make_tree_1()[0]

        btree.note_rename("grandparent/parent/file", 

                          "grandparent/alt_parent/file")

        self.assertEqual(btree.id2path("c"), "grandparent/alt_parent/file")

        self.assertEqual(btree.path2id("grandparent/alt_parent/file"), "c")

        assert btree.path2id("grandparent/parent/file") is None

    def unified_diff(self, old, new):

        out = StringIO()

        internal_diff("old", old, "new", new, out)

        out.seek(0,0)

        return out.read()

    def make_tree_2(self):

        btree = self.make_tree_1()[0]

        btree.note_rename("grandparent/parent/file", 

                          "grandparent/alt_parent/file")

        assert btree.id2path("e") is None

        assert btree.path2id("grandparent/parent/file") is None

        btree.note_id("e", "grandparent/parent/file")

        return btree

    def test_adds(self):

        """File/inventory adds"""

        btree = self.make_tree_2()

        add_patch = self.unified_diff([], ["Extra cheese\n"])

        btree.note_patch("grandparent/parent/file", add_patch)

        btree.note_id('f', 'grandparent/parent/symlink', kind='symlink')

        btree.note_target('grandparent/parent/symlink', 'venus')

        self.adds_test(btree)

    def adds_test(self, btree):

        self.assertEqual(btree.id2path("e"), "grandparent/parent/file")

        self.assertEqual(btree.path2id("grandparent/parent/file"), "e")

        self.assertEqual(btree.get_file("e").read(), "Extra cheese\n")

        self.assertEqual(btree.get_symlink_target('f'), 'venus')

    def test_adds2(self):

        """File/inventory adds, with patch-compatibile renames"""

        btree = self.make_tree_2()

        btree.contents_by_id = False

        add_patch = self.unified_diff(["Hello\n"], ["Extra cheese\n"])

        btree.note_patch("grandparent/parent/file", add_patch)

        btree.note_id('f', 'grandparent/parent/symlink', kind='symlink')

        btree.note_target('grandparent/parent/symlink', 'venus')

        self.adds_test(btree)

    def make_tree_3(self):

        btree, mtree = self.make_tree_1()

        mtree.add_file("e", "grandparent/parent/topping", "Anchovies\n")

        btree.note_rename("grandparent/parent/file", 

                          "grandparent/alt_parent/file")

        btree.note_rename("grandparent/parent/topping", 

                          "grandparent/alt_parent/stopping")

        return btree

    def get_file_test(self, btree):

        self.assertEqual(btree.get_file("e").read(), "Lemon\n")

        self.assertEqual(btree.get_file("c").read(), "Hello\n")

    def test_get_file(self):

        """Get file contents"""

        btree = self.make_tree_3()

        mod_patch = self.unified_diff(["Anchovies\n"], ["Lemon\n"])

        btree.note_patch("grandparent/alt_parent/stopping", mod_patch)

        self.get_file_test(btree)

    def test_get_file2(self):

        """Get file contents, with patch-compatibile renames"""

        btree = self.make_tree_3()

        btree.contents_by_id = False

        mod_patch = self.unified_diff([], ["Lemon\n"])

        btree.note_patch("grandparent/alt_parent/stopping", mod_patch)

        mod_patch = self.unified_diff([], ["Hello\n"])

        btree.note_patch("grandparent/alt_parent/file", mod_patch)

        self.get_file_test(btree)

    def test_delete(self):

        "Deletion by bundle"

        btree = self.make_tree_1()[0]

        self.assertEqual(btree.get_file("c").read(), "Hello\n")

        btree.note_deletion("grandparent/parent/file")

        assert btree.id2path("c") is None

        assert btree.path2id("grandparent/parent/file") is None

    def sorted_ids(self, tree):

        ids = list(tree)

        ids.sort()

        return ids

    def test_iteration(self):

        """Ensure that iteration through ids works properly"""

        btree = self.make_tree_1()[0]

        self.assertEqual(self.sorted_ids(btree),

            [inventory.ROOT_ID, 'a', 'b', 'c', 'd'])

        btree.note_deletion("grandparent/parent/file")

        btree.note_id("e", "grandparent/alt_parent/fool", kind="directory")

        btree.note_last_changed("grandparent/alt_parent/fool", 

                                "revisionidiguess")

        self.assertEqual(self.sorted_ids(btree),

            [inventory.ROOT_ID, 'a', 'b', 'd', 'e'])

class BundleTester1(TestCaseWithTransport):

    def test_mismatched_bundle(self):

        format = bzrdir.BzrDirMetaFormat1()

        format.repository_format = repository.RepositoryFormatKnit2()

        serializer = BundleSerializerV08('0.8')

        b = self.make_branch('.', format=format)

        self.assertRaises(errors.IncompatibleBundleFormat, serializer.write, 

                          b.repository, [], {}, StringIO())

    def test_matched_bundle(self):

        """Don't raise IncompatibleBundleFormat for knit2 and bundle0.9"""

        format = bzrdir.BzrDirMetaFormat1()

        format.repository_format = repository.RepositoryFormatKnit2()

        serializer = BundleSerializerV09('0.9')

        b = self.make_branch('.', format=format)

        serializer.write(b.repository, [], {}, StringIO())

    def test_mismatched_model(self):

        """Try copying a bundle from knit2 to knit1"""

        format = bzrdir.BzrDirMetaFormat1()

        format.repository_format = repository.RepositoryFormatKnit2()

        source = self.make_branch_and_tree('source', format=format)

        source.commit('one', rev_id='one-id')

        source.commit('two', rev_id='two-id')

        text = StringIO()

        write_bundle(source.branch.repository, 'two-id', None, text, 

                     format='0.9')

        text.seek(0)

        format = bzrdir.BzrDirMetaFormat1()

        format.repository_format = repository.RepositoryFormatKnit1()

        target = self.make_branch('target', format=format)

        self.assertRaises(errors.IncompatibleRevision, install_bundle, 

                          target.repository, read_bundle(text))

class V08BundleTester(TestCaseWithTransport):

    format = '0.8'

    def bzrdir_format(self):

        format = bzrdir.BzrDirMetaFormat1()

        format.repository_format = repository.RepositoryFormatKnit1()

        return format

    def make_branch_and_tree(self, path, format=None):

        if format is None:

            format = self.bzrdir_format()

        return TestCaseWithTransport.make_branch_and_tree(self, path, format)

    def make_branch(self, path, format=None):

        if format is None:

            format = self.bzrdir_format()

        return TestCaseWithTransport.make_branch(self, path, format)

    def create_bundle_text(self, base_rev_id, rev_id):

        bundle_txt = StringIO()

        rev_ids = write_bundle(self.b1.repository, rev_id, base_rev_id, 

                               bundle_txt, format=self.format)

        bundle_txt.seek(0)

        self.assertEqual(bundle_txt.readline(), 

                         '# Bazaar revision bundle v%s\n' % self.format)

        self.assertEqual(bundle_txt.readline(), '#\n')

        rev = self.b1.repository.get_revision(rev_id)

        self.assertEqual(bundle_txt.readline().decode('utf-8'),

                         u'# message:\n')

        open(',,bundle', 'wb').write(bundle_txt.getvalue())

        bundle_txt.seek(0)

        return bundle_txt, rev_ids

    def get_valid_bundle(self, base_rev_id, rev_id, checkout_dir=None):

        """Create a bundle from base_rev_id -> rev_id in built-in branch.

        Make sure that the text generated is valid, and that it

        can be applied against the base, and generate the same information.

        :return: The in-memory bundle 

        """

        bundle_txt, rev_ids = self.create_bundle_text(base_rev_id, rev_id)

        # This should also validate the generated bundle 

        bundle = read_bundle(bundle_txt)

        repository = self.b1.repository

        for bundle_rev in bundle.real_revisions:

            # These really should have already been checked when we read the

            # bundle, since it computes the sha1 hash for the revision, which

            # only will match if everything is okay, but lets be explicit about

            # it

            branch_rev = repository.get_revision(bundle_rev.revision_id)

            for a in ('inventory_sha1', 'revision_id', 'parent_ids',

                      'timestamp', 'timezone', 'message', 'committer', 

                      'parent_ids', 'properties'):

                self.assertEqual(getattr(branch_rev, a), 

                                 getattr(bundle_rev, a))

            self.assertEqual(len(branch_rev.parent_ids), 

                             len(bundle_rev.parent_ids))

        self.assertEqual(rev_ids, 

                         [r.revision_id for r in bundle.real_revisions])

        self.valid_apply_bundle(base_rev_id, bundle,

                                   checkout_dir=checkout_dir)

        return bundle

    def get_invalid_bundle(self, base_rev_id, rev_id):

        """Create a bundle from base_rev_id -> rev_id in built-in branch.

        Munge the text so that it's invalid.

        :return: The in-memory bundle

        """

        bundle_txt, rev_ids = self.create_bundle_text(base_rev_id, rev_id)

        new_text = bundle_txt.getvalue().replace('executable:no', 

                                               'executable:yes')

        bundle_txt = StringIO(new_text)

        bundle = read_bundle(bundle_txt)

        self.valid_apply_bundle(base_rev_id, bundle)

        return bundle 

    def test_non_bundle(self):

        self.assertRaises(NotABundle, read_bundle, StringIO('#!/bin/sh\n'))

    def test_malformed(self):

        self.assertRaises(BadBundle, read_bundle, 

                          StringIO('# Bazaar revision bundle v'))

    def test_crlf_bundle(self):

        try:

            read_bundle(StringIO('# Bazaar revision bundle v0.8\r\n'))

        except BadBundle:

            # It is currently permitted for bundles with crlf line endings to

            # make read_bundle raise a BadBundle, but this should be fixed.

            # Anything else, especially NotABundle, is an error.

            pass

    def get_checkout(self, rev_id, checkout_dir=None):

        """Get a new tree, with the specified revision in it.

        """

        if checkout_dir is None:

            checkout_dir = tempfile.mkdtemp(prefix='test-branch-', dir='.')

        else:

            if not os.path.exists(checkout_dir):

                os.mkdir(checkout_dir)

        tree = self.make_branch_and_tree(checkout_dir)

        s = StringIO()

        ancestors = write_bundle(self.b1.repository, rev_id, None, s,

                                 format=self.format)

        s.seek(0)

        assert isinstance(s.getvalue(), str), (

            "Bundle isn't a bytestring:\n %s..." % repr(s.getvalue())[:40])

        install_bundle(tree.branch.repository, read_bundle(s))

        for ancestor in ancestors:

            old = self.b1.repository.revision_tree(ancestor)

            new = tree.branch.repository.revision_tree(ancestor)

            # Check that there aren't any inventory level changes

            delta = new.changes_from(old)

            self.assertFalse(delta.has_changed(),

                             'Revision %s not copied correctly.'

                             % (ancestor,))

            # Now check that the file contents are all correct

            for inventory_id in old:

                try:

                    old_file = old.get_file(inventory_id)

                except NoSuchFile:

                    continue

                if old_file is None:

                    continue

                self.assertEqual(old_file.read(),

                                 new.get_file(inventory_id).read())

        if rev_id is not None:

            rh = self.b1.revision_history()

            tree.branch.set_revision_history(rh[:rh.index(rev_id)+1])

            tree.update()

            delta = tree.changes_from(self.b1.repository.revision_tree(rev_id))

            self.assertFalse(delta.has_changed(),

                             'Working tree has modifications')

        return tree

    def valid_apply_bundle(self, base_rev_id, info, checkout_dir=None):

        """Get the base revision, apply the changes, and make

        sure everything matches the builtin branch.

        """

        to_tree = self.get_checkout(base_rev_id, checkout_dir=checkout_dir)

        original_parents = to_tree.get_parent_ids()

        repository = to_tree.branch.repository

        original_parents = to_tree.get_parent_ids()

        self.assertIs(repository.has_revision(base_rev_id), True)

        for rev in info.real_revisions:

            self.assert_(not repository.has_revision(rev.revision_id),

                'Revision {%s} present before applying bundle' 

                % rev.revision_id)

        merge_bundle(info, to_tree, True, Merge3Merger, False, False)

        for rev in info.real_revisions:

            self.assert_(repository.has_revision(rev.revision_id),

                'Missing revision {%s} after applying bundle' 

                % rev.revision_id)

        self.assert_(to_tree.branch.repository.has_revision(info.target))

        # Do we also want to verify that all the texts have been added?

        self.assertEqual(original_parents + [info.target],

            to_tree.get_parent_ids())

        rev = info.real_revisions[-1]

        base_tree = self.b1.repository.revision_tree(rev.revision_id)

        to_tree = to_tree.branch.repository.revision_tree(rev.revision_id)

        # TODO: make sure the target tree is identical to base tree

        #       we might also check the working tree.

        base_files = list(base_tree.list_files())

        to_files = list(to_tree.list_files())

        self.assertEqual(len(base_files), len(to_files))

        for base_file, to_file in zip(base_files, to_files):

            self.assertEqual(base_file, to_file)

        for path, status, kind, fileid, entry in base_files:

            # Check that the meta information is the same

            self.assertEqual(base_tree.get_file_size(fileid),

                    to_tree.get_file_size(fileid))

            self.assertEqual(base_tree.get_file_sha1(fileid),

                    to_tree.get_file_sha1(fileid))

            # Check that the contents are the same

            # This is pretty expensive

            # self.assertEqual(base_tree.get_file(fileid).read(),

            #         to_tree.get_file(fileid).read())

    def test_bundle(self):

        self.tree1 = self.make_branch_and_tree('b1')

        self.b1 = self.tree1.branch

        open('b1/one', 'wb').write('one\n')

        self.tree1.add('one')

        self.tree1.commit('add one', rev_id='a@cset-0-1')

        bundle = self.get_valid_bundle(None, 'a@cset-0-1')

        # FIXME: The current write_bundle api no longer supports

        #        setting a custom summary message

        #        We should re-introduce the ability, and update

        #        the tests to make sure it works.

        # bundle = self.get_valid_bundle(None, 'a@cset-0-1',

        #         message='With a specialized message')

        # Make sure we can handle files with spaces, tabs, other

        # bogus characters

        self.build_tree([

                'b1/with space.txt'

                , 'b1/dir/'

                , 'b1/dir/filein subdir.c'

                , 'b1/dir/WithCaps.txt'

                , 'b1/dir/ pre space'

                , 'b1/sub/'

                , 'b1/sub/sub/'

                , 'b1/sub/sub/nonempty.txt'

                ])

        open('b1/sub/sub/emptyfile.txt', 'wb').close()

        open('b1/dir/nolastnewline.txt', 'wb').write('bloop')

        tt = TreeTransform(self.tree1)

        tt.new_file('executable', tt.root, '#!/bin/sh\n', 'exe-1', True)

        tt.apply()

        self.tree1.add([

                'with space.txt'

                , 'dir'

                , 'dir/filein subdir.c'

                , 'dir/WithCaps.txt'

                , 'dir/ pre space'

                , 'dir/nolastnewline.txt'