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
19
from bzrlib.osutils import relpath
22
class WebserverNotAvailable(Exception):
25
class TestingHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
26
def log_message(self, format, *args):
27
self.server.test_case.log("webserver - %s - - [%s] %s",
28
self.address_string(),
29
self.log_date_time_string(),
32
def handle_one_request(self):
33
"""Handle a single HTTP request.
35
You normally don't need to override this method; see the class
36
__doc__ string for information on how to handle specific HTTP
37
commands such as GET and POST.
40
for i in xrange(1,11): # Don't try more than 10 times
42
self.raw_requestline = self.rfile.readline()
43
except socket.error, e:
44
if e.args[0] in (errno.EAGAIN, errno.EWOULDBLOCK):
45
# omitted for now because some tests look at the log of
46
# the server and expect to see no errors. see recent
47
# email thread. -- mbp 20051021.
48
## self.log_message('EAGAIN (%d) while reading from raw_requestline' % i)
54
if not self.raw_requestline:
55
self.close_connection = 1
57
if not self.parse_request(): # An error code has been sent, just exit
59
mname = 'do_' + self.command
60
if not hasattr(self, mname):
61
self.send_error(501, "Unsupported method (%r)" % self.command)
63
method = getattr(self, mname)
66
class TestingHTTPServer(BaseHTTPServer.HTTPServer):
67
def __init__(self, server_address, RequestHandlerClass, test_case):
68
BaseHTTPServer.HTTPServer.__init__(self, server_address,
70
self.test_case = test_case
73
class TestCaseWithWebserver(TestCaseInTempDir):
74
"""Derived class that starts a localhost-only webserver
75
(in addition to what TestCaseInTempDir does).
77
This is useful for testing RemoteBranch.
80
_HTTP_PORTS = range(13000, 0x8000)
82
def _http_start(self):
83
import SimpleHTTPServer, BaseHTTPServer, socket, errno
85
for port in self._HTTP_PORTS:
87
httpd = TestingHTTPServer(('localhost', port),
88
TestingHTTPRequestHandler,
90
except socket.error, e:
91
if e.args[0] == errno.EADDRINUSE:
93
print >>sys.stderr, "Cannot run webserver :-("
99
raise WebserverNotAvailable("Cannot run webserver :-( "
100
"no free ports in range %s..%s" %
101
(_HTTP_PORTS[0], _HTTP_PORTS[-1]))
103
self._http_base_url = 'http://localhost:%s/' % port
104
self._http_starting.release()
105
httpd.socket.settimeout(0.1)
107
while self._http_running:
109
httpd.handle_request()
110
except socket.timeout:
113
def get_remote_url(self, path):
116
if os.path.isabs(path):
117
remote_path = relpath(self.test_dir, path)
121
self._http_starting.acquire()
122
self._http_starting.release()
123
return self._http_base_url + remote_path
126
TestCaseInTempDir.setUp(self)
128
self._http_starting = threading.Lock()
129
self._http_starting.acquire()
130
self._http_running = True
131
self._http_base_url = None
132
self._http_thread = threading.Thread(target=self._http_start)
133
self._http_thread.setDaemon(True)
134
self._http_thread.start()
135
self._http_proxy = os.environ.get("http_proxy")
136
if self._http_proxy is not None:
137
del os.environ["http_proxy"]
140
self._http_running = False
141
self._http_thread.join()
142
if self._http_proxy is not None:
144
os.environ["http_proxy"] = self._http_proxy
145
TestCaseInTempDir.tearDown(self)
added
removed
bzrlib/tests/HTTPTestUtil.py
