~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2006-06-29 18:21:57 UTC
  • mfrom: (1711.2.72 jam-integration)
  • Revision ID: pqm@pqm.ubuntu.com-20060629182157-cdc39b5f9e4fac85
(jam) add run_bzr_error, clean up remerge tests

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
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
                           InvalidURL,
 
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
            # 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')
 
166
 
 
167
    def test_parameters(self):
 
168
        transport = MemoryTransport()
 
169
        self.assertEqual(True, transport.listable())
 
170
        self.assertEqual(False, transport.should_cache())
 
171
        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
 
 
189
        
 
190
class ReadonlyDecoratorTransportTest(TestCase):
 
191
    """Readonly decoration specific tests."""
 
192
 
 
193
    def test_local_parameters(self):
 
194
        import bzrlib.transport.readonly as readonly
 
195
        # connect to . in readonly mode
 
196
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
 
197
        self.assertEqual(True, transport.listable())
 
198
        self.assertEqual(False, transport.should_cache())
 
199
        self.assertEqual(True, transport.is_readonly())
 
200
 
 
201
    def test_http_parameters(self):
 
202
        import bzrlib.transport.readonly as readonly
 
203
        from bzrlib.transport.http import HttpServer
 
204
        # connect to . via http which is not listable
 
205
        server = HttpServer()
 
206
        server.setUp()
 
207
        try:
 
208
            transport = get_transport('readonly+' + server.get_url())
 
209
            self.failUnless(isinstance(transport,
 
210
                                       readonly.ReadonlyTransportDecorator))
 
211
            self.assertEqual(False, transport.listable())
 
212
            self.assertEqual(True, transport.should_cache())
 
213
            self.assertEqual(True, transport.is_readonly())
 
214
        finally:
 
215
            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