← 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: Danny van Heumen
  • Date: 2010-03-09 16:38:10 UTC
  • mto: (4634.139.5 2.0)
  • mto: This revision was merged to the branch mainline in revision 5160.
  • Revision ID: danny@dannyvanheumen.nl-20100309163810-ujn8hcx08f75nlf1
Refined test to make use of locking hooks and also validate if lock is truly a checkout-lock.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/http_utils.py
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005 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
 
    transport,
32
 
    )
33
 
from bzrlib.smart import (
34
 
    medium,
35
 
    protocol,
36
 
    )
 
31
    )
 
32
from bzrlib.smart import medium, protocol
37
33
from bzrlib.tests import http_server
38
 
from bzrlib.transport import chroot
 
34
from bzrlib.transport import (
 
35
    chroot,
 
36
    get_transport,
 
37
    )
39
38
 
40
39
 
41
40
class HTTPServerWithSmarts(http_server.HttpServer):
57
56
 
58
57
    def do_POST(self):
59
58
        """Hand the request off to a smart server instance."""
60
 
        backing = transport.get_transport(
61
 
            self.server.test_case_server._home_dir)
 
59
        backing = get_transport(self.server.test_case_server._home_dir)
62
60
        chroot_server = chroot.ChrootServer(backing)
63
 
        chroot_server.start_server()
 
61
        chroot_server.setUp()
64
62
        try:
65
 
            t = transport.get_transport(chroot_server.get_url())
 
63
            t = get_transport(chroot_server.get_url())
66
64
            self.do_POST_inner(t)
67
65
        finally:
68
 
            chroot_server.stop_server()
 
66
            chroot_server.tearDown()
69
67
 
70
68
    def do_POST_inner(self, chrooted_transport):
71
69
        self.send_response(200)
108
106
    one. This will currently fail if the primary transport is not
109
107
    backed by regular disk files.
110
108
    """
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
 
 
118
109
    def setUp(self):
119
110
        super(TestCaseWithWebserver, self).setUp()
120
111
        self.transport_readonly_server = http_server.HttpServer
121
112
 
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
 
 
128
113
 
129
114
class TestCaseWithTwoWebservers(TestCaseWithWebserver):
130
115
    """A support class providing readonly urls on two servers that are http://.
142
127
 
143
128
        This is mostly a hook for daughter classes.
144
129
        """
145
 
        server = self.transport_secondary_server(
146
 
            protocol_version=self._protocol_version)
147
 
        server._url_protocol = self._url_protocol
148
 
        return server
 
130
        return self.transport_secondary_server()
149
131
 
150
132
    def get_secondary_server(self):
151
133
        """Get the server instance for the secondary transport."""
152
134
        if self.__secondary_server is None:
153
135
            self.__secondary_server = self.create_transport_secondary_server()
154
 
            self.start_server(self.__secondary_server)
 
136
            self.__secondary_server.setUp()
 
137
            self.addCleanup(self.__secondary_server.tearDown)
155
138
        return self.__secondary_server
156
139
 
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
 
 
166
140
 
167
141
class ProxyServer(http_server.HttpServer):
168
142
    """A proxy test server for http transports."""
242
216
   The 'old' server is redirected to the 'new' server.
243
217
   """
244
218
 
 
219
   def create_transport_secondary_server(self):
 
220
       """Create the secondary server redirecting to the primary server"""
 
221
       new = self.get_readonly_server()
 
222
       redirecting = HTTPServerRedirecting()
 
223
       redirecting.redirect_to(new.host, new.port)
 
224
       return redirecting
 
225
 
245
226
   def setUp(self):
246
227
       super(TestCaseWithRedirectedWebserver, self).setUp()
247
228
       # The redirections will point to the new server
248
229
       self.new_server = self.get_readonly_server()
249
 
       # The requests to the old server will be redirected to the new server
 
230
       # The requests to the old server will be redirected
250
231
       self.old_server = self.get_secondary_server()
251
232
 
252
 
   def create_transport_secondary_server(self):
253
 
       """Create the secondary server redirecting to the primary server"""
254
 
       new = self.get_readonly_server()
255
 
       redirecting = HTTPServerRedirecting(
256
 
           protocol_version=self._protocol_version)
257
 
       redirecting.redirect_to(new.host, new.port)
258
 
       redirecting._url_protocol = self._url_protocol
259
 
       return redirecting
260
 
 
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
278
 
 
279
233
 
280
234
class AuthRequestHandler(http_server.TestingHTTPRequestHandler):
281
235
    """Requires an authentication to process requests.