15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
29
from bzrlib.inventory import (CHKInventory, Inventory, ROOT_ID, InventoryFile,
30
InventoryDirectory, InventoryEntry, TreeReference)
31
from bzrlib.tests import (
31
from ..bzr.inventory import (
39
mutable_inventory_from_tree,
33
43
TestCaseWithTransport,
36
split_suite_by_condition,
38
from bzrlib.tests.per_workingtree import workingtree_formats
41
def load_tests(standard_tests, module, loader):
42
"""Parameterise some inventory tests."""
43
to_adapt, result = split_suite_by_condition(standard_tests,
44
condition_isinstance(TestDeltaApplication))
45
from .scenarios import load_tests_apply_scenarios
48
load_tests = load_tests_apply_scenarios
51
def delta_application_scenarios():
46
53
('Inventory', {'apply_delta':apply_inventory_Inventory}),
107
119
# Fresh object, reads disk again.
108
tree = tree.bzrdir.open_workingtree()
120
tree = tree.controldir.open_workingtree()
109
121
tree.lock_write()
111
123
tree.apply_inventory_delta(delta)
114
126
# reload tree - ensure we get what was written.
115
tree = tree.bzrdir.open_workingtree()
127
tree = tree.controldir.open_workingtree()
117
129
self.addCleanup(tree.unlock)
118
# One could add 'tree._validate' here but that would cause 'early' failues
119
# as far as higher level code is concerned. Possibly adding an
120
# expect_fail parameter to this function and if that is False then do a
122
return tree.inventory
125
def apply_inventory_WT_basis(self, basis, delta):
130
if not invalid_delta:
132
return tree.root_inventory
135
def _create_repo_revisions(repo, basis, delta, invalid_delta):
136
repo.start_write_group()
138
rev = revision.Revision('basis', timestamp=0, timezone=None,
139
message="", committer="foo@example.com")
140
basis.revision_id = 'basis'
141
create_texts_for_inv(repo, basis)
142
repo.add_revision('basis', rev, basis)
144
# We don't want to apply the delta to the basis, because we expect
145
# the delta is invalid.
147
result_inv.revision_id = 'result'
148
target_entries = None
150
result_inv = basis.create_by_apply_delta(delta, 'result')
151
create_texts_for_inv(repo, result_inv)
152
target_entries = list(result_inv.iter_entries_by_dir())
153
rev = revision.Revision('result', timestamp=0, timezone=None,
154
message="", committer="foo@example.com")
155
repo.add_revision('result', rev, result_inv)
156
repo.commit_write_group()
158
repo.abort_write_group()
160
return target_entries
163
def _get_basis_entries(tree):
164
basis_tree = tree.basis_tree()
165
basis_tree.lock_read()
166
basis_tree_entries = list(basis_tree.inventory.iter_entries_by_dir())
168
return basis_tree_entries
171
def _populate_different_tree(tree, basis, delta):
172
"""Put all entries into tree, but at a unique location."""
175
tree.add(['unique-dir'], ['unique-dir-id'], ['directory'])
176
for path, ie in basis.iter_entries_by_dir():
177
if ie.file_id in added_ids:
179
# We want a unique path for each of these, we use the file-id
180
tree.add(['unique-dir/' + ie.file_id], [ie.file_id], [ie.kind])
181
added_ids.add(ie.file_id)
182
for old_path, new_path, file_id, ie in delta:
183
if file_id in added_ids:
185
tree.add(['unique-dir/' + file_id], [file_id], [ie.kind])
188
def apply_inventory_WT_basis(test, basis, delta, invalid_delta=True):
126
189
"""Apply delta to basis and return the result.
128
191
This sets the parent and then calls update_basis_by_delta.
130
193
allow safety checks made by the WT to succeed, and finally ensures that all
131
194
items in the delta with a new path are present in the WT before calling
132
195
update_basis_by_delta.
134
197
:param basis: An inventory to be used as the basis.
135
198
:param delta: The inventory delta to apply:
136
199
:return: An inventory resulting from the application.
138
control = self.make_bzrdir('tree', format=self.format._matchingbzrdir)
201
control = test.make_controldir('tree', format=test.format._matchingcontroldir)
139
202
control.create_repository()
140
203
control.create_branch()
141
tree = self.format.initialize(control)
204
tree = test.format.initialize(control)
142
205
tree.lock_write()
144
repo = tree.branch.repository
145
repo.start_write_group()
147
rev = revision.Revision('basis', timestamp=0, timezone=None,
148
message="", committer="foo@example.com")
149
basis.revision_id = 'basis'
150
create_texts_for_inv(tree.branch.repository, basis)
151
repo.add_revision('basis', rev, basis)
152
# Add a revision for the result, with the basis content -
153
# update_basis_by_delta doesn't check that the delta results in
154
# result, and we want inconsistent deltas to get called on the
155
# tree, or else the code isn't actually checked.
156
rev = revision.Revision('result', timestamp=0, timezone=None,
157
message="", committer="foo@example.com")
158
basis.revision_id = 'result'
159
repo.add_revision('result', rev, basis)
160
repo.commit_write_group()
162
repo.abort_write_group()
207
target_entries = _create_repo_revisions(tree.branch.repository, basis,
208
delta, invalid_delta)
164
209
# Set the basis state as the trees current state
165
210
tree._write_inventory(basis)
166
211
# This reads basis from the repo and puts it into the tree's local
167
212
# cache, if it has one.
168
213
tree.set_parent_ids(['basis'])
171
for old, new, id, entry in delta:
172
if None in (new, entry):
174
paths[new] = (entry.file_id, entry.kind)
175
parents.add(osutils.dirname(new))
176
parents = osutils.minimum_path_selection(parents)
178
# Put place holders in the tree to permit adding the other entries.
179
for pos, parent in enumerate(parents):
180
if not tree.path2id(parent):
181
# add a synthetic directory in the tree so we can can put the
182
# tree0 entries in place for dirstate.
183
tree.add([parent], ["id%d" % pos], ["directory"])
185
# Many deltas may cause this mini-apply to fail, but we want to see what
186
# the delta application code says, not the prep that we do to deal with
187
# limitations of dirstate's update_basis code.
188
for path, (file_id, kind) in sorted(paths.items()):
190
tree.add([path], [file_id], [kind])
191
except (KeyboardInterrupt, SystemExit):
197
216
# Fresh lock, reads disk again.
198
217
tree.lock_write()
200
219
tree.update_basis_by_delta('result', delta)
220
if not invalid_delta:
203
224
# reload tree - ensure we get what was written.
204
tree = tree.bzrdir.open_workingtree()
225
tree = tree.controldir.open_workingtree()
205
226
basis_tree = tree.basis_tree()
206
227
basis_tree.lock_read()
207
self.addCleanup(basis_tree.unlock)
208
# Note, that if the tree does not have a local cache, the trick above of
209
# setting the result as the basis, will come back to bite us. That said,
210
# all the implementations in bzr do have a local cache.
211
return basis_tree.inventory
214
def apply_inventory_Repository_add_inventory_by_delta(self, basis, delta):
228
test.addCleanup(basis_tree.unlock)
229
basis_inv = basis_tree.root_inventory
231
basis_entries = list(basis_inv.iter_entries_by_dir())
232
test.assertEqual(target_entries, basis_entries)
236
def apply_inventory_Repository_add_inventory_by_delta(self, basis, delta,
215
238
"""Apply delta to basis and return the result.
217
240
This inserts basis as a whole inventory and then uses
264
287
def test_creation_from_root_id(self):
265
288
# iff a root id is passed to the constructor, a root directory is made
266
inv = inventory.Inventory(root_id='tree-root')
289
inv = inventory.Inventory(root_id=b'tree-root')
267
290
self.assertNotEqual(None, inv.root)
268
self.assertEqual('tree-root', inv.root.file_id)
291
self.assertEqual(b'tree-root', inv.root.file_id)
270
293
def test_add_path_of_root(self):
271
294
# if no root id is given at creation time, there is no root directory
272
295
inv = inventory.Inventory(root_id=None)
273
296
self.assertIs(None, inv.root)
274
297
# add a root entry by adding its path
275
ie = inv.add_path("", "directory", "my-root")
276
ie.revision = 'test-rev'
277
self.assertEqual("my-root", ie.file_id)
298
ie = inv.add_path(u"", "directory", b"my-root")
299
ie.revision = b'test-rev'
300
self.assertEqual(b"my-root", ie.file_id)
278
301
self.assertIs(ie, inv.root)
280
303
def test_add_path(self):
281
inv = inventory.Inventory(root_id='tree_root')
282
ie = inv.add_path('hello', 'file', 'hello-id')
283
self.assertEqual('hello-id', ie.file_id)
304
inv = inventory.Inventory(root_id=b'tree_root')
305
ie = inv.add_path(u'hello', 'file', b'hello-id')
306
self.assertEqual(b'hello-id', ie.file_id)
284
307
self.assertEqual('file', ie.kind)
286
309
def test_copy(self):
287
310
"""Make sure copy() works and creates a deep copy."""
288
inv = inventory.Inventory(root_id='some-tree-root')
289
ie = inv.add_path('hello', 'file', 'hello-id')
311
inv = inventory.Inventory(root_id=b'some-tree-root')
312
ie = inv.add_path(u'hello', 'file', b'hello-id')
290
313
inv2 = inv.copy()
291
inv.root.file_id = 'some-new-root'
293
self.assertEqual('some-tree-root', inv2.root.file_id)
294
self.assertEqual('hello', inv2['hello-id'].name)
314
inv.root.file_id = b'some-new-root'
316
self.assertEqual(b'some-tree-root', inv2.root.file_id)
317
self.assertEqual(u'hello', inv2[b'hello-id'].name)
296
319
def test_copy_empty(self):
297
320
"""Make sure an empty inventory can be copied."""
302
325
def test_copy_copies_root_revision(self):
303
326
"""Make sure the revision of the root gets copied."""
304
inv = inventory.Inventory(root_id='someroot')
305
inv.root.revision = 'therev'
327
inv = inventory.Inventory(root_id=b'someroot')
328
inv.root.revision = b'therev'
306
329
inv2 = inv.copy()
307
self.assertEquals('someroot', inv2.root.file_id)
308
self.assertEquals('therev', inv2.root.revision)
330
self.assertEqual(b'someroot', inv2.root.file_id)
331
self.assertEqual(b'therev', inv2.root.revision)
310
333
def test_create_tree_reference(self):
311
inv = inventory.Inventory('tree-root-123')
312
inv.add(TreeReference('nested-id', 'nested', parent_id='tree-root-123',
313
revision='rev', reference_revision='rev2'))
334
inv = inventory.Inventory(b'tree-root-123')
335
inv.add(TreeReference(
336
b'nested-id', 'nested', parent_id=b'tree-root-123',
337
revision=b'rev', reference_revision=b'rev2'))
315
339
def test_error_encoding(self):
316
340
inv = inventory.Inventory('tree-root')
574
603
self.assertRaises(errors.InconsistentDelta, self.apply_delta, self,
578
class TestInventory(TestCase):
606
def test_add_file(self):
607
inv = self.get_empty_inventory()
608
file1 = inventory.InventoryFile('file-id', 'path', inv.root.file_id)
609
file1.revision = 'result'
612
delta = [(None, u'path', 'file-id', file1)]
613
res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
614
self.assertEqual('file-id', res_inv['file-id'].file_id)
616
def test_remove_file(self):
617
inv = self.get_empty_inventory()
618
file1 = inventory.InventoryFile('file-id', 'path', inv.root.file_id)
619
file1.revision = 'result'
623
delta = [(u'path', None, 'file-id', None)]
624
res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
625
self.assertEqual(None, res_inv.path2id('path'))
626
self.assertRaises(errors.NoSuchId, res_inv.id2path, 'file-id')
628
def test_rename_file(self):
629
inv = self.get_empty_inventory()
630
file1 = self.make_file_ie(name='path', parent_id=inv.root.file_id)
632
file2 = self.make_file_ie(name='path2', parent_id=inv.root.file_id)
633
delta = [(u'path', 'path2', 'file-id', file2)]
634
res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
635
self.assertEqual(None, res_inv.path2id('path'))
636
self.assertEqual('file-id', res_inv.path2id('path2'))
638
def test_replaced_at_new_path(self):
639
inv = self.get_empty_inventory()
640
file1 = self.make_file_ie(file_id='id1', parent_id=inv.root.file_id)
642
file2 = self.make_file_ie(file_id='id2', parent_id=inv.root.file_id)
643
delta = [(u'name', None, 'id1', None),
644
(None, u'name', 'id2', file2)]
645
res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
646
self.assertEqual('id2', res_inv.path2id('name'))
648
def test_rename_dir(self):
649
inv = self.get_empty_inventory()
650
dir1 = inventory.InventoryDirectory('dir-id', 'dir1', inv.root.file_id)
651
dir1.revision = 'basis'
652
file1 = self.make_file_ie(parent_id='dir-id')
655
dir2 = inventory.InventoryDirectory('dir-id', 'dir2', inv.root.file_id)
656
dir2.revision = 'result'
657
delta = [('dir1', 'dir2', 'dir-id', dir2)]
658
res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
659
# The file should be accessible under the new path
660
self.assertEqual('file-id', res_inv.path2id('dir2/name'))
662
def test_renamed_dir_with_renamed_child(self):
663
inv = self.get_empty_inventory()
664
dir1 = inventory.InventoryDirectory('dir-id', 'dir1', inv.root.file_id)
665
dir1.revision = 'basis'
666
file1 = self.make_file_ie('file-id-1', 'name1', parent_id='dir-id')
667
file2 = self.make_file_ie('file-id-2', 'name2', parent_id='dir-id')
671
dir2 = inventory.InventoryDirectory('dir-id', 'dir2', inv.root.file_id)
672
dir2.revision = 'result'
673
file2b = self.make_file_ie('file-id-2', 'name2', inv.root.file_id)
674
delta = [('dir1', 'dir2', 'dir-id', dir2),
675
('dir1/name2', 'name2', 'file-id-2', file2b)]
676
res_inv = self.apply_delta(self, inv, delta, invalid_delta=False)
677
# The file should be accessible under the new path
678
self.assertEqual('file-id-1', res_inv.path2id('dir2/name1'))
679
self.assertEqual(None, res_inv.path2id('dir2/name2'))
680
self.assertEqual('file-id-2', res_inv.path2id('name2'))
580
682
def test_is_root(self):
581
683
"""Ensure our root-checking code is accurate."""
903
1001
def test___getitem__(self):
904
1002
inv = Inventory()
905
inv.revision_id = "revid"
906
inv.root.revision = "rootrev"
907
inv.add(InventoryFile("fileid", "file", inv.root.file_id))
908
inv["fileid"].revision = "filerev"
909
inv["fileid"].executable = True
910
inv["fileid"].text_sha1 = "ffff"
911
inv["fileid"].text_size = 1
1003
inv.revision_id = b"revid"
1004
inv.root.revision = b"rootrev"
1005
inv.add(InventoryFile(b"fileid", u"file", inv.root.file_id))
1006
inv[b"fileid"].revision = b"filerev"
1007
inv[b"fileid"].executable = True
1008
inv[b"fileid"].text_sha1 = b"ffff"
1009
inv[b"fileid"].text_size = 1
912
1010
chk_bytes = self.get_chk_bytes()
913
1011
chk_inv = CHKInventory.from_inventory(chk_bytes, inv)
914
bytes = ''.join(chk_inv.to_lines())
915
new_inv = CHKInventory.deserialise(chk_bytes, bytes, ("revid",))
1012
data = b''.join(chk_inv.to_lines())
1013
new_inv = CHKInventory.deserialise(chk_bytes, data, (b"revid",))
916
1014
root_entry = new_inv[inv.root.file_id]
917
file_entry = new_inv["fileid"]
1015
file_entry = new_inv[b"fileid"]
918
1016
self.assertEqual("directory", root_entry.kind)
919
1017
self.assertEqual(inv.root.file_id, root_entry.file_id)
920
1018
self.assertEqual(inv.root.parent_id, root_entry.parent_id)
921
1019
self.assertEqual(inv.root.name, root_entry.name)
922
self.assertEqual("rootrev", root_entry.revision)
1020
self.assertEqual(b"rootrev", root_entry.revision)
923
1021
self.assertEqual("file", file_entry.kind)
924
self.assertEqual("fileid", file_entry.file_id)
1022
self.assertEqual(b"fileid", file_entry.file_id)
925
1023
self.assertEqual(inv.root.file_id, file_entry.parent_id)
926
self.assertEqual("file", file_entry.name)
927
self.assertEqual("filerev", file_entry.revision)
928
self.assertEqual("ffff", file_entry.text_sha1)
1024
self.assertEqual(u"file", file_entry.name)
1025
self.assertEqual(b"filerev", file_entry.revision)
1026
self.assertEqual(b"ffff", file_entry.text_sha1)
929
1027
self.assertEqual(1, file_entry.text_size)
930
1028
self.assertEqual(True, file_entry.executable)
931
1029
self.assertRaises(errors.NoSuchId, new_inv.__getitem__, 'missing')
1215
1313
self.assertEqual(('tree\xce\xa9name', 'tree-root-id', 'tree-rev-id'),
1216
1314
inv._bytes_to_utf8name_key(bytes))
1316
def make_basic_utf8_inventory(self):
1318
inv.revision_id = "revid"
1319
inv.root.revision = "rootrev"
1320
root_id = inv.root.file_id
1321
inv.add(InventoryFile("fileid", u'f\xefle', root_id))
1322
inv["fileid"].revision = "filerev"
1323
inv["fileid"].text_sha1 = "ffff"
1324
inv["fileid"].text_size = 0
1325
inv.add(InventoryDirectory("dirid", u'dir-\N{EURO SIGN}', root_id))
1326
inv.add(InventoryFile("childid", u'ch\xefld', "dirid"))
1327
inv["childid"].revision = "filerev"
1328
inv["childid"].text_sha1 = "ffff"
1329
inv["childid"].text_size = 0
1330
chk_bytes = self.get_chk_bytes()
1331
chk_inv = CHKInventory.from_inventory(chk_bytes, inv)
1332
bytes = ''.join(chk_inv.to_lines())
1333
return CHKInventory.deserialise(chk_bytes, bytes, ("revid",))
1335
def test__preload_handles_utf8(self):
1336
new_inv = self.make_basic_utf8_inventory()
1337
self.assertEqual({}, new_inv._fileid_to_entry_cache)
1338
self.assertFalse(new_inv._fully_cached)
1339
new_inv._preload_cache()
1341
sorted([new_inv.root_id, "fileid", "dirid", "childid"]),
1342
sorted(new_inv._fileid_to_entry_cache.keys()))
1343
ie_root = new_inv._fileid_to_entry_cache[new_inv.root_id]
1344
self.assertEqual([u'dir-\N{EURO SIGN}', u'f\xefle'],
1345
sorted(ie_root._children.keys()))
1346
ie_dir = new_inv._fileid_to_entry_cache['dirid']
1347
self.assertEqual([u'ch\xefld'], sorted(ie_dir._children.keys()))
1349
def test__preload_populates_cache(self):
1351
inv.revision_id = "revid"
1352
inv.root.revision = "rootrev"
1353
root_id = inv.root.file_id
1354
inv.add(InventoryFile("fileid", "file", root_id))
1355
inv["fileid"].revision = "filerev"
1356
inv["fileid"].executable = True
1357
inv["fileid"].text_sha1 = "ffff"
1358
inv["fileid"].text_size = 1
1359
inv.add(InventoryDirectory("dirid", "dir", root_id))
1360
inv.add(InventoryFile("childid", "child", "dirid"))
1361
inv["childid"].revision = "filerev"
1362
inv["childid"].executable = False
1363
inv["childid"].text_sha1 = "dddd"
1364
inv["childid"].text_size = 1
1365
chk_bytes = self.get_chk_bytes()
1366
chk_inv = CHKInventory.from_inventory(chk_bytes, inv)
1367
bytes = ''.join(chk_inv.to_lines())
1368
new_inv = CHKInventory.deserialise(chk_bytes, bytes, ("revid",))
1369
self.assertEqual({}, new_inv._fileid_to_entry_cache)
1370
self.assertFalse(new_inv._fully_cached)
1371
new_inv._preload_cache()
1373
sorted([root_id, "fileid", "dirid", "childid"]),
1374
sorted(new_inv._fileid_to_entry_cache.keys()))
1375
self.assertTrue(new_inv._fully_cached)
1376
ie_root = new_inv._fileid_to_entry_cache[root_id]
1377
self.assertEqual(['dir', 'file'], sorted(ie_root._children.keys()))
1378
ie_dir = new_inv._fileid_to_entry_cache['dirid']
1379
self.assertEqual(['child'], sorted(ie_dir._children.keys()))
1381
def test__preload_handles_partially_evaluated_inventory(self):
1382
new_inv = self.make_basic_utf8_inventory()
1383
ie = new_inv[new_inv.root_id]
1384
self.assertIs(None, ie._children)
1385
self.assertEqual([u'dir-\N{EURO SIGN}', u'f\xefle'],
1386
sorted(ie.children.keys()))
1387
# Accessing .children loads _children
1388
self.assertEqual([u'dir-\N{EURO SIGN}', u'f\xefle'],
1389
sorted(ie._children.keys()))
1390
new_inv._preload_cache()
1392
self.assertEqual([u'dir-\N{EURO SIGN}', u'f\xefle'],
1393
sorted(ie._children.keys()))
1394
ie_dir = new_inv["dirid"]
1395
self.assertEqual([u'ch\xefld'],
1396
sorted(ie_dir._children.keys()))
1398
def test_filter_change_in_renamed_subfolder(self):
1399
inv = Inventory('tree-root')
1400
src_ie = inv.add_path('src', 'directory', 'src-id')
1401
inv.add_path('src/sub/', 'directory', 'sub-id')
1402
a_ie = inv.add_path('src/sub/a', 'file', 'a-id')
1403
a_ie.text_sha1 = osutils.sha_string('content\n')
1404
a_ie.text_size = len('content\n')
1405
chk_bytes = self.get_chk_bytes()
1406
inv = CHKInventory.from_inventory(chk_bytes, inv)
1407
inv = inv.create_by_apply_delta([
1408
("src/sub/a", "src/sub/a", "a-id", a_ie),
1409
("src", "src2", "src-id", src_ie),
1411
new_inv = inv.filter(['a-id', 'src-id'])
1415
('src/sub', 'sub-id'),
1416
('src/sub/a', 'a-id'),
1417
], [(path, ie.file_id) for path, ie in new_inv.iter_entries()])
1219
1419
class TestCHKInventoryExpand(tests.TestCaseWithMemoryTransport):
1346
1546
inv = self.make_simple_inventory()
1347
1547
self.assertExpand(['TREE_ROOT', 'dir1-id', 'sub-dir1-id', 'top-id',
1348
1548
'subsub-file1-id'], inv, ['top-id', 'subsub-file1-id'])
1551
class TestMutableInventoryFromTree(TestCaseWithTransport):
1553
def test_empty(self):
1554
repository = self.make_repository('.')
1555
tree = repository.revision_tree(revision.NULL_REVISION)
1556
inv = mutable_inventory_from_tree(tree)
1557
self.assertEqual(revision.NULL_REVISION, inv.revision_id)
1558
self.assertEqual(0, len(inv))
1560
def test_some_files(self):
1561
wt = self.make_branch_and_tree('.')
1562
self.build_tree(['a'])
1563
wt.add(['a'], ['thefileid'])
1564
revid = wt.commit("commit")
1565
tree = wt.branch.repository.revision_tree(revid)
1566
inv = mutable_inventory_from_tree(tree)
1567
self.assertEqual(revid, inv.revision_id)
1568
self.assertEqual(2, len(inv))
1569
self.assertEqual("a", inv['thefileid'].name)
1570
# The inventory should be mutable and independent of
1572
self.assertFalse(tree.root_inventory['thefileid'].executable)
1573
inv['thefileid'].executable = True
1574
self.assertFalse(tree.root_inventory['thefileid'].executable)