← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transform.py

(vila) Fix bug #701212,
 don't set the tag dict of the master branch you are updating from. (John A
 Meinel)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transform.py
1
 
# Copyright (C) 2006-2011 Canonical Ltd
 
1
# Copyright (C) 2006-2010 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
 
import errno
18
17
import os
19
18
from StringIO import StringIO
20
19
import sys
93
92
        self.addCleanup(transform.finalize)
94
93
        return transform, transform.root
95
94
 
96
 
    def get_transform_for_sha1_test(self):
97
 
        trans, root = self.get_transform()
98
 
        self.wt.lock_tree_write()
99
 
        self.addCleanup(self.wt.unlock)
100
 
        contents = ['just some content\n']
101
 
        sha1 = osutils.sha_strings(contents)
102
 
        # Roll back the clock
103
 
        trans._creation_mtime = time.time() - 20.0
104
 
        return trans, root, contents, sha1
105
 
 
106
95
    def test_existing_limbo(self):
107
96
        transform, root = self.get_transform()
108
97
        limbo_name = transform._limbodir
171
160
        transform.finalize()
172
161
        transform.finalize()
173
162
 
174
 
    def test_apply_informs_tree_of_observed_sha1(self):
175
 
        trans, root, contents, sha1 = self.get_transform_for_sha1_test()
176
 
        trans_id = trans.new_file('file1', root, contents, file_id='file1-id',
177
 
                                  sha1=sha1)
178
 
        calls = []
179
 
        orig = self.wt._observed_sha1
180
 
        def _observed_sha1(*args):
181
 
            calls.append(args)
182
 
            orig(*args)
183
 
        self.wt._observed_sha1 = _observed_sha1
184
 
        trans.apply()
185
 
        self.assertEqual([(None, 'file1', trans._observed_sha1s[trans_id])],
186
 
                         calls)
187
 
 
188
 
    def test_create_file_caches_sha1(self):
189
 
        trans, root, contents, sha1 = self.get_transform_for_sha1_test()
190
 
        trans_id = trans.create_path('file1', root)
191
 
        trans.create_file(contents, trans_id, sha1=sha1)
192
 
        st_val = osutils.lstat(trans._limbo_name(trans_id))
193
 
        o_sha1, o_st_val = trans._observed_sha1s[trans_id]
194
 
        self.assertEqual(o_sha1, sha1)
195
 
        self.assertEqualStat(o_st_val, st_val)
196
 
 
197
 
    def test__apply_insertions_updates_sha1(self):
198
 
        trans, root, contents, sha1 = self.get_transform_for_sha1_test()
199
 
        trans_id = trans.create_path('file1', root)
200
 
        trans.create_file(contents, trans_id, sha1=sha1)
201
 
        st_val = osutils.lstat(trans._limbo_name(trans_id))
202
 
        o_sha1, o_st_val = trans._observed_sha1s[trans_id]
203
 
        self.assertEqual(o_sha1, sha1)
204
 
        self.assertEqualStat(o_st_val, st_val)
205
 
        creation_mtime = trans._creation_mtime + 10.0
206
 
        # We fake a time difference from when the file was created until now it
207
 
        # is being renamed by using os.utime. Note that the change we actually
208
 
        # want to see is the real ctime change from 'os.rename()', but as long
209
 
        # as we observe a new stat value, we should be fine.
210
 
        os.utime(trans._limbo_name(trans_id), (creation_mtime, creation_mtime))
211
 
        trans.apply()
212
 
        new_st_val = osutils.lstat(self.wt.abspath('file1'))
213
 
        o_sha1, o_st_val = trans._observed_sha1s[trans_id]
214
 
        self.assertEqual(o_sha1, sha1)
215
 
        self.assertEqualStat(o_st_val, new_st_val)
216
 
        self.assertNotEqual(st_val.st_mtime, new_st_val.st_mtime)
217
 
 
218
 
    def test_new_file_caches_sha1(self):
219
 
        trans, root, contents, sha1 = self.get_transform_for_sha1_test()
220
 
        trans_id = trans.new_file('file1', root, contents, file_id='file1-id',
221
 
                                  sha1=sha1)
222
 
        st_val = osutils.lstat(trans._limbo_name(trans_id))
223
 
        o_sha1, o_st_val = trans._observed_sha1s[trans_id]
224
 
        self.assertEqual(o_sha1, sha1)
225
 
        self.assertEqualStat(o_st_val, st_val)
226
 
 
227
 
    def test_cancel_creation_removes_observed_sha1(self):
228
 
        trans, root, contents, sha1 = self.get_transform_for_sha1_test()
229
 
        trans_id = trans.new_file('file1', root, contents, file_id='file1-id',
230
 
                                  sha1=sha1)
231
 
        self.assertTrue(trans_id in trans._observed_sha1s)
232
 
        trans.cancel_creation(trans_id)
233
 
        self.assertFalse(trans_id in trans._observed_sha1s)
234
 
 
235
163
    def test_create_files_same_timestamp(self):
236
164
        transform, root = self.get_transform()
237
165
        self.wt.lock_tree_write()
285
213
        new_trans_id = transform.new_directory('', ROOT_PARENT, 'alt-root-id')
286
214
        self.assertRaises(ValueError, transform.fixup_new_roots)
287
215
 
288
 
    def test_retain_existing_root(self):
289
 
        tt, root = self.get_transform()
290
 
        with tt:
291
 
            tt.new_directory('', ROOT_PARENT, 'new-root-id')
292
 
            tt.fixup_new_roots()
293
 
            self.assertNotEqual('new-root-id', tt.final_file_id(tt.root))
294
 
 
295
 
    def test_retain_existing_root_added_file(self):
296
 
        tt, root = self.get_transform()
297
 
        new_trans_id = tt.new_directory('', ROOT_PARENT, 'new-root-id')
298
 
        child = tt.new_directory('child', new_trans_id, 'child-id')
299
 
        tt.fixup_new_roots()
300
 
        self.assertEqual(tt.root, tt.final_parent(child))
301
 
 
302
 
    def test_add_unversioned_root(self):
303
 
        transform, root = self.get_transform()
304
 
        new_trans_id = transform.new_directory('', ROOT_PARENT, None)
305
 
        transform.delete_contents(transform.root)
306
 
        transform.fixup_new_roots()
307
 
        self.assertNotIn(transform.root, transform._new_id)
308
 
 
309
 
    def test_remove_root_fixup(self):
310
 
        transform, root = self.get_transform()
311
 
        old_root_id = self.wt.get_root_id()
312
 
        self.assertNotEqual('new-root-id', old_root_id)
313
 
        transform.delete_contents(root)
314
 
        transform.unversion_file(root)
315
 
        transform.fixup_new_roots()
316
 
        transform.apply()
317
 
        self.assertEqual(old_root_id, self.wt.get_root_id())
318
 
 
319
 
        transform, root = self.get_transform()
320
 
        new_trans_id = transform.new_directory('', ROOT_PARENT, 'new-root-id')
321
 
        new_trans_id = transform.new_directory('', ROOT_PARENT, 'alt-root-id')
322
 
        self.assertRaises(ValueError, transform.fixup_new_roots)
323
 
 
324
 
    def test_apply_retains_root_directory(self):
325
 
        # Do not attempt to delete the physical root directory, because that
326
 
        # is impossible.
327
 
        transform, root = self.get_transform()
328
 
        with transform:
329
 
            transform.delete_contents(root)
330
 
            e = self.assertRaises(AssertionError, self.assertRaises,
331
 
                                  errors.TransformRenameFailed,
332
 
                                  transform.apply)
333
 
        self.assertContainsRe('TransformRenameFailed not raised', str(e))
334
 
 
335
216
    def test_hardlink(self):
336
217
        self.requireFeature(HardlinkFeature)
337
218
        transform, root = self.get_transform()
342
223
        trans_id = target_transform.create_path('file1', target_transform.root)
343
224
        target_transform.create_hardlink(self.wt.abspath('file1'), trans_id)
344
225
        target_transform.apply()
345
 
        self.assertPathExists('target/file1')
 
226
        self.failUnlessExists('target/file1')
346
227
        source_stat = os.stat(self.wt.abspath('file1'))
347
228
        target_stat = os.stat('target/file1')
348
229
        self.assertEqual(source_stat, target_stat)
514
395
        transform.new_file('FiLe', transform.root, 'content')
515
396
        resolve_conflicts(transform)
516
397
        transform.apply()
517
 
        self.assertPathExists('tree/file')
518
 
        self.assertPathExists('tree/FiLe.moved')
 
398
        self.failUnlessExists('tree/file')
 
399
        self.failUnlessExists('tree/FiLe.moved')
519
400
 
520
401
    def test_resolve_checkout_case_conflict(self):
521
402
        tree = self.make_branch_and_tree('tree')
530
411
        resolve_conflicts(transform,
531
412
                          pass_func=lambda t, c: resolve_checkout(t, c, []))
532
413
        transform.apply()
533
 
        self.assertPathExists('tree/file')
534
 
        self.assertPathExists('tree/FiLe.moved')
 
414
        self.failUnlessExists('tree/file')
 
415
        self.failUnlessExists('tree/FiLe.moved')
535
416
 
536
417
    def test_apply_case_conflict(self):
537
418
        """Ensure that a transform with case conflicts can always be applied"""
545
426
        transform.new_file('dirFiLe', dir, 'content')
546
427
        resolve_conflicts(transform)
547
428
        transform.apply()
548
 
        self.assertPathExists('tree/file')
 
429
        self.failUnlessExists('tree/file')
549
430
        if not os.path.exists('tree/FiLe.moved'):
550
 
            self.assertPathExists('tree/FiLe')
551
 
        self.assertPathExists('tree/dir/dirfile')
 
431
            self.failUnlessExists('tree/FiLe')
 
432
        self.failUnlessExists('tree/dir/dirfile')
552
433
        if not os.path.exists('tree/dir/dirFiLe.moved'):
553
 
            self.assertPathExists('tree/dir/dirFiLe')
 
434
            self.failUnlessExists('tree/dir/dirFiLe')
554
435
 
555
436
    def test_case_insensitive_limbo(self):
556
437
        tree = self.make_branch_and_tree('tree')
858
739
        raw_conflicts = resolve_conflicts(tt)
859
740
        cooked_conflicts = cook_conflicts(raw_conflicts, tt)
860
741
        tt.finalize()
861
 
        conflicts_s = [unicode(c) for c in cooked_conflicts]
 
742
        conflicts_s = [str(c) for c in cooked_conflicts]
862
743
        self.assertEqual(len(cooked_conflicts), len(conflicts_s))
863
744
        self.assertEqual(conflicts_s[0], 'Conflict adding file dorothy.  '
864
745
                                         'Moved existing file to '
1016
897
        # On windows looks like:
1017
898
        # "Failed to rename .../work/myfile to 
1018
899
        # .../work/.bzr/checkout/limbo/new-1: [Errno 13] Permission denied"
1019
 
        # This test isn't concerned with exactly what the error looks like,
1020
 
        # and the strerror will vary across OS and locales, but the assert
1021
 
        # that the exeception attributes are what we expect
1022
 
        self.assertEqual(e.errno, errno.EACCES)
1023
 
        if os.name == "posix":
1024
 
            self.assertEndsWith(e.to_path, "/first-dir/newname")
1025
 
        else:
1026
 
            self.assertEqual(os.path.basename(e.from_path), "myfile")
 
900
        # The strerror will vary per OS and language so it's not checked here
 
901
        self.assertContainsRe(str(e),
 
902
            "Failed to rename .*(first-dir.newname:|myfile)")
1027
903
 
1028
904
    def test_set_executability_order(self):
1029
905
        """Ensure that executability behaves the same, no matter what order.
1302
1178
        parent2 = transform.new_directory('parent2', root)
1303
1179
        transform.adjust_path('child1', parent2, child1)
1304
1180
        transform.apply()
1305
 
        self.assertPathDoesNotExist(self.wt.abspath('parent1/child1'))
1306
 
        self.assertPathExists(self.wt.abspath('parent2/child1'))
 
1181
        self.failIfExists(self.wt.abspath('parent1/child1'))
 
1182
        self.failUnlessExists(self.wt.abspath('parent2/child1'))
1307
1183
        # rename limbo/new-1 => parent1, rename limbo/new-3 => parent2
1308
1184
        # no rename for child1 (counting only renames during apply)
1309
 
        self.assertEqual(2, transform.rename_count)
 
1185
        self.failUnlessEqual(2, transform.rename_count)
1310
1186
 
1311
1187
    def test_cancel_parent(self):
1312
1188
        """Cancelling a parent doesn't cause deletion of a non-empty directory
1335
1211
        parent2 = transform.new_directory('parent2', root)
1336
1212
        transform.adjust_path('child1', parent2, child1)
1337
1213
        transform.apply()
1338
 
        self.assertPathDoesNotExist(self.wt.abspath('parent1'))
1339
 
        self.assertPathExists(self.wt.abspath('parent2/child1'))
 
1214
        self.failIfExists(self.wt.abspath('parent1'))
 
1215
        self.failUnlessExists(self.wt.abspath('parent2/child1'))
1340
1216
        # rename limbo/new-3 => parent2, rename limbo/new-2 => child1
1341
 
        self.assertEqual(2, transform.rename_count)
 
1217
        self.failUnlessEqual(2, transform.rename_count)
1342
1218
 
1343
1219
    def test_adjust_and_cancel(self):
1344
1220
        """Make sure adjust_path keeps track of limbo children properly"""
1377
1253
        child = transform.new_directory('child', parent)
1378
1254
        transform.adjust_path('parent', root, parent)
1379
1255
        transform.apply()
1380
 
        self.assertPathExists(self.wt.abspath('parent/child'))
 
1256
        self.failUnlessExists(self.wt.abspath('parent/child'))
1381
1257
        self.assertEqual(1, transform.rename_count)
1382
1258
 
1383
1259
    def test_reuse_name(self):
1514
1390
        tt.create_file(["aa\n"], bar_trans_id)
1515
1391
        tt.version_file("bar-1", bar_trans_id)
1516
1392
        tt.apply()
1517
 
        self.assertPathExists("foo/bar")
 
1393
        self.failUnlessExists("foo/bar")
1518
1394
        wt.lock_read()
1519
1395
        try:
1520
1396
            self.assertEqual(wt.inventory.get_file_kind(wt.path2id("foo")),
1537
1413
        tt.delete_contents(foo_trans_id)
1538
1414
        tt.create_symlink("bar", foo_trans_id)
1539
1415
        tt.apply()
1540
 
        self.assertPathExists("foo")
 
1416
        self.failUnlessExists("foo")
1541
1417
        wt.lock_read()
1542
1418
        self.addCleanup(wt.unlock)
1543
1419
        self.assertEqual(wt.inventory.get_file_kind(wt.path2id("foo")),
1556
1432
        tt.delete_versioned(bar_trans_id)
1557
1433
        tt.create_file(["aa\n"], foo_trans_id)
1558
1434
        tt.apply()
1559
 
        self.assertPathExists("foo")
 
1435
        self.failUnlessExists("foo")
1560
1436
        wt.lock_read()
1561
1437
        self.addCleanup(wt.unlock)
1562
1438
        self.assertEqual(wt.inventory.get_file_kind(wt.path2id("foo")),
1577
1453
        self.build_tree(['baz'])
1578
1454
        tt.create_hardlink("baz", foo_trans_id)
1579
1455
        tt.apply()
1580
 
        self.assertPathExists("foo")
1581
 
        self.assertPathExists("baz")
 
1456
        self.failUnlessExists("foo")
 
1457
        self.failUnlessExists("baz")
1582
1458
        wt.lock_read()
1583
1459
        self.addCleanup(wt.unlock)
1584
1460
        self.assertEqual(wt.inventory.get_file_kind(wt.path2id("foo")),
1646
1522
    return template % ('<' * 7, tree, '=' * 7, merge, '>' * 7)
1647
1523
 
1648
1524
 
1649
 
class TestInventoryAltered(tests.TestCaseWithTransport):
1650
 
 
1651
 
    def test_inventory_altered_unchanged(self):
1652
 
        tree = self.make_branch_and_tree('tree')
1653
 
        self.build_tree(['tree/foo'])
1654
 
        tree.add('foo', 'foo-id')
1655
 
        with TransformPreview(tree) as tt:
1656
 
            self.assertEqual([], tt._inventory_altered())
1657
 
 
1658
 
    def test_inventory_altered_changed_parent_id(self):
1659
 
        tree = self.make_branch_and_tree('tree')
1660
 
        self.build_tree(['tree/foo'])
1661
 
        tree.add('foo', 'foo-id')
1662
 
        with TransformPreview(tree) as tt:
1663
 
            tt.unversion_file(tt.root)
1664
 
            tt.version_file('new-id', tt.root)
1665
 
            foo_trans_id = tt.trans_id_tree_file_id('foo-id')
1666
 
            foo_tuple = ('foo', foo_trans_id)
1667
 
            root_tuple = ('', tt.root)
1668
 
            self.assertEqual([root_tuple, foo_tuple], tt._inventory_altered())
1669
 
 
1670
 
    def test_inventory_altered_noop_changed_parent_id(self):
1671
 
        tree = self.make_branch_and_tree('tree')
1672
 
        self.build_tree(['tree/foo'])
1673
 
        tree.add('foo', 'foo-id')
1674
 
        with TransformPreview(tree) as tt:
1675
 
            tt.unversion_file(tt.root)
1676
 
            tt.version_file(tree.get_root_id(), tt.root)
1677
 
            foo_trans_id = tt.trans_id_tree_file_id('foo-id')
1678
 
            self.assertEqual([], tt._inventory_altered())
1679
 
 
1680
 
 
1681
1525
class TestTransformMerge(TestCaseInTempDir):
1682
1526
 
1683
1527
    def test_text_merge(self):
1893
1737
        tree.add_reference(subtree)
1894
1738
        tree.commit('a revision')
1895
1739
        tree.branch.create_checkout('target')
1896
 
        self.assertPathExists('target')
1897
 
        self.assertPathExists('target/subtree')
 
1740
        self.failUnlessExists('target')
 
1741
        self.failUnlessExists('target/subtree')
1898
1742
 
1899
1743
    def test_file_conflict_handling(self):
1900
1744
        """Ensure that when building trees, conflict handling is done"""
1947
1791
        source.commit('added file')
1948
1792
        build_tree(source.basis_tree(), target)
1949
1793
        self.assertEqual([], target.conflicts())
1950
 
        self.assertPathExists('target/dir1/file')
 
1794
        self.failUnlessExists('target/dir1/file')
1951
1795
 
1952
1796
        # Ensure contents are merged
1953
1797
        target = self.make_branch_and_tree('target2')
1954
1798
        self.build_tree(['target2/dir1/', 'target2/dir1/file2'])
1955
1799
        build_tree(source.basis_tree(), target)
1956
1800
        self.assertEqual([], target.conflicts())
1957
 
        self.assertPathExists('target2/dir1/file2')
1958
 
        self.assertPathExists('target2/dir1/file')
 
1801
        self.failUnlessExists('target2/dir1/file2')
 
1802
        self.failUnlessExists('target2/dir1/file')
1959
1803
 
1960
1804
        # Ensure new contents are suppressed for existing branches
1961
1805
        target = self.make_branch_and_tree('target3')
1962
1806
        self.make_branch('target3/dir1')
1963
1807
        self.build_tree(['target3/dir1/file2'])
1964
1808
        build_tree(source.basis_tree(), target)
1965
 
        self.assertPathDoesNotExist('target3/dir1/file')
1966
 
        self.assertPathExists('target3/dir1/file2')
1967
 
        self.assertPathExists('target3/dir1.diverted/file')
 
1809
        self.failIfExists('target3/dir1/file')
 
1810
        self.failUnlessExists('target3/dir1/file2')
 
1811
        self.failUnlessExists('target3/dir1.diverted/file')
1968
1812
        self.assertEqual([DuplicateEntry('Diverted to',
1969
1813
            'dir1.diverted', 'dir1', 'new-dir1', None)],
1970
1814
            target.conflicts())
1973
1817
        self.build_tree(['target4/dir1/'])
1974
1818
        self.make_branch('target4/dir1/file')
1975
1819
        build_tree(source.basis_tree(), target)
1976
 
        self.assertPathExists('target4/dir1/file')
 
1820
        self.failUnlessExists('target4/dir1/file')
1977
1821
        self.assertEqual('directory', file_kind('target4/dir1/file'))
1978
 
        self.assertPathExists('target4/dir1/file.diverted')
 
1822
        self.failUnlessExists('target4/dir1/file.diverted')
1979
1823
        self.assertEqual([DuplicateEntry('Diverted to',
1980
1824
            'dir1/file.diverted', 'dir1/file', 'new-file', None)],
1981
1825
            target.conflicts())
2049
1893
        self.addCleanup(target.unlock)
2050
1894
        self.assertEqual([], list(target.iter_changes(revision_tree)))
2051
1895
 
2052
 
    def test_build_tree_accelerator_tree_observes_sha1(self):
2053
 
        source = self.create_ab_tree()
2054
 
        sha1 = osutils.sha_string('A')
2055
 
        target = self.make_branch_and_tree('target')
2056
 
        target.lock_write()
2057
 
        self.addCleanup(target.unlock)
2058
 
        state = target.current_dirstate()
2059
 
        state._cutoff_time = time.time() + 60
2060
 
        build_tree(source.basis_tree(), target, source)
2061
 
        entry = state._get_entry(0, path_utf8='file1')
2062
 
        self.assertEqual(sha1, entry[1][0][1])
2063
 
 
2064
1896
    def test_build_tree_accelerator_tree_missing_file(self):
2065
1897
        source = self.create_ab_tree()
2066
1898
        os.unlink('source/file1')
2224
2056
        self.assertEqual('file.moved', target.id2path('lower-id'))
2225
2057
        self.assertEqual('FILE', target.id2path('upper-id'))
2226
2058
 
2227
 
    def test_build_tree_observes_sha(self):
2228
 
        source = self.make_branch_and_tree('source')
2229
 
        self.build_tree(['source/file1', 'source/dir/', 'source/dir/file2'])
2230
 
        source.add(['file1', 'dir', 'dir/file2'],
2231
 
                   ['file1-id', 'dir-id', 'file2-id'])
2232
 
        source.commit('new files')
2233
 
        target = self.make_branch_and_tree('target')
2234
 
        target.lock_write()
2235
 
        self.addCleanup(target.unlock)
2236
 
        # We make use of the fact that DirState caches its cutoff time. So we
2237
 
        # set the 'safe' time to one minute in the future.
2238
 
        state = target.current_dirstate()
2239
 
        state._cutoff_time = time.time() + 60
2240
 
        build_tree(source.basis_tree(), target)
2241
 
        entry1_sha = osutils.sha_file_by_name('source/file1')
2242
 
        entry2_sha = osutils.sha_file_by_name('source/dir/file2')
2243
 
        # entry[1] is the state information, entry[1][0] is the state of the
2244
 
        # working tree, entry[1][0][1] is the sha value for the current working
2245
 
        # tree
2246
 
        entry1 = state._get_entry(0, path_utf8='file1')
2247
 
        self.assertEqual(entry1_sha, entry1[1][0][1])
2248
 
        # The 'size' field must also be set.
2249
 
        self.assertEqual(25, entry1[1][0][2])
2250
 
        entry1_state = entry1[1][0]
2251
 
        entry2 = state._get_entry(0, path_utf8='dir/file2')
2252
 
        self.assertEqual(entry2_sha, entry2[1][0][1])
2253
 
        self.assertEqual(29, entry2[1][0][2])
2254
 
        entry2_state = entry2[1][0]
2255
 
        # Now, make sure that we don't have to re-read the content. The
2256
 
        # packed_stat should match exactly.
2257
 
        self.assertEqual(entry1_sha, target.get_file_sha1('file1-id', 'file1'))
2258
 
        self.assertEqual(entry2_sha,
2259
 
                         target.get_file_sha1('file2-id', 'dir/file2'))
2260
 
        self.assertEqual(entry1_state, entry1[1][0])
2261
 
        self.assertEqual(entry2_state, entry2[1][0])
2262
 
 
2263
2059
 
2264
2060
class TestCommitTransform(tests.TestCaseWithTransport):
2265
2061
 
2430
2226
        self.build_tree(['a/', 'a/b', 'c/', 'c/d'])
2431
2227
        mover = _FileMover()
2432
2228
        mover.rename('a', 'q')
2433
 
        self.assertPathExists('q')
2434
 
        self.assertPathDoesNotExist('a')
2435
 
        self.assertPathExists('q/b')
2436
 
        self.assertPathExists('c')
2437
 
        self.assertPathExists('c/d')
 
2229
        self.failUnlessExists('q')
 
2230
        self.failIfExists('a')
 
2231
        self.failUnlessExists('q/b')
 
2232
        self.failUnlessExists('c')
 
2233
        self.failUnlessExists('c/d')
2438
2234
 
2439
2235
    def test_pre_delete_rollback(self):
2440
2236
        self.build_tree(['a/'])
2441
2237
        mover = _FileMover()
2442
2238
        mover.pre_delete('a', 'q')
2443
 
        self.assertPathExists('q')
2444
 
        self.assertPathDoesNotExist('a')
 
2239
        self.failUnlessExists('q')
 
2240
        self.failIfExists('a')
2445
2241
        mover.rollback()
2446
 
        self.assertPathDoesNotExist('q')
2447
 
        self.assertPathExists('a')
 
2242
        self.failIfExists('q')
 
2243
        self.failUnlessExists('a')
2448
2244
 
2449
2245
    def test_apply_deletions(self):
2450
2246
        self.build_tree(['a/', 'b/'])
2451
2247
        mover = _FileMover()
2452
2248
        mover.pre_delete('a', 'q')
2453
2249
        mover.pre_delete('b', 'r')
2454
 
        self.assertPathExists('q')
2455
 
        self.assertPathExists('r')
2456
 
        self.assertPathDoesNotExist('a')
2457
 
        self.assertPathDoesNotExist('b')
 
2250
        self.failUnlessExists('q')
 
2251
        self.failUnlessExists('r')
 
2252
        self.failIfExists('a')
 
2253
        self.failIfExists('b')
2458
2254
        mover.apply_deletions()
2459
 
        self.assertPathDoesNotExist('q')
2460
 
        self.assertPathDoesNotExist('r')
2461
 
        self.assertPathDoesNotExist('a')
2462
 
        self.assertPathDoesNotExist('b')
 
2255
        self.failIfExists('q')
 
2256
        self.failIfExists('r')
 
2257
        self.failIfExists('a')
 
2258
        self.failIfExists('b')
2463
2259
 
2464
2260
    def test_file_mover_rollback(self):
2465
2261
        self.build_tree(['a/', 'a/b', 'c/', 'c/d/', 'c/e/'])
2470
2266
            mover.rename('a', 'c')
2471
2267
        except errors.FileExists, e:
2472
2268
            mover.rollback()
2473
 
        self.assertPathExists('a')
2474
 
        self.assertPathExists('c/d')
 
2269
        self.failUnlessExists('a')
 
2270
        self.failUnlessExists('c/d')
2475
2271
 
2476
2272
 
2477
2273
class Bogus(Exception):
2507
2303
        tt.adjust_path('d', a_id, tt.trans_id_tree_path('a/b'))
2508
2304
        self.assertRaises(Bogus, tt.apply,
2509
2305
                          _mover=self.ExceptionFileMover(bad_source='a'))
2510
 
        self.assertPathExists('a')
2511
 
        self.assertPathExists('a/b')
 
2306
        self.failUnlessExists('a')
 
2307
        self.failUnlessExists('a/b')
2512
2308
        tt.apply()
2513
 
        self.assertPathExists('c')
2514
 
        self.assertPathExists('c/d')
 
2309
        self.failUnlessExists('c')
 
2310
        self.failUnlessExists('c/d')
2515
2311
 
2516
2312
    def test_rollback_rename_into_place(self):
2517
2313
        tree = self.make_branch_and_tree('.')
2523
2319
        tt.adjust_path('d', a_id, tt.trans_id_tree_path('a/b'))
2524
2320
        self.assertRaises(Bogus, tt.apply,
2525
2321
                          _mover=self.ExceptionFileMover(bad_target='c/d'))
2526
 
        self.assertPathExists('a')
2527
 
        self.assertPathExists('a/b')
 
2322
        self.failUnlessExists('a')
 
2323
        self.failUnlessExists('a/b')
2528
2324
        tt.apply()
2529
 
        self.assertPathExists('c')
2530
 
        self.assertPathExists('c/d')
 
2325
        self.failUnlessExists('c')
 
2326
        self.failUnlessExists('c/d')
2531
2327
 
2532
2328
    def test_rollback_deletion(self):
2533
2329
        tree = self.make_branch_and_tree('.')
2539
2335
        tt.adjust_path('d', tt.root, tt.trans_id_tree_path('a/b'))
2540
2336
        self.assertRaises(Bogus, tt.apply,
2541
2337
                          _mover=self.ExceptionFileMover(bad_target='d'))
2542
 
        self.assertPathExists('a')
2543
 
        self.assertPathExists('a/b')
2544
 
 
2545
 
 
2546
 
class TestFinalizeRobustness(tests.TestCaseWithTransport):
2547
 
    """Ensure treetransform creation errors can be safely cleaned up after"""
2548
 
 
2549
 
    def _override_globals_in_method(self, instance, method_name, globals):
2550
 
        """Replace method on instance with one with updated globals"""
2551
 
        import types
2552
 
        func = getattr(instance, method_name).im_func
2553
 
        new_globals = dict(func.func_globals)
2554
 
        new_globals.update(globals)
2555
 
        new_func = types.FunctionType(func.func_code, new_globals,
2556
 
            func.func_name, func.func_defaults)
2557
 
        setattr(instance, method_name,
2558
 
            types.MethodType(new_func, instance, instance.__class__))
2559
 
        self.addCleanup(delattr, instance, method_name)
2560
 
 
2561
 
    @staticmethod
2562
 
    def _fake_open_raises_before(name, mode):
2563
 
        """Like open() but raises before doing anything"""
2564
 
        raise RuntimeError
2565
 
 
2566
 
    @staticmethod
2567
 
    def _fake_open_raises_after(name, mode):
2568
 
        """Like open() but raises after creating file without returning"""
2569
 
        open(name, mode).close()
2570
 
        raise RuntimeError
2571
 
 
2572
 
    def create_transform_and_root_trans_id(self):
2573
 
        """Setup a transform creating a file in limbo"""
2574
 
        tree = self.make_branch_and_tree('.')
2575
 
        tt = TreeTransform(tree)
2576
 
        return tt, tt.create_path("a", tt.root)
2577
 
 
2578
 
    def create_transform_and_subdir_trans_id(self):
2579
 
        """Setup a transform creating a directory containing a file in limbo"""
2580
 
        tree = self.make_branch_and_tree('.')
2581
 
        tt = TreeTransform(tree)
2582
 
        d_trans_id = tt.create_path("d", tt.root)
2583
 
        tt.create_directory(d_trans_id)
2584
 
        f_trans_id = tt.create_path("a", d_trans_id)
2585
 
        tt.adjust_path("a", d_trans_id, f_trans_id)
2586
 
        return tt, f_trans_id
2587
 
 
2588
 
    def test_root_create_file_open_raises_before_creation(self):
2589
 
        tt, trans_id = self.create_transform_and_root_trans_id()
2590
 
        self._override_globals_in_method(tt, "create_file",
2591
 
            {"open": self._fake_open_raises_before})
2592
 
        self.assertRaises(RuntimeError, tt.create_file, ["contents"], trans_id)
2593
 
        path = tt._limbo_name(trans_id)
2594
 
        self.assertPathDoesNotExist(path)
2595
 
        tt.finalize()
2596
 
        self.assertPathDoesNotExist(tt._limbodir)
2597
 
 
2598
 
    def test_root_create_file_open_raises_after_creation(self):
2599
 
        tt, trans_id = self.create_transform_and_root_trans_id()
2600
 
        self._override_globals_in_method(tt, "create_file",
2601
 
            {"open": self._fake_open_raises_after})
2602
 
        self.assertRaises(RuntimeError, tt.create_file, ["contents"], trans_id)
2603
 
        path = tt._limbo_name(trans_id)
2604
 
        self.assertPathExists(path)
2605
 
        tt.finalize()
2606
 
        self.assertPathDoesNotExist(path)
2607
 
        self.assertPathDoesNotExist(tt._limbodir)
2608
 
 
2609
 
    def test_subdir_create_file_open_raises_before_creation(self):
2610
 
        tt, trans_id = self.create_transform_and_subdir_trans_id()
2611
 
        self._override_globals_in_method(tt, "create_file",
2612
 
            {"open": self._fake_open_raises_before})
2613
 
        self.assertRaises(RuntimeError, tt.create_file, ["contents"], trans_id)
2614
 
        path = tt._limbo_name(trans_id)
2615
 
        self.assertPathDoesNotExist(path)
2616
 
        tt.finalize()
2617
 
        self.assertPathDoesNotExist(tt._limbodir)
2618
 
 
2619
 
    def test_subdir_create_file_open_raises_after_creation(self):
2620
 
        tt, trans_id = self.create_transform_and_subdir_trans_id()
2621
 
        self._override_globals_in_method(tt, "create_file",
2622
 
            {"open": self._fake_open_raises_after})
2623
 
        self.assertRaises(RuntimeError, tt.create_file, ["contents"], trans_id)
2624
 
        path = tt._limbo_name(trans_id)
2625
 
        self.assertPathExists(path)
2626
 
        tt.finalize()
2627
 
        self.assertPathDoesNotExist(path)
2628
 
        self.assertPathDoesNotExist(tt._limbodir)
2629
 
 
2630
 
    def test_rename_in_limbo_rename_raises_after_rename(self):
2631
 
        tt, trans_id = self.create_transform_and_root_trans_id()
2632
 
        parent1 = tt.new_directory('parent1', tt.root)
2633
 
        child1 = tt.new_file('child1', parent1, 'contents')
2634
 
        parent2 = tt.new_directory('parent2', tt.root)
2635
 
 
2636
 
        class FakeOSModule(object):
2637
 
            def rename(self, old, new):
2638
 
                os.rename(old, new)
2639
 
                raise RuntimeError
2640
 
        self._override_globals_in_method(tt, "_rename_in_limbo",
2641
 
            {"os": FakeOSModule()})
2642
 
        self.assertRaises(
2643
 
            RuntimeError, tt.adjust_path, "child1", parent2, child1)
2644
 
        path = osutils.pathjoin(tt._limbo_name(parent2), "child1")
2645
 
        self.assertPathExists(path)
2646
 
        tt.finalize()
2647
 
        self.assertPathDoesNotExist(path)
2648
 
        self.assertPathDoesNotExist(tt._limbodir)
2649
 
 
2650
 
    def test_rename_in_limbo_rename_raises_before_rename(self):
2651
 
        tt, trans_id = self.create_transform_and_root_trans_id()
2652
 
        parent1 = tt.new_directory('parent1', tt.root)
2653
 
        child1 = tt.new_file('child1', parent1, 'contents')
2654
 
        parent2 = tt.new_directory('parent2', tt.root)
2655
 
 
2656
 
        class FakeOSModule(object):
2657
 
            def rename(self, old, new):
2658
 
                raise RuntimeError
2659
 
        self._override_globals_in_method(tt, "_rename_in_limbo",
2660
 
            {"os": FakeOSModule()})
2661
 
        self.assertRaises(
2662
 
            RuntimeError, tt.adjust_path, "child1", parent2, child1)
2663
 
        path = osutils.pathjoin(tt._limbo_name(parent1), "child1")
2664
 
        self.assertPathExists(path)
2665
 
        tt.finalize()
2666
 
        self.assertPathDoesNotExist(path)
2667
 
        self.assertPathDoesNotExist(tt._limbodir)
 
2338
        self.failUnlessExists('a')
 
2339
        self.failUnlessExists('a/b')
2668
2340
 
2669
2341
 
2670
2342
class TestTransformMissingParent(tests.TestCaseWithTransport):
2854
2526
        preview_mtime = preview_tree.get_file_mtime('file-id', 'renamed')
2855
2527
        work_mtime = work_tree.get_file_mtime('file-id', 'file')
2856
2528
 
2857
 
    def test_get_file_size(self):
2858
 
        work_tree = self.make_branch_and_tree('tree')
2859
 
        self.build_tree_contents([('tree/old', 'old')])
2860
 
        work_tree.add('old', 'old-id')
2861
 
        preview = TransformPreview(work_tree)
2862
 
        self.addCleanup(preview.finalize)
2863
 
        new_id = preview.new_file('name', preview.root, 'contents', 'new-id',
2864
 
                                  'executable')
2865
 
        tree = preview.get_preview_tree()
2866
 
        self.assertEqual(len('old'), tree.get_file_size('old-id'))
2867
 
        self.assertEqual(len('contents'), tree.get_file_size('new-id'))
2868
 
 
2869
2529
    def test_get_file(self):
2870
2530
        preview = self.get_empty_preview()
2871
2531
        preview.new_file('file', preview.root, 'contents', 'file-id')
3283
2943
        merger.merge_type = Merge3Merger
3284
2944
        merger.do_merge()
3285
2945
 
3286
 
    def test_has_filename(self):
3287
 
        wt = self.make_branch_and_tree('tree')
3288
 
        self.build_tree(['tree/unmodified', 'tree/removed', 'tree/modified'])
3289
 
        tt = TransformPreview(wt)
3290
 
        removed_id = tt.trans_id_tree_path('removed')
3291
 
        tt.delete_contents(removed_id)
3292
 
        tt.new_file('new', tt.root, 'contents')
3293
 
        modified_id = tt.trans_id_tree_path('modified')
3294
 
        tt.delete_contents(modified_id)
3295
 
        tt.create_file('modified-contents', modified_id)
3296
 
        self.addCleanup(tt.finalize)
3297
 
        tree = tt.get_preview_tree()
3298
 
        self.assertTrue(tree.has_filename('unmodified'))
3299
 
        self.assertFalse(tree.has_filename('not-present'))
3300
 
        self.assertFalse(tree.has_filename('removed'))
3301
 
        self.assertTrue(tree.has_filename('new'))
3302
 
        self.assertTrue(tree.has_filename('modified'))
3303
 
 
3304
2946
    def test_is_executable(self):
3305
2947
        tree = self.make_branch_and_tree('tree')
3306
2948
        preview = TransformPreview(tree)