~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/HTTPTestUtil.py

  • Committer: Robey Pointer
  • Date: 2006-09-08 18:46:29 UTC
  • mto: This revision was merged to the branch mainline in revision 1996.
  • Revision ID: robey@lag.net-20060908184629-e3fc4c61ca21508c
pychecker is on crack; go back to using 'is None'.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
1
# Copyright (C) 2005 by 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
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
7
 
 
 
7
#
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
12
 
 
 
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
 
import BaseHTTPServer, SimpleHTTPServer, socket, errno, time
18
 
from bzrlib.tests import TestCaseInTempDir
19
 
 
20
 
 
21
 
class WebserverNotAvailable(Exception):
22
 
    pass
23
 
 
24
 
class BadWebserverPath(ValueError):
25
 
    def __str__(self):
26
 
        return 'path %s is not in %s' % self.args
27
 
 
28
 
class TestingHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
29
 
    def log_message(self, format, *args):
30
 
        self.server.test_case.log("webserver - %s - - [%s] %s",
31
 
                                  self.address_string(),
32
 
                                  self.log_date_time_string(),
33
 
                                  format%args)
34
 
 
35
 
    def handle_one_request(self):
36
 
        """Handle a single HTTP request.
37
 
 
38
 
        You normally don't need to override this method; see the class
39
 
        __doc__ string for information on how to handle specific HTTP
40
 
        commands such as GET and POST.
41
 
 
42
 
        """
43
 
        for i in xrange(1,11): # Don't try more than 10 times
44
 
            try:
45
 
                self.raw_requestline = self.rfile.readline()
46
 
            except socket.error, e:
47
 
                if e.args[0] in (errno.EAGAIN, errno.EWOULDBLOCK):
48
 
                    # omitted for now because some tests look at the log of
49
 
                    # the server and expect to see no errors.  see recent
50
 
                    # email thread. -- mbp 20051021. 
51
 
                    ## self.log_message('EAGAIN (%d) while reading from raw_requestline' % i)
52
 
                    time.sleep(0.01)
53
 
                    continue
54
 
                raise
55
 
            else:
56
 
                break
57
 
        if not self.raw_requestline:
58
 
            self.close_connection = 1
59
 
            return
60
 
        if not self.parse_request(): # An error code has been sent, just exit
61
 
            return
62
 
        mname = 'do_' + self.command
63
 
        if not hasattr(self, mname):
64
 
            self.send_error(501, "Unsupported method (%r)" % self.command)
65
 
            return
66
 
        method = getattr(self, mname)
67
 
        method()
68
 
 
69
 
class TestingHTTPServer(BaseHTTPServer.HTTPServer):
70
 
    def __init__(self, server_address, RequestHandlerClass, test_case):
71
 
        BaseHTTPServer.HTTPServer.__init__(self, server_address,
72
 
                                                RequestHandlerClass)
73
 
        self.test_case = test_case
74
 
 
75
 
 
76
 
class TestCaseWithWebserver(TestCaseInTempDir):
77
 
    """Derived class that starts a localhost-only webserver
78
 
    (in addition to what TestCaseInTempDir does).
79
 
 
80
 
    This is useful for testing RemoteBranch.
 
17
import os
 
18
 
 
19
import bzrlib
 
20
from bzrlib.tests import TestCaseWithTransport
 
21
 
 
22
 
 
23
class TestCaseWithWebserver(TestCaseWithTransport):
 
24
    """A support class that provides readonly urls that are http://.
 
25
 
 
26
    This is done by forcing the readonly server to be an http one. This 
 
27
    will current fail if the primary transport is not backed by regular disk
 
28
    files.
81
29
    """
82
30
 
83
 
    _HTTP_PORTS = range(13000, 0x8000)
84
 
 
85
 
    def _http_start(self):
86
 
        import SimpleHTTPServer, BaseHTTPServer, socket, errno
87
 
        httpd = None
88
 
        for port in self._HTTP_PORTS:
89
 
            try:
90
 
                httpd = TestingHTTPServer(('localhost', port),
91
 
                                          TestingHTTPRequestHandler,
92
 
                                          self)
93
 
            except socket.error, e:
94
 
                if e.args[0] == errno.EADDRINUSE:
95
 
                    continue
96
 
                print >>sys.stderr, "Cannot run webserver :-("
97
 
                raise
98
 
            else:
99
 
                break
100
 
 
101
 
        if httpd is None:
102
 
            raise WebserverNotAvailable("Cannot run webserver :-( "
103
 
                                        "no free ports in range %s..%s" %
104
 
                                        (_HTTP_PORTS[0], _HTTP_PORTS[-1]))
105
 
 
106
 
        self._http_base_url = 'http://localhost:%s/' % port
107
 
        self._http_starting.release()
108
 
        httpd.socket.settimeout(0.1)
109
 
 
110
 
        while self._http_running:
111
 
            try:
112
 
                httpd.handle_request()
113
 
            except socket.timeout:
114
 
                pass
115
 
 
116
 
    def get_remote_url(self, path):
117
 
        import os
118
 
 
119
 
        path_parts = path.split(os.path.sep)
120
 
        if os.path.isabs(path):
121
 
            if path_parts[:len(self._local_path_parts)] != \
122
 
                   self._local_path_parts:
123
 
                raise BadWebserverPath(path, self.test_dir)
124
 
            remote_path = '/'.join(path_parts[len(self._local_path_parts):])
125
 
        else:
126
 
            remote_path = '/'.join(path_parts)
127
 
 
128
 
        self._http_starting.acquire()
129
 
        self._http_starting.release()
130
 
        return self._http_base_url + remote_path
131
 
 
132
31
    def setUp(self):
133
 
        TestCaseInTempDir.setUp(self)
134
 
        import threading, os
135
 
        self._local_path_parts = self.test_dir.split(os.path.sep)
136
 
        self._http_starting = threading.Lock()
137
 
        self._http_starting.acquire()
138
 
        self._http_running = True
139
 
        self._http_base_url = None
140
 
        self._http_thread = threading.Thread(target=self._http_start)
141
 
        self._http_thread.setDaemon(True)
142
 
        self._http_thread.start()
143
 
        self._http_proxy = os.environ.get("http_proxy")
144
 
        if self._http_proxy is not None:
145
 
            del os.environ["http_proxy"]
146
 
 
147
 
    def tearDown(self):
148
 
        self._http_running = False
149
 
        self._http_thread.join()
150
 
        if self._http_proxy is not None:
151
 
            import os
152
 
            os.environ["http_proxy"] = self._http_proxy
153
 
        TestCaseInTempDir.tearDown(self)
154
 
 
 
32
        super(TestCaseWithWebserver, self).setUp()
 
33
        self.transport_readonly_server = bzrlib.transport.http.HttpServer