~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/http_utils.py

  • Committer: Vincent Ladeuil
  • Date: 2010-08-30 07:42:12 UTC
  • mfrom: (5247.2.39 catch-them-all)
  • mto: This revision was merged to the branch mainline in revision 5396.
  • Revision ID: v.ladeuil+lp@free.fr-20100830074212-y0vj6t3sm1zr706y
Fix most of the leaking tests

Show diffs side-by-side

added added

removed removed

Lines of Context:
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()
107
109
    backed by regular disk files.
108
110
    """
109
111
 
110
 
    # This can be overriden or parametrized by daughter clasess if needed, but
111
 
    # it must exist so that the create_transport_readonly_server() method can
112
 
    # propagate it.
 
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.
113
115
    _protocol_version = None
 
116
    _url_protocol = 'http'
114
117
 
115
118
    def setUp(self):
116
119
        super(TestCaseWithWebserver, self).setUp()
117
120
        self.transport_readonly_server = http_server.HttpServer
118
121
 
119
122
    def create_transport_readonly_server(self):
120
 
        return self.transport_readonly_server(
 
123
        server = self.transport_readonly_server(
121
124
            protocol_version=self._protocol_version)
 
125
        server._url_protocol = self._url_protocol
 
126
        return server
122
127
 
123
128
 
124
129
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
137
142
 
138
143
        This is mostly a hook for daughter classes.
139
144
        """
140
 
        return self.transport_secondary_server(
 
145
        server = self.transport_secondary_server(
141
146
            protocol_version=self._protocol_version)
 
147
        server._url_protocol = self._url_protocol
 
148
        return server
142
149
 
143
150
    def get_secondary_server(self):
144
151
        """Get the server instance for the secondary transport."""
147
154
            self.start_server(self.__secondary_server)
148
155
        return self.__secondary_server
149
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
 
150
166
 
151
167
class ProxyServer(http_server.HttpServer):
152
168
    """A proxy test server for http transports."""
226
242
   The 'old' server is redirected to the 'new' server.
227
243
   """
228
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
 
229
252
   def create_transport_secondary_server(self):
230
253
       """Create the secondary server redirecting to the primary server"""
231
254
       new = self.get_readonly_server()
232
255
       redirecting = HTTPServerRedirecting(
233
256
           protocol_version=self._protocol_version)
234
257
       redirecting.redirect_to(new.host, new.port)
 
258
       redirecting._url_protocol = self._url_protocol
235
259
       return redirecting
236
260
 
237
 
   def setUp(self):
238
 
       super(TestCaseWithRedirectedWebserver, self).setUp()
239
 
       # The redirections will point to the new server
240
 
       self.new_server = self.get_readonly_server()
241
 
       # The requests to the old server will be redirected
242
 
       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
243
278
 
244
279
 
245
280
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):