← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Martin Pool
  • Date: 2006-06-15 05:36:34 UTC
  • mto: This revision was merged to the branch mainline in revision 1797.
  • Revision ID: mbp@sourcefrog.net-20060615053634-4fd52ba691855659
Clean up many exception classes.

Errors indicating a user error are now shown with is_user_error on the
exception; use this rather than hardcoding a list of exceptions that should be
handled this way.

Exceptions now inherit from BzrNewException where possible to use consistent
formatting method.

Remove rather obsolete docstring test on Branch.missing_revisions.

Remove dead code from find_merge_base.


Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
 
1
# Copyright (C) 2004, 2005, 2006 by Canonical 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
 
 
18
import os
 
19
import sys
 
20
import stat
 
21
from cStringIO import StringIO
 
22
 
 
23
import bzrlib
 
24
from bzrlib.errors import (NoSuchFile, FileExists,
 
25
                           TransportNotPossible,
 
26
                           ConnectionError,
 
27
                           DependencyNotPresent,
 
28
                           )
 
29
from bzrlib.tests import TestCase, TestCaseInTempDir
 
30
from bzrlib.transport import (_get_protocol_handlers,
 
31
                              _get_transport_modules,
 
32
                              get_transport,
 
33
                              register_lazy_transport,
 
34
                              _set_protocol_handlers,
 
35
                              urlescape,
 
36
                              Transport,
 
37
                              )
 
38
from bzrlib.transport.memory import MemoryTransport
 
39
from bzrlib.transport.local import LocalTransport
 
40
 
 
41
 
 
42
class TestTransport(TestCase):
 
43
    """Test the non transport-concrete class functionality."""
 
44
 
 
45
    def test_urlescape(self):
 
46
        self.assertEqual('%25', urlescape('%'))
 
47
 
 
48
    def test__get_set_protocol_handlers(self):
 
49
        handlers = _get_protocol_handlers()
 
50
        self.assertNotEqual({}, handlers)
 
51
        try:
 
52
            _set_protocol_handlers({})
 
53
            self.assertEqual({}, _get_protocol_handlers())
 
54
        finally:
 
55
            _set_protocol_handlers(handlers)
 
56
 
 
57
    def test_get_transport_modules(self):
 
58
        handlers = _get_protocol_handlers()
 
59
        class SampleHandler(object):
 
60
            """I exist, isnt that enough?"""
 
61
        try:
 
62
            my_handlers = {}
 
63
            _set_protocol_handlers(my_handlers)
 
64
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
65
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
66
            self.assertEqual([SampleHandler.__module__],
 
67
                             _get_transport_modules())
 
68
        finally:
 
69
            _set_protocol_handlers(handlers)
 
70
 
 
71
    def test_transport_dependency(self):
 
72
        """Transport with missing dependency causes no error"""
 
73
        saved_handlers = _get_protocol_handlers()
 
74
        try:
 
75
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
76
                    'BadTransportHandler')
 
77
            t = get_transport('foo://fooserver/foo')
 
78
            # because we failed to load the transport
 
79
            self.assertTrue(isinstance(t, LocalTransport))
 
80
        finally:
 
81
            # restore original values
 
82
            _set_protocol_handlers(saved_handlers)
 
83
            
 
84
    def test_transport_fallback(self):
 
85
        """Transport with missing dependency causes no error"""
 
86
        saved_handlers = _get_protocol_handlers()
 
87
        try:
 
88
            _set_protocol_handlers({})
 
89
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
90
                    'BackupTransportHandler')
 
91
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
92
                    'BadTransportHandler')
 
93
            t = get_transport('foo://fooserver/foo')
 
94
            self.assertTrue(isinstance(t, BackupTransportHandler))
 
95
        finally:
 
96
            _set_protocol_handlers(saved_handlers)
 
97
            
 
98
 
 
99
class TestMemoryTransport(TestCase):
 
100
 
 
101
    def test_get_transport(self):
 
102
        MemoryTransport()
 
103
 
 
104
    def test_clone(self):
 
105
        transport = MemoryTransport()
 
106
        self.assertTrue(isinstance(transport, MemoryTransport))
 
107
 
 
108
    def test_abspath(self):
 
109
        transport = MemoryTransport()
 
110
        self.assertEqual("memory:/relpath", transport.abspath('relpath'))
 
111
 
 
112
    def test_relpath(self):
 
113
        transport = MemoryTransport()
 
114
 
 
115
    def test_append_and_get(self):
 
116
        transport = MemoryTransport()
 
117
        transport.append('path', StringIO('content'))
 
118
        self.assertEqual(transport.get('path').read(), 'content')
 
119
        transport.append('path', StringIO('content'))
 
120
        self.assertEqual(transport.get('path').read(), 'contentcontent')
 
121
 
 
122
    def test_put_and_get(self):
 
123
        transport = MemoryTransport()
 
124
        transport.put('path', StringIO('content'))
 
125
        self.assertEqual(transport.get('path').read(), 'content')
 
126
        transport.put('path', StringIO('content'))
 
127
        self.assertEqual(transport.get('path').read(), 'content')
 
128
 
 
129
    def test_append_without_dir_fails(self):
 
130
        transport = MemoryTransport()
 
131
        self.assertRaises(NoSuchFile,
 
132
                          transport.append, 'dir/path', StringIO('content'))
 
133
 
 
134
    def test_put_without_dir_fails(self):
 
135
        transport = MemoryTransport()
 
136
        self.assertRaises(NoSuchFile,
 
137
                          transport.put, 'dir/path', StringIO('content'))
 
138
 
 
139
    def test_get_missing(self):
 
140
        transport = MemoryTransport()
 
141
        self.assertRaises(NoSuchFile, transport.get, 'foo')
 
142
 
 
143
    def test_has_missing(self):
 
144
        transport = MemoryTransport()
 
145
        self.assertEquals(False, transport.has('foo'))
 
146
 
 
147
    def test_has_present(self):
 
148
        transport = MemoryTransport()
 
149
        transport.append('foo', StringIO('content'))
 
150
        self.assertEquals(True, transport.has('foo'))
 
151
 
 
152
    def test_mkdir(self):
 
153
        transport = MemoryTransport()
 
154
        transport.mkdir('dir')
 
155
        transport.append('dir/path', StringIO('content'))
 
156
        self.assertEqual(transport.get('dir/path').read(), 'content')
 
157
 
 
158
    def test_mkdir_missing_parent(self):
 
159
        transport = MemoryTransport()
 
160
        self.assertRaises(NoSuchFile,
 
161
                          transport.mkdir, 'dir/dir')
 
162
 
 
163
    def test_mkdir_twice(self):
 
164
        transport = MemoryTransport()
 
165
        transport.mkdir('dir')
 
166
        self.assertRaises(FileExists, transport.mkdir, 'dir')
 
167
 
 
168
    def test_parameters(self):
 
169
        transport = MemoryTransport()
 
170
        self.assertEqual(True, transport.listable())
 
171
        self.assertEqual(False, transport.should_cache())
 
172
        self.assertEqual(False, transport.is_readonly())
 
173
 
 
174
    def test_iter_files_recursive(self):
 
175
        transport = MemoryTransport()
 
176
        transport.mkdir('dir')
 
177
        transport.put('dir/foo', StringIO('content'))
 
178
        transport.put('dir/bar', StringIO('content'))
 
179
        transport.put('bar', StringIO('content'))
 
180
        paths = set(transport.iter_files_recursive())
 
181
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
 
182
 
 
183
    def test_stat(self):
 
184
        transport = MemoryTransport()
 
185
        transport.put('foo', StringIO('content'))
 
186
        transport.put('bar', StringIO('phowar'))
 
187
        self.assertEqual(7, transport.stat('foo').st_size)
 
188
        self.assertEqual(6, transport.stat('bar').st_size)
 
189
 
 
190
        
 
191
class ReadonlyDecoratorTransportTest(TestCase):
 
192
    """Readonly decoration specific tests."""
 
193
 
 
194
    def test_local_parameters(self):
 
195
        import bzrlib.transport.readonly as readonly
 
196
        # connect to . in readonly mode
 
197
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
 
198
        self.assertEqual(True, transport.listable())
 
199
        self.assertEqual(False, transport.should_cache())
 
200
        self.assertEqual(True, transport.is_readonly())
 
201
 
 
202
    def test_http_parameters(self):
 
203
        import bzrlib.transport.readonly as readonly
 
204
        from bzrlib.transport.http import HttpServer
 
205
        # connect to . via http which is not listable
 
206
        server = HttpServer()
 
207
        server.setUp()
 
208
        try:
 
209
            transport = get_transport('readonly+' + server.get_url())
 
210
            self.failUnless(isinstance(transport,
 
211
                                       readonly.ReadonlyTransportDecorator))
 
212
            self.assertEqual(False, transport.listable())
 
213
            self.assertEqual(True, transport.should_cache())
 
214
            self.assertEqual(True, transport.is_readonly())
 
215
        finally:
 
216
            server.tearDown()
 
217
 
 
218
 
 
219
class FakeNFSDecoratorTests(TestCaseInTempDir):
 
220
    """NFS decorator specific tests."""
 
221
 
 
222
    def get_nfs_transport(self, url):
 
223
        import bzrlib.transport.fakenfs as fakenfs
 
224
        # connect to url with nfs decoration
 
225
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
 
226
 
 
227
    def test_local_parameters(self):
 
228
        # the listable, should_cache and is_readonly parameters
 
229
        # are not changed by the fakenfs decorator
 
230
        transport = self.get_nfs_transport('.')
 
231
        self.assertEqual(True, transport.listable())
 
232
        self.assertEqual(False, transport.should_cache())
 
233
        self.assertEqual(False, transport.is_readonly())
 
234
 
 
235
    def test_http_parameters(self):
 
236
        # the listable, should_cache and is_readonly parameters
 
237
        # are not changed by the fakenfs decorator
 
238
        from bzrlib.transport.http import HttpServer
 
239
        # connect to . via http which is not listable
 
240
        server = HttpServer()
 
241
        server.setUp()
 
242
        try:
 
243
            transport = self.get_nfs_transport(server.get_url())
 
244
            self.assertIsInstance(
 
245
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
 
246
            self.assertEqual(False, transport.listable())
 
247
            self.assertEqual(True, transport.should_cache())
 
248
            self.assertEqual(True, transport.is_readonly())
 
249
        finally:
 
250
            server.tearDown()
 
251
 
 
252
    def test_fakenfs_server_default(self):
 
253
        # a FakeNFSServer() should bring up a local relpath server for itself
 
254
        import bzrlib.transport.fakenfs as fakenfs
 
255
        server = fakenfs.FakeNFSServer()
 
256
        server.setUp()
 
257
        try:
 
258
            # the server should be a relpath localhost server
 
259
            self.assertEqual(server.get_url(), 'fakenfs+.')
 
260
            # and we should be able to get a transport for it
 
261
            transport = get_transport(server.get_url())
 
262
            # which must be a FakeNFSTransportDecorator instance.
 
263
            self.assertIsInstance(
 
264
                transport, fakenfs.FakeNFSTransportDecorator)
 
265
        finally:
 
266
            server.tearDown()
 
267
 
 
268
    def test_fakenfs_rename_semantics(self):
 
269
        # a FakeNFS transport must mangle the way rename errors occur to
 
270
        # look like NFS problems.
 
271
        transport = self.get_nfs_transport('.')
 
272
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
 
273
                        transport=transport)
 
274
        self.assertRaises(bzrlib.errors.ResourceBusy,
 
275
                          transport.rename, 'from', 'to')
 
276
 
 
277
 
 
278
class FakeVFATDecoratorTests(TestCaseInTempDir):
 
279
    """Tests for simulation of VFAT restrictions"""
 
280
 
 
281
    def get_vfat_transport(self, url):
 
282
        """Return vfat-backed transport for test directory"""
 
283
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
 
284
        return FakeVFATTransportDecorator('vfat+' + url)
 
285
 
 
286
    def test_transport_creation(self):
 
287
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
 
288
        transport = self.get_vfat_transport('.')
 
289
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
 
290
 
 
291
    def test_transport_mkdir(self):
 
292
        transport = self.get_vfat_transport('.')
 
293
        transport.mkdir('HELLO')
 
294
        self.assertTrue(transport.has('hello'))
 
295
        self.assertTrue(transport.has('Hello'))
 
296
 
 
297
    def test_forbidden_chars(self):
 
298
        transport = self.get_vfat_transport('.')
 
299
        self.assertRaises(ValueError, transport.has, "<NU>")
 
300
 
 
301
 
 
302
class BadTransportHandler(Transport):
 
303
    def __init__(self, base_url):
 
304
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
 
305
 
 
306
 
 
307
class BackupTransportHandler(Transport):
 
308
    """Test transport that works as a backup for the BadTransportHandler"""
 
309
    pass