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.selftest 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] == errno.EAGAIN:
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
path_parts = path.split(os.path.sep)
117
if os.path.isabs(path):
118
if path_parts[:len(self._local_path_parts)] != \
119
self._local_path_parts:
120
raise BadWebserverPath(path, self.test_dir)
121
remote_path = '/'.join(path_parts[len(self._local_path_parts):])
123
remote_path = '/'.join(path_parts)
125
self._http_starting.acquire()
126
self._http_starting.release()
127
return self._http_base_url + remote_path
130
TestCaseInTempDir.setUp(self)
132
self._local_path_parts = self.test_dir.split(os.path.sep)
133
self._http_starting = threading.Lock()
134
self._http_starting.acquire()
135
self._http_running = True
136
self._http_base_url = None
137
self._http_thread = threading.Thread(target=self._http_start)
138
self._http_thread.setDaemon(True)
139
self._http_thread.start()
142
self._http_running = False
143
self._http_thread.join()
144
TestCaseInTempDir.tearDown(self)
added
removed
bzrlib/selftest/HTTPTestUtil.py
