/brz/remove-bazaar

# Copyright (C) 2006-2012, 2016 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

import codecs

import errno

import os

import sys

import time

from .. import (

    bencode,

    errors,

    filters,

    generate_ids,

    osutils,

    revision as _mod_revision,

    rules,

    tests,

    trace,

    transform,

    urlutils,

    )

from ..conflicts import (

    DeletingParent,

    DuplicateEntry,

    DuplicateID,

    MissingParent,

    NonDirectoryParent,

    ParentLoop,

    UnversionedParent,

)

from ..controldir import ControlDir

from ..diff import show_diff_trees

from ..errors import (

    DuplicateKey,

    ExistingLimbo,

    ExistingPendingDeletion,

    ImmortalLimbo,

    ImmortalPendingDeletion,

    LockError,

    MalformedTransform,

    ReusingTransform,

)

from ..osutils import (

    file_kind,

    pathjoin,

)

from ..merge import Merge3Merger, Merger

from ..mutabletree import MutableTree

from ..sixish import (

    BytesIO,

    PY3,

    text_type,

    )

from . import (

    features,

    TestCaseInTempDir,

    TestSkipped,

    )

from .features import (

    HardlinkFeature,

    SymlinkFeature,

    )

from ..transform import (

    build_tree,

    create_from_tree,

    cook_conflicts,

    _FileMover,

    FinalPaths,

    resolve_conflicts,

    resolve_checkout,

    ROOT_PARENT,

    TransformPreview,

    TreeTransform,

)

class TestTreeTransform(tests.TestCaseWithTransport):

    def setUp(self):

        super(TestTreeTransform, self).setUp()

        self.wt = self.make_branch_and_tree('.', format='development-subtree')

        os.chdir('..')

    def get_transform(self):

        transform = TreeTransform(self.wt)

        self.addCleanup(transform.finalize)

        return transform, transform.root

    def get_transform_for_sha1_test(self):

        trans, root = self.get_transform()

        self.wt.lock_tree_write()

        self.addCleanup(self.wt.unlock)

        contents = [b'just some content\n']

        sha1 = osutils.sha_strings(contents)

        # Roll back the clock

        trans._creation_mtime = time.time() - 20.0

        return trans, root, contents, sha1

    def test_existing_limbo(self):

        transform, root = self.get_transform()

        limbo_name = transform._limbodir

        deletion_path = transform._deletiondir

        os.mkdir(pathjoin(limbo_name, 'hehe'))

        self.assertRaises(ImmortalLimbo, transform.apply)

        self.assertRaises(LockError, self.wt.unlock)

        self.assertRaises(ExistingLimbo, self.get_transform)

        self.assertRaises(LockError, self.wt.unlock)

        os.rmdir(pathjoin(limbo_name, 'hehe'))

        os.rmdir(limbo_name)

        os.rmdir(deletion_path)

        transform, root = self.get_transform()

        transform.apply()

    def test_existing_pending_deletion(self):

        transform, root = self.get_transform()

        deletion_path = self._limbodir = urlutils.local_path_from_url(

            transform._tree._transport.abspath('pending-deletion'))

        os.mkdir(pathjoin(deletion_path, 'blocking-directory'))

        self.assertRaises(ImmortalPendingDeletion, transform.apply)

        self.assertRaises(LockError, self.wt.unlock)

        self.assertRaises(ExistingPendingDeletion, self.get_transform)

    def test_build(self):

        transform, root = self.get_transform()

        self.wt.lock_tree_write()

        self.addCleanup(self.wt.unlock)

        self.assertIs(transform.get_tree_parent(root), ROOT_PARENT)

        imaginary_id = transform.trans_id_tree_path('imaginary')

        imaginary_id2 = transform.trans_id_tree_path('imaginary/')

        self.assertEqual(imaginary_id, imaginary_id2)

        self.assertEqual(root, transform.get_tree_parent(imaginary_id))

        self.assertEqual('directory', transform.final_kind(root))

        self.assertEqual(self.wt.get_root_id(), transform.final_file_id(root))

        trans_id = transform.create_path('name', root)

        self.assertIs(transform.final_file_id(trans_id), None)

        self.assertIs(None, transform.final_kind(trans_id))

        transform.create_file([b'contents'], trans_id)

        transform.set_executability(True, trans_id)

        transform.version_file(b'my_pretties', trans_id)

        self.assertRaises(DuplicateKey, transform.version_file,

                          b'my_pretties', trans_id)

        self.assertEqual(transform.final_file_id(trans_id), b'my_pretties')

        self.assertEqual(transform.final_parent(trans_id), root)

        self.assertIs(transform.final_parent(root), ROOT_PARENT)

        self.assertIs(transform.get_tree_parent(root), ROOT_PARENT)

        oz_id = transform.create_path('oz', root)

        transform.create_directory(oz_id)

        transform.version_file(b'ozzie', oz_id)

        trans_id2 = transform.create_path('name2', root)

        transform.create_file([b'contents'], trans_id2)

        transform.set_executability(False, trans_id2)

        transform.version_file(b'my_pretties2', trans_id2)

        modified_paths = transform.apply().modified_paths

        with self.wt.get_file('name') as f:

            self.assertEqual(b'contents', f.read())

        self.assertEqual(self.wt.path2id('name'), b'my_pretties')

        self.assertIs(self.wt.is_executable('name'), True)

        self.assertIs(self.wt.is_executable('name2'), False)

        self.assertEqual('directory', file_kind(self.wt.abspath('oz')))

        self.assertEqual(len(modified_paths), 3)

        tree_mod_paths = [self.wt.abspath(self.wt.id2path(f)) for f in

                          (b'ozzie', b'my_pretties', b'my_pretties2')]

        self.assertSubset(tree_mod_paths, modified_paths)

        # is it safe to finalize repeatedly?

        transform.finalize()

        transform.finalize()

    def test_apply_informs_tree_of_observed_sha1(self):

        trans, root, contents, sha1 = self.get_transform_for_sha1_test()

        trans_id = trans.new_file('file1', root, contents, file_id=b'file1-id',

                                  sha1=sha1)

        calls = []

        orig = self.wt._observed_sha1

        def _observed_sha1(*args):

            calls.append(args)

            orig(*args)

        self.wt._observed_sha1 = _observed_sha1

        trans.apply()

        self.assertEqual([(None, 'file1', trans._observed_sha1s[trans_id])],

                         calls)

    def test_create_file_caches_sha1(self):

        trans, root, contents, sha1 = self.get_transform_for_sha1_test()

        trans_id = trans.create_path('file1', root)

        trans.create_file(contents, trans_id, sha1=sha1)

        st_val = osutils.lstat(trans._limbo_name(trans_id))

        o_sha1, o_st_val = trans._observed_sha1s[trans_id]

        self.assertEqual(o_sha1, sha1)

        self.assertEqualStat(o_st_val, st_val)

    def test__apply_insertions_updates_sha1(self):

        trans, root, contents, sha1 = self.get_transform_for_sha1_test()

        trans_id = trans.create_path('file1', root)

        trans.create_file(contents, trans_id, sha1=sha1)

        st_val = osutils.lstat(trans._limbo_name(trans_id))

        o_sha1, o_st_val = trans._observed_sha1s[trans_id]

        self.assertEqual(o_sha1, sha1)

        self.assertEqualStat(o_st_val, st_val)

        creation_mtime = trans._creation_mtime + 10.0

        # We fake a time difference from when the file was created until now it

        # is being renamed by using os.utime. Note that the change we actually

        # want to see is the real ctime change from 'os.rename()', but as long

        # as we observe a new stat value, we should be fine.

        os.utime(trans._limbo_name(trans_id), (creation_mtime, creation_mtime))

        trans.apply()

        new_st_val = osutils.lstat(self.wt.abspath('file1'))

        o_sha1, o_st_val = trans._observed_sha1s[trans_id]

        self.assertEqual(o_sha1, sha1)

        self.assertEqualStat(o_st_val, new_st_val)

        self.assertNotEqual(st_val.st_mtime, new_st_val.st_mtime)

    def test_new_file_caches_sha1(self):

        trans, root, contents, sha1 = self.get_transform_for_sha1_test()

        trans_id = trans.new_file('file1', root, contents, file_id=b'file1-id',

                                  sha1=sha1)

        st_val = osutils.lstat(trans._limbo_name(trans_id))

        o_sha1, o_st_val = trans._observed_sha1s[trans_id]

        self.assertEqual(o_sha1, sha1)

        self.assertEqualStat(o_st_val, st_val)

    def test_cancel_creation_removes_observed_sha1(self):

        trans, root, contents, sha1 = self.get_transform_for_sha1_test()

        trans_id = trans.new_file('file1', root, contents, file_id=b'file1-id',

                                  sha1=sha1)

        self.assertTrue(trans_id in trans._observed_sha1s)

        trans.cancel_creation(trans_id)

        self.assertFalse(trans_id in trans._observed_sha1s)

    def test_create_files_same_timestamp(self):

        transform, root = self.get_transform()

        self.wt.lock_tree_write()

        self.addCleanup(self.wt.unlock)

        # Roll back the clock, so that we know everything is being set to the

        # exact time

        transform._creation_mtime = creation_mtime = time.time() - 20.0

        transform.create_file([b'content-one'],

                              transform.create_path('one', root))

        time.sleep(1) # *ugly*

        transform.create_file([b'content-two'],

                              transform.create_path('two', root))

        transform.apply()

        fo, st1 = self.wt.get_file_with_stat('one', filtered=False)

        fo.close()

        fo, st2 = self.wt.get_file_with_stat('two', filtered=False)

        fo.close()

        # We only guarantee 2s resolution

        self.assertTrue(abs(creation_mtime - st1.st_mtime) < 2.0,

            "%s != %s within 2 seconds" % (creation_mtime, st1.st_mtime))

        # But if we have more than that, all files should get the same result

        self.assertEqual(st1.st_mtime, st2.st_mtime)

    def test_change_root_id(self):

        transform, root = self.get_transform()

        self.assertNotEqual(b'new-root-id', self.wt.get_root_id())

        transform.new_directory('', ROOT_PARENT, b'new-root-id')

        transform.delete_contents(root)

        transform.unversion_file(root)

        transform.fixup_new_roots()

        transform.apply()

        self.assertEqual(b'new-root-id', self.wt.get_root_id())

    def test_change_root_id_add_files(self):

        transform, root = self.get_transform()

        self.assertNotEqual(b'new-root-id', self.wt.get_root_id())

        new_trans_id = transform.new_directory('', ROOT_PARENT, b'new-root-id')

        transform.new_file('file', new_trans_id, [b'new-contents\n'],

                           b'new-file-id')

        transform.delete_contents(root)

        transform.unversion_file(root)

        transform.fixup_new_roots()

        transform.apply()

        self.assertEqual(b'new-root-id', self.wt.get_root_id())

        self.assertEqual(b'new-file-id', self.wt.path2id('file'))

        self.assertFileEqual(b'new-contents\n', self.wt.abspath('file'))

    def test_add_two_roots(self):

        transform, root = self.get_transform()

        new_trans_id = transform.new_directory('', ROOT_PARENT, b'new-root-id')

        new_trans_id = transform.new_directory('', ROOT_PARENT, b'alt-root-id')

        self.assertRaises(ValueError, transform.fixup_new_roots)

    def test_retain_existing_root(self):

        tt, root = self.get_transform()

        with tt:

            tt.new_directory('', ROOT_PARENT, b'new-root-id')

            tt.fixup_new_roots()

            self.assertNotEqual(b'new-root-id', tt.final_file_id(tt.root))

    def test_retain_existing_root_added_file(self):

        tt, root = self.get_transform()

        new_trans_id = tt.new_directory('', ROOT_PARENT, b'new-root-id')

        child = tt.new_directory('child', new_trans_id, b'child-id')

        tt.fixup_new_roots()

        self.assertEqual(tt.root, tt.final_parent(child))

    def test_add_unversioned_root(self):

        transform, root = self.get_transform()

        new_trans_id = transform.new_directory('', ROOT_PARENT, None)

        transform.delete_contents(transform.root)

        transform.fixup_new_roots()

        self.assertNotIn(transform.root, transform._new_id)

    def test_remove_root_fixup(self):

        transform, root = self.get_transform()

        old_root_id = self.wt.get_root_id()

        self.assertNotEqual(b'new-root-id', old_root_id)

        transform.delete_contents(root)

        transform.unversion_file(root)

        transform.fixup_new_roots()

        transform.apply()

        self.assertEqual(old_root_id, self.wt.get_root_id())

        transform, root = self.get_transform()

        new_trans_id = transform.new_directory('', ROOT_PARENT, b'new-root-id')

        new_trans_id = transform.new_directory('', ROOT_PARENT, b'alt-root-id')

        self.assertRaises(ValueError, transform.fixup_new_roots)

    def test_fixup_new_roots_permits_empty_tree(self):

        transform, root = self.get_transform()

        transform.delete_contents(root)

        transform.unversion_file(root)

        transform.fixup_new_roots()

        self.assertIs(None, transform.final_kind(root))

        self.assertIs(None, transform.final_file_id(root))

    def test_apply_retains_root_directory(self):

        # Do not attempt to delete the physical root directory, because that

        # is impossible.

        transform, root = self.get_transform()

        with transform:

            transform.delete_contents(root)

            e = self.assertRaises(AssertionError, self.assertRaises,

                                  errors.TransformRenameFailed,

                                  transform.apply)

        self.assertContainsRe('TransformRenameFailed not raised', str(e))

    def test_apply_retains_file_id(self):

        transform, root = self.get_transform()

        old_root_id = transform.tree_file_id(root)

        transform.unversion_file(root)

        transform.apply()

        self.assertEqual(old_root_id, self.wt.get_root_id())

    def test_hardlink(self):

        self.requireFeature(HardlinkFeature)

        transform, root = self.get_transform()

        transform.new_file('file1', root, [b'contents'])

        transform.apply()

        target = self.make_branch_and_tree('target')

        target_transform = TreeTransform(target)

        trans_id = target_transform.create_path('file1', target_transform.root)

        target_transform.create_hardlink(self.wt.abspath('file1'), trans_id)

        target_transform.apply()

        self.assertPathExists('target/file1')

        source_stat = os.stat(self.wt.abspath('file1'))

        target_stat = os.stat('target/file1')

        self.assertEqual(source_stat, target_stat)

    def test_convenience(self):

        transform, root = self.get_transform()

        self.wt.lock_tree_write()

        self.addCleanup(self.wt.unlock)

        trans_id = transform.new_file('name', root, [b'contents'],

                                      b'my_pretties', True)

        oz = transform.new_directory('oz', root, b'oz-id')

        dorothy = transform.new_directory('dorothy', oz, b'dorothy-id')

        toto = transform.new_file('toto', dorothy, [b'toto-contents'],

                                  b'toto-id', False)

        self.assertEqual(len(transform.find_conflicts()), 0)

        transform.apply()

        self.assertRaises(ReusingTransform, transform.find_conflicts)

        with open(self.wt.abspath('name'), 'r') as f:

            self.assertEqual('contents', f.read())

        self.assertEqual(self.wt.path2id('name'), b'my_pretties')

        self.assertIs(self.wt.is_executable('name'), True)

        self.assertEqual(self.wt.path2id('oz'), b'oz-id')

        self.assertEqual(self.wt.path2id('oz/dorothy'), b'dorothy-id')

        self.assertEqual(self.wt.path2id('oz/dorothy/toto'), b'toto-id')

        self.assertEqual(b'toto-contents',

                         self.wt.get_file('oz/dorothy/toto').read())

        self.assertIs(self.wt.is_executable('oz/dorothy/toto'), False)

    def test_tree_reference(self):

        transform, root = self.get_transform()

        tree = transform._tree

        trans_id = transform.new_directory('reference', root, b'subtree-id')

        transform.set_tree_reference(b'subtree-revision', trans_id)

        transform.apply()

        tree.lock_read()

        self.addCleanup(tree.unlock)

        self.assertEqual(b'subtree-revision',

                         tree.root_inventory.get_entry(b'subtree-id').reference_revision)

    def test_conflicts(self):

        transform, root = self.get_transform()

        trans_id = transform.new_file('name', root, [b'contents'],

                                      b'my_pretties')

        self.assertEqual(len(transform.find_conflicts()), 0)

        trans_id2 = transform.new_file('name', root, [b'Crontents'], b'toto')

        self.assertEqual(transform.find_conflicts(),

                         [('duplicate', trans_id, trans_id2, 'name')])

        self.assertRaises(MalformedTransform, transform.apply)

        transform.adjust_path('name', trans_id, trans_id2)

        self.assertEqual(transform.find_conflicts(),

                         [('non-directory parent', trans_id)])

        tinman_id = transform.trans_id_tree_path('tinman')

        transform.adjust_path('name', tinman_id, trans_id2)

        self.assertEqual(transform.find_conflicts(),

                         [('unversioned parent', tinman_id),

                          ('missing parent', tinman_id)])

        lion_id = transform.create_path('lion', root)

        self.assertEqual(transform.find_conflicts(),

                         [('unversioned parent', tinman_id),

                          ('missing parent', tinman_id)])

        transform.adjust_path('name', lion_id, trans_id2)

        self.assertEqual(transform.find_conflicts(),

                         [('unversioned parent', lion_id),

                          ('missing parent', lion_id)])

        transform.version_file(b"Courage", lion_id)

        self.assertEqual(transform.find_conflicts(),

                         [('missing parent', lion_id),

                          ('versioning no contents', lion_id)])

        transform.adjust_path('name2', root, trans_id2)

        self.assertEqual(transform.find_conflicts(),

                         [('versioning no contents', lion_id)])

        transform.create_file([b'Contents, okay?'], lion_id)

        transform.adjust_path('name2', trans_id2, trans_id2)

        self.assertEqual(transform.find_conflicts(),

                         [('parent loop', trans_id2),

                          ('non-directory parent', trans_id2)])

        transform.adjust_path('name2', root, trans_id2)

        oz_id = transform.new_directory('oz', root)

        transform.set_executability(True, oz_id)

        self.assertEqual(transform.find_conflicts(),

                         [('unversioned executability', oz_id)])

        transform.version_file(b'oz-id', oz_id)

        self.assertEqual(transform.find_conflicts(),

                         [('non-file executability', oz_id)])

        transform.set_executability(None, oz_id)

        tip_id = transform.new_file('tip', oz_id, [b'ozma'], b'tip-id')

        transform.apply()

        self.assertEqual(self.wt.path2id('name'), b'my_pretties')

        with open(self.wt.abspath('name'), 'rb') as f:

            self.assertEqual(b'contents', f.read())

        transform2, root = self.get_transform()

        oz_id = transform2.trans_id_tree_path('oz')

        newtip = transform2.new_file('tip', oz_id, [b'other'], b'tip-id')

        result = transform2.find_conflicts()

        fp = FinalPaths(transform2)

        self.assertTrue('oz/tip' in transform2._tree_path_ids)

        self.assertEqual(fp.get_path(newtip), pathjoin('oz', 'tip'))

        self.assertEqual(len(result), 2)

        self.assertEqual((result[0][0], result[0][1]),

                         ('duplicate', newtip))

        self.assertEqual((result[1][0], result[1][2]),

                         ('duplicate id', newtip))

        transform2.finalize()

        transform3 = TreeTransform(self.wt)

        self.addCleanup(transform3.finalize)

        oz_id = transform3.trans_id_tree_path('oz')

        transform3.delete_contents(oz_id)

        self.assertEqual(transform3.find_conflicts(),

                         [('missing parent', oz_id)])

        root_id = transform3.root

        tip_id = transform3.trans_id_tree_path('oz/tip')

        transform3.adjust_path('tip', root_id, tip_id)

        transform3.apply()

    def test_conflict_on_case_insensitive(self):

        tree = self.make_branch_and_tree('tree')

        # Don't try this at home, kids!

        # Force the tree to report that it is case sensitive, for conflict

        # resolution tests

        tree.case_sensitive = True

        transform = TreeTransform(tree)

        self.addCleanup(transform.finalize)

        transform.new_file('file', transform.root, [b'content'])

        transform.new_file('FiLe', transform.root, [b'content'])

        result = transform.find_conflicts()

        self.assertEqual([], result)

        transform.finalize()

        # Force the tree to report that it is case insensitive, for conflict

        # generation tests

        tree.case_sensitive = False

        transform = TreeTransform(tree)

        self.addCleanup(transform.finalize)

        transform.new_file('file', transform.root, [b'content'])

        transform.new_file('FiLe', transform.root, [b'content'])

        result = transform.find_conflicts()

        self.assertEqual([('duplicate', 'new-1', 'new-2', 'file')], result)

    def test_conflict_on_case_insensitive_existing(self):

        tree = self.make_branch_and_tree('tree')

        self.build_tree(['tree/FiLe'])

        # Don't try this at home, kids!

        # Force the tree to report that it is case sensitive, for conflict

        # resolution tests

        tree.case_sensitive = True

        transform = TreeTransform(tree)

        self.addCleanup(transform.finalize)

        transform.new_file('file', transform.root, [b'content'])

        result = transform.find_conflicts()

        self.assertEqual([], result)