/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Robert Collins
  • Date: 2006-07-24 02:07:58 UTC
  • mto: (1879.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 1880.
  • Revision ID: robertc@robertcollins.net-20060724020758-22705e712fcbfc4a
Review feedback.

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