← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_knit.py

  • Committer: John Arbash Meinel
  • Date: 2006-08-09 14:43:27 UTC
  • mto: This revision was merged to the branch mainline in revision 1912.
  • Revision ID: john@arbash-meinel.com-20060809144327-d604af2edf646794
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_knit.py
272
272
        self.assertEquals(origins[0], ('text-c', 'z\n'))
273
273
        self.assertEquals(origins[1], ('text-b', 'c\n'))
274
274
 
275
 
    def test_extraction_reads_components_once(self):
276
 
        t = MemoryTransport()
277
 
        instrumented_t = TransportLogger(t)
278
 
        k1 = KnitVersionedFile('id', instrumented_t, create=True, delta=True)
279
 
        # should read the index
280
 
        self.assertEqual([('id.kndx',)], instrumented_t._calls)
281
 
        instrumented_t._calls = []
282
 
        # add a text       
283
 
        k1.add_lines('base', [], ['text\n'])
284
 
        # should not have read at all
285
 
        self.assertEqual([], instrumented_t._calls)
286
 
 
287
 
        # add a text
288
 
        k1.add_lines('sub', ['base'], ['text\n', 'text2\n'])
289
 
        # should not have read at all
290
 
        self.assertEqual([], instrumented_t._calls)
291
 
        
292
 
        # read a text
293
 
        k1.get_lines('sub')
294
 
        # should not have read at all
295
 
        self.assertEqual([], instrumented_t._calls)
296
 
 
297
 
        # clear the cache
298
 
        k1.clear_cache()
299
 
 
300
 
        # read a text
301
 
        k1.get_lines('base')
302
 
        # should have read a component
303
 
        # should not have read the first component only
304
 
        self.assertEqual([('id.knit', [(0, 87)])], instrumented_t._calls)
305
 
        instrumented_t._calls = []
306
 
        # read again
307
 
        k1.get_lines('base')
308
 
        # should not have read at all
309
 
        self.assertEqual([], instrumented_t._calls)
310
 
        # and now read the other component
311
 
        k1.get_lines('sub')
312
 
        # should have read the second component
313
 
        self.assertEqual([('id.knit', [(87, 93)])], instrumented_t._calls)
314
 
        instrumented_t._calls = []
315
 
 
316
 
        # clear the cache
317
 
        k1.clear_cache()
318
 
        # add a text cold 
319
 
        k1.add_lines('sub2', ['base'], ['text\n', 'text3\n'])
320
 
        # should read the first component only
321
 
        self.assertEqual([('id.knit', [(0, 87)])], instrumented_t._calls)
 
275
    def test_get_line_delta_texts(self):
 
276
        """Make sure we can call get_texts on text with reused line deltas"""
 
277
        k1 = KnitVersionedFile('test1', get_transport('.'), 
 
278
                               factory=KnitPlainFactory(), create=True)
 
279
        for t in range(3):
 
280
            if t == 0:
 
281
                parents = []
 
282
            else:
 
283
                parents = ['%d' % (t-1)]
 
284
            k1.add_lines('%d' % t, parents, ['hello\n'] * t)
 
285
        k1.get_texts(('%d' % t) for t in range(3))
322
286
        
323
287
    def test_iter_lines_reads_in_order(self):
324
288
        t = MemoryTransport()
486
450
        self.failIf(WeaveToKnit.is_compatible(k, w))
487
451
        self.failIf(WeaveToKnit.is_compatible(w, w))
488
452
        self.failIf(WeaveToKnit.is_compatible(k, k))
 
453
 
 
454
 
 
455
class TestKnitCaching(KnitTests):
 
456
    
 
457
    def create_knit(self, cache_add=False):
 
458
        k = self.make_test_knit(True)
 
459
        if cache_add:
 
460
            k.enable_cache()
 
461
 
 
462
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
463
        k.add_lines('text-2', [], split_lines(TEXT_2))
 
464
        return k
 
465
 
 
466
    def test_no_caching(self):
 
467
        k = self.create_knit()
 
468
        # Nothing should be cached without setting 'enable_cache'
 
469
        self.assertEqual({}, k._data._cache)
 
470
 
 
471
    def test_cache_add_and_clear(self):
 
472
        k = self.create_knit(True)
 
473
 
 
474
        self.assertEqual(['text-1', 'text-2'], sorted(k._data._cache.keys()))
 
475
 
 
476
        k.clear_cache()
 
477
        self.assertEqual({}, k._data._cache)
 
478
 
 
479
    def test_cache_data_read_raw(self):
 
480
        k = self.create_knit()
 
481
 
 
482
        # Now cache and read
 
483
        k.enable_cache()
 
484
 
 
485
        def read_one_raw(version):
 
486
            pos_map = k._get_components_positions([version])
 
487
            method, pos, size, next = pos_map[version]
 
488
            lst = list(k._data.read_records_iter_raw([(version, pos, size)]))
 
489
            self.assertEqual(1, len(lst))
 
490
            return lst[0]
 
491
 
 
492
        val = read_one_raw('text-1')
 
493
        self.assertEqual({'text-1':val[1]}, k._data._cache)
 
494
 
 
495
        k.clear_cache()
 
496
        # After clear, new reads are not cached
 
497
        self.assertEqual({}, k._data._cache)
 
498
 
 
499
        val2 = read_one_raw('text-1')
 
500
        self.assertEqual(val, val2)
 
501
        self.assertEqual({}, k._data._cache)
 
502
 
 
503
    def test_cache_data_read(self):
 
504
        k = self.create_knit()
 
505
 
 
506
        def read_one(version):
 
507
            pos_map = k._get_components_positions([version])
 
508
            method, pos, size, next = pos_map[version]
 
509
            lst = list(k._data.read_records_iter([(version, pos, size)]))
 
510
            self.assertEqual(1, len(lst))
 
511
            return lst[0]
 
512
 
 
513
        # Now cache and read
 
514
        k.enable_cache()
 
515
 
 
516
        val = read_one('text-2')
 
517
        self.assertEqual(['text-2'], k._data._cache.keys())
 
518
        self.assertEqual('text-2', val[0])
 
519
        content, digest = k._data._parse_record('text-2',
 
520
                                                k._data._cache['text-2'])
 
521
        self.assertEqual(content, val[1])
 
522
        self.assertEqual(digest, val[2])
 
523
 
 
524
        k.clear_cache()
 
525
        self.assertEqual({}, k._data._cache)
 
526
 
 
527
        val2 = read_one('text-2')
 
528
        self.assertEqual(val, val2)
 
529
        self.assertEqual({}, k._data._cache)
 
530
 
 
531
    def test_cache_read(self):
 
532
        k = self.create_knit()
 
533
        k.enable_cache()
 
534
 
 
535
        text = k.get_text('text-1')
 
536
        self.assertEqual(TEXT_1, text)
 
537
        self.assertEqual(['text-1'], k._data._cache.keys())
 
538
 
 
539
        k.clear_cache()
 
540
        self.assertEqual({}, k._data._cache)
 
541
 
 
542
        text = k.get_text('text-1')
 
543
        self.assertEqual(TEXT_1, text)
 
544
        self.assertEqual({}, k._data._cache)