/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_store.py

  • Committer: Aaron Bentley
  • Date: 2006-04-07 22:46:52 UTC
  • mfrom: (1645 +trunk)
  • mto: This revision was merged to the branch mainline in revision 1727.
  • Revision ID: aaron.bentley@utoronto.ca-20060407224652-4925bc3735b926f8
Merged latest bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
20
20
import os
21
21
import gzip
22
22
 
 
23
import bzrlib.errors as errors
23
24
from bzrlib.errors import BzrError, UnlistableStore, NoSuchFile
24
 
from bzrlib.store import copy_all
25
25
from bzrlib.transport.local import LocalTransport
26
26
from bzrlib.store.text import TextStore
27
 
from bzrlib.tests import TestCase, TestCaseInTempDir
 
27
from bzrlib.tests import TestCase, TestCaseInTempDir, TestCaseWithTransport
28
28
import bzrlib.store as store
 
29
import bzrlib.transactions as transactions
29
30
import bzrlib.transport as transport
30
31
from bzrlib.transport.memory import MemoryTransport
31
32
 
50
51
        store_a.add('foo', '1')
51
52
        os.mkdir('b')
52
53
        store_b = self.get_store('b')
53
 
        copy_all(store_a, store_b)
 
54
        store_b.copy_all_ids(store_a)
54
55
        self.assertEqual(store_a.get('1').read(), 'foo')
55
56
        self.assertEqual(store_b.get('1').read(), 'foo')
56
57
        # TODO: Switch the exception form UnlistableStore to
92
93
        self.assertEqual(store.total_size(), (2, 55))
93
94
        
94
95
    def test__relpath_suffixed(self):
95
 
        my_store = TextStore(MockTransport(), True, compressed=True)
 
96
        my_store = TextStore(MockTransport(),
 
97
                             prefixed=True, compressed=True)
96
98
        my_store.register_suffix('dsc')
97
99
        self.assertEqual('45/foo.dsc', my_store._relpath('foo', ['dsc']))
98
100
 
357
359
                         my_store.get('missing', 'sig').read())
358
360
 
359
361
    def test___iter__no_suffix(self):
360
 
        my_store = TextStore(MemoryTransport(), False, compressed=False)
 
362
        my_store = TextStore(MemoryTransport(),
 
363
                             prefixed=False, compressed=False)
361
364
        stream = StringIO("content")
362
365
        my_store.add(stream, "foo")
363
366
        self.assertEqual(set(['foo']),
382
385
 
383
386
    def test_copy_suffixes(self):
384
387
        from_store = self.get_populated_store()
385
 
        to_store = TextStore(MemoryTransport(), True, compressed=True)
 
388
        to_store = TextStore(MemoryTransport(),
 
389
                             prefixed=True, compressed=True)
386
390
        to_store.register_suffix('sig')
387
 
        copy_all(from_store, to_store)
 
391
        to_store.copy_all_ids(from_store)
388
392
        self.assertEqual(1, len(to_store))
389
393
        self.assertEqual(set(['foo']), set(to_store.__iter__()))
390
394
        self.assertEqual('content', to_store.get('foo').read())
394
398
    def test_relpath_escaped(self):
395
399
        my_store = store.TransportStore(MemoryTransport())
396
400
        self.assertEqual('%25', my_store._relpath('%'))
 
401
 
 
402
 
 
403
class TestVersionFileStore(TestCaseWithTransport):
 
404
 
 
405
    def setUp(self):
 
406
        super(TestVersionFileStore, self).setUp()
 
407
        self.vfstore = store.versioned.VersionedFileStore(MemoryTransport())
 
408
 
 
409
    def test_get_weave_registers_dirty_in_write(self):
 
410
        transaction = transactions.WriteTransaction()
 
411
        vf = self.vfstore.get_weave_or_empty('id', transaction)
 
412
        transaction.finish()
 
413
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
 
414
        transaction = transactions.WriteTransaction()
 
415
        vf = self.vfstore.get_weave('id', transaction)
 
416
        transaction.finish()
 
417
        self.assertRaises(errors.OutSideTransaction, vf.add_lines, 'b', [], [])
 
418
 
 
419
    def test_get_weave_or_empty_readonly_fails(self):
 
420
        transaction = transactions.ReadOnlyTransaction()
 
421
        vf = self.assertRaises(errors.ReadOnlyError,
 
422
                               self.vfstore.get_weave_or_empty,
 
423
                               'id',
 
424
                               transaction)
 
425
 
 
426
    def test_get_weave_readonly_cant_write(self):
 
427
        transaction = transactions.WriteTransaction()
 
428
        vf = self.vfstore.get_weave_or_empty('id', transaction)
 
429
        transaction.finish()
 
430
        transaction = transactions.ReadOnlyTransaction()
 
431
        vf = self.vfstore.get_weave_or_empty('id', transaction)
 
432
        self.assertRaises(errors.ReadOnlyError, vf.add_lines, 'b', [], [])
 
433