~bzr-pqm/bzr/bzr.dev

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
# Copyright (C) 2009, 2010, 2011 Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Implementation tests for bzrlib.merge.Merger."""

import os

from bzrlib.conflicts import TextConflict
from bzrlib import (
    errors,
    merge as _mod_merge,
    option,
    )
from bzrlib.tests import (
    multiply_tests,
    TestCaseWithTransport,
    )
from bzrlib.tests.test_merge_core import MergeBuilder
from bzrlib.transform import TreeTransform



def load_tests(standard_tests, module, loader):
    """Multiply tests for tranport implementations."""
    result = loader.suiteClass()
    scenarios = [
        (name, {'merge_type': merger})
        for name, merger in option._merge_type_registry.items()]
    return multiply_tests(standard_tests, scenarios, result)


class TestMergeImplementation(TestCaseWithTransport):

    def do_merge(self, target_tree, source_tree, **kwargs):
        merger = _mod_merge.Merger.from_revision_ids(None,
            target_tree, source_tree.last_revision(),
            other_branch=source_tree.branch)
        merger.merge_type=self.merge_type
        for name, value in kwargs.items():
            setattr(merger, name, value)
        merger.do_merge()

    def test_merge_specific_file(self):
        this_tree = self.make_branch_and_tree('this')
        this_tree.lock_write()
        self.addCleanup(this_tree.unlock)
        self.build_tree_contents([
            ('this/file1', 'a\nb\n'),
            ('this/file2', 'a\nb\n')
        ])
        this_tree.add(['file1', 'file2'])
        this_tree.commit('Added files')
        other_tree = this_tree.bzrdir.sprout('other').open_workingtree()
        self.build_tree_contents([
            ('other/file1', 'a\nb\nc\n'),
            ('other/file2', 'a\nb\nc\n')
        ])
        other_tree.commit('modified both')
        self.build_tree_contents([
            ('this/file1', 'd\na\nb\n'),
            ('this/file2', 'd\na\nb\n')
        ])
        this_tree.commit('modified both')
        self.do_merge(this_tree, other_tree, interesting_files=['file1'])
        self.assertFileEqual('d\na\nb\nc\n', 'this/file1')
        self.assertFileEqual('d\na\nb\n', 'this/file2')

    def test_merge_move_and_change(self):
        this_tree = self.make_branch_and_tree('this')
        this_tree.lock_write()
        self.addCleanup(this_tree.unlock)
        self.build_tree_contents([
            ('this/file1', 'line 1\nline 2\nline 3\nline 4\n'),
        ])
        this_tree.add('file1',)
        this_tree.commit('Added file')
        other_tree = this_tree.bzrdir.sprout('other').open_workingtree()
        self.build_tree_contents([
            ('other/file1', 'line 1\nline 2 to 2.1\nline 3\nline 4\n'),
        ])
        other_tree.commit('Changed 2 to 2.1')
        self.build_tree_contents([
            ('this/file1', 'line 1\nline 3\nline 2\nline 4\n'),
        ])
        this_tree.commit('Swapped 2 & 3')
        self.do_merge(this_tree, other_tree)
        if self.merge_type is _mod_merge.LCAMerger:
            self.expectFailure(
                "lca merge doesn't conflict for move and change",
                self.assertFileEqual,
                'line 1\n'
                '<<<<<<< TREE\n'
                'line 3\n'
                'line 2\n'
                '=======\n'
                'line 2 to 2.1\n'
                'line 3\n'
                '>>>>>>> MERGE-SOURCE\n'
                'line 4\n', 'this/file1')
        else:
            self.assertFileEqual('line 1\n'
                '<<<<<<< TREE\n'
                'line 3\n'
                'line 2\n'
                '=======\n'
                'line 2 to 2.1\n'
                'line 3\n'
                '>>>>>>> MERGE-SOURCE\n'
                'line 4\n', 'this/file1')

    def test_modify_conflicts_with_delete(self):
        # If one side deletes a line, and the other modifies that line, then
        # the modification should be considered a conflict
        builder = self.make_branch_builder('test')
        builder.start_series()
        builder.build_snapshot('BASE-id', None,
            [('add', ('', None, 'directory', None)),
             ('add', ('foo', 'foo-id', 'file', 'a\nb\nc\nd\ne\n')),
            ])
        # Delete 'b\n'
        builder.build_snapshot('OTHER-id', ['BASE-id'],
            [('modify', ('foo-id', 'a\nc\nd\ne\n'))])
        # Modify 'b\n', add 'X\n'
        builder.build_snapshot('THIS-id', ['BASE-id'],
            [('modify', ('foo-id', 'a\nb2\nc\nd\nX\ne\n'))])
        builder.finish_series()
        branch = builder.get_branch()
        this_tree = branch.bzrdir.create_workingtree()
        this_tree.lock_write()
        self.addCleanup(this_tree.unlock)
        other_tree = this_tree.bzrdir.sprout('other',
                                             'OTHER-id').open_workingtree()
        self.do_merge(this_tree, other_tree)
        if self.merge_type is _mod_merge.LCAMerger:
            self.expectFailure("lca merge doesn't track deleted lines",
                self.assertFileEqual,
                    'a\n'
                    '<<<<<<< TREE\n'
                    'b2\n'
                    '=======\n'
                    '>>>>>>> MERGE-SOURCE\n'
                    'c\n'
                    'd\n'
                    'X\n'
                    'e\n', 'test/foo')
        else:
            self.assertFileEqual(
                'a\n'
                '<<<<<<< TREE\n'
                'b2\n'
                '=======\n'
                '>>>>>>> MERGE-SOURCE\n'
                'c\n'
                'd\n'
                'X\n'
                'e\n', 'test/foo')

    def get_limbodir_deletiondir(self, wt):
        transform = TreeTransform(wt)
        limbodir = transform._limbodir
        deletiondir = transform._deletiondir
        transform.finalize()
        return (limbodir, deletiondir)

    def test_merge_with_existing_limbo(self):
        wt = self.make_branch_and_tree('this')
        (limbodir, deletiondir) =  self.get_limbodir_deletiondir(wt)
        os.mkdir(limbodir)
        self.assertRaises(errors.ExistingLimbo, self.do_merge, wt, wt)
        self.assertRaises(errors.LockError, wt.unlock)

    def test_merge_with_pending_deletion(self):
        wt = self.make_branch_and_tree('this')
        (limbodir, deletiondir) =  self.get_limbodir_deletiondir(wt)
        os.mkdir(deletiondir)
        self.assertRaises(errors.ExistingPendingDeletion, self.do_merge, wt, wt)
        self.assertRaises(errors.LockError, wt.unlock)


class TestHookMergeFileContent(TestCaseWithTransport):
    """Tests that the 'merge_file_content' hook is invoked."""

    def setUp(self):
        TestCaseWithTransport.setUp(self)
        self.hook_log = []

    def install_hook_inactive(self):
        def inactive_factory(merger):
            # This hook is never active
            self.hook_log.append(('inactive',))
            return None
        _mod_merge.Merger.hooks.install_named_hook(
            'merge_file_content', inactive_factory, 'test hook (inactive)')

    def install_hook_noop(self):
        test = self
        class HookNA(_mod_merge.AbstractPerFileMerger):
            def merge_contents(self, merge_params):
                # This hook unconditionally does nothing.
                test.hook_log.append(('no-op',))
                return 'not_applicable', None
        def hook_na_factory(merger):
            return HookNA(merger)
        _mod_merge.Merger.hooks.install_named_hook(
            'merge_file_content', hook_na_factory, 'test hook (no-op)')

    def install_hook_success(self):
        test = self
        class HookSuccess(_mod_merge.AbstractPerFileMerger):
            def merge_contents(self, merge_params):
                test.hook_log.append(('success',))
                if merge_params.file_id == '1':
                    return 'success', ['text-merged-by-hook']
                return 'not_applicable', None
        def hook_success_factory(merger):
            return HookSuccess(merger)
        _mod_merge.Merger.hooks.install_named_hook(
            'merge_file_content', hook_success_factory, 'test hook (success)')

    def install_hook_conflict(self):
        test = self
        class HookConflict(_mod_merge.AbstractPerFileMerger):
            def merge_contents(self, merge_params):
                test.hook_log.append(('conflict',))
                if merge_params.file_id == '1':
                    return ('conflicted',
                        ['text-with-conflict-markers-from-hook'])
                return 'not_applicable', None
        def hook_conflict_factory(merger):
            return HookConflict(merger)
        _mod_merge.Merger.hooks.install_named_hook(
            'merge_file_content', hook_conflict_factory, 'test hook (delete)')

    def install_hook_delete(self):
        test = self
        class HookDelete(_mod_merge.AbstractPerFileMerger):
            def merge_contents(self, merge_params):
                test.hook_log.append(('delete',))
                if merge_params.file_id == '1':
                    return 'delete', None
                return 'not_applicable', None
        def hook_delete_factory(merger):
            return HookDelete(merger)
        _mod_merge.Merger.hooks.install_named_hook(
            'merge_file_content', hook_delete_factory, 'test hook (delete)')

    def install_hook_log_lines(self):
        """Install a hook that saves the get_lines for the this, base and other
        versions of the file.
        """
        test = self
        class HookLogLines(_mod_merge.AbstractPerFileMerger):
            def merge_contents(self, merge_params):
                test.hook_log.append((
                    'log_lines',
                    merge_params.this_lines,
                    merge_params.other_lines,
                    merge_params.base_lines,
                    ))
                return 'not_applicable', None
        def hook_log_lines_factory(merger):
            return HookLogLines(merger)
        _mod_merge.Merger.hooks.install_named_hook(
            'merge_file_content', hook_log_lines_factory,
            'test hook (log_lines)')

    def make_merge_builder(self):
        builder = MergeBuilder(self.test_base_dir)
        self.addCleanup(builder.cleanup)
        return builder

    def create_file_needing_contents_merge(self, builder, file_id):
        builder.add_file(file_id, builder.tree_root, "name1", "text1", True)
        builder.change_contents(file_id, other="text4", this="text3")

    def test_change_vs_change(self):
        """Hook is used for (changed, changed)"""
        self.install_hook_success()
        builder = self.make_merge_builder()
        builder.add_file("1", builder.tree_root, "name1", "text1", True)
        builder.change_contents("1", other="text4", this="text3")
        conflicts = builder.merge(self.merge_type)
        self.assertEqual(conflicts, [])
        self.assertEqual(
            builder.this.get_file('1').read(), 'text-merged-by-hook')

    def test_change_vs_deleted(self):
        """Hook is used for (changed, deleted)"""
        self.install_hook_success()
        builder = self.make_merge_builder()
        builder.add_file("1", builder.tree_root, "name1", "text1", True)
        builder.change_contents("1", this="text2")
        builder.remove_file("1", other=True)
        conflicts = builder.merge(self.merge_type)
        self.assertEqual(conflicts, [])
        self.assertEqual(
            builder.this.get_file('1').read(), 'text-merged-by-hook')

    def test_result_can_be_delete(self):
        """A hook's result can be the deletion of a file."""
        self.install_hook_delete()
        builder = self.make_merge_builder()
        self.create_file_needing_contents_merge(builder, "1")
        conflicts = builder.merge(self.merge_type)
        self.assertEqual(conflicts, [])
        self.assertRaises(errors.NoSuchId, builder.this.id2path, '1')
        self.assertEqual([], list(builder.this.list_files()))

    def test_result_can_be_conflict(self):
        """A hook's result can be a conflict."""
        self.install_hook_conflict()
        builder = self.make_merge_builder()
        self.create_file_needing_contents_merge(builder, "1")
        conflicts = builder.merge(self.merge_type)
        self.assertEqual(conflicts, [TextConflict('name1', file_id='1')])
        # The hook still gets to set the file contents in this case, so that it
        # can insert custom conflict markers.
        self.assertEqual(
            builder.this.get_file('1').read(),
            'text-with-conflict-markers-from-hook')

    def test_can_access_this_other_and_base_versions(self):
        """The hook function can call params.merger.get_lines to access the
        THIS/OTHER/BASE versions of the file.
        """
        self.install_hook_log_lines()
        builder = self.make_merge_builder()
        builder.add_file("1", builder.tree_root, "name1", "text1", True)
        builder.change_contents("1", this="text2", other="text3")
        conflicts = builder.merge(self.merge_type)
        self.assertEqual(
            [('log_lines', ['text2'], ['text3'], ['text1'])], self.hook_log)

    def test_chain_when_not_active(self):
        """When a hook function returns None, merging still works."""
        self.install_hook_inactive()
        self.install_hook_success()
        builder = self.make_merge_builder()
        self.create_file_needing_contents_merge(builder, "1")
        conflicts = builder.merge(self.merge_type)
        self.assertEqual(conflicts, [])
        self.assertEqual(
            builder.this.get_file('1').read(), 'text-merged-by-hook')
        self.assertEqual([('inactive',), ('success',)], self.hook_log)

    def test_chain_when_not_applicable(self):
        """When a hook function returns not_applicable, the next function is
        tried (when one exists).
        """
        self.install_hook_noop()
        self.install_hook_success()
        builder = self.make_merge_builder()
        self.create_file_needing_contents_merge(builder, "1")
        conflicts = builder.merge(self.merge_type)
        self.assertEqual(conflicts, [])
        self.assertEqual(
            builder.this.get_file('1').read(), 'text-merged-by-hook')
        self.assertEqual([('no-op',), ('success',)], self.hook_log)

    def test_chain_stops_after_success(self):
        """When a hook function returns success, no later functions are tried.
        """
        self.install_hook_success()
        self.install_hook_noop()
        builder = self.make_merge_builder()
        self.create_file_needing_contents_merge(builder, "1")
        conflicts = builder.merge(self.merge_type)
        self.assertEqual([('success',)], self.hook_log)

    def test_chain_stops_after_conflict(self):
        """When a hook function returns conflict, no later functions are tried.
        """
        self.install_hook_conflict()
        self.install_hook_noop()
        builder = self.make_merge_builder()
        self.create_file_needing_contents_merge(builder, "1")
        conflicts = builder.merge(self.merge_type)
        self.assertEqual([('conflict',)], self.hook_log)

    def test_chain_stops_after_delete(self):
        """When a hook function returns delete, no later functions are tried.
        """
        self.install_hook_delete()
        self.install_hook_noop()
        builder = self.make_merge_builder()
        self.create_file_needing_contents_merge(builder, "1")
        conflicts = builder.merge(self.merge_type)
        self.assertEqual([('delete',)], self.hook_log)