← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

  • Committer: Robert Collins
  • Date: 2005-10-02 21:51:29 UTC
  • mfrom: (1396)
  • mto: This revision was merged to the branch mainline in revision 1397.
  • Revision ID: robertc@robertcollins.net-20051002215128-5686c7d24bf9bdb9
merge from martins newformat branch - brings in transport abstraction

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/teststore.py
16
16
 
17
17
"""Test Store implementation
18
18
"""
19
 
 
20
 
from StringIO import StringIO
 
19
from cStringIO import StringIO
21
20
import os
22
21
 
23
 
from bzrlib.store import copy_all, ImmutableStore, RemoteStore
 
22
from bzrlib.store import copy_all
 
23
from bzrlib.transport.local import LocalTransport
 
24
from bzrlib.transport import NoSuchFile
 
25
from bzrlib.store.compressed_text import CompressedTextStore
 
26
from bzrlib.store.text import TextStore
24
27
from bzrlib.selftest import TestCase, TestCaseInTempDir
25
28
from bzrlib.errors import BzrError, UnlistableStore
26
29
import bzrlib.store
27
30
 
28
31
 
29
 
class TestStore(TestCaseInTempDir):
 
32
def fill_store(store):
 
33
    store.add(StringIO('hello'), 'a')
 
34
    store.add(StringIO('other'), 'b')
 
35
    store.add(StringIO('something'), 'c')
 
36
    store.add(StringIO('goodbye'), '123123')
 
37
 
 
38
def check_equals(tester, store, files, values, permit_failure=False):
 
39
    files = store.get(files, permit_failure=permit_failure)
 
40
    count = 0
 
41
    for f, v in zip(files, values):
 
42
        count += 1
 
43
        if v is None:
 
44
            tester.assert_(f is None)
 
45
        else:
 
46
            tester.assertEquals(f.read(), v)
 
47
    tester.assertEquals(count, len(values))
 
48
    # We need to check to make sure there are no more
 
49
    # files to be returned, I'm using a cheezy way
 
50
    # Convert to a list, and there shouldn't be any left
 
51
    tester.assertEquals(len(list(files)), 0)
 
52
 
 
53
def test_multiple_add(tester, store):
 
54
    fill_store(store)
 
55
    tester.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
 
56
 
 
57
def test_get(tester, store):
 
58
    fill_store(store)
 
59
 
 
60
    check_equals(tester, store, ['a'], ['hello'])
 
61
    check_equals(tester, store, ['b', 'c'], ['other', 'something'])
 
62
 
 
63
    # Make sure that requesting a non-existing file fails
 
64
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
 
65
            ['d'], [None])
 
66
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
 
67
            ['a', 'd'], ['hello', None])
 
68
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
 
69
            ['d', 'a'], [None, 'hello'])
 
70
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
 
71
            ['d', 'd', 'd'], [None, None, None])
 
72
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
 
73
            ['a', 'd', 'b'], ['hello', None, 'other'])
 
74
 
 
75
def test_ignore_get(tester, store):
 
76
    fill_store(store)
 
77
 
 
78
    files = store.get(['d'], permit_failure=True)
 
79
    files = list(files)
 
80
    tester.assertEquals(len(files), 1)
 
81
    tester.assert_(files[0] is None)
 
82
 
 
83
    check_equals(tester, store, ['a', 'd'], ['hello', None],
 
84
            permit_failure=True)
 
85
    check_equals(tester, store, ['d', 'a'], [None, 'hello'],
 
86
            permit_failure=True)
 
87
    check_equals(tester, store, ['d', 'd'], [None, None],
 
88
            permit_failure=True)
 
89
    check_equals(tester, store, ['a', 'd', 'b'], ['hello', None, 'other'],
 
90
            permit_failure=True)
 
91
    check_equals(tester, store, ['a', 'd', 'b'], ['hello', None, 'other'],
 
92
            permit_failure=True)
 
93
    check_equals(tester, store, ['b', 'd', 'c'], ['other', None, 'something'],
 
94
            permit_failure=True)
 
95
 
 
96
 
 
97
def get_compressed_store(path='.'):
 
98
    t = LocalTransport(path)
 
99
    return CompressedTextStore(t)
 
100
 
 
101
 
 
102
def get_text_store(path='.'):
 
103
    t = LocalTransport(path)
 
104
    return TextStore(t)
 
105
 
 
106
 
 
107
class TestCompressedTextStore(TestCaseInTempDir):
30
108
 
31
109
    def test_multiple_add(self):
32
110
        """Multiple add with same ID should raise a BzrError"""
33
 
        store = ImmutableStore('.')
34
 
        store.add(StringIO('goodbye'), '123123')
35
 
        self.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
 
111
        store = get_compressed_store()
 
112
        test_multiple_add(self, store)
 
113
 
 
114
    def test_get(self):
 
115
        store = get_compressed_store()
 
116
        test_get(self, store)
 
117
 
 
118
    def test_ignore_get(self):
 
119
        store = get_compressed_store()
 
120
        test_ignore_get(self, store)
 
121
 
36
122
 
37
123
    def test_total_size(self):
38
 
        store = ImmutableStore('.')
 
124
        store = get_compressed_store('.')
39
125
        store.add(StringIO('goodbye'), '123123')
40
126
        store.add(StringIO('goodbye2'), '123123.dsc')
41
127
        # these get gzipped - content should be stable
44
130
    def test_copy_all(self):
45
131
        """Test copying"""
46
132
        os.mkdir('a')
47
 
        store_a = ImmutableStore('a')
 
133
        store_a = get_text_store('a')
48
134
        store_a.add('foo', '1')
49
135
        os.mkdir('b')
50
 
        store_b = ImmutableStore('b')
 
136
        store_b = get_text_store('b')
51
137
        copy_all(store_a, store_b)
52
138
        self.assertEqual(store_a['1'].read(), 'foo')
53
139
        self.assertEqual(store_b['1'].read(), 'foo')
54
 
        store_c = RemoteStore('http://example.com/')
55
 
        self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
56
 
 
57
140
 
58
141
class TestMemoryStore(TestCase):
59
142
    
87
170
        store.add(StringIO('goodbye'), '123123')
88
171
        store.add(StringIO('goodbye2'), '123123.dsc')
89
172
        self.assertEqual(store.total_size(), (2, 15))
 
173
        # TODO: Switch the exception form UnlistableStore to
 
174
        #       or make Stores throw UnlistableStore if their
 
175
        #       Transport doesn't support listing
 
176
        # store_c = RemoteStore('http://example.com/')
 
177
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
 
178
 
 
179
class TestTextStore(TestCaseInTempDir):
 
180
    def test_multiple_add(self):
 
181
        """Multiple add with same ID should raise a BzrError"""
 
182
        store = get_text_store()
 
183
        test_multiple_add(self, store)
 
184
 
 
185
    def test_get(self):
 
186
        store = get_text_store()
 
187
        test_get(self, store)
 
188
 
 
189
    def test_ignore_get(self):
 
190
        store = get_text_store()
 
191
        test_ignore_get(self, store)
 
192
 
 
193
 
 
194
    def test_copy_all(self):
 
195
        """Test copying"""
 
196
        os.mkdir('a')
 
197
        store_a = get_text_store('a')
 
198
        store_a.add('foo', '1')
 
199
        os.mkdir('b')
 
200
        store_b = get_text_store('b')
 
201
        copy_all(store_a, store_b)
 
202
        self.assertEqual(store_a['1'].read(), 'foo')
 
203
        self.assertEqual(store_b['1'].read(), 'foo')
 
204
        # TODO: Switch the exception form UnlistableStore to
 
205
        #       or make Stores throw UnlistableStore if their
 
206
        #       Transport doesn't support listing
 
207
        # store_c = RemoteStore('http://example.com/')
 
208
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)