← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

  • Committer: Martin Pool
  • Date: 2005-07-07 10:22:02 UTC
  • Revision ID: mbp@sourcefrog.net-20050707102201-2d2a13a25098b101
- rearrange and clear up merged weave

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/teststore.py
1
 
# Copyright (C) 2005 by Canonical Development Ltd
2
 
 
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
 
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
 
13
 
# You should have received a copy of the GNU General Public License
14
 
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
 
 
17
 
"""Test Store implementation
18
 
"""
19
 
from cStringIO import StringIO
20
 
import os
21
 
 
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
27
 
from bzrlib.selftest import TestCase, TestCaseInTempDir
28
 
from bzrlib.errors import BzrError, UnlistableStore
29
 
import bzrlib.store
30
 
 
31
 
 
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):
108
 
 
109
 
    def test_multiple_add(self):
110
 
        """Multiple add with same ID should raise a BzrError"""
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
 
 
122
 
    def test_total_size(self):
123
 
        store = get_compressed_store('.')
124
 
        store.add(StringIO('goodbye'), '123123')
125
 
        store.add(StringIO('goodbye2'), '123123.dsc')
126
 
        # these get gzipped - content should be stable
127
 
        self.assertEqual(store.total_size(), (2, 55))
128
 
        
129
 
    def test_copy_all(self):
130
 
        """Test copying"""
131
 
        os.mkdir('a')
132
 
        store_a = get_text_store('a')
133
 
        store_a.add('foo', '1')
134
 
        os.mkdir('b')
135
 
        store_b = get_text_store('b')
136
 
        copy_all(store_a, store_b)
137
 
        self.assertEqual(store_a['1'].read(), 'foo')
138
 
        self.assertEqual(store_b['1'].read(), 'foo')
139
 
 
140
 
 
141
 
class TestMemoryStore(TestCase):
142
 
    
143
 
    def get_store(self):
144
 
        return bzrlib.store.ImmutableMemoryStore()
145
 
    
146
 
    def test_imports(self):
147
 
        from bzrlib.store import ImmutableMemoryStore
148
 
 
149
 
    def test_add_and_retrieve(self):
150
 
        store = self.get_store()
151
 
        store.add(StringIO('hello'), 'aa')
152
 
        self.assertNotEqual(store['aa'], None)
153
 
        self.assertEqual(store['aa'].read(), 'hello')
154
 
        store.add(StringIO('hello world'), 'bb')
155
 
        self.assertNotEqual(store['bb'], None)
156
 
        self.assertEqual(store['bb'].read(), 'hello world')
157
 
 
158
 
    def test_missing_is_absent(self):
159
 
        store = self.get_store()
160
 
        self.failIf('aa' in store)
161
 
 
162
 
    def test_adding_fails_when_present(self):
163
 
        store = self.get_store()
164
 
        store.add(StringIO('hello'), 'aa')
165
 
        self.assertRaises(bzrlib.store.StoreError,
166
 
                          store.add, StringIO('hello'), 'aa')
167
 
 
168
 
    def test_total_size(self):
169
 
        store = self.get_store()
170
 
        store.add(StringIO('goodbye'), '123123')
171
 
        store.add(StringIO('goodbye2'), '123123.dsc')
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
 
 
180
 
class TestTextStore(TestCaseInTempDir):
181
 
    def test_multiple_add(self):
182
 
        """Multiple add with same ID should raise a BzrError"""
183
 
        store = get_text_store()
184
 
        test_multiple_add(self, store)
185
 
 
186
 
    def test_get(self):
187
 
        store = get_text_store()
188
 
        test_get(self, store)
189
 
 
190
 
    def test_ignore_get(self):
191
 
        store = get_text_store()
192
 
        test_ignore_get(self, store)
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)