← 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: Canonical.com Patch Queue Manager
  • Date: 2010-09-01 08:02:42 UTC
  • mfrom: (5390.3.3 faster-revert-593560)
  • Revision ID: pqm@pqm.ubuntu.com-20100901080242-esg62ody4frwmy66
(spiv) Avoid repeatedly calling self.target.all_file_ids() in
 InterTree.iter_changes. (Andrew Bennetts)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_inv.py
1
 
# Copyright (C) 2005, 2006, 2007 Canonical Ltd
 
1
# Copyright (C) 2005-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
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
 
18
 
from bzrlib import errors, chk_map, inventory, osutils
 
18
from bzrlib import (
 
19
    chk_map,
 
20
    groupcompress,
 
21
    errors,
 
22
    inventory,
 
23
    osutils,
 
24
    repository,
 
25
    revision,
 
26
    tests,
 
27
    )
19
28
from bzrlib.inventory import (CHKInventory, Inventory, ROOT_ID, InventoryFile,
20
29
    InventoryDirectory, InventoryEntry, TreeReference)
21
 
from bzrlib.tests import TestCase, TestCaseWithTransport
 
30
from bzrlib.tests import (
 
31
    TestCase,
 
32
    TestCaseWithTransport,
 
33
    condition_isinstance,
 
34
    multiply_tests,
 
35
    split_suite_by_condition,
 
36
    )
 
37
from bzrlib.tests.per_workingtree import workingtree_formats
 
38
 
 
39
 
 
40
def load_tests(standard_tests, module, loader):
 
41
    """Parameterise some inventory tests."""
 
42
    to_adapt, result = split_suite_by_condition(standard_tests,
 
43
        condition_isinstance(TestDeltaApplication))
 
44
    scenarios = [
 
45
        ('Inventory', {'apply_delta':apply_inventory_Inventory}),
 
46
        ]
 
47
    # Working tree basis delta application
 
48
    # Repository add_inv_by_delta.
 
49
    # Reduce form of the per_repository test logic - that logic needs to be
 
50
    # be able to get /just/ repositories whereas these tests are fine with
 
51
    # just creating trees.
 
52
    formats = set()
 
53
    for _, format in repository.format_registry.iteritems():
 
54
        scenarios.append((str(format.__name__), {
 
55
            'apply_delta':apply_inventory_Repository_add_inventory_by_delta,
 
56
            'format':format}))
 
57
    for format in workingtree_formats():
 
58
        scenarios.append(
 
59
            (str(format.__class__.__name__) + ".update_basis_by_delta", {
 
60
            'apply_delta':apply_inventory_WT_basis,
 
61
            'format':format}))
 
62
        scenarios.append(
 
63
            (str(format.__class__.__name__) + ".apply_inventory_delta", {
 
64
            'apply_delta':apply_inventory_WT,
 
65
            'format':format}))
 
66
    return multiply_tests(to_adapt, scenarios, result)
 
67
 
 
68
 
 
69
def create_texts_for_inv(repo, inv):
 
70
    for path, ie in inv.iter_entries():
 
71
        if ie.text_size:
 
72
            lines = ['a' * ie.text_size]
 
73
        else:
 
74
            lines = []
 
75
        repo.texts.add_lines((ie.file_id, ie.revision), [], lines)
 
76
    
 
77
def apply_inventory_Inventory(self, basis, delta):
 
78
    """Apply delta to basis and return the result.
 
79
    
 
80
    :param basis: An inventory to be used as the basis.
 
81
    :param delta: The inventory delta to apply:
 
82
    :return: An inventory resulting from the application.
 
83
    """
 
84
    basis.apply_delta(delta)
 
85
    return basis
 
86
 
 
87
 
 
88
def apply_inventory_WT(self, basis, delta):
 
89
    """Apply delta to basis and return the result.
 
90
 
 
91
    This sets the tree state to be basis, and then calls apply_inventory_delta.
 
92
    
 
93
    :param basis: An inventory to be used as the basis.
 
94
    :param delta: The inventory delta to apply:
 
95
    :return: An inventory resulting from the application.
 
96
    """
 
97
    control = self.make_bzrdir('tree', format=self.format._matchingbzrdir)
 
98
    control.create_repository()
 
99
    control.create_branch()
 
100
    tree = self.format.initialize(control)
 
101
    tree.lock_write()
 
102
    try:
 
103
        tree._write_inventory(basis)
 
104
    finally:
 
105
        tree.unlock()
 
106
    # Fresh object, reads disk again.
 
107
    tree = tree.bzrdir.open_workingtree()
 
108
    tree.lock_write()
 
109
    try:
 
110
        tree.apply_inventory_delta(delta)
 
111
    finally:
 
112
        tree.unlock()
 
113
    # reload tree - ensure we get what was written.
 
114
    tree = tree.bzrdir.open_workingtree()
 
115
    tree.lock_read()
 
116
    self.addCleanup(tree.unlock)
 
117
    # One could add 'tree._validate' here but that would cause 'early' failues 
 
118
    # as far as higher level code is concerned. Possibly adding an
 
119
    # expect_fail parameter to this function and if that is False then do a
 
120
    # validate call.
 
121
    return tree.inventory
 
122
 
 
123
 
 
124
def apply_inventory_WT_basis(self, basis, delta):
 
125
    """Apply delta to basis and return the result.
 
126
 
 
127
    This sets the parent and then calls update_basis_by_delta.
 
128
    It also puts the basis in the repository under both 'basis' and 'result' to
 
129
    allow safety checks made by the WT to succeed, and finally ensures that all
 
130
    items in the delta with a new path are present in the WT before calling
 
131
    update_basis_by_delta.
 
132
    
 
133
    :param basis: An inventory to be used as the basis.
 
134
    :param delta: The inventory delta to apply:
 
135
    :return: An inventory resulting from the application.
 
136
    """
 
137
    control = self.make_bzrdir('tree', format=self.format._matchingbzrdir)
 
138
    control.create_repository()
 
139
    control.create_branch()
 
140
    tree = self.format.initialize(control)
 
141
    tree.lock_write()
 
142
    try:
 
143
        repo = tree.branch.repository
 
144
        repo.start_write_group()
 
145
        try:
 
146
            rev = revision.Revision('basis', timestamp=0, timezone=None,
 
147
                message="", committer="foo@example.com")
 
148
            basis.revision_id = 'basis'
 
149
            create_texts_for_inv(tree.branch.repository, basis)
 
150
            repo.add_revision('basis', rev, basis)
 
151
            # Add a revision for the result, with the basis content - 
 
152
            # update_basis_by_delta doesn't check that the delta results in
 
153
            # result, and we want inconsistent deltas to get called on the
 
154
            # tree, or else the code isn't actually checked.
 
155
            rev = revision.Revision('result', timestamp=0, timezone=None,
 
156
                message="", committer="foo@example.com")
 
157
            basis.revision_id = 'result'
 
158
            repo.add_revision('result', rev, basis)
 
159
            repo.commit_write_group()
 
160
        except:
 
161
            repo.abort_write_group()
 
162
            raise
 
163
        # Set the basis state as the trees current state
 
164
        tree._write_inventory(basis)
 
165
        # This reads basis from the repo and puts it into the tree's local
 
166
        # cache, if it has one.
 
167
        tree.set_parent_ids(['basis'])
 
168
        paths = {}
 
169
        parents = set()
 
170
        for old, new, id, entry in delta:
 
171
            if None in (new, entry):
 
172
                continue
 
173
            paths[new] = (entry.file_id, entry.kind)
 
174
            parents.add(osutils.dirname(new))
 
175
        parents = osutils.minimum_path_selection(parents)
 
176
        parents.discard('')
 
177
        # Put place holders in the tree to permit adding the other entries.
 
178
        for pos, parent in enumerate(parents):
 
179
            if not tree.path2id(parent):
 
180
                # add a synthetic directory in the tree so we can can put the
 
181
                # tree0 entries in place for dirstate.
 
182
                tree.add([parent], ["id%d" % pos], ["directory"])
 
183
        if paths:
 
184
            # Many deltas may cause this mini-apply to fail, but we want to see what
 
185
            # the delta application code says, not the prep that we do to deal with 
 
186
            # limitations of dirstate's update_basis code.
 
187
            for path, (file_id, kind) in sorted(paths.items()):
 
188
                try:
 
189
                    tree.add([path], [file_id], [kind])
 
190
                except (KeyboardInterrupt, SystemExit):
 
191
                    raise
 
192
                except:
 
193
                    pass
 
194
    finally:
 
195
        tree.unlock()
 
196
    # Fresh lock, reads disk again.
 
197
    tree.lock_write()
 
198
    try:
 
199
        tree.update_basis_by_delta('result', delta)
 
200
    finally:
 
201
        tree.unlock()
 
202
    # reload tree - ensure we get what was written.
 
203
    tree = tree.bzrdir.open_workingtree()
 
204
    basis_tree = tree.basis_tree()
 
205
    basis_tree.lock_read()
 
206
    self.addCleanup(basis_tree.unlock)
 
207
    # Note, that if the tree does not have a local cache, the trick above of
 
208
    # setting the result as the basis, will come back to bite us. That said,
 
209
    # all the implementations in bzr do have a local cache.
 
210
    return basis_tree.inventory
 
211
 
 
212
 
 
213
def apply_inventory_Repository_add_inventory_by_delta(self, basis, delta):
 
214
    """Apply delta to basis and return the result.
 
215
    
 
216
    This inserts basis as a whole inventory and then uses
 
217
    add_inventory_by_delta to add delta.
 
218
 
 
219
    :param basis: An inventory to be used as the basis.
 
220
    :param delta: The inventory delta to apply:
 
221
    :return: An inventory resulting from the application.
 
222
    """
 
223
    format = self.format()
 
224
    control = self.make_bzrdir('tree', format=format._matchingbzrdir)
 
225
    repo = format.initialize(control)
 
226
    repo.lock_write()
 
227
    try:
 
228
        repo.start_write_group()
 
229
        try:
 
230
            rev = revision.Revision('basis', timestamp=0, timezone=None,
 
231
                message="", committer="foo@example.com")
 
232
            basis.revision_id = 'basis'
 
233
            create_texts_for_inv(repo, basis)
 
234
            repo.add_revision('basis', rev, basis)
 
235
            repo.commit_write_group()
 
236
        except:
 
237
            repo.abort_write_group()
 
238
            raise
 
239
    finally:
 
240
        repo.unlock()
 
241
    repo.lock_write()
 
242
    try:
 
243
        repo.start_write_group()
 
244
        try:
 
245
            inv_sha1 = repo.add_inventory_by_delta('basis', delta,
 
246
                'result', ['basis'])
 
247
        except:
 
248
            repo.abort_write_group()
 
249
            raise
 
250
        else:
 
251
            repo.commit_write_group()
 
252
    finally:
 
253
        repo.unlock()
 
254
    # Fresh lock, reads disk again.
 
255
    repo = repo.bzrdir.open_repository()
 
256
    repo.lock_read()
 
257
    self.addCleanup(repo.unlock)
 
258
    return repo.get_inventory('result')
 
259
 
 
260
 
 
261
class TestInventoryUpdates(TestCase):
 
262
 
 
263
    def test_creation_from_root_id(self):
 
264
        # iff a root id is passed to the constructor, a root directory is made
 
265
        inv = inventory.Inventory(root_id='tree-root')
 
266
        self.assertNotEqual(None, inv.root)
 
267
        self.assertEqual('tree-root', inv.root.file_id)
 
268
 
 
269
    def test_add_path_of_root(self):
 
270
        # if no root id is given at creation time, there is no root directory
 
271
        inv = inventory.Inventory(root_id=None)
 
272
        self.assertIs(None, inv.root)
 
273
        # add a root entry by adding its path
 
274
        ie = inv.add_path("", "directory", "my-root")
 
275
        ie.revision = 'test-rev'
 
276
        self.assertEqual("my-root", ie.file_id)
 
277
        self.assertIs(ie, inv.root)
 
278
 
 
279
    def test_add_path(self):
 
280
        inv = inventory.Inventory(root_id='tree_root')
 
281
        ie = inv.add_path('hello', 'file', 'hello-id')
 
282
        self.assertEqual('hello-id', ie.file_id)
 
283
        self.assertEqual('file', ie.kind)
 
284
 
 
285
    def test_copy(self):
 
286
        """Make sure copy() works and creates a deep copy."""
 
287
        inv = inventory.Inventory(root_id='some-tree-root')
 
288
        ie = inv.add_path('hello', 'file', 'hello-id')
 
289
        inv2 = inv.copy()
 
290
        inv.root.file_id = 'some-new-root'
 
291
        ie.name = 'file2'
 
292
        self.assertEqual('some-tree-root', inv2.root.file_id)
 
293
        self.assertEqual('hello', inv2['hello-id'].name)
 
294
 
 
295
    def test_copy_empty(self):
 
296
        """Make sure an empty inventory can be copied."""
 
297
        inv = inventory.Inventory(root_id=None)
 
298
        inv2 = inv.copy()
 
299
        self.assertIs(None, inv2.root)
 
300
 
 
301
    def test_copy_copies_root_revision(self):
 
302
        """Make sure the revision of the root gets copied."""
 
303
        inv = inventory.Inventory(root_id='someroot')
 
304
        inv.root.revision = 'therev'
 
305
        inv2 = inv.copy()
 
306
        self.assertEquals('someroot', inv2.root.file_id)
 
307
        self.assertEquals('therev', inv2.root.revision)
 
308
 
 
309
    def test_create_tree_reference(self):
 
310
        inv = inventory.Inventory('tree-root-123')
 
311
        inv.add(TreeReference('nested-id', 'nested', parent_id='tree-root-123',
 
312
                              revision='rev', reference_revision='rev2'))
 
313
 
 
314
    def test_error_encoding(self):
 
315
        inv = inventory.Inventory('tree-root')
 
316
        inv.add(InventoryFile('a-id', u'\u1234', 'tree-root'))
 
317
        e = self.assertRaises(errors.InconsistentDelta, inv.add,
 
318
            InventoryFile('b-id', u'\u1234', 'tree-root'))
 
319
        self.assertContainsRe(str(e), r'\\u1234')
 
320
 
 
321
    def test_add_recursive(self):
 
322
        parent = InventoryDirectory('src-id', 'src', 'tree-root')
 
323
        child = InventoryFile('hello-id', 'hello.c', 'src-id')
 
324
        parent.children[child.file_id] = child
 
325
        inv = inventory.Inventory('tree-root')
 
326
        inv.add(parent)
 
327
        self.assertEqual('src/hello.c', inv.id2path('hello-id'))
 
328
 
 
329
 
 
330
 
 
331
class TestDeltaApplication(TestCaseWithTransport):
 
332
 
 
333
    def get_empty_inventory(self, reference_inv=None):
 
334
        """Get an empty inventory.
 
335
 
 
336
        Note that tests should not depend on the revision of the root for
 
337
        setting up test conditions, as it has to be flexible to accomodate non
 
338
        rich root repositories.
 
339
 
 
340
        :param reference_inv: If not None, get the revision for the root from
 
341
            this inventory. This is useful for dealing with older repositories
 
342
            that routinely discarded the root entry data. If None, the root's
 
343
            revision is set to 'basis'.
 
344
        """
 
345
        inv = inventory.Inventory()
 
346
        if reference_inv is not None:
 
347
            inv.root.revision = reference_inv.root.revision
 
348
        else:
 
349
            inv.root.revision = 'basis'
 
350
        return inv
 
351
 
 
352
    def test_empty_delta(self):
 
353
        inv = self.get_empty_inventory()
 
354
        delta = []
 
355
        inv = self.apply_delta(self, inv, delta)
 
356
        inv2 = self.get_empty_inventory(inv)
 
357
        self.assertEqual([], inv2._make_delta(inv))
 
358
 
 
359
    def test_None_file_id(self):
 
360
        inv = self.get_empty_inventory()
 
361
        dir1 = inventory.InventoryDirectory(None, 'dir1', inv.root.file_id)
 
362
        dir1.revision = 'result'
 
363
        delta = [(None, u'dir1', None, dir1)]
 
364
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
365
            inv, delta)
 
366
 
 
367
    def test_unicode_file_id(self):
 
368
        inv = self.get_empty_inventory()
 
369
        dir1 = inventory.InventoryDirectory(u'dirid', 'dir1', inv.root.file_id)
 
370
        dir1.revision = 'result'
 
371
        delta = [(None, u'dir1', dir1.file_id, dir1)]
 
372
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
373
            inv, delta)
 
374
 
 
375
    def test_repeated_file_id(self):
 
376
        inv = self.get_empty_inventory()
 
377
        file1 = inventory.InventoryFile('id', 'path1', inv.root.file_id)
 
378
        file1.revision = 'result'
 
379
        file1.text_size = 0
 
380
        file1.text_sha1 = ""
 
381
        file2 = inventory.InventoryFile('id', 'path2', inv.root.file_id)
 
382
        file2.revision = 'result'
 
383
        file2.text_size = 0
 
384
        file2.text_sha1 = ""
 
385
        delta = [(None, u'path1', 'id', file1), (None, u'path2', 'id', file2)]
 
386
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
387
            inv, delta)
 
388
 
 
389
    def test_repeated_new_path(self):
 
390
        inv = self.get_empty_inventory()
 
391
        file1 = inventory.InventoryFile('id1', 'path', inv.root.file_id)
 
392
        file1.revision = 'result'
 
393
        file1.text_size = 0
 
394
        file1.text_sha1 = ""
 
395
        file2 = inventory.InventoryFile('id2', 'path', inv.root.file_id)
 
396
        file2.revision = 'result'
 
397
        file2.text_size = 0
 
398
        file2.text_sha1 = ""
 
399
        delta = [(None, u'path', 'id1', file1), (None, u'path', 'id2', file2)]
 
400
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
401
            inv, delta)
 
402
 
 
403
    def test_repeated_old_path(self):
 
404
        inv = self.get_empty_inventory()
 
405
        file1 = inventory.InventoryFile('id1', 'path', inv.root.file_id)
 
406
        file1.revision = 'result'
 
407
        file1.text_size = 0
 
408
        file1.text_sha1 = ""
 
409
        # We can't *create* a source inventory with the same path, but
 
410
        # a badly generated partial delta might claim the same source twice.
 
411
        # This would be buggy in two ways: the path is repeated in the delta,
 
412
        # And the path for one of the file ids doesn't match the source
 
413
        # location. Alternatively, we could have a repeated fileid, but that
 
414
        # is separately checked for.
 
415
        file2 = inventory.InventoryFile('id2', 'path2', inv.root.file_id)
 
416
        file2.revision = 'result'
 
417
        file2.text_size = 0
 
418
        file2.text_sha1 = ""
 
419
        inv.add(file1)
 
420
        inv.add(file2)
 
421
        delta = [(u'path', None, 'id1', None), (u'path', None, 'id2', None)]
 
422
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
423
            inv, delta)
 
424
 
 
425
    def test_mismatched_id_entry_id(self):
 
426
        inv = self.get_empty_inventory()
 
427
        file1 = inventory.InventoryFile('id1', 'path', inv.root.file_id)
 
428
        file1.revision = 'result'
 
429
        file1.text_size = 0
 
430
        file1.text_sha1 = ""
 
431
        delta = [(None, u'path', 'id', file1)]
 
432
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
433
            inv, delta)
 
434
 
 
435
    def test_mismatched_new_path_entry_None(self):
 
436
        inv = self.get_empty_inventory()
 
437
        delta = [(None, u'path', 'id', None)]
 
438
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
439
            inv, delta)
 
440
 
 
441
    def test_mismatched_new_path_None_entry(self):
 
442
        inv = self.get_empty_inventory()
 
443
        file1 = inventory.InventoryFile('id1', 'path', inv.root.file_id)
 
444
        file1.revision = 'result'
 
445
        file1.text_size = 0
 
446
        file1.text_sha1 = ""
 
447
        delta = [(u"path", None, 'id1', file1)]
 
448
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
449
            inv, delta)
 
450
 
 
451
    def test_parent_is_not_directory(self):
 
452
        inv = self.get_empty_inventory()
 
453
        file1 = inventory.InventoryFile('id1', 'path', inv.root.file_id)
 
454
        file1.revision = 'result'
 
455
        file1.text_size = 0
 
456
        file1.text_sha1 = ""
 
457
        file2 = inventory.InventoryFile('id2', 'path2', 'id1')
 
458
        file2.revision = 'result'
 
459
        file2.text_size = 0
 
460
        file2.text_sha1 = ""
 
461
        inv.add(file1)
 
462
        delta = [(None, u'path/path2', 'id2', file2)]
 
463
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
464
            inv, delta)
 
465
 
 
466
    def test_parent_is_missing(self):
 
467
        inv = self.get_empty_inventory()
 
468
        file2 = inventory.InventoryFile('id2', 'path2', 'missingparent')
 
469
        file2.revision = 'result'
 
470
        file2.text_size = 0
 
471
        file2.text_sha1 = ""
 
472
        delta = [(None, u'path/path2', 'id2', file2)]
 
473
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
474
            inv, delta)
 
475
 
 
476
    def test_new_parent_path_has_wrong_id(self):
 
477
        inv = self.get_empty_inventory()
 
478
        parent1 = inventory.InventoryDirectory('p-1', 'dir', inv.root.file_id)
 
479
        parent1.revision = 'result'
 
480
        parent2 = inventory.InventoryDirectory('p-2', 'dir2', inv.root.file_id)
 
481
        parent2.revision = 'result'
 
482
        file1 = inventory.InventoryFile('id', 'path', 'p-2')
 
483
        file1.revision = 'result'
 
484
        file1.text_size = 0
 
485
        file1.text_sha1 = ""
 
486
        inv.add(parent1)
 
487
        inv.add(parent2)
 
488
        # This delta claims that file1 is at dir/path, but actually its at
 
489
        # dir2/path if you follow the inventory parent structure.
 
490
        delta = [(None, u'dir/path', 'id', file1)]
 
491
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
492
            inv, delta)
 
493
 
 
494
    def test_old_parent_path_is_wrong(self):
 
495
        inv = self.get_empty_inventory()
 
496
        parent1 = inventory.InventoryDirectory('p-1', 'dir', inv.root.file_id)
 
497
        parent1.revision = 'result'
 
498
        parent2 = inventory.InventoryDirectory('p-2', 'dir2', inv.root.file_id)
 
499
        parent2.revision = 'result'
 
500
        file1 = inventory.InventoryFile('id', 'path', 'p-2')
 
501
        file1.revision = 'result'
 
502
        file1.text_size = 0
 
503
        file1.text_sha1 = ""
 
504
        inv.add(parent1)
 
505
        inv.add(parent2)
 
506
        inv.add(file1)
 
507
        # This delta claims that file1 was at dir/path, but actually it was at
 
508
        # dir2/path if you follow the inventory parent structure.
 
509
        delta = [(u'dir/path', None, 'id', None)]
 
510
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
511
            inv, delta)
 
512
 
 
513
    def test_old_parent_path_is_for_other_id(self):
 
514
        inv = self.get_empty_inventory()
 
515
        parent1 = inventory.InventoryDirectory('p-1', 'dir', inv.root.file_id)
 
516
        parent1.revision = 'result'
 
517
        parent2 = inventory.InventoryDirectory('p-2', 'dir2', inv.root.file_id)
 
518
        parent2.revision = 'result'
 
519
        file1 = inventory.InventoryFile('id', 'path', 'p-2')
 
520
        file1.revision = 'result'
 
521
        file1.text_size = 0
 
522
        file1.text_sha1 = ""
 
523
        file2 = inventory.InventoryFile('id2', 'path', 'p-1')
 
524
        file2.revision = 'result'
 
525
        file2.text_size = 0
 
526
        file2.text_sha1 = ""
 
527
        inv.add(parent1)
 
528
        inv.add(parent2)
 
529
        inv.add(file1)
 
530
        inv.add(file2)
 
531
        # This delta claims that file1 was at dir/path, but actually it was at
 
532
        # dir2/path if you follow the inventory parent structure. At dir/path
 
533
        # is another entry we should not delete.
 
534
        delta = [(u'dir/path', None, 'id', None)]
 
535
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
536
            inv, delta)
 
537
 
 
538
    def test_add_existing_id_new_path(self):
 
539
        inv = self.get_empty_inventory()
 
540
        parent1 = inventory.InventoryDirectory('p-1', 'dir1', inv.root.file_id)
 
541
        parent1.revision = 'result'
 
542
        parent2 = inventory.InventoryDirectory('p-1', 'dir2', inv.root.file_id)
 
543
        parent2.revision = 'result'
 
544
        inv.add(parent1)
 
545
        delta = [(None, u'dir2', 'p-1', parent2)]
 
546
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
547
            inv, delta)
 
548
 
 
549
    def test_add_new_id_existing_path(self):
 
550
        inv = self.get_empty_inventory()
 
551
        parent1 = inventory.InventoryDirectory('p-1', 'dir1', inv.root.file_id)
 
552
        parent1.revision = 'result'
 
553
        parent2 = inventory.InventoryDirectory('p-2', 'dir1', inv.root.file_id)
 
554
        parent2.revision = 'result'
 
555
        inv.add(parent1)
 
556
        delta = [(None, u'dir1', 'p-2', parent2)]
 
557
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
558
            inv, delta)
 
559
 
 
560
    def test_remove_dir_leaving_dangling_child(self):
 
561
        inv = self.get_empty_inventory()
 
562
        dir1 = inventory.InventoryDirectory('p-1', 'dir1', inv.root.file_id)
 
563
        dir1.revision = 'result'
 
564
        dir2 = inventory.InventoryDirectory('p-2', 'child1', 'p-1')
 
565
        dir2.revision = 'result'
 
566
        dir3 = inventory.InventoryDirectory('p-3', 'child2', 'p-1')
 
567
        dir3.revision = 'result'
 
568
        inv.add(dir1)
 
569
        inv.add(dir2)
 
570
        inv.add(dir3)
 
571
        delta = [(u'dir1', None, 'p-1', None),
 
572
            (u'dir1/child2', None, 'p-3', None)]
 
573
        self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
 
574
            inv, delta)
 
575
 
 
576
 
 
577
class TestInventory(TestCase):
 
578
 
 
579
    def test_is_root(self):
 
580
        """Ensure our root-checking code is accurate."""
 
581
        inv = inventory.Inventory('TREE_ROOT')
 
582
        self.assertTrue(inv.is_root('TREE_ROOT'))
 
583
        self.assertFalse(inv.is_root('booga'))
 
584
        inv.root.file_id = 'booga'
 
585
        self.assertFalse(inv.is_root('TREE_ROOT'))
 
586
        self.assertTrue(inv.is_root('booga'))
 
587
        # works properly even if no root is set
 
588
        inv.root = None
 
589
        self.assertFalse(inv.is_root('TREE_ROOT'))
 
590
        self.assertFalse(inv.is_root('booga'))
22
591
 
23
592
 
24
593
class TestInventoryEntry(TestCase):
37
606
 
38
607
    def test_dir_detect_changes(self):
39
608
        left = inventory.InventoryDirectory('123', 'hello.c', ROOT_ID)
40
 
        left.text_sha1 = 123
41
 
        left.executable = True
42
 
        left.symlink_target='foo'
43
609
        right = inventory.InventoryDirectory('123', 'hello.c', ROOT_ID)
44
 
        right.text_sha1 = 321
45
 
        right.symlink_target='bar'
46
610
        self.assertEqual((False, False), left.detect_changes(right))
47
611
        self.assertEqual((False, False), right.detect_changes(left))
48
612
 
62
626
 
63
627
    def test_symlink_detect_changes(self):
64
628
        left = inventory.InventoryLink('123', 'hello.c', ROOT_ID)
65
 
        left.text_sha1 = 123
66
 
        left.executable = True
67
629
        left.symlink_target='foo'
68
630
        right = inventory.InventoryLink('123', 'hello.c', ROOT_ID)
69
 
        right.text_sha1 = 321
70
631
        right.symlink_target='foo'
71
632
        self.assertEqual((False, False), left.detect_changes(right))
72
633
        self.assertEqual((False, False), right.detect_changes(left))
168
729
        self.assertEqual(expected_change, change)
169
730
 
170
731
 
171
 
class TestCHKInventory(TestCaseWithTransport):
 
732
class TestCHKInventory(tests.TestCaseWithMemoryTransport):
172
733
 
173
734
    def get_chk_bytes(self):
174
 
        # The easiest way to get a CHK store is a development5 repository and
175
 
        # then work with the chk_bytes attribute directly.
176
 
        repo = self.make_repository(".", format="development5")
177
 
        repo.lock_write()
178
 
        self.addCleanup(repo.unlock)
179
 
        repo.start_write_group()
180
 
        self.addCleanup(repo.abort_write_group)
181
 
        return repo.chk_bytes
 
735
        factory = groupcompress.make_pack_factory(True, True, 1)
 
736
        trans = self.get_transport('')
 
737
        return factory(trans)
182
738
 
183
739
    def read_bytes(self, chk_bytes, key):
184
740
        stream = chk_bytes.get_record_stream([key], 'unordered', True)
297
853
        inv.root.revision = "rootrev"
298
854
        chk_bytes = self.get_chk_bytes()
299
855
        chk_inv = CHKInventory.from_inventory(chk_bytes, inv, 120)
 
856
        chk_inv.id_to_entry._ensure_root()
300
857
        self.assertEqual(120, chk_inv.id_to_entry._root_node.maximum_size)
 
858
        self.assertEqual(1, chk_inv.id_to_entry._root_node._key_width)
 
859
        p_id_basename = chk_inv.parent_id_basename_to_file_id
 
860
        p_id_basename._ensure_root()
 
861
        self.assertEqual(120, p_id_basename._root_node.maximum_size)
 
862
        self.assertEqual(2, p_id_basename._root_node._key_width)
301
863
 
302
864
    def test___iter__(self):
303
865
        inv = Inventory()
431
993
        inv.add_path("", "directory", "myrootid", None)
432
994
        inv.revision_id = "expectedid"
433
995
        reference_inv = CHKInventory.from_inventory(chk_bytes, inv)
434
 
        delta = [(None, "",  "myrootid", inv.root)]
 
996
        delta = [("", None, base_inv.root.file_id, None),
 
997
            (None, "",  "myrootid", inv.root)]
435
998
        new_inv = base_inv.create_by_apply_delta(delta, "expectedid")
436
999
        self.assertEquals(reference_inv.root, new_inv.root)
437
1000
 
454
1017
        # new_inv should be the same as reference_inv.
455
1018
        self.assertEqual(reference_inv.revision_id, new_inv.revision_id)
456
1019
        self.assertEqual(reference_inv.root_id, new_inv.root_id)
 
1020
        reference_inv.id_to_entry._ensure_root()
 
1021
        new_inv.id_to_entry._ensure_root()
457
1022
        self.assertEqual(reference_inv.id_to_entry._root_node._key,
458
1023
            new_inv.id_to_entry._root_node._key)
459
1024
 
473
1038
        reference_inv = CHKInventory.from_inventory(chk_bytes, inv)
474
1039
        delta = [(None, "A",  "A-id", a_entry)]
475
1040
        new_inv = base_inv.create_by_apply_delta(delta, "expectedid")
 
1041
        reference_inv.id_to_entry._ensure_root()
 
1042
        reference_inv.parent_id_basename_to_file_id._ensure_root()
 
1043
        new_inv.id_to_entry._ensure_root()
 
1044
        new_inv.parent_id_basename_to_file_id._ensure_root()
476
1045
        # new_inv should be the same as reference_inv.
477
1046
        self.assertEqual(reference_inv.revision_id, new_inv.revision_id)
478
1047
        self.assertEqual(reference_inv.root_id, new_inv.root_id)
636
1205
        self.assertIsInstance(ie2.name, unicode)
637
1206
        self.assertEqual(('tree\xce\xa9name', 'tree-root-id', 'tree-rev-id'),
638
1207
                         inv._bytes_to_utf8name_key(bytes))
 
1208
 
 
1209
 
 
1210
class TestCHKInventoryExpand(tests.TestCaseWithMemoryTransport):
 
1211
 
 
1212
    def get_chk_bytes(self):
 
1213
        factory = groupcompress.make_pack_factory(True, True, 1)
 
1214
        trans = self.get_transport('')
 
1215
        return factory(trans)
 
1216
 
 
1217
    def make_dir(self, inv, name, parent_id):
 
1218
        inv.add(inv.make_entry('directory', name, parent_id, name + '-id'))
 
1219
 
 
1220
    def make_file(self, inv, name, parent_id, content='content\n'):
 
1221
        ie = inv.make_entry('file', name, parent_id, name + '-id')
 
1222
        ie.text_sha1 = osutils.sha_string(content)
 
1223
        ie.text_size = len(content)
 
1224
        inv.add(ie)
 
1225
 
 
1226
    def make_simple_inventory(self):
 
1227
        inv = Inventory('TREE_ROOT')
 
1228
        inv.revision_id = "revid"
 
1229
        inv.root.revision = "rootrev"
 
1230
        # /                 TREE_ROOT
 
1231
        # dir1/             dir1-id
 
1232
        #   sub-file1       sub-file1-id
 
1233
        #   sub-file2       sub-file2-id
 
1234
        #   sub-dir1/       sub-dir1-id
 
1235
        #     subsub-file1  subsub-file1-id
 
1236
        # dir2/             dir2-id
 
1237
        #   sub2-file1      sub2-file1-id
 
1238
        # top               top-id
 
1239
        self.make_dir(inv, 'dir1', 'TREE_ROOT')
 
1240
        self.make_dir(inv, 'dir2', 'TREE_ROOT')
 
1241
        self.make_dir(inv, 'sub-dir1', 'dir1-id')
 
1242
        self.make_file(inv, 'top', 'TREE_ROOT')
 
1243
        self.make_file(inv, 'sub-file1', 'dir1-id')
 
1244
        self.make_file(inv, 'sub-file2', 'dir1-id')
 
1245
        self.make_file(inv, 'subsub-file1', 'sub-dir1-id')
 
1246
        self.make_file(inv, 'sub2-file1', 'dir2-id')
 
1247
        chk_bytes = self.get_chk_bytes()
 
1248
        #  use a small maximum_size to force internal paging structures
 
1249
        chk_inv = CHKInventory.from_inventory(chk_bytes, inv,
 
1250
                        maximum_size=100,
 
1251
                        search_key_name='hash-255-way')
 
1252
        bytes = ''.join(chk_inv.to_lines())
 
1253
        return CHKInventory.deserialise(chk_bytes, bytes, ("revid",))
 
1254
 
 
1255
    def assert_Getitems(self, expected_fileids, inv, file_ids):
 
1256
        self.assertEqual(sorted(expected_fileids),
 
1257
                         sorted([ie.file_id for ie in inv._getitems(file_ids)]))
 
1258
 
 
1259
    def assertExpand(self, all_ids, inv, file_ids):
 
1260
        (val_all_ids,
 
1261
         val_children) = inv._expand_fileids_to_parents_and_children(file_ids)
 
1262
        self.assertEqual(set(all_ids), val_all_ids)
 
1263
        entries = inv._getitems(val_all_ids)
 
1264
        expected_children = {}
 
1265
        for entry in entries:
 
1266
            s = expected_children.setdefault(entry.parent_id, [])
 
1267
            s.append(entry.file_id)
 
1268
        val_children = dict((k, sorted(v)) for k, v
 
1269
                            in val_children.iteritems())
 
1270
        expected_children = dict((k, sorted(v)) for k, v
 
1271
                            in expected_children.iteritems())
 
1272
        self.assertEqual(expected_children, val_children)
 
1273
 
 
1274
    def test_make_simple_inventory(self):
 
1275
        inv = self.make_simple_inventory()
 
1276
        layout = []
 
1277
        for path, entry in inv.iter_entries_by_dir():
 
1278
            layout.append((path, entry.file_id))
 
1279
        self.assertEqual([
 
1280
            ('', 'TREE_ROOT'),
 
1281
            ('dir1', 'dir1-id'),
 
1282
            ('dir2', 'dir2-id'),
 
1283
            ('top', 'top-id'),
 
1284
            ('dir1/sub-dir1', 'sub-dir1-id'),
 
1285
            ('dir1/sub-file1', 'sub-file1-id'),
 
1286
            ('dir1/sub-file2', 'sub-file2-id'),
 
1287
            ('dir1/sub-dir1/subsub-file1', 'subsub-file1-id'),
 
1288
            ('dir2/sub2-file1', 'sub2-file1-id'),
 
1289
            ], layout)
 
1290
 
 
1291
    def test__getitems(self):
 
1292
        inv = self.make_simple_inventory()
 
1293
        # Reading from disk
 
1294
        self.assert_Getitems(['dir1-id'], inv, ['dir1-id'])
 
1295
        self.assertTrue('dir1-id' in inv._fileid_to_entry_cache)
 
1296
        self.assertFalse('sub-file2-id' in inv._fileid_to_entry_cache)
 
1297
        # From cache
 
1298
        self.assert_Getitems(['dir1-id'], inv, ['dir1-id'])
 
1299
        # Mixed
 
1300
        self.assert_Getitems(['dir1-id', 'sub-file2-id'], inv,
 
1301
                             ['dir1-id', 'sub-file2-id'])
 
1302
        self.assertTrue('dir1-id' in inv._fileid_to_entry_cache)
 
1303
        self.assertTrue('sub-file2-id' in inv._fileid_to_entry_cache)
 
1304
 
 
1305
    def test_single_file(self):
 
1306
        inv = self.make_simple_inventory()
 
1307
        self.assertExpand(['TREE_ROOT', 'top-id'], inv, ['top-id'])
 
1308
 
 
1309
    def test_get_all_parents(self):
 
1310
        inv = self.make_simple_inventory()
 
1311
        self.assertExpand(['TREE_ROOT', 'dir1-id', 'sub-dir1-id',
 
1312
                           'subsub-file1-id',
 
1313
                          ], inv, ['subsub-file1-id'])
 
1314
 
 
1315
    def test_get_children(self):
 
1316
        inv = self.make_simple_inventory()
 
1317
        self.assertExpand(['TREE_ROOT', 'dir1-id', 'sub-dir1-id',
 
1318
                           'sub-file1-id', 'sub-file2-id', 'subsub-file1-id',
 
1319
                          ], inv, ['dir1-id'])
 
1320
 
 
1321
    def test_from_root(self):
 
1322
        inv = self.make_simple_inventory()
 
1323
        self.assertExpand(['TREE_ROOT', 'dir1-id', 'dir2-id', 'sub-dir1-id',
 
1324
                           'sub-file1-id', 'sub-file2-id', 'sub2-file1-id',
 
1325
                           'subsub-file1-id', 'top-id'], inv, ['TREE_ROOT'])
 
1326
 
 
1327
    def test_top_level_file(self):
 
1328
        inv = self.make_simple_inventory()
 
1329
        self.assertExpand(['TREE_ROOT', 'top-id'], inv, ['top-id'])
 
1330
 
 
1331
    def test_subsub_file(self):
 
1332
        inv = self.make_simple_inventory()
 
1333
        self.assertExpand(['TREE_ROOT', 'dir1-id', 'sub-dir1-id',
 
1334
                           'subsub-file1-id'], inv, ['subsub-file1-id'])
 
1335
 
 
1336
    def test_sub_and_root(self):
 
1337
        inv = self.make_simple_inventory()
 
1338
        self.assertExpand(['TREE_ROOT', 'dir1-id', 'sub-dir1-id', 'top-id',
 
1339
                           'subsub-file1-id'], inv, ['top-id', 'subsub-file1-id'])