← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/http_utils.py

Add bzrlib.pyutils, which has get_named_object, a wrapper around __import__.

This is used to replace various ad hoc implementations of the same logic,
notably the version used in registry's _LazyObjectGetter which had a bug when
getting a module without also getting a member.  And of course, this new
function has unit tests, unlike the replaced code.

This also adds a KnownHooksRegistry subclass to provide a more natural home for
some other logic.

I'm not thrilled about the name of the new module or the new functions, but it's
hard to think of good names for such generic functionality.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/http_utils.py
28
28
    errors,
29
29
    osutils,
30
30
    tests,
31
 
    )
32
 
from bzrlib.smart import medium, protocol
 
31
    transport,
 
32
    )
 
33
from bzrlib.smart import (
 
34
    medium,
 
35
    protocol,
 
36
    )
33
37
from bzrlib.tests import http_server
34
 
from bzrlib.transport import (
35
 
    chroot,
36
 
    get_transport,
37
 
    )
 
38
from bzrlib.transport import chroot
38
39
 
39
40
 
40
41
class HTTPServerWithSmarts(http_server.HttpServer):
56
57
 
57
58
    def do_POST(self):
58
59
        """Hand the request off to a smart server instance."""
59
 
        backing = get_transport(self.server.test_case_server._home_dir)
 
60
        backing = transport.get_transport(
 
61
            self.server.test_case_server._home_dir)
60
62
        chroot_server = chroot.ChrootServer(backing)
61
63
        chroot_server.start_server()
62
64
        try:
63
 
            t = get_transport(chroot_server.get_url())
 
65
            t = transport.get_transport(chroot_server.get_url())
64
66
            self.do_POST_inner(t)
65
67
        finally:
66
68
            chroot_server.stop_server()
106
108
    one. This will currently fail if the primary transport is not
107
109
    backed by regular disk files.
108
110
    """
 
111
 
 
112
    # These attributes can be overriden or parametrized by daughter clasess if
 
113
    # needed, but must exist so that the create_transport_readonly_server()
 
114
    # method (or any method creating an http(s) server) can propagate it.
 
115
    _protocol_version = None
 
116
    _url_protocol = 'http'
 
117
 
109
118
    def setUp(self):
110
119
        super(TestCaseWithWebserver, self).setUp()
111
120
        self.transport_readonly_server = http_server.HttpServer
112
121
 
 
122
    def create_transport_readonly_server(self):
 
123
        server = self.transport_readonly_server(
 
124
            protocol_version=self._protocol_version)
 
125
        server._url_protocol = self._url_protocol
 
126
        return server
 
127
 
113
128
 
114
129
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
115
130
    """A support class providing readonly urls on two servers that are http://.
127
142
 
128
143
        This is mostly a hook for daughter classes.
129
144
        """
130
 
        return self.transport_secondary_server()
 
145
        server = self.transport_secondary_server(
 
146
            protocol_version=self._protocol_version)
 
147
        server._url_protocol = self._url_protocol
 
148
        return server
131
149
 
132
150
    def get_secondary_server(self):
133
151
        """Get the server instance for the secondary transport."""
136
154
            self.start_server(self.__secondary_server)
137
155
        return self.__secondary_server
138
156
 
 
157
    def get_secondary_url(self, relpath=None):
 
158
        base = self.get_secondary_server().get_url()
 
159
        return self._adjust_url(base, relpath)
 
160
 
 
161
    def get_secondary_transport(self, relpath=None):
 
162
        t = transport.get_transport(self.get_secondary_url(relpath))
 
163
        self.assertTrue(t.is_readonly())
 
164
        return t
 
165
 
139
166
 
140
167
class ProxyServer(http_server.HttpServer):
141
168
    """A proxy test server for http transports."""
215
242
   The 'old' server is redirected to the 'new' server.
216
243
   """
217
244
 
 
245
   def setUp(self):
 
246
       super(TestCaseWithRedirectedWebserver, self).setUp()
 
247
       # The redirections will point to the new server
 
248
       self.new_server = self.get_readonly_server()
 
249
       # The requests to the old server will be redirected to the new server
 
250
       self.old_server = self.get_secondary_server()
 
251
 
218
252
   def create_transport_secondary_server(self):
219
253
       """Create the secondary server redirecting to the primary server"""
220
254
       new = self.get_readonly_server()
221
 
       redirecting = HTTPServerRedirecting()
 
255
       redirecting = HTTPServerRedirecting(
 
256
           protocol_version=self._protocol_version)
222
257
       redirecting.redirect_to(new.host, new.port)
 
258
       redirecting._url_protocol = self._url_protocol
223
259
       return redirecting
224
260
 
225
 
   def setUp(self):
226
 
       super(TestCaseWithRedirectedWebserver, self).setUp()
227
 
       # The redirections will point to the new server
228
 
       self.new_server = self.get_readonly_server()
229
 
       # The requests to the old server will be redirected
230
 
       self.old_server = self.get_secondary_server()
 
261
   def get_old_url(self, relpath=None):
 
262
        base = self.old_server.get_url()
 
263
        return self._adjust_url(base, relpath)
 
264
 
 
265
   def get_old_transport(self, relpath=None):
 
266
        t = transport.get_transport(self.get_old_url(relpath))
 
267
        self.assertTrue(t.is_readonly())
 
268
        return t
 
269
 
 
270
   def get_new_url(self, relpath=None):
 
271
        base = self.new_server.get_url()
 
272
        return self._adjust_url(base, relpath)
 
273
 
 
274
   def get_new_transport(self, relpath=None):
 
275
        t = transport.get_transport(self.get_new_url(relpath))
 
276
        self.assertTrue(t.is_readonly())
 
277
        return t
231
278
 
232
279
 
233
280
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):