~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

Merge from bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2004, 2005, 2006 by Canonical Ltd
 
1
# Copyright (C) 2004, 2005 by Canonical Ltd
2
2
 
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
20
20
import stat
21
21
from cStringIO import StringIO
22
22
 
23
 
import bzrlib
24
23
from bzrlib.errors import (NoSuchFile, FileExists,
25
 
                           TransportNotPossible,
26
 
                           ConnectionError,
27
 
                           DependencyNotPresent,
28
 
                           )
29
 
from bzrlib.tests import TestCase, TestCaseInTempDir
 
24
                           TransportNotPossible, ConnectionError)
 
25
from bzrlib.tests import TestCase
30
26
from bzrlib.transport import (_get_protocol_handlers,
31
27
                              _get_transport_modules,
32
28
                              get_transport,
33
29
                              register_lazy_transport,
34
30
                              _set_protocol_handlers,
35
31
                              urlescape,
36
 
                              Transport,
37
32
                              )
38
 
from bzrlib.transport.memory import MemoryTransport
39
 
from bzrlib.transport.local import LocalTransport
40
33
 
41
34
 
42
35
class TestTransport(TestCase):
67
60
                             _get_transport_modules())
68
61
        finally:
69
62
            _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')
 
63
            
 
64
 
 
65
class MemoryTransportTest(TestCase):
 
66
    """Memory transport specific tests."""
167
67
 
168
68
    def test_parameters(self):
169
 
        transport = MemoryTransport()
 
69
        import bzrlib.transport.memory as memory
 
70
        transport = memory.MemoryTransport()
170
71
        self.assertEqual(True, transport.listable())
171
72
        self.assertEqual(False, transport.should_cache())
172
73
        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
 
 
 
74
        
190
75
        
191
76
class ReadonlyDecoratorTransportTest(TestCase):
192
77
    """Readonly decoration specific tests."""
214
99
            self.assertEqual(True, transport.is_readonly())
215
100
        finally:
216
101
            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