← 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: Canonical.com Patch Queue Manager
  • Date: 2006-09-17 21:19:56 UTC
  • mfrom: (1997.1.6 bind-does-not-push-or-pull)
  • Revision ID: pqm@pqm.ubuntu.com-20060917211956-6e30d07da410fd1a
(Robert Collins) Change the Branch bind method to just bind rather than binding and pushing (fixes #43744 and #39542)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_knit.py
20
20
import difflib
21
21
 
22
22
 
23
 
from bzrlib.errors import KnitError, RevisionAlreadyPresent
 
23
from bzrlib.errors import KnitError, RevisionAlreadyPresent, NoSuchFile
24
24
from bzrlib.knit import (
25
25
    KnitVersionedFile,
26
26
    KnitPlainFactory,
36
36
class KnitTests(TestCaseWithTransport):
37
37
    """Class containing knit test helper routines."""
38
38
 
39
 
    def make_test_knit(self, annotate=False):
 
39
    def make_test_knit(self, annotate=False, delay_create=False):
40
40
        if not annotate:
41
41
            factory = KnitPlainFactory()
42
42
        else:
43
43
            factory = None
44
 
        return KnitVersionedFile('test', get_transport('.'), access_mode='w', factory=factory, create=True)
 
44
        return KnitVersionedFile('test', get_transport('.'), access_mode='w',
 
45
                                 factory=factory, create=True,
 
46
                                 delay_create=delay_create)
45
47
 
46
48
 
47
49
class BasicKnitTests(KnitTests):
272
274
        self.assertEquals(origins[0], ('text-c', 'z\n'))
273
275
        self.assertEquals(origins[1], ('text-b', 'c\n'))
274
276
 
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)
 
277
    def test_get_line_delta_texts(self):
 
278
        """Make sure we can call get_texts on text with reused line deltas"""
 
279
        k1 = KnitVersionedFile('test1', get_transport('.'), 
 
280
                               factory=KnitPlainFactory(), create=True)
 
281
        for t in range(3):
 
282
            if t == 0:
 
283
                parents = []
 
284
            else:
 
285
                parents = ['%d' % (t-1)]
 
286
            k1.add_lines('%d' % t, parents, ['hello\n'] * t)
 
287
        k1.get_texts(('%d' % t) for t in range(3))
322
288
        
323
289
    def test_iter_lines_reads_in_order(self):
324
290
        t = MemoryTransport()
348
314
        # this tests that a new knit index file has the expected content
349
315
        # and that is writes the data we expect as records are added.
350
316
        knit = self.make_test_knit(True)
 
317
        # Now knit files are not created until we first add data to them
351
318
        self.assertFileEqual("# bzr knit index 8\n", 'test.kndx')
352
319
        knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
353
320
        self.assertFileEqual(
378
345
        self.assertEqual(['revid', 'revid2', 'revid3'], knit.versions())
379
346
        self.assertEqual(['revid2'], knit.get_parents('revid3'))
380
347
 
 
348
    def test_delay_create(self):
 
349
        """Test that passing delay_create=True creates files late"""
 
350
        knit = self.make_test_knit(annotate=True, delay_create=True)
 
351
        self.failIfExists('test.knit')
 
352
        self.failIfExists('test.kndx')
 
353
        knit.add_lines_with_ghosts('revid', ['a_ghost'], ['a\n'])
 
354
        self.failUnlessExists('test.knit')
 
355
        self.assertFileEqual(
 
356
            "# bzr knit index 8\n"
 
357
            "\n"
 
358
            "revid fulltext 0 84 .a_ghost :",
 
359
            'test.kndx')
 
360
 
 
361
    def test_create_parent_dir(self):
 
362
        """create_parent_dir can create knits in nonexistant dirs"""
 
363
        # Has no effect if we don't set 'delay_create'
 
364
        trans = get_transport('.')
 
365
        self.assertRaises(NoSuchFile, KnitVersionedFile, 'dir/test',
 
366
                          trans, access_mode='w', factory=None,
 
367
                          create=True, create_parent_dir=True)
 
368
        # Nothing should have changed yet
 
369
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
 
370
                                 factory=None, create=True,
 
371
                                 create_parent_dir=True,
 
372
                                 delay_create=True)
 
373
        self.failIfExists('dir/test.knit')
 
374
        self.failIfExists('dir/test.kndx')
 
375
        self.failIfExists('dir')
 
376
        knit.add_lines('revid', [], ['a\n'])
 
377
        self.failUnlessExists('dir')
 
378
        self.failUnlessExists('dir/test.knit')
 
379
        self.assertFileEqual(
 
380
            "# bzr knit index 8\n"
 
381
            "\n"
 
382
            "revid fulltext 0 84  :",
 
383
            'dir/test.kndx')
 
384
 
 
385
    def test_create_mode_700(self):
 
386
        trans = get_transport('.')
 
387
        if not trans._can_roundtrip_unix_modebits():
 
388
            # Can't roundtrip, so no need to run this test
 
389
            return
 
390
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
 
391
                                 factory=None, create=True,
 
392
                                 create_parent_dir=True,
 
393
                                 delay_create=True,
 
394
                                 file_mode=0600,
 
395
                                 dir_mode=0700)
 
396
        knit.add_lines('revid', [], ['a\n'])
 
397
        self.assertTransportMode(trans, 'dir', 0700)
 
398
        self.assertTransportMode(trans, 'dir/test.knit', 0600)
 
399
        self.assertTransportMode(trans, 'dir/test.kndx', 0600)
 
400
 
 
401
    def test_create_mode_770(self):
 
402
        trans = get_transport('.')
 
403
        if not trans._can_roundtrip_unix_modebits():
 
404
            # Can't roundtrip, so no need to run this test
 
405
            return
 
406
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
 
407
                                 factory=None, create=True,
 
408
                                 create_parent_dir=True,
 
409
                                 delay_create=True,
 
410
                                 file_mode=0660,
 
411
                                 dir_mode=0770)
 
412
        knit.add_lines('revid', [], ['a\n'])
 
413
        self.assertTransportMode(trans, 'dir', 0770)
 
414
        self.assertTransportMode(trans, 'dir/test.knit', 0660)
 
415
        self.assertTransportMode(trans, 'dir/test.kndx', 0660)
 
416
 
 
417
    def test_create_mode_777(self):
 
418
        trans = get_transport('.')
 
419
        if not trans._can_roundtrip_unix_modebits():
 
420
            # Can't roundtrip, so no need to run this test
 
421
            return
 
422
        knit = KnitVersionedFile('dir/test', trans, access_mode='w',
 
423
                                 factory=None, create=True,
 
424
                                 create_parent_dir=True,
 
425
                                 delay_create=True,
 
426
                                 file_mode=0666,
 
427
                                 dir_mode=0777)
 
428
        knit.add_lines('revid', [], ['a\n'])
 
429
        self.assertTransportMode(trans, 'dir', 0777)
 
430
        self.assertTransportMode(trans, 'dir/test.knit', 0666)
 
431
        self.assertTransportMode(trans, 'dir/test.kndx', 0666)
 
432
 
381
433
    def test_plan_merge(self):
382
434
        my_knit = self.make_test_knit(annotate=True)
383
435
        my_knit.add_lines('text1', [], split_lines(TEXT_1))
486
538
        self.failIf(WeaveToKnit.is_compatible(k, w))
487
539
        self.failIf(WeaveToKnit.is_compatible(w, w))
488
540
        self.failIf(WeaveToKnit.is_compatible(k, k))
 
541
 
 
542
 
 
543
class TestKnitCaching(KnitTests):
 
544
    
 
545
    def create_knit(self, cache_add=False):
 
546
        k = self.make_test_knit(True)
 
547
        if cache_add:
 
548
            k.enable_cache()
 
549
 
 
550
        k.add_lines('text-1', [], split_lines(TEXT_1))
 
551
        k.add_lines('text-2', [], split_lines(TEXT_2))
 
552
        return k
 
553
 
 
554
    def test_no_caching(self):
 
555
        k = self.create_knit()
 
556
        # Nothing should be cached without setting 'enable_cache'
 
557
        self.assertEqual({}, k._data._cache)
 
558
 
 
559
    def test_cache_add_and_clear(self):
 
560
        k = self.create_knit(True)
 
561
 
 
562
        self.assertEqual(['text-1', 'text-2'], sorted(k._data._cache.keys()))
 
563
 
 
564
        k.clear_cache()
 
565
        self.assertEqual({}, k._data._cache)
 
566
 
 
567
    def test_cache_data_read_raw(self):
 
568
        k = self.create_knit()
 
569
 
 
570
        # Now cache and read
 
571
        k.enable_cache()
 
572
 
 
573
        def read_one_raw(version):
 
574
            pos_map = k._get_components_positions([version])
 
575
            method, pos, size, next = pos_map[version]
 
576
            lst = list(k._data.read_records_iter_raw([(version, pos, size)]))
 
577
            self.assertEqual(1, len(lst))
 
578
            return lst[0]
 
579
 
 
580
        val = read_one_raw('text-1')
 
581
        self.assertEqual({'text-1':val[1]}, k._data._cache)
 
582
 
 
583
        k.clear_cache()
 
584
        # After clear, new reads are not cached
 
585
        self.assertEqual({}, k._data._cache)
 
586
 
 
587
        val2 = read_one_raw('text-1')
 
588
        self.assertEqual(val, val2)
 
589
        self.assertEqual({}, k._data._cache)
 
590
 
 
591
    def test_cache_data_read(self):
 
592
        k = self.create_knit()
 
593
 
 
594
        def read_one(version):
 
595
            pos_map = k._get_components_positions([version])
 
596
            method, pos, size, next = pos_map[version]
 
597
            lst = list(k._data.read_records_iter([(version, pos, size)]))
 
598
            self.assertEqual(1, len(lst))
 
599
            return lst[0]
 
600
 
 
601
        # Now cache and read
 
602
        k.enable_cache()
 
603
 
 
604
        val = read_one('text-2')
 
605
        self.assertEqual(['text-2'], k._data._cache.keys())
 
606
        self.assertEqual('text-2', val[0])
 
607
        content, digest = k._data._parse_record('text-2',
 
608
                                                k._data._cache['text-2'])
 
609
        self.assertEqual(content, val[1])
 
610
        self.assertEqual(digest, val[2])
 
611
 
 
612
        k.clear_cache()
 
613
        self.assertEqual({}, k._data._cache)
 
614
 
 
615
        val2 = read_one('text-2')
 
616
        self.assertEqual(val, val2)
 
617
        self.assertEqual({}, k._data._cache)
 
618
 
 
619
    def test_cache_read(self):
 
620
        k = self.create_knit()
 
621
        k.enable_cache()
 
622
 
 
623
        text = k.get_text('text-1')
 
624
        self.assertEqual(TEXT_1, text)
 
625
        self.assertEqual(['text-1'], k._data._cache.keys())
 
626
 
 
627
        k.clear_cache()
 
628
        self.assertEqual({}, k._data._cache)
 
629
 
 
630
        text = k.get_text('text-1')
 
631
        self.assertEqual(TEXT_1, text)
 
632
        self.assertEqual({}, k._data._cache)