← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/teststore.py

  • Committer: mbp at sourcefrog
  • Date: 2005-04-05 08:24:51 UTC
  • Revision ID: mbp@sourcefrog.net-20050405082451-408ebb0fd108440f
start adding quotes

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 StringIO 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 TestCaseInTempDir
28
 
from bzrlib.errors import BzrError, UnlistableStore
29
 
 
30
 
 
31
 
def fill_store(store):
32
 
    store.add(StringIO('hello'), 'a')
33
 
    store.add(StringIO('other'), 'b')
34
 
    store.add(StringIO('something'), 'c')
35
 
    store.add(StringIO('goodbye'), '123123')
36
 
 
37
 
def check_equals(tester, store, files, values, permit_failure=False):
38
 
    files = store.get(files, permit_failure=permit_failure)
39
 
    count = 0
40
 
    for f, v in zip(files, values):
41
 
        count += 1
42
 
        if v is None:
43
 
            tester.assert_(f is None)
44
 
        else:
45
 
            tester.assertEquals(f.read(), v)
46
 
    tester.assertEquals(count, len(values))
47
 
    # We need to check to make sure there are no more
48
 
    # files to be returned, I'm using a cheezy way
49
 
    # Convert to a list, and there shouldn't be any left
50
 
    tester.assertEquals(len(list(files)), 0)
51
 
 
52
 
def test_multiple_add(tester, store):
53
 
    fill_store(store)
54
 
    tester.assertRaises(BzrError, store.add, StringIO('goodbye'), '123123')
55
 
 
56
 
def test_get(tester, store):
57
 
    fill_store(store)
58
 
 
59
 
    check_equals(tester, store, ['a'], ['hello'])
60
 
    check_equals(tester, store, ['b', 'c'], ['other', 'something'])
61
 
 
62
 
    # Make sure that requesting a non-existing file fails
63
 
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
64
 
            ['d'], [None])
65
 
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
66
 
            ['a', 'd'], ['hello', None])
67
 
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
68
 
            ['d', 'a'], [None, 'hello'])
69
 
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
70
 
            ['d', 'd', 'd'], [None, None, None])
71
 
    tester.assertRaises(NoSuchFile, check_equals, tester, store,
72
 
            ['a', 'd', 'b'], ['hello', None, 'other'])
73
 
 
74
 
def test_ignore_get(tester, store):
75
 
    fill_store(store)
76
 
 
77
 
    files = store.get(['d'], permit_failure=True)
78
 
    files = list(files)
79
 
    tester.assertEquals(len(files), 1)
80
 
    tester.assert_(files[0] is None)
81
 
 
82
 
    check_equals(tester, store, ['a', 'd'], ['hello', None],
83
 
            permit_failure=True)
84
 
    check_equals(tester, store, ['d', 'a'], [None, 'hello'],
85
 
            permit_failure=True)
86
 
    check_equals(tester, store, ['d', 'd'], [None, None],
87
 
            permit_failure=True)
88
 
    check_equals(tester, store, ['a', 'd', 'b'], ['hello', None, 'other'],
89
 
            permit_failure=True)
90
 
    check_equals(tester, store, ['a', 'd', 'b'], ['hello', None, 'other'],
91
 
            permit_failure=True)
92
 
    check_equals(tester, store, ['b', 'd', 'c'], ['other', None, 'something'],
93
 
            permit_failure=True)
94
 
 
95
 
def get_compressed_store(path='.'):
96
 
    t = LocalTransport(path)
97
 
    return CompressedTextStore(t)
98
 
 
99
 
def get_text_store(path='.'):
100
 
    t = LocalTransport(path)
101
 
    return TextStore(t)
102
 
 
103
 
class TestCompressedTextStore(TestCaseInTempDir):
104
 
    def test_multiple_add(self):
105
 
        """Multiple add with same ID should raise a BzrError"""
106
 
        store = get_compressed_store()
107
 
        test_multiple_add(self, store)
108
 
 
109
 
    def test_get(self):
110
 
        store = get_compressed_store()
111
 
        test_get(self, store)
112
 
 
113
 
    def test_ignore_get(self):
114
 
        store = get_compressed_store()
115
 
        test_ignore_get(self, store)
116
 
 
117
 
 
118
 
    def test_copy_all(self):
119
 
        """Test copying"""
120
 
        os.mkdir('a')
121
 
        store_a = get_text_store('a')
122
 
        store_a.add('foo', '1')
123
 
        os.mkdir('b')
124
 
        store_b = get_text_store('b')
125
 
        copy_all(store_a, store_b)
126
 
        self.assertEqual(store_a['1'].read(), 'foo')
127
 
        self.assertEqual(store_b['1'].read(), 'foo')
128
 
        # TODO: Switch the exception form UnlistableStore to
129
 
        #       or make Stores throw UnlistableStore if their
130
 
        #       Transport doesn't support listing
131
 
        # store_c = RemoteStore('http://example.com/')
132
 
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
133
 
 
134
 
class TestTextStore(TestCaseInTempDir):
135
 
    def test_multiple_add(self):
136
 
        """Multiple add with same ID should raise a BzrError"""
137
 
        store = get_text_store()
138
 
        test_multiple_add(self, store)
139
 
 
140
 
    def test_get(self):
141
 
        store = get_text_store()
142
 
        test_get(self, store)
143
 
 
144
 
    def test_ignore_get(self):
145
 
        store = get_text_store()
146
 
        test_ignore_get(self, store)
147
 
 
148
 
 
149
 
    def test_copy_all(self):
150
 
        """Test copying"""
151
 
        os.mkdir('a')
152
 
        store_a = get_text_store('a')
153
 
        store_a.add('foo', '1')
154
 
        os.mkdir('b')
155
 
        store_b = get_text_store('b')
156
 
        copy_all(store_a, store_b)
157
 
        self.assertEqual(store_a['1'].read(), 'foo')
158
 
        self.assertEqual(store_b['1'].read(), 'foo')
159
 
        # TODO: Switch the exception form UnlistableStore to
160
 
        #       or make Stores throw UnlistableStore if their
161
 
        #       Transport doesn't support listing
162
 
        # store_c = RemoteStore('http://example.com/')
163
 
        # self.assertRaises(UnlistableStore, copy_all, store_c, store_b)
164