← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

first cut at merge from integration.

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