← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/http_utils.py

Cleanup asserts

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.
343
297
 
344
298
    def authorized(self):
345
299
        tcs = self.server.test_case_server
 
300
        if tcs.auth_scheme != 'digest':
 
301
            return False
346
302
 
347
303
        auth_header = self.headers.get(tcs.auth_header_recv, None)
348
304
        if auth_header is None:
363
319
        self.send_header(tcs.auth_header_sent,header)
364
320
 
365
321
 
366
 
class DigestAndBasicAuthRequestHandler(DigestAuthRequestHandler):
367
 
    """Implements a digest and basic authentication of a request.
368
 
 
369
 
    I.e. the server proposes both schemes and the client should choose the best
370
 
    one it can handle, which, in that case, should be digest, the only scheme
371
 
    accepted here.
372
 
    """
373
 
 
374
 
    def send_header_auth_reqed(self):
375
 
        tcs = self.server.test_case_server
376
 
        self.send_header(tcs.auth_header_sent,
377
 
                         'Basic realm="%s"' % tcs.auth_realm)
378
 
        header = 'Digest realm="%s", ' % tcs.auth_realm
379
 
        header += 'nonce="%s", algorithm="%s", qop="auth"' % (tcs.auth_nonce,
380
 
                                                              'MD5')
381
 
        self.send_header(tcs.auth_header_sent,header)
382
 
 
383
 
 
384
322
class AuthServer(http_server.HttpServer):
385
323
    """Extends HttpServer with a dictionary of passwords.
386
324
 
472
410
 
473
411
        return response_digest == auth['response']
474
412
 
475
 
 
476
413
class HTTPAuthServer(AuthServer):
477
414
    """An HTTP server requiring authentication"""
478
415
 
510
447
        self.init_http_auth()
511
448
 
512
449
 
513
 
class HTTPBasicAndDigestAuthServer(DigestAuthServer, HTTPAuthServer):
514
 
    """An HTTP server requiring basic or digest authentication"""
515
 
 
516
 
    def __init__(self, protocol_version=None):
517
 
        DigestAuthServer.__init__(self, DigestAndBasicAuthRequestHandler,
518
 
                                  'basicdigest',
519
 
                                  protocol_version=protocol_version)
520
 
        self.init_http_auth()
521
 
        # We really accept Digest only
522
 
        self.auth_scheme = 'digest'
523
 
 
524
 
 
525
450
class ProxyBasicAuthServer(ProxyAuthServer):
526
451
    """A proxy server requiring basic authentication"""
527
452
 
540
465
        self.init_proxy_auth()
541
466
 
542
467
 
543
 
class ProxyBasicAndDigestAuthServer(DigestAuthServer, ProxyAuthServer):
544
 
    """An proxy server requiring basic or digest authentication"""
545
 
 
546
 
    def __init__(self, protocol_version=None):
547
 
        DigestAuthServer.__init__(self, DigestAndBasicAuthRequestHandler,
548
 
                                  'basicdigest',
549
 
                                  protocol_version=protocol_version)
550
 
        self.init_proxy_auth()
551
 
        # We really accept Digest only
552
 
        self.auth_scheme = 'digest'
553
 
 
554