← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_knit.py

Merge from bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_knit.py
37
37
from bzrlib.index import *
38
38
from bzrlib.knit import (
39
39
    AnnotatedKnitContent,
 
40
    DATA_SUFFIX,
 
41
    INDEX_SUFFIX,
40
42
    KnitContent,
41
43
    KnitGraphIndex,
42
44
    KnitVersionedFile,
45
47
    _KnitAccess,
46
48
    _KnitData,
47
49
    _KnitIndex,
 
50
    make_file_knit,
48
51
    _PackAccess,
49
52
    PlainKnitContent,
50
53
    _StreamAccess,
53
56
    KnitSequenceMatcher,
54
57
    )
55
58
from bzrlib.osutils import split_lines
 
59
from bzrlib.symbol_versioning import one_four
56
60
from bzrlib.tests import (
57
61
    Feature,
58
62
    TestCase,
158
162
        self.assertEqual(content.annotate(),
159
163
            [("bogus", "text1"), ("bogus", "text2")])
160
164
 
161
 
    def test_annotate_iter(self):
162
 
        content = self._make_content([])
163
 
        it = content.annotate_iter()
164
 
        self.assertRaises(StopIteration, it.next)
165
 
 
166
 
        content = self._make_content([("bogus", "text1"), ("bogus", "text2")])
167
 
        it = content.annotate_iter()
168
 
        self.assertEqual(it.next(), ("bogus", "text1"))
169
 
        self.assertEqual(it.next(), ("bogus", "text2"))
170
 
        self.assertRaises(StopIteration, it.next)
171
 
 
172
165
    def test_line_delta(self):
173
166
        content1 = self._make_content([("", "a"), ("", "b")])
174
167
        content2 = self._make_content([("", "a"), ("", "a"), ("", "c")])
196
189
        self.assertEqual(content.annotate(),
197
190
            [("origin1", "text1"), ("origin2", "text2")])
198
191
 
199
 
    def test_annotate_iter(self):
200
 
        content = self._make_content([])
201
 
        it = content.annotate_iter()
202
 
        self.assertRaises(StopIteration, it.next)
203
 
 
204
 
        content = self._make_content([("origin1", "text1"), ("origin2", "text2")])
205
 
        it = content.annotate_iter()
206
 
        self.assertEqual(it.next(), ("origin1", "text1"))
207
 
        self.assertEqual(it.next(), ("origin2", "text2"))
208
 
        self.assertRaises(StopIteration, it.next)
209
 
 
210
192
    def test_line_delta(self):
211
193
        content1 = self._make_content([("", "a"), ("", "b")])
212
194
        content2 = self._make_content([("", "a"), ("", "a"), ("", "c")])
492
474
        self.addCleanup(reset)
493
475
        from bzrlib._knit_load_data_py import _load_data_py
494
476
        knit._load_data = _load_data_py
495
 
        return _KnitIndex(*args, **kwargs)
 
477
        return _KnitIndex(get_scope=lambda:None, *args, **kwargs)
496
478
 
497
479
    def test_no_such_file(self):
498
480
        transport = MockTransport()
628
610
            {}),
629
611
            transport.calls.pop(0))
630
612
 
631
 
    def test_get_graph(self):
632
 
        transport = MockTransport()
633
 
        index = self.get_knit_index(transport, "filename", "w", create=True)
634
 
        self.assertEqual([], index.get_graph())
635
 
 
636
 
        index.add_version("a", ["option"], (None, 0, 1), ["b"])
637
 
        self.assertEqual([("a", ("b",))], index.get_graph())
638
 
 
639
 
        index.add_version("c", ["option"], (None, 0, 1), ["d"])
640
 
        self.assertEqual([("a", ("b",)), ("c", ("d",))],
641
 
            sorted(index.get_graph()))
642
 
 
643
613
    def test_get_ancestry(self):
644
614
        transport = MockTransport([
645
615
            _KnitIndex.HEADER,
689
659
        self.assertRaises(RevisionNotPresent,
690
660
            index.get_ancestry_with_ghosts, ["e"])
691
661
 
692
 
    def test_iter_parents(self):
693
 
        transport = MockTransport()
694
 
        index = self.get_knit_index(transport, "filename", "w", create=True)
695
 
        # no parents
696
 
        index.add_version('r0', ['option'], (None, 0, 1), [])
697
 
        # 1 parent
698
 
        index.add_version('r1', ['option'], (None, 0, 1), ['r0'])
699
 
        # 2 parents
700
 
        index.add_version('r2', ['option'], (None, 0, 1), ['r1', 'r0'])
701
 
        # XXX TODO a ghost
702
 
        # cases: each sample data individually:
703
 
        self.assertEqual(set([('r0', ())]),
704
 
            set(index.iter_parents(['r0'])))
705
 
        self.assertEqual(set([('r1', ('r0', ))]),
706
 
            set(index.iter_parents(['r1'])))
707
 
        self.assertEqual(set([('r2', ('r1', 'r0'))]),
708
 
            set(index.iter_parents(['r2'])))
709
 
        # no nodes returned for a missing node
710
 
        self.assertEqual(set(),
711
 
            set(index.iter_parents(['missing'])))
712
 
        # 1 node returned with missing nodes skipped
713
 
        self.assertEqual(set([('r1', ('r0', ))]),
714
 
            set(index.iter_parents(['ghost1', 'r1', 'ghost'])))
715
 
        # 2 nodes returned
716
 
        self.assertEqual(set([('r0', ()), ('r1', ('r0', ))]),
717
 
            set(index.iter_parents(['r0', 'r1'])))
718
 
        # 2 nodes returned, missing skipped
719
 
        self.assertEqual(set([('r0', ()), ('r1', ('r0', ))]),
720
 
            set(index.iter_parents(['a', 'r0', 'b', 'r1', 'c'])))
721
 
 
722
662
    def test_num_versions(self):
723
663
        transport = MockTransport([
724
664
            _KnitIndex.HEADER
1087
1027
        self.addCleanup(reset)
1088
1028
        from bzrlib._knit_load_data_c import _load_data_c
1089
1029
        knit._load_data = _load_data_c
1090
 
        return _KnitIndex(*args, **kwargs)
1091
 
 
 
1030
        return _KnitIndex(get_scope=lambda:None, *args, **kwargs)
1092
1031
 
1093
1032
 
1094
1033
class KnitTests(TestCaseWithTransport):
1095
1034
    """Class containing knit test helper routines."""
1096
1035
 
1097
1036
    def make_test_knit(self, annotate=False, delay_create=False, index=None,
1098
 
                       name='test'):
 
1037
                       name='test', delta=True, access_mode='w'):
1099
1038
        if not annotate:
1100
1039
            factory = KnitPlainFactory()
1101
1040
        else:
1102
1041
            factory = None
1103
 
        return KnitVersionedFile(name, get_transport('.'), access_mode='w',
1104
 
                                 factory=factory, create=True,
1105
 
                                 delay_create=delay_create, index=index)
 
1042
        if index is None:
 
1043
            index = _KnitIndex(get_transport('.'), name + INDEX_SUFFIX,
 
1044
                access_mode, create=True, file_mode=None,
 
1045
                create_parent_dir=False, delay_create=delay_create,
 
1046
                dir_mode=None, get_scope=lambda:None)
 
1047
        access = _KnitAccess(get_transport('.'), name + DATA_SUFFIX, None,
 
1048
            None, delay_create, False)
 
1049
        return KnitVersionedFile(name, get_transport('.'), factory=factory,
 
1050
            create=True, delay_create=delay_create, index=index,
 
1051
            access_method=access, delta=delta)
1106
1052
 
1107
1053
    def assertRecordContentEqual(self, knit, version_id, candidate_content):
1108
1054
        """Assert that some raw record content matches the raw record content
1127
1073
    def test_make_explicit_index(self):
1128
1074
        """We can supply an index to use."""
1129
1075
        knit = KnitVersionedFile('test', get_transport('.'),
1130
 
            index='strangelove')
 
1076
            index='strangelove', access_method="a")
1131
1077
        self.assertEqual(knit._index, 'strangelove')
1132
1078
 
1133
1079
    def test_knit_add(self):
1174
1120
        k = self.make_test_knit()
1175
1121
        k.add_lines('text-1', [], split_lines(TEXT_1))
1176
1122
        del k
1177
 
        k2 = KnitVersionedFile('test', get_transport('.'), access_mode='r', factory=KnitPlainFactory(), create=True)
 
1123
        k2 = make_file_knit('test', get_transport('.'), access_mode='r',
 
1124
            factory=KnitPlainFactory(), create=True)
1178
1125
        self.assertTrue(k2.has_version('text-1'))
1179
1126
        self.assertEqualDiff(''.join(k2.get_lines('text-1')), TEXT_1)
1180
1127
 
1202
1149
    def test_incomplete(self):
1203
1150
        """Test if texts without a ending line-end can be inserted and
1204
1151
        extracted."""
1205
 
        k = KnitVersionedFile('test', get_transport('.'), delta=False, create=True)
 
1152
        k = make_file_knit('test', get_transport('.'), delta=False, create=True)
1206
1153
        k.add_lines('text-1', [], ['a\n',    'b'  ])
1207
1154
        k.add_lines('text-2', ['text-1'], ['a\rb\n', 'b\n'])
1208
1155
        # reopening ensures maximum room for confusion
1209
 
        k = KnitVersionedFile('test', get_transport('.'), delta=False, create=True)
 
1156
        k = make_file_knit('test', get_transport('.'), delta=False, create=True)
1210
1157
        self.assertEquals(k.get_lines('text-1'), ['a\n',    'b'  ])
1211
1158
        self.assertEquals(k.get_lines('text-2'), ['a\rb\n', 'b\n'])
1212
1159
 
1234
1181
 
1235
1182
    def test_add_delta(self):
1236
1183
        """Store in knit with parents"""
1237
 
        k = KnitVersionedFile('test', get_transport('.'), factory=KnitPlainFactory(),
1238
 
            delta=True, create=True)
 
1184
        k = self.make_test_knit(annotate=False)
1239
1185
        self.add_stock_one_and_one_a(k)
1240
 
        k.clear_cache()
1241
1186
        self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
1242
1187
 
1243
1188
    def test_add_delta_knit_graph_index(self):
1245
1190
        index = InMemoryGraphIndex(2)
1246
1191
        knit_index = KnitGraphIndex(index, add_callback=index.add_nodes,
1247
1192
            deltas=True)
1248
 
        k = KnitVersionedFile('test', get_transport('.'),
1249
 
            delta=True, create=True, index=knit_index)
 
1193
        k = self.make_test_knit(annotate=True, index=knit_index)
1250
1194
        self.add_stock_one_and_one_a(k)
1251
 
        k.clear_cache()
1252
1195
        self.assertEqualDiff(''.join(k.get_lines('text-1a')), TEXT_1A)
1253
1196
        # check the index had the right data added.
1254
1197
        self.assertEqual(set([
1260
1203
 
1261
1204
    def test_annotate(self):
1262
1205
        """Annotations"""
1263
 
        k = KnitVersionedFile('knit', get_transport('.'), factory=KnitAnnotateFactory(),
1264
 
            delta=True, create=True)
 
1206
        k = self.make_test_knit(annotate=True, name='knit')
1265
1207
        self.insert_and_test_small_annotate(k)
1266
1208
 
1267
1209
    def insert_and_test_small_annotate(self, k):
1275
1217
 
1276
1218
    def test_annotate_fulltext(self):
1277
1219
        """Annotations"""
1278
 
        k = KnitVersionedFile('knit', get_transport('.'), factory=KnitAnnotateFactory(),
1279
 
            delta=False, create=True)
 
1220
        k = self.make_test_knit(annotate=True, name='knit', delta=False)
1280
1221
        self.insert_and_test_small_annotate(k)
1281
1222
 
1282
1223
    def test_annotate_merge_1(self):
1353
1294
        self.assertEquals(origins[2], ('text-1', 'c\n'))
1354
1295
 
1355
1296
    def _test_join_with_factories(self, k1_factory, k2_factory):
1356
 
        k1 = KnitVersionedFile('test1', get_transport('.'), factory=k1_factory, create=True)
 
1297
        k1 = make_file_knit('test1', get_transport('.'), factory=k1_factory, create=True)
1357
1298
        k1.add_lines('text-a', [], ['a1\n', 'a2\n', 'a3\n'])
1358
1299
        k1.add_lines('text-b', ['text-a'], ['a1\n', 'b2\n', 'a3\n'])
1359
1300
        k1.add_lines('text-c', [], ['c1\n', 'c2\n', 'c3\n'])
1360
1301
        k1.add_lines('text-d', ['text-c'], ['c1\n', 'd2\n', 'd3\n'])
1361
1302
        k1.add_lines('text-m', ['text-b', 'text-d'], ['a1\n', 'b2\n', 'd3\n'])
1362
 
        k2 = KnitVersionedFile('test2', get_transport('.'), factory=k2_factory, create=True)
 
1303
        k2 = make_file_knit('test2', get_transport('.'), factory=k2_factory, create=True)
1363
1304
        count = k2.join(k1, version_ids=['text-m'])
1364
1305
        self.assertEquals(count, 5)
1365
1306
        self.assertTrue(k2.has_version('text-a'))
1386
1327
        self._test_join_with_factories(KnitPlainFactory(), None)
1387
1328
 
1388
1329
    def test_reannotate(self):
1389
 
        k1 = KnitVersionedFile('knit1', get_transport('.'),
 
1330
        k1 = make_file_knit('knit1', get_transport('.'),
1390
1331
                               factory=KnitAnnotateFactory(), create=True)
1391
1332
        # 0
1392
1333
        k1.add_lines('text-a', [], ['a\n', 'b\n'])
1393
1334
        # 1
1394
1335
        k1.add_lines('text-b', ['text-a'], ['a\n', 'c\n'])
1395
1336
 
1396
 
        k2 = KnitVersionedFile('test2', get_transport('.'),
 
1337
        k2 = make_file_knit('test2', get_transport('.'),
1397
1338
                               factory=KnitAnnotateFactory(), create=True)
1398
1339
        k2.join(k1, version_ids=['text-b'])
1399
1340
 
1416
1357
 
1417
1358
    def test_get_line_delta_texts(self):
1418
1359
        """Make sure we can call get_texts on text with reused line deltas"""
1419
 
        k1 = KnitVersionedFile('test1', get_transport('.'), 
1420
 
                               factory=KnitPlainFactory(), create=True)
 
1360
        k1 = make_file_knit('test1', get_transport('.'),
 
1361
            factory=KnitPlainFactory(), create=True)
1421
1362
        for t in range(3):
1422
1363
            if t == 0:
1423
1364
                parents = []
1428
1369
        
1429
1370
    def test_iter_lines_reads_in_order(self):
1430
1371
        instrumented_t = get_transport('trace+memory:///')
1431
 
        k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
 
1372
        k1 = make_file_knit('id', instrumented_t, create=True, delta=True)
1432
1373
        self.assertEqual([('get', 'id.kndx',)], instrumented_t._activity)
1433
1374
        # add texts with no required ordering
1434
1375
        k1.add_lines('base', [], ['text\n'])
1435
1376
        k1.add_lines('base2', [], ['text2\n'])
1436
 
        k1.clear_cache()
1437
1377
        # clear the logged activity, but preserve the list instance in case of
1438
1378
        # clones pointing at it.
1439
1379
        del instrumented_t._activity[:]
1445
1385
            instrumented_t._activity)
1446
1386
        self.assertEqual([('text\n', 'base'), ('text2\n', 'base2')], results)
1447
1387
 
1448
 
    def test_create_empty_annotated(self):
1449
 
        k1 = self.make_test_knit(True)
1450
 
        # 0
1451
 
        k1.add_lines('text-a', [], ['a\n', 'b\n'])
1452
 
        k2 = k1.create_empty('t', MemoryTransport())
1453
 
        self.assertTrue(isinstance(k2.factory, KnitAnnotateFactory))
1454
 
        self.assertEqual(k1.delta, k2.delta)
1455
 
        # the generic test checks for empty content and file class
1456
 
 
1457
1388
    def test_knit_format(self):
1458
1389
        # this tests that a new knit index file has the expected content
1459
1390
        # and that is writes the data we expect as records are added.
1473
1404
            "\nrevid2 line-delta 84 82 0 :",
1474
1405
            'test.kndx')
1475
1406
        # we should be able to load this file again
1476
 
        knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
 
1407
        knit = make_file_knit('test', get_transport('.'), access_mode='r')
1477
1408
        self.assertEqual(['revid', 'revid2'], knit.versions())
1478
1409
        # write a short write to the file and ensure that its ignored
1479
1410
        indexfile = file('test.kndx', 'ab')
1480
1411
        indexfile.write('\nrevid3 line-delta 166 82 1 2 3 4 5 .phwoar:demo ')
1481
1412
        indexfile.close()
1482
1413
        # we should be able to load this file again
1483
 
        knit = KnitVersionedFile('test', get_transport('.'), access_mode='w')
 
1414
        knit = make_file_knit('test', get_transport('.'), access_mode='w')
1484
1415
        self.assertEqual(['revid', 'revid2'], knit.versions())
1485
1416
        # and add a revision with the same id the failed write had
1486
1417
        knit.add_lines('revid3', ['revid2'], ['a\n'])
1487
1418
        # and when reading it revid3 should now appear.
1488
 
        knit = KnitVersionedFile('test', get_transport('.'), access_mode='r')
 
1419
        knit = make_file_knit('test', get_transport('.'), access_mode='r')
1489
1420
        self.assertEqual(['revid', 'revid2', 'revid3'], knit.versions())
1490
1421
        self.assertEqual({'revid3':('revid2',)}, knit.get_parent_map(['revid3']))
1491
1422
 
1506
1437
        """create_parent_dir can create knits in nonexistant dirs"""
1507
1438
        # Has no effect if we don't set 'delay_create'
1508
1439
        trans = get_transport('.')
1509
 
        self.assertRaises(NoSuchFile, KnitVersionedFile, 'dir/test',
 
1440
        self.assertRaises(NoSuchFile, make_file_knit, 'dir/test',
1510
1441
                          trans, access_mode='w', factory=None,
1511
1442
                          create=True, create_parent_dir=True)
1512
1443
        # Nothing should have changed yet
1513
 
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
 
1444
        knit = make_file_knit('dir/test', trans, access_mode='w',
1514
1445
                                 factory=None, create=True,
1515
1446
                                 create_parent_dir=True,
1516
1447
                                 delay_create=True)
1531
1462
        if not trans._can_roundtrip_unix_modebits():
1532
1463
            # Can't roundtrip, so no need to run this test
1533
1464
            return
1534
 
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1535
 
                                 factory=None, create=True,
1536
 
                                 create_parent_dir=True,
1537
 
                                 delay_create=True,
1538
 
                                 file_mode=0600,
1539
 
                                 dir_mode=0700)
 
1465
        knit = make_file_knit('dir/test', trans, access_mode='w', factory=None,
 
1466
            create=True, create_parent_dir=True, delay_create=True,
 
1467
            file_mode=0600, dir_mode=0700)
1540
1468
        knit.add_lines('revid', [], ['a\n'])
1541
1469
        self.assertTransportMode(trans, 'dir', 0700)
1542
1470
        self.assertTransportMode(trans, 'dir/test.knit', 0600)
1547
1475
        if not trans._can_roundtrip_unix_modebits():
1548
1476
            # Can't roundtrip, so no need to run this test
1549
1477
            return
1550
 
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1551
 
                                 factory=None, create=True,
1552
 
                                 create_parent_dir=True,
1553
 
                                 delay_create=True,
1554
 
                                 file_mode=0660,
1555
 
                                 dir_mode=0770)
 
1478
        knit = make_file_knit('dir/test', trans, access_mode='w', factory=None,
 
1479
            create=True, create_parent_dir=True, delay_create=True,
 
1480
            file_mode=0660, dir_mode=0770)
1556
1481
        knit.add_lines('revid', [], ['a\n'])
1557
1482
        self.assertTransportMode(trans, 'dir', 0770)
1558
1483
        self.assertTransportMode(trans, 'dir/test.knit', 0660)
1563
1488
        if not trans._can_roundtrip_unix_modebits():
1564
1489
            # Can't roundtrip, so no need to run this test
1565
1490
            return
1566
 
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
1567
 
                                 factory=None, create=True,
1568
 
                                 create_parent_dir=True,
1569
 
                                 delay_create=True,
1570
 
                                 file_mode=0666,
1571
 
                                 dir_mode=0777)
 
1491
        knit = make_file_knit('dir/test', trans, access_mode='w', factory=None,
 
1492
            create=True, create_parent_dir=True, delay_create=True,
 
1493
            file_mode=0666, dir_mode=0777)
1572
1494
        knit.add_lines('revid', [], ['a\n'])
1573
1495
        self.assertTransportMode(trans, 'dir', 0777)
1574
1496
        self.assertTransportMode(trans, 'dir/test.knit', 0666)
1897
1819
        # The target knit object is in a consistent state, i.e. the record we
1898
1820
        # just added is immediately visible.
1899
1821
        self.assertTrue(target.has_version('text-a'))
1900
 
        self.assertTrue(target.has_ghost('text-ghost'))
 
1822
        self.assertFalse(target.has_version('text-ghost'))
 
1823
        self.assertEqual({'text-a':('text-ghost',)},
 
1824
            target.get_parent_map(['text-a', 'text-ghost']))
1901
1825
        self.assertEqual(split_lines(TEXT_1), target.get_lines('text-a'))
1902
1826
 
1903
1827
    def test_insert_data_stream_inconsistent_version_lines(self):
1943
1867
            errors.KnitDataStreamUnknown,
1944
1868
            target.insert_data_stream, data_stream)
1945
1869
 
 
1870
    def test_insert_data_stream_bug_208418(self):
 
1871
        """You can insert a stream with an incompatible format, even when:
 
1872
          * the stream has a line-delta record,
 
1873
          * whose parent is in the target, also stored as a line-delta
 
1874
 
 
1875
        See <https://launchpad.net/bugs/208418>.
 
1876
        """
 
1877
        base_lines = split_lines(TEXT_1)
 
1878
        # Make the target
 
1879
        target = self.make_test_knit(name='target', annotate=True)
 
1880
        target.add_lines('version-1', [], base_lines)
 
1881
        target.add_lines('version-2', ['version-1'], base_lines + ['a\n'])
 
1882
        # The second record should be a delta.
 
1883
        self.assertEqual('line-delta', target._index.get_method('version-2'))
 
1884
        
 
1885
        # Make a source, with a different format, but the same data
 
1886
        source = self.make_test_knit(name='source', annotate=False)
 
1887
        source.add_lines('version-1', [], base_lines)
 
1888
        source.add_lines('version-2', ['version-1'], base_lines + ['a\n'])
 
1889
        # Now add another record, which should be stored as a delta against
 
1890
        # version-2.
 
1891
        source.add_lines('version-3', ['version-2'], base_lines + ['b\n'])
 
1892
        self.assertEqual('line-delta', source._index.get_method('version-3'))
 
1893
 
 
1894
        # Make a stream of the new version
 
1895
        data_stream = source.get_data_stream(['version-3'])
 
1896
        # And insert into the target
 
1897
        target.insert_data_stream(data_stream)
 
1898
        # No errors should have been raised.
 
1899
 
 
1900
 
1946
1901
    #  * test that a stream of "already present version, then new version"
1947
1902
    #    inserts correctly.
1948
1903
 
2075
2030
    def test_weave_to_knit_matches(self):
2076
2031
        # check that the WeaveToKnit is_compatible function
2077
2032
        # registers True for a Weave to a Knit.
2078
 
        w = Weave()
 
2033
        w = Weave(get_scope=lambda:None)
2079
2034
        k = self.make_test_knit()
2080
2035
        self.failUnless(WeaveToKnit.is_compatible(w, k))
2081
2036
        self.failIf(WeaveToKnit.is_compatible(k, w))
2083
2038
        self.failIf(WeaveToKnit.is_compatible(k, k))
2084
2039
 
2085
2040
 
2086
 
class TestKnitCaching(KnitTests):
2087
 
    
2088
 
    def create_knit(self):
2089
 
        k = self.make_test_knit(True)
2090
 
        k.add_lines('text-1', [], split_lines(TEXT_1))
2091
 
        k.add_lines('text-2', [], split_lines(TEXT_2))
2092
 
        return k
2093
 
 
2094
 
    def test_no_caching(self):
2095
 
        k = self.create_knit()
2096
 
        # Nothing should be cached without setting 'enable_cache'
2097
 
        self.assertEqual({}, k._data._cache)
2098
 
 
2099
 
    def test_cache_data_read_raw(self):
2100
 
        k = self.create_knit()
2101
 
 
2102
 
        # Now cache and read
2103
 
        k.enable_cache()
2104
 
 
2105
 
        def read_one_raw(version):
2106
 
            pos_map = k._get_components_positions([version])
2107
 
            method, index_memo, next = pos_map[version]
2108
 
            lst = list(k._data.read_records_iter_raw([(version, index_memo)]))
2109
 
            self.assertEqual(1, len(lst))
2110
 
            return lst[0]
2111
 
 
2112
 
        val = read_one_raw('text-1')
2113
 
        self.assertEqual({'text-1':val[1]}, k._data._cache)
2114
 
 
2115
 
        k.clear_cache()
2116
 
        # After clear, new reads are not cached
2117
 
        self.assertEqual({}, k._data._cache)
2118
 
 
2119
 
        val2 = read_one_raw('text-1')
2120
 
        self.assertEqual(val, val2)
2121
 
        self.assertEqual({}, k._data._cache)
2122
 
 
2123
 
    def test_cache_data_read(self):
2124
 
        k = self.create_knit()
2125
 
 
2126
 
        def read_one(version):
2127
 
            pos_map = k._get_components_positions([version])
2128
 
            method, index_memo, next = pos_map[version]
2129
 
            lst = list(k._data.read_records_iter([(version, index_memo)]))
2130
 
            self.assertEqual(1, len(lst))
2131
 
            return lst[0]
2132
 
 
2133
 
        # Now cache and read
2134
 
        k.enable_cache()
2135
 
 
2136
 
        val = read_one('text-2')
2137
 
        self.assertEqual(['text-2'], k._data._cache.keys())
2138
 
        self.assertEqual('text-2', val[0])
2139
 
        content, digest = k._data._parse_record('text-2',
2140
 
                                                k._data._cache['text-2'])
2141
 
        self.assertEqual(content, val[1])
2142
 
        self.assertEqual(digest, val[2])
2143
 
 
2144
 
        k.clear_cache()
2145
 
        self.assertEqual({}, k._data._cache)
2146
 
 
2147
 
        val2 = read_one('text-2')
2148
 
        self.assertEqual(val, val2)
2149
 
        self.assertEqual({}, k._data._cache)
2150
 
 
2151
 
    def test_cache_read(self):
2152
 
        k = self.create_knit()
2153
 
        k.enable_cache()
2154
 
 
2155
 
        text = k.get_text('text-1')
2156
 
        self.assertEqual(TEXT_1, text)
2157
 
        self.assertEqual(['text-1'], k._data._cache.keys())
2158
 
 
2159
 
        k.clear_cache()
2160
 
        self.assertEqual({}, k._data._cache)
2161
 
 
2162
 
        text = k.get_text('text-1')
2163
 
        self.assertEqual(TEXT_1, text)
2164
 
        self.assertEqual({}, k._data._cache)
2165
 
 
2166
 
 
2167
2041
class TestKnitIndex(KnitTests):
2168
2042
 
2169
2043
    def test_add_versions_dictionary_compresses(self):
2289
2163
        return KnitGraphIndex(combined_index, deltas=deltas,
2290
2164
            add_callback=add_callback)
2291
2165
 
2292
 
    def test_get_graph(self):
2293
 
        index = self.two_graph_index()
2294
 
        self.assertEqual(set([
2295
 
            ('tip', ('parent', )),
2296
 
            ('tail', ()),
2297
 
            ('parent', ('tail', 'ghost')),
2298
 
            ('separate', ()),
2299
 
            ]), set(index.get_graph()))
2300
 
 
2301
2166
    def test_get_ancestry(self):
2302
2167
        # get_ancestry is defined as eliding ghosts, not erroring.
2303
2168
        index = self.two_graph_index()
2513
2378
             ('tip', 'no-eol,line-delta', (None, 0, 100), ['parent'])])
2514
2379
        self.assertEqual([], self.caught_entries)
2515
2380
 
2516
 
    def test_iter_parents(self):
2517
 
        index1 = self.make_g_index('1', 1, [
2518
 
        # no parents
2519
 
            (('r0', ), 'N0 100', ([], )),
2520
 
        # 1 parent
2521
 
            (('r1', ), '', ([('r0', )], ))])
2522
 
        index2 = self.make_g_index('2', 1, [
2523
 
        # 2 parents
2524
 
            (('r2', ), 'N0 100', ([('r1', ), ('r0', )], )),
2525
 
            ])
2526
 
        combined_index = CombinedGraphIndex([index1, index2])
2527
 
        index = KnitGraphIndex(combined_index)
2528
 
        # XXX TODO a ghost
2529
 
        # cases: each sample data individually:
2530
 
        self.assertEqual(set([('r0', ())]),
2531
 
            set(index.iter_parents(['r0'])))
2532
 
        self.assertEqual(set([('r1', ('r0', ))]),
2533
 
            set(index.iter_parents(['r1'])))
2534
 
        self.assertEqual(set([('r2', ('r1', 'r0'))]),
2535
 
            set(index.iter_parents(['r2'])))
2536
 
        # no nodes returned for a missing node
2537
 
        self.assertEqual(set(),
2538
 
            set(index.iter_parents(['missing'])))
2539
 
        # 1 node returned with missing nodes skipped
2540
 
        self.assertEqual(set([('r1', ('r0', ))]),
2541
 
            set(index.iter_parents(['ghost1', 'r1', 'ghost'])))
2542
 
        # 2 nodes returned
2543
 
        self.assertEqual(set([('r0', ()), ('r1', ('r0', ))]),
2544
 
            set(index.iter_parents(['r0', 'r1'])))
2545
 
        # 2 nodes returned, missing skipped
2546
 
        self.assertEqual(set([('r0', ()), ('r1', ('r0', ))]),
2547
 
            set(index.iter_parents(['a', 'r0', 'b', 'r1', 'c'])))
2548
 
 
2549
 
 
2550
2381
class TestNoParentsGraphIndexKnit(KnitTests):
2551
2382
    """Tests for knits using KnitGraphIndex with no parents."""
2552
2383
 
2587
2418
        return KnitGraphIndex(combined_index, parents=False,
2588
2419
            add_callback=add_callback)
2589
2420
 
2590
 
    def test_get_graph(self):
2591
 
        index = self.two_graph_index()
2592
 
        self.assertEqual(set([
2593
 
            ('tip', ()),
2594
 
            ('tail', ()),
2595
 
            ('parent', ()),
2596
 
            ('separate', ()),
2597
 
            ]), set(index.get_graph()))
2598
 
 
2599
2421
    def test_get_ancestry(self):
2600
2422
        # with no parents, ancestry is always just the key.
2601
2423
        index = self.two_graph_index()
2773
2595
             ('tip', 'no-eol,line-delta', (None, 0, 100), [])])
2774
2596
        self.assertEqual([], self.caught_entries)
2775
2597
 
2776
 
    def test_iter_parents(self):
2777
 
        index = self.two_graph_index()
2778
 
        self.assertEqual(set([
2779
 
            ('tip', ()), ('tail', ()), ('parent', ()), ('separate', ())
2780
 
            ]),
2781
 
            set(index.iter_parents(['tip', 'tail', 'ghost', 'parent', 'separate'])))
2782
 
        self.assertEqual(set([('tip', ())]),
2783
 
            set(index.iter_parents(['tip'])))
2784
 
        self.assertEqual(set(),
2785
 
            set(index.iter_parents([])))
2786
 
 
2787
 
 
2788
2598
class TestPackKnits(KnitTests):
2789
2599
    """Tests that use a _PackAccess and KnitGraphIndex."""
2790
2600
 
2839
2649
            set(result),
2840
2650
            set(index.get_ancestry(ancestry_versions, False)))
2841
2651
 
2842
 
    def assertIterParents(self, knit, versions, parent_versions, result):
2843
 
        """Check the result of an iter_parents call on knit."""
2844
 
        index = self.get_index(knit, knit.get_data_stream(versions))
2845
 
        self.assertEqual(result, index.iter_parents(parent_versions))
2846
 
 
2847
2652
    def assertGetMethod(self, knit, versions, version, result):
2848
2653
        index = self.get_index(knit, knit.get_data_stream(versions))
2849
2654
        self.assertEqual(result, index.get_method(version))
2914
2719
        # we thunk across.
2915
2720
        self.assertGetMethod(knit, ['c'], 'b', 'fulltext')
2916
2721
 
2917
 
    def test_iter_parents(self):
2918
 
        knit = self.make_knit_with_4_versions_2_dags()
2919
 
        self.assertIterParents(knit, ['a'], ['a'], [('a', ())])
2920
 
        self.assertIterParents(knit, ['a', 'b'], ['a', 'b'],
2921
 
            [('a', ()), ('b', ())])
2922
 
        self.assertIterParents(knit, ['a', 'b', 'c'], ['a', 'b', 'c'],
2923
 
            [('a', ()), ('b', ()), ('c', ('b', 'a'))])
2924
 
        self.assertIterParents(knit, ['a', 'b', 'c', 'd'],
2925
 
            ['a', 'b', 'c', 'd'],
2926
 
            [('a', ()), ('b', ()), ('c', ('b', 'a')), ('d', ('e', 'f'))])
2927
 
        self.assertIterParents(knit, ['c'], ['a', 'b', 'c'],
2928
 
            [('c', ('b', 'a'))])
2929
 
 
2930
2722
    def test_get_options(self):
2931
2723
        knit = self.make_knit_with_4_versions_2_dags()
2932
2724
        self.assertGetOptions(knit, 'a', ['no-eol', 'fulltext'])
3018
2810
            knit.get_data_stream([]))
3019
2811
        self.assertRaises(errors.KnitCorrupt,
3020
2812
            list, access.get_raw_records([(True, "A", None, None)]))
 
2813
 
 
2814
 
 
2815
class TestFormatSignatures(KnitTests):
 
2816
 
 
2817
    def test_knit_format_signatures(self):
 
2818
        """Different formats of knit have different signature strings."""
 
2819
        knit = self.make_test_knit(name='a', annotate=True)
 
2820
        self.assertEqual('knit-annotated', knit.get_format_signature())
 
2821
        knit = self.make_test_knit(name='p', annotate=False)
 
2822
        self.assertEqual('knit-plain', knit.get_format_signature())