/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

from io import BytesIO, StringIO

import os

import sys

import time

from .. import (

    bencode,

    errors,

    filters,

    osutils,

    revision as _mod_revision,

    rules,

    tests,

    trace,

    transform,

    urlutils,

    )

from ..bzr import (

    generate_ids,

    )

from ..bzr.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,

    ImmortalPendingDeletion,

    LockError,

)

from ..osutils import (

    file_kind,

    pathjoin,

)

from ..merge import Merge3Merger, Merger

from ..mutabletree import MutableTree

from . import (

    features,

    TestCaseInTempDir,

    TestSkipped,

    )

from .features import (

    HardlinkFeature,

    SymlinkFeature,

    )

from ..transform import (

    create_from_tree,

    _FileMover,

    FinalPaths,

    resolve_conflicts,

    ROOT_PARENT,

    ImmortalLimbo,

    MalformedTransform,

    NoFinalPath,

    ReusingTransform,

    TransformRenameFailed,

)

class TransformGroup(object):

    def __init__(self, dirname, root_id):

        self.name = dirname

        os.mkdir(dirname)

        self.wt = ControlDir.create_standalone_workingtree(dirname)

        self.wt.set_root_id(root_id)

        self.b = self.wt.branch

        self.tt = self.wt.transform()

        self.root = self.tt.trans_id_tree_path('')

def conflict_text(tree, merge):

    template = b'%s TREE\n%s%s\n%s%s MERGE-SOURCE\n'

    return template % (b'<' * 7, tree, b'=' * 7, merge, b'>' * 7)

class TestTransformMerge(TestCaseInTempDir):

    def test_text_merge(self):

        root_id = generate_ids.gen_root_id()

        base = TransformGroup("base", root_id)

        base.tt.new_file('a', base.root, [b'a\nb\nc\nd\be\n'], b'a')

        base.tt.new_file('b', base.root, [b'b1'], b'b')

        base.tt.new_file('c', base.root, [b'c'], b'c')

        base.tt.new_file('d', base.root, [b'd'], b'd')

        base.tt.new_file('e', base.root, [b'e'], b'e')

        base.tt.new_file('f', base.root, [b'f'], b'f')

        base.tt.new_directory('g', base.root, b'g')

        base.tt.new_directory('h', base.root, b'h')

        base.tt.apply()

        other = TransformGroup("other", root_id)

        other.tt.new_file('a', other.root, [b'y\nb\nc\nd\be\n'], b'a')

        other.tt.new_file('b', other.root, [b'b2'], b'b')

        other.tt.new_file('c', other.root, [b'c2'], b'c')

        other.tt.new_file('d', other.root, [b'd'], b'd')

        other.tt.new_file('e', other.root, [b'e2'], b'e')

        other.tt.new_file('f', other.root, [b'f'], b'f')

        other.tt.new_file('g', other.root, [b'g'], b'g')

        other.tt.new_file('h', other.root, [b'h\ni\nj\nk\n'], b'h')

        other.tt.new_file('i', other.root, [b'h\ni\nj\nk\n'], b'i')

        other.tt.apply()

        this = TransformGroup("this", root_id)

        this.tt.new_file('a', this.root, [b'a\nb\nc\nd\bz\n'], b'a')

        this.tt.new_file('b', this.root, [b'b'], b'b')

        this.tt.new_file('c', this.root, [b'c'], b'c')

        this.tt.new_file('d', this.root, [b'd2'], b'd')

        this.tt.new_file('e', this.root, [b'e2'], b'e')

        this.tt.new_file('f', this.root, [b'f'], b'f')

        this.tt.new_file('g', this.root, [b'g'], b'g')

        this.tt.new_file('h', this.root, [b'1\n2\n3\n4\n'], b'h')

        this.tt.new_file('i', this.root, [b'1\n2\n3\n4\n'], b'i')

        this.tt.apply()

        Merge3Merger(this.wt, this.wt, base.wt, other.wt)

        # textual merge

        with this.wt.get_file(this.wt.id2path(b'a')) as f:

            self.assertEqual(f.read(), b'y\nb\nc\nd\bz\n')

        # three-way text conflict

        with this.wt.get_file(this.wt.id2path(b'b')) as f:

            self.assertEqual(f.read(), conflict_text(b'b', b'b2'))

        # OTHER wins

        self.assertEqual(this.wt.get_file(this.wt.id2path(b'c')).read(), b'c2')

        # THIS wins

        self.assertEqual(this.wt.get_file(this.wt.id2path(b'd')).read(), b'd2')

        # Ambigious clean merge

        self.assertEqual(this.wt.get_file(this.wt.id2path(b'e')).read(), b'e2')

        # No change

        self.assertEqual(this.wt.get_file(this.wt.id2path(b'f')).read(), b'f')

        # Correct correct results when THIS == OTHER

        self.assertEqual(this.wt.get_file(this.wt.id2path(b'g')).read(), b'g')

        # Text conflict when THIS & OTHER are text and BASE is dir

        self.assertEqual(this.wt.get_file(this.wt.id2path(b'h')).read(),

                         conflict_text(b'1\n2\n3\n4\n', b'h\ni\nj\nk\n'))

        self.assertEqual(this.wt.get_file('h.THIS').read(),

                         b'1\n2\n3\n4\n')

        self.assertEqual(this.wt.get_file('h.OTHER').read(),

                         b'h\ni\nj\nk\n')

        self.assertEqual(file_kind(this.wt.abspath('h.BASE')), 'directory')

        self.assertEqual(this.wt.get_file(this.wt.id2path(b'i')).read(),

                         conflict_text(b'1\n2\n3\n4\n', b'h\ni\nj\nk\n'))

        self.assertEqual(this.wt.get_file('i.THIS').read(),

                         b'1\n2\n3\n4\n')

        self.assertEqual(this.wt.get_file('i.OTHER').read(),

                         b'h\ni\nj\nk\n')

        self.assertEqual(os.path.exists(this.wt.abspath('i.BASE')), False)

        modified = ['a', 'b', 'c', 'h', 'i']

        merge_modified = this.wt.merge_modified()

        self.assertSubset(merge_modified, modified)

        self.assertEqual(len(merge_modified), len(modified))

        with open(this.wt.abspath(this.wt.id2path(b'a')), 'wb') as f:

            f.write(b'booga')

        modified.pop(0)

        merge_modified = this.wt.merge_modified()

        self.assertSubset(merge_modified, modified)

        self.assertEqual(len(merge_modified), len(modified))

        this.wt.remove('b')

        this.wt.revert()

    def test_file_merge(self):

        self.requireFeature(SymlinkFeature)

        root_id = generate_ids.gen_root_id()

        base = TransformGroup("BASE", root_id)

        this = TransformGroup("THIS", root_id)

        other = TransformGroup("OTHER", root_id)

        for tg in this, base, other:

            tg.tt.new_directory('a', tg.root, b'a')

            tg.tt.new_symlink('b', tg.root, 'b', b'b')

            tg.tt.new_file('c', tg.root, [b'c'], b'c')

            tg.tt.new_symlink('d', tg.root, tg.name, b'd')

        targets = ((base, 'base-e', 'base-f', None, None),

                   (this, 'other-e', 'this-f', 'other-g', 'this-h'),

                   (other, 'other-e', None, 'other-g', 'other-h'))

        for tg, e_target, f_target, g_target, h_target in targets:

            for link, target in (('e', e_target), ('f', f_target),

                                 ('g', g_target), ('h', h_target)):

                if target is not None:

                    tg.tt.new_symlink(link, tg.root, target,

                                      link.encode('ascii'))

        for tg in this, base, other:

            tg.tt.apply()

        Merge3Merger(this.wt, this.wt, base.wt, other.wt)

        self.assertIs(os.path.isdir(this.wt.abspath('a')), True)

        self.assertIs(os.path.islink(this.wt.abspath('b')), True)

        self.assertIs(os.path.isfile(this.wt.abspath('c')), True)

        for suffix in ('THIS', 'BASE', 'OTHER'):

            self.assertEqual(os.readlink(

                this.wt.abspath('d.' + suffix)), suffix)

        self.assertIs(os.path.lexists(this.wt.abspath('d')), False)

        self.assertEqual(this.wt.id2path(b'd'), 'd.OTHER')

        self.assertEqual(this.wt.id2path(b'f'), 'f.THIS')

        self.assertEqual(os.readlink(this.wt.abspath('e')), 'other-e')

        self.assertIs(os.path.lexists(this.wt.abspath('e.THIS')), False)

        self.assertIs(os.path.lexists(this.wt.abspath('e.OTHER')), False)

        self.assertIs(os.path.lexists(this.wt.abspath('e.BASE')), False)

        self.assertIs(os.path.lexists(this.wt.abspath('g')), True)

        self.assertIs(os.path.lexists(this.wt.abspath('g.BASE')), False)

        self.assertIs(os.path.lexists(this.wt.abspath('h')), False)

        self.assertIs(os.path.lexists(this.wt.abspath('h.BASE')), False)

        self.assertIs(os.path.lexists(this.wt.abspath('h.THIS')), True)

        self.assertIs(os.path.lexists(this.wt.abspath('h.OTHER')), True)

    def test_filename_merge(self):

        root_id = generate_ids.gen_root_id()

        base = TransformGroup("BASE", root_id)

        this = TransformGroup("THIS", root_id)

        other = TransformGroup("OTHER", root_id)

        base_a, this_a, other_a = [t.tt.new_directory('a', t.root, b'a')

                                   for t in [base, this, other]]

        base_b, this_b, other_b = [t.tt.new_directory('b', t.root, b'b')

                                   for t in [base, this, other]]

        base.tt.new_directory('c', base_a, b'c')

        this.tt.new_directory('c1', this_a, b'c')

        other.tt.new_directory('c', other_b, b'c')

        base.tt.new_directory('d', base_a, b'd')

        this.tt.new_directory('d1', this_b, b'd')

        other.tt.new_directory('d', other_a, b'd')

        base.tt.new_directory('e', base_a, b'e')

        this.tt.new_directory('e', this_a, b'e')

        other.tt.new_directory('e1', other_b, b'e')

        base.tt.new_directory('f', base_a, b'f')

        this.tt.new_directory('f1', this_b, b'f')

        other.tt.new_directory('f1', other_b, b'f')

        for tg in [this, base, other]:

            tg.tt.apply()

        Merge3Merger(this.wt, this.wt, base.wt, other.wt)

        self.assertEqual(this.wt.id2path(b'c'), pathjoin('b/c1'))

        self.assertEqual(this.wt.id2path(b'd'), pathjoin('b/d1'))

        self.assertEqual(this.wt.id2path(b'e'), pathjoin('b/e1'))

        self.assertEqual(this.wt.id2path(b'f'), pathjoin('b/f1'))

    def test_filename_merge_conflicts(self):

        root_id = generate_ids.gen_root_id()

        base = TransformGroup("BASE", root_id)

        this = TransformGroup("THIS", root_id)

        other = TransformGroup("OTHER", root_id)

        base_a, this_a, other_a = [t.tt.new_directory('a', t.root, b'a')

                                   for t in [base, this, other]]

        base_b, this_b, other_b = [t.tt.new_directory('b', t.root, b'b')

                                   for t in [base, this, other]]

        base.tt.new_file('g', base_a, [b'g'], b'g')

        other.tt.new_file('g1', other_b, [b'g1'], b'g')

        base.tt.new_file('h', base_a, [b'h'], b'h')

        this.tt.new_file('h1', this_b, [b'h1'], b'h')

        base.tt.new_file('i', base.root, [b'i'], b'i')

        other.tt.new_directory('i1', this_b, b'i')

        for tg in [this, base, other]:

            tg.tt.apply()

        Merge3Merger(this.wt, this.wt, base.wt, other.wt)

        self.assertEqual(this.wt.id2path(b'g'), pathjoin('b/g1.OTHER'))

        self.assertIs(os.path.lexists(this.wt.abspath('b/g1.BASE')), True)

        self.assertIs(os.path.lexists(this.wt.abspath('b/g1.THIS')), False)

        self.assertEqual(this.wt.id2path(b'h'), pathjoin('b/h1.THIS'))

        self.assertIs(os.path.lexists(this.wt.abspath('b/h1.BASE')), True)

        self.assertIs(os.path.lexists(this.wt.abspath('b/h1.OTHER')), False)

        self.assertEqual(this.wt.id2path(b'i'), pathjoin('b/i1.OTHER'))

class TestCommitTransform(tests.TestCaseWithTransport):

    def get_branch(self):

        tree = self.make_branch_and_tree('tree')

        tree.lock_write()

        self.addCleanup(tree.unlock)

        tree.commit('empty commit')

        return tree.branch

    def get_branch_and_transform(self):

        branch = self.get_branch()

        tt = branch.basis_tree().preview_transform()

        self.addCleanup(tt.finalize)

        return branch, tt

    def test_commit_wrong_basis(self):

        branch = self.get_branch()

        basis = branch.repository.revision_tree(

            _mod_revision.NULL_REVISION)

        tt = basis.preview_transform()

        self.addCleanup(tt.finalize)

        e = self.assertRaises(ValueError, tt.commit, branch, '')

        self.assertEqual('TreeTransform not based on branch basis: null:',

                         str(e))

    def test_empy_commit(self):

        branch, tt = self.get_branch_and_transform()

        rev = tt.commit(branch, 'my message')

        self.assertEqual(2, branch.revno())

        repo = branch.repository

        self.assertEqual('my message', repo.get_revision(rev).message)

    def test_merge_parents(self):

        branch, tt = self.get_branch_and_transform()

        tt.commit(branch, 'my message', [b'rev1b', b'rev1c'])

        self.assertEqual([b'rev1b', b'rev1c'],

                         branch.basis_tree().get_parent_ids()[1:])

    def test_first_commit(self):

        branch = self.make_branch('branch')

        branch.lock_write()

        self.addCleanup(branch.unlock)

        tt = branch.basis_tree().preview_transform()

        self.addCleanup(tt.finalize)

        tt.new_directory('', ROOT_PARENT, b'TREE_ROOT')

        tt.commit(branch, 'my message')

        self.assertEqual([], branch.basis_tree().get_parent_ids())

        self.assertNotEqual(_mod_revision.NULL_REVISION,

                            branch.last_revision())

    def test_first_commit_with_merge_parents(self):

        branch = self.make_branch('branch')

        branch.lock_write()

        self.addCleanup(branch.unlock)

        tt = branch.basis_tree().preview_transform()

        self.addCleanup(tt.finalize)

        e = self.assertRaises(ValueError, tt.commit, branch,

                              'my message', [b'rev1b-id'])

        self.assertEqual('Cannot supply merge parents for first commit.',

                         str(e))

        self.assertEqual(_mod_revision.NULL_REVISION, branch.last_revision())

    def test_add_files(self):

        branch, tt = self.get_branch_and_transform()

        tt.new_file('file', tt.root, [b'contents'], b'file-id')

        trans_id = tt.new_directory('dir', tt.root, b'dir-id')

        if SymlinkFeature.available():

            tt.new_symlink('symlink', trans_id, 'target', b'symlink-id')

        tt.commit(branch, 'message')

        tree = branch.basis_tree()

        self.assertEqual('file', tree.id2path(b'file-id'))

        self.assertEqual(b'contents', tree.get_file_text('file'))

        self.assertEqual('dir', tree.id2path(b'dir-id'))

        if SymlinkFeature.available():

            self.assertEqual('dir/symlink', tree.id2path(b'symlink-id'))

            self.assertEqual('target', tree.get_symlink_target('dir/symlink'))

    def test_add_unversioned(self):

        branch, tt = self.get_branch_and_transform()

        tt.new_file('file', tt.root, [b'contents'])

        self.assertRaises(errors.StrictCommitFailed, tt.commit, branch,

                          'message', strict=True)

    def test_modify_strict(self):

        branch, tt = self.get_branch_and_transform()

        tt.new_file('file', tt.root, [b'contents'], b'file-id')

        tt.commit(branch, 'message', strict=True)

        tt = branch.basis_tree().preview_transform()

        self.addCleanup(tt.finalize)

        trans_id = tt.trans_id_file_id(b'file-id')

        tt.delete_contents(trans_id)

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

        tt.commit(branch, 'message', strict=True)

    def test_commit_malformed(self):

        """Committing a malformed transform should raise an exception.

        In this case, we are adding a file without adding its parent.

        """

        branch, tt = self.get_branch_and_transform()

        parent_id = tt.trans_id_file_id(b'parent-id')

        tt.new_file('file', parent_id, [b'contents'], b'file-id')

        self.assertRaises(MalformedTransform, tt.commit, branch,

                          'message')

    def test_commit_rich_revision_data(self):

        branch, tt = self.get_branch_and_transform()

        rev_id = tt.commit(branch, 'message', timestamp=1, timezone=43201,

                           committer='me <me@example.com>',

                           revprops={u'foo': 'bar'}, revision_id=b'revid-1',

                           authors=['Author1 <author1@example.com>',

                                    'Author2 <author2@example.com>',

                                    ])

        self.assertEqual(b'revid-1', rev_id)

        revision = branch.repository.get_revision(rev_id)

        self.assertEqual(1, revision.timestamp)

        self.assertEqual(43201, revision.timezone)

        self.assertEqual('me <me@example.com>', revision.committer)

        self.assertEqual(['Author1 <author1@example.com>',

                          'Author2 <author2@example.com>'],

                         revision.get_apparent_authors())

        del revision.properties['authors']

        self.assertEqual({'foo': 'bar',

                          'branch-nick': 'tree'},

                         revision.properties)

    def test_no_explicit_revprops(self):

        branch, tt = self.get_branch_and_transform()

        rev_id = tt.commit(branch, 'message', authors=[

            'Author1 <author1@example.com>',

            'Author2 <author2@example.com>', ])

        revision = branch.repository.get_revision(rev_id)

        self.assertEqual(['Author1 <author1@example.com>',

                          'Author2 <author2@example.com>'],

                         revision.get_apparent_authors())

        self.assertEqual('tree', revision.properties['branch-nick'])

class TestFileMover(tests.TestCaseWithTransport):

    def test_file_mover(self):

        self.build_tree(['a/', 'a/b', 'c/', 'c/d'])

        mover = _FileMover()

        mover.rename('a', 'q')

        self.assertPathExists('q')

        self.assertPathDoesNotExist('a')

        self.assertPathExists('q/b')

        self.assertPathExists('c')

        self.assertPathExists('c/d')

    def test_pre_delete_rollback(self):

        self.build_tree(['a/'])

        mover = _FileMover()

        mover.pre_delete('a', 'q')

        self.assertPathExists('q')

        self.assertPathDoesNotExist('a')

        mover.rollback()

        self.assertPathDoesNotExist('q')

        self.assertPathExists('a')

    def test_apply_deletions(self):

        self.build_tree(['a/', 'b/'])

        mover = _FileMover()

        mover.pre_delete('a', 'q')

        mover.pre_delete('b', 'r')

        self.assertPathExists('q')

        self.assertPathExists('r')

        self.assertPathDoesNotExist('a')

        self.assertPathDoesNotExist('b')

        mover.apply_deletions()

        self.assertPathDoesNotExist('q')

        self.assertPathDoesNotExist('r')

        self.assertPathDoesNotExist('a')

        self.assertPathDoesNotExist('b')