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
17
import BaseHTTPServer, SimpleHTTPServer, socket, errno, time
18
20
from bzrlib.tests import TestCaseInTempDir
21
from bzrlib.transport.http import HttpServer
19
22
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
25
class TestCaseWithWebserver(TestCaseInTempDir):
74
"""Derived class that starts a localhost-only webserver
26
"""Derived class that starts a localhost-only webserver.
75
27
(in addition to what TestCaseInTempDir does).
77
This is useful for testing RemoteBranch.
29
This is useful for testing things with a web server.
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
32
def get_remote_url(self, path):
116
34
if os.path.isabs(path):
117
35
remote_path = relpath(self.test_dir, path)
119
37
remote_path = path
121
self._http_starting.acquire()
122
self._http_starting.release()
123
return self._http_base_url + remote_path
38
return self.server.get_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)
41
super(TestCaseWithWebserver, self).setUp()
42
self.server = HttpServer()
44
self.addCleanup(self.server.tearDown)
added
removed
bzrlib/tests/HTTPTestUtil.py
