1315
1315
self.assertEquals(origins[1], ('text-1', 'b\n'))
1316
1316
self.assertEquals(origins[2], ('text-1', 'c\n'))
1318
def test_knit_join(self):
1319
"""Store in knit with parents"""
1320
k1 = KnitVersionedFile('test1', get_transport('.'), factory=KnitPlainFactory(), create=True)
1321
k1.add_lines('text-a', [], split_lines(TEXT_1))
1322
k1.add_lines('text-b', ['text-a'], split_lines(TEXT_1))
1324
k1.add_lines('text-c', [], split_lines(TEXT_1))
1325
k1.add_lines('text-d', ['text-c'], split_lines(TEXT_1))
1327
k1.add_lines('text-m', ['text-b', 'text-d'], split_lines(TEXT_1))
1329
k2 = KnitVersionedFile('test2', get_transport('.'), factory=KnitPlainFactory(), create=True)
1318
def _test_join_with_factories(self, k1_factory, k2_factory):
1319
k1 = KnitVersionedFile('test1', get_transport('.'), factory=k1_factory, create=True)
1320
k1.add_lines('text-a', [], ['a1\n', 'a2\n', 'a3\n'])
1321
k1.add_lines('text-b', ['text-a'], ['a1\n', 'b2\n', 'a3\n'])
1322
k1.add_lines('text-c', [], ['c1\n', 'c2\n', 'c3\n'])
1323
k1.add_lines('text-d', ['text-c'], ['c1\n', 'd2\n', 'd3\n'])
1324
k1.add_lines('text-m', ['text-b', 'text-d'], ['a1\n', 'b2\n', 'd3\n'])
1325
k2 = KnitVersionedFile('test2', get_transport('.'), factory=k2_factory, create=True)
1330
1326
count = k2.join(k1, version_ids=['text-m'])
1331
1327
self.assertEquals(count, 5)
1332
1328
self.assertTrue(k2.has_version('text-a'))
1333
1329
self.assertTrue(k2.has_version('text-c'))
1330
origins = k2.annotate('text-m')
1331
self.assertEquals(origins[0], ('text-a', 'a1\n'))
1332
self.assertEquals(origins[1], ('text-b', 'b2\n'))
1333
self.assertEquals(origins[2], ('text-d', 'd3\n'))
1335
def test_knit_join_plain_to_plain(self):
1336
"""Test joining a plain knit with a plain knit."""
1337
self._test_join_with_factories(KnitPlainFactory(), KnitPlainFactory())
1339
def test_knit_join_anno_to_anno(self):
1340
"""Test joining an annotated knit with an annotated knit."""
1341
self._test_join_with_factories(None, None)
1343
def test_knit_join_anno_to_plain(self):
1344
"""Test joining an annotated knit with a plain knit."""
1345
self._test_join_with_factories(None, KnitPlainFactory())
1347
def test_knit_join_plain_to_anno(self):
1348
"""Test joining a plain knit with an annotated knit."""
1349
self._test_join_with_factories(KnitPlainFactory(), None)
1335
1351
def test_reannotate(self):
1336
1352
k1 = KnitVersionedFile('knit1', get_transport('.'),
1374
1390
k1.get_texts(('%d' % t) for t in range(3))
1376
1392
def test_iter_lines_reads_in_order(self):
1377
t = MemoryTransport()
1378
instrumented_t = TransportLogger(t)
1393
instrumented_t = get_transport('trace+memory:///')
1379
1394
k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
1380
self.assertEqual([('id.kndx',)], instrumented_t._calls)
1395
self.assertEqual([('get', 'id.kndx',)], instrumented_t._activity)
1381
1396
# add texts with no required ordering
1382
1397
k1.add_lines('base', [], ['text\n'])
1383
1398
k1.add_lines('base2', [], ['text2\n'])
1384
1399
k1.clear_cache()
1385
instrumented_t._calls = []
1400
# clear the logged activity, but preserve the list instance in case of
1401
# clones pointing at it.
1402
del instrumented_t._activity[:]
1386
1403
# request a last-first iteration
1387
results = list(k1.iter_lines_added_or_present_in_versions(['base2', 'base']))
1388
self.assertEqual([('id.knit', [(0, 87), (87, 89)])], instrumented_t._calls)
1404
results = list(k1.iter_lines_added_or_present_in_versions(
1407
[('readv', 'id.knit', [(0, 87), (87, 89)], False, None)],
1408
instrumented_t._activity)
1389
1409
self.assertEqual(['text\n', 'text2\n'], results)
1391
1411
def test_create_empty_annotated(self):