/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: Martin Pool
  • Date: 2006-03-16 20:02:36 UTC
  • mto: (1610.1.10 bzr.mbp.integration)
  • mto: This revision was merged to the branch mainline in revision 1614.
  • Revision ID: mbp@sourcefrog.net-20060316200236-36b7bbaafd2cef9a
Raise errors correctly on pycurl connection failure

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
from bzrlib.errors import (NoSuchFile, FileExists,
 
24
                           TransportNotPossible,
 
25
                           ConnectionError,
 
26
                           DependencyNotPresent,
 
27
                           )
 
28
from bzrlib.tests import TestCase
 
29
from bzrlib.transport import (_get_protocol_handlers,
 
30
                              _get_transport_modules,
 
31
                              get_transport,
 
32
                              register_lazy_transport,
 
33
                              _set_protocol_handlers,
 
34
                              urlescape,
 
35
                              Transport,
 
36
                              )
 
37
from bzrlib.transport.memory import MemoryTransport
 
38
from bzrlib.transport.local import LocalTransport
 
39
 
 
40
 
 
41
class TestTransport(TestCase):
 
42
    """Test the non transport-concrete class functionality."""
 
43
 
 
44
    def test_urlescape(self):
 
45
        self.assertEqual('%25', urlescape('%'))
 
46
 
 
47
    def test__get_set_protocol_handlers(self):
 
48
        handlers = _get_protocol_handlers()
 
49
        self.assertNotEqual({}, handlers)
 
50
        try:
 
51
            _set_protocol_handlers({})
 
52
            self.assertEqual({}, _get_protocol_handlers())
 
53
        finally:
 
54
            _set_protocol_handlers(handlers)
 
55
 
 
56
    def test_get_transport_modules(self):
 
57
        handlers = _get_protocol_handlers()
 
58
        class SampleHandler(object):
 
59
            """I exist, isnt that enough?"""
 
60
        try:
 
61
            my_handlers = {}
 
62
            _set_protocol_handlers(my_handlers)
 
63
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
64
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
65
            self.assertEqual([SampleHandler.__module__],
 
66
                             _get_transport_modules())
 
67
        finally:
 
68
            _set_protocol_handlers(handlers)
 
69
 
 
70
    def test_transport_dependency(self):
 
71
        """Transport with missing dependency causes no error"""
 
72
        saved_handlers = _get_protocol_handlers()
 
73
        try:
 
74
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
75
                    'BadTransportHandler')
 
76
            t = get_transport('foo://fooserver/foo')
 
77
            # because we failed to load the transport
 
78
            self.assertTrue(isinstance(t, LocalTransport))
 
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 BadTransportHandler(Transport):
 
219
    def __init__(self, base_url):
 
220
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
 
221
 
 
222
 
 
223
class BackupTransportHandler(Transport):
 
224
    """Test transport that works as a backup for the BadTransportHandler"""
 
225
    pass