← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_inv.py

  • Committer: INADA Naoki
  • Date: 2011-05-17 00:45:09 UTC
  • mfrom: (5875 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5891.
  • Revision ID: songofacandy@gmail.com-20110517004509-q58negjbdjh7t6u1
merge from lp:bzr

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_inv.py
85
85
        repo.texts.add_lines((ie.file_id, ie.revision), [], lines)
86
86
 
87
87
 
88
 
def apply_inventory_Inventory(self, basis, delta, invalid_delta=True):
 
88
def apply_inventory_Inventory(self, basis, delta):
89
89
    """Apply delta to basis and return the result.
90
 
 
 
90
    
91
91
    :param basis: An inventory to be used as the basis.
92
92
    :param delta: The inventory delta to apply:
93
93
    :return: An inventory resulting from the application.
96
96
    return basis
97
97
 
98
98
 
99
 
def apply_inventory_WT(self, basis, delta, invalid_delta=True):
 
99
def apply_inventory_WT(self, basis, delta):
100
100
    """Apply delta to basis and return the result.
101
101
 
102
102
    This sets the tree state to be basis, and then calls apply_inventory_delta.
103
 
 
 
103
    
104
104
    :param basis: An inventory to be used as the basis.
105
105
    :param delta: The inventory delta to apply:
106
106
    :return: An inventory resulting from the application.
125
125
    tree = tree.bzrdir.open_workingtree()
126
126
    tree.lock_read()
127
127
    self.addCleanup(tree.unlock)
128
 
    if not invalid_delta:
129
 
        tree._validate()
 
128
    # One could add 'tree._validate' here but that would cause 'early' failues 
 
129
    # as far as higher level code is concerned. Possibly adding an
 
130
    # expect_fail parameter to this function and if that is False then do a
 
131
    # validate call.
130
132
    return tree.inventory
131
133
 
132
134
 
133
 
def _create_repo_revisions(repo, basis, delta, invalid_delta):
134
 
    repo.start_write_group()
135
 
    try:
136
 
        rev = revision.Revision('basis', timestamp=0, timezone=None,
137
 
            message="", committer="foo@example.com")
138
 
        basis.revision_id = 'basis'
139
 
        create_texts_for_inv(repo, basis)
140
 
        repo.add_revision('basis', rev, basis)
141
 
        if invalid_delta:
142
 
            # We don't want to apply the delta to the basis, because we expect
143
 
            # the delta is invalid.
144
 
            result_inv = basis
145
 
            result_inv.revision_id = 'result'
146
 
            target_entries = None
147
 
        else:
148
 
            result_inv = basis.create_by_apply_delta(delta, 'result')
149
 
            create_texts_for_inv(repo, result_inv)
150
 
            target_entries = list(result_inv.iter_entries_by_dir())
151
 
        rev = revision.Revision('result', timestamp=0, timezone=None,
152
 
            message="", committer="foo@example.com")
153
 
        repo.add_revision('result', rev, result_inv)
154
 
        repo.commit_write_group()
155
 
    except:
156
 
        repo.abort_write_group()
157
 
        raise
158
 
    return target_entries
159
 
 
160
 
 
161
 
def _get_basis_entries(tree):
162
 
    basis_tree = tree.basis_tree()
163
 
    basis_tree.lock_read()
164
 
    basis_tree_entries = list(basis_tree.inventory.iter_entries_by_dir())
165
 
    basis_tree.unlock()
166
 
    return basis_tree_entries
167
 
 
168
 
 
169
 
def _populate_different_tree(tree, basis, delta):
170
 
    """Put all entries into tree, but at a unique location."""
171
 
    added_ids = set()
172
 
    added_paths = set()
173
 
    tree.add(['unique-dir'], ['unique-dir-id'], ['directory'])
174
 
    for path, ie in basis.iter_entries_by_dir():
175
 
        if ie.file_id in added_ids:
176
 
            continue
177
 
        # We want a unique path for each of these, we use the file-id
178
 
        tree.add(['unique-dir/' + ie.file_id], [ie.file_id], [ie.kind])
179
 
        added_ids.add(ie.file_id)
180
 
    for old_path, new_path, file_id, ie in delta:
181
 
        if file_id in added_ids:
182
 
            continue
183
 
        tree.add(['unique-dir/' + file_id], [file_id], [ie.kind])
184
 
 
185
 
 
186
 
def apply_inventory_WT_basis(test, basis, delta, invalid_delta=True):
 
135
def apply_inventory_WT_basis(self, basis, delta):
187
136
    """Apply delta to basis and return the result.
188
137
 
189
138
    This sets the parent and then calls update_basis_by_delta.
191
140
    allow safety checks made by the WT to succeed, and finally ensures that all
192
141
    items in the delta with a new path are present in the WT before calling
193
142
    update_basis_by_delta.
194
 
 
 
143
    
195
144
    :param basis: An inventory to be used as the basis.
196
145
    :param delta: The inventory delta to apply:
197
146
    :return: An inventory resulting from the application.
198
147
    """
199
 
    control = test.make_bzrdir('tree', format=test.format._matchingbzrdir)
 
148
    control = self.make_bzrdir('tree', format=self.format._matchingbzrdir)
200
149
    control.create_repository()
201
150
    control.create_branch()
202
 
    tree = test.format.initialize(control)
 
151
    tree = self.format.initialize(control)
203
152
    tree.lock_write()
204
153
    try:
205
 
        target_entries = _create_repo_revisions(tree.branch.repository, basis,
206
 
                                                delta, invalid_delta)
 
154
        repo = tree.branch.repository
 
155
        repo.start_write_group()
 
156
        try:
 
157
            rev = revision.Revision('basis', timestamp=0, timezone=None,
 
158
                message="", committer="foo@example.com")
 
159
            basis.revision_id = 'basis'
 
160
            create_texts_for_inv(tree.branch.repository, basis)
 
161
            repo.add_revision('basis', rev, basis)
 
162
            # Add a revision for the result, with the basis content - 
 
163
            # update_basis_by_delta doesn't check that the delta results in
 
164
            # result, and we want inconsistent deltas to get called on the
 
165
            # tree, or else the code isn't actually checked.
 
166
            rev = revision.Revision('result', timestamp=0, timezone=None,
 
167
                message="", committer="foo@example.com")
 
168
            basis.revision_id = 'result'
 
169
            repo.add_revision('result', rev, basis)
 
170
            repo.commit_write_group()
 
171
        except:
 
172
            repo.abort_write_group()
 
173
            raise
207
174
        # Set the basis state as the trees current state
208
175
        tree._write_inventory(basis)
209
176
        # This reads basis from the repo and puts it into the tree's local
210
177
        # cache, if it has one.
211
178
        tree.set_parent_ids(['basis'])
 
179
        paths = {}
 
180
        parents = set()
 
181
        for old, new, id, entry in delta:
 
182
            if None in (new, entry):
 
183
                continue
 
184
            paths[new] = (entry.file_id, entry.kind)
 
185
            parents.add(osutils.dirname(new))
 
186
        parents = osutils.minimum_path_selection(parents)
 
187
        parents.discard('')
 
188
        # Put place holders in the tree to permit adding the other entries.
 
189
        for pos, parent in enumerate(parents):
 
190
            if not tree.path2id(parent):
 
191
                # add a synthetic directory in the tree so we can can put the
 
192
                # tree0 entries in place for dirstate.
 
193
                tree.add([parent], ["id%d" % pos], ["directory"])
 
194
        if paths:
 
195
            # Many deltas may cause this mini-apply to fail, but we want to see what
 
196
            # the delta application code says, not the prep that we do to deal with 
 
197
            # limitations of dirstate's update_basis code.
 
198
            for path, (file_id, kind) in sorted(paths.items()):
 
199
                try:
 
200
                    tree.add([path], [file_id], [kind])
 
201
                except (KeyboardInterrupt, SystemExit):
 
202
                    raise
 
203
                except:
 
204
                    pass
212
205
    finally:
213
206
        tree.unlock()
214
207
    # Fresh lock, reads disk again.
215
208
    tree.lock_write()
216
209
    try:
217
210
        tree.update_basis_by_delta('result', delta)
218
 
        if not invalid_delta:
219
 
            tree._validate()
220
211
    finally:
221
212
        tree.unlock()
222
213
    # reload tree - ensure we get what was written.
223
214
    tree = tree.bzrdir.open_workingtree()
224
215
    basis_tree = tree.basis_tree()
225
216
    basis_tree.lock_read()
226
 
    test.addCleanup(basis_tree.unlock)
227
 
    basis_inv = basis_tree.inventory
228
 
    if target_entries:
229
 
        basis_entries = list(basis_inv.iter_entries_by_dir())
230
 
        test.assertEqual(target_entries, basis_entries)
231
 
    return basis_inv
232
 
 
233
 
 
234
 
def apply_inventory_Repository_add_inventory_by_delta(self, basis, delta,
235
 
                                                      invalid_delta=True):
 
217
    self.addCleanup(basis_tree.unlock)
 
218
    # Note, that if the tree does not have a local cache, the trick above of
 
219
    # setting the result as the basis, will come back to bite us. That said,
 
220
    # all the implementations in bzr do have a local cache.
 
221
    return basis_tree.inventory
 
222
 
 
223
 
 
224
def apply_inventory_Repository_add_inventory_by_delta(self, basis, delta):
236
225
    """Apply delta to basis and return the result.
237
226
    
238
227
    This inserts basis as a whole inventory and then uses
373
362
            inv.root.revision = 'basis'
374
363
        return inv
375
364
 
376
 
    def make_file_ie(self, file_id='file-id', name='name', parent_id=None):
377
 
        ie_file = inventory.InventoryFile(file_id, name, parent_id)
378
 
        ie_file.revision = 'result'
379
 
        ie_file.text_size = 0
380
 
        ie_file.text_sha1 = ''
381
 
        return ie_file
382
 
 
383
365
    def test_empty_delta(self):
384
366
        inv = self.get_empty_inventory()
385
367
        delta = []
409
391
        file1.revision = 'result'
410
392
        file1.text_size = 0
411
393
        file1.text_sha1 = ""
412
 
        file2 = file1.copy()
413
 
        file2.name = 'path2'
 
394
        file2 = inventory.InventoryFile('id', 'path2', inv.root.file_id)
 
395
        file2.revision = 'result'
 
396
        file2.text_size = 0
 
397
        file2.text_sha1 = ""
414
398
        delta = [(None, u'path1', 'id', file1), (None, u'path2', 'id', file2)]
415
399
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
416
400
            inv, delta)
421
405
        file1.revision = 'result'
422
406
        file1.text_size = 0
423
407
        file1.text_sha1 = ""
424
 
        file2 = file1.copy()
425
 
        file2.file_id = 'id2'
 
408
        file2 = inventory.InventoryFile('id2', 'path', inv.root.file_id)
 
409
        file2.revision = 'result'
 
410
        file2.text_size = 0
 
411
        file2.text_sha1 = ""
426
412
        delta = [(None, u'path', 'id1', file1), (None, u'path', 'id2', file2)]
427
413
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
428
414
            inv, delta)
600
586
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
601
587
            inv, delta)
602
588
 
603
 
    def test_add_file(self):
604
 
        inv = self.get_empty_inventory()
605
 
        file1 = inventory.InventoryFile('file-id', 'path', inv.root.file_id)
606
 
        file1.revision = 'result'
607
 
        file1.text_size = 0
608
 
        file1.text_sha1 = ''
609
 
        delta = [(None, u'path', 'file-id', file1)]
610
 
        res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
611
 
        self.assertEqual('file-id', res_inv['file-id'].file_id)
612
 
 
613
 
    def test_remove_file(self):
614
 
        inv = self.get_empty_inventory()
615
 
        file1 = inventory.InventoryFile('file-id', 'path', inv.root.file_id)
616
 
        file1.revision = 'result'
617
 
        file1.text_size = 0
618
 
        file1.text_sha1 = ''
619
 
        inv.add(file1)
620
 
        delta = [(u'path', None, 'file-id', None)]
621
 
        res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
622
 
        self.assertEqual(None, res_inv.path2id('path'))
623
 
        self.assertRaises(errors.NoSuchId, res_inv.id2path, 'file-id')
624
 
 
625
 
    def test_rename_file(self):
626
 
        inv = self.get_empty_inventory()
627
 
        file1 = self.make_file_ie(name='path', parent_id=inv.root.file_id)
628
 
        inv.add(file1)
629
 
        file2 = self.make_file_ie(name='path2', parent_id=inv.root.file_id)
630
 
        delta = [(u'path', 'path2', 'file-id', file2)]
631
 
        res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
632
 
        self.assertEqual(None, res_inv.path2id('path'))
633
 
        self.assertEqual('file-id', res_inv.path2id('path2'))
634
 
 
635
 
    def test_replaced_at_new_path(self):
636
 
        inv = self.get_empty_inventory()
637
 
        file1 = self.make_file_ie(file_id='id1', parent_id=inv.root.file_id)
638
 
        inv.add(file1)
639
 
        file2 = self.make_file_ie(file_id='id2', parent_id=inv.root.file_id)
640
 
        delta = [(u'name', None, 'id1', None),
641
 
                 (None, u'name', 'id2', file2)]
642
 
        res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
643
 
        self.assertEqual('id2', res_inv.path2id('name'))
644
 
 
645
 
    def test_rename_dir(self):
646
 
        inv = self.get_empty_inventory()
647
 
        dir1 = inventory.InventoryDirectory('dir-id', 'dir1', inv.root.file_id)
648
 
        dir1.revision = 'basis'
649
 
        file1 = self.make_file_ie(parent_id='dir-id')
650
 
        inv.add(dir1)
651
 
        inv.add(file1)
652
 
        dir2 = inventory.InventoryDirectory('dir-id', 'dir2', inv.root.file_id)
653
 
        dir2.revision = 'result'
654
 
        delta = [('dir1', 'dir2', 'dir-id', dir2)]
655
 
        res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
656
 
        # The file should be accessible under the new path
657
 
        self.assertEqual('file-id', res_inv.path2id('dir2/name'))
658
 
 
659
 
    def test_renamed_dir_with_renamed_child(self):
660
 
        inv = self.get_empty_inventory()
661
 
        dir1 = inventory.InventoryDirectory('dir-id', 'dir1', inv.root.file_id)
662
 
        dir1.revision = 'basis'
663
 
        file1 = self.make_file_ie('file-id-1', 'name1', parent_id='dir-id')
664
 
        file2 = self.make_file_ie('file-id-2', 'name2', parent_id='dir-id')
665
 
        inv.add(dir1)
666
 
        inv.add(file1)
667
 
        inv.add(file2)
668
 
        dir2 = inventory.InventoryDirectory('dir-id', 'dir2', inv.root.file_id)
669
 
        dir2.revision = 'result'
670
 
        file2b = self.make_file_ie('file-id-2', 'name2', inv.root.file_id)
671
 
        delta = [('dir1', 'dir2', 'dir-id', dir2),
672
 
                 ('dir1/name2', 'name2', 'file-id-2', file2b)]
673
 
        res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
674
 
        # The file should be accessible under the new path
675
 
        self.assertEqual('file-id-1', res_inv.path2id('dir2/name1'))
676
 
        self.assertEqual(None, res_inv.path2id('dir2/name2'))
677
 
        self.assertEqual('file-id-2', res_inv.path2id('name2'))
 
589
 
 
590
class TestInventory(TestCase):
678
591
 
679
592
    def test_is_root(self):
680
593
        """Ensure our root-checking code is accurate."""