← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/http_utils.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-09-01 08:02:42 UTC
  • mfrom: (5390.3.3 faster-revert-593560)
  • Revision ID: pqm@pqm.ubuntu.com-20100901080242-esg62ody4frwmy66
(spiv) Avoid repeatedly calling self.target.all_file_ids() in
 InterTree.iter_changes. (Andrew Bennetts)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/http_utils.py
1
 
# Copyright (C) 2005 Canonical Ltd
 
1
# Copyright (C) 2005-2010 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
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
 
        chroot_server.setUp()
 
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
 
            chroot_server.tearDown()
 
68
            chroot_server.stop_server()
67
69
 
68
70
    def do_POST_inner(self, chrooted_transport):
69
71
        self.send_response(200)
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."""
134
152
        if self.__secondary_server is None:
135
153
            self.__secondary_server = self.create_transport_secondary_server()
136
 
            self.__secondary_server.setUp()
137
 
            self.addCleanup(self.__secondary_server.tearDown)
 
154
            self.start_server(self.__secondary_server)
138
155
        return self.__secondary_server
139
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
 
140
166
 
141
167
class ProxyServer(http_server.HttpServer):
142
168
    """A proxy test server for http transports."""
216
242
   The 'old' server is redirected to the 'new' server.
217
243
   """
218
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
 
219
252
   def create_transport_secondary_server(self):
220
253
       """Create the secondary server redirecting to the primary server"""
221
254
       new = self.get_readonly_server()
222
 
       redirecting = HTTPServerRedirecting()
 
255
       redirecting = HTTPServerRedirecting(
 
256
           protocol_version=self._protocol_version)
223
257
       redirecting.redirect_to(new.host, new.port)
 
258
       redirecting._url_protocol = self._url_protocol
224
259
       return redirecting
225
260
 
226
 
   def setUp(self):
227
 
       super(TestCaseWithRedirectedWebserver, self).setUp()
228
 
       # The redirections will point to the new server
229
 
       self.new_server = self.get_readonly_server()
230
 
       # The requests to the old server will be redirected
231
 
       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
232
278
 
233
279
 
234
280
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):