1
# Copyright (C) 2005 by Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
import BaseHTTPServer, SimpleHTTPServer, socket, errno, time
18
from bzrlib.tests import TestCaseInTempDir
21
class WebserverNotAvailable(Exception):
24
class BadWebserverPath(ValueError):
26
return 'path %s is not in %s' % self.args
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(),
35
def handle_one_request(self):
36
"""Handle a single HTTP request.
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.
43
for i in xrange(1,11): # Don't try more than 10 times
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)
57
if not self.raw_requestline:
58
self.close_connection = 1
60
if not self.parse_request(): # An error code has been sent, just exit
62
mname = 'do_' + self.command
63
if not hasattr(self, mname):
64
self.send_error(501, "Unsupported method (%r)" % self.command)
66
method = getattr(self, mname)
69
class TestingHTTPServer(BaseHTTPServer.HTTPServer):
70
def __init__(self, server_address, RequestHandlerClass, test_case):
71
BaseHTTPServer.HTTPServer.__init__(self, server_address,
73
self.test_case = test_case
76
class TestCaseWithWebserver(TestCaseInTempDir):
77
"""Derived class that starts a localhost-only webserver
78
(in addition to what TestCaseInTempDir does).
80
This is useful for testing RemoteBranch.
83
_HTTP_PORTS = range(13000, 0x8000)
85
def _http_start(self):
86
import SimpleHTTPServer, BaseHTTPServer, socket, errno
88
for port in self._HTTP_PORTS:
90
httpd = TestingHTTPServer(('localhost', port),
91
TestingHTTPRequestHandler,
93
except socket.error, e:
94
if e.args[0] == errno.EADDRINUSE:
96
print >>sys.stderr, "Cannot run webserver :-("
102
raise WebserverNotAvailable("Cannot run webserver :-( "
103
"no free ports in range %s..%s" %
104
(_HTTP_PORTS[0], _HTTP_PORTS[-1]))
106
self._http_base_url = 'http://localhost:%s/' % port
107
self._http_starting.release()
108
httpd.socket.settimeout(0.1)
110
while self._http_running:
112
httpd.handle_request()
113
except socket.timeout:
116
def get_remote_url(self, path):
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):])
126
remote_path = '/'.join(path_parts)
128
self._http_starting.acquire()
129
self._http_starting.release()
130
return self._http_base_url + remote_path
133
TestCaseInTempDir.setUp(self)
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"]
148
self._http_running = False
149
self._http_thread.join()
150
if self._http_proxy is not None:
152
os.environ["http_proxy"] = self._http_proxy
153
TestCaseInTempDir.tearDown(self)
added
removed
bzrlib/tests/HTTPTestUtil.py
