mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-03 19:34:08 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 9

More buildbot cleanup - let the OS assign the port for test_urllib2_localnet

Browse source
This commit is contained in:
Nick Coghlan 2007-07-25 12:57:48 +00:00
parent 16ed5b4bfe
commit d44a5afaa9
1 changed files with 11 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

40
Lib/test/test_urllib2_localnet.py
View file

@ -40,14 +40,16 @@ class LoopbackHttpServer(BaseHTTPServer.HTTPServer):
class LoopbackHttpServerThread(threading.Thread): class LoopbackHttpServerThread(threading.Thread):
"""Stoppable thread that runs a loopback http server.""" """Stoppable thread that runs a loopback http server."""
def __init__(self, port, RequestHandlerClass): def __init__(self, request_handler):
threading.Thread.__init__(self) threading.Thread.__init__(self)
self._RequestHandlerClass = RequestHandlerClass
self._stop = False self._stop = False
self._port = port
self._server_address = ('127.0.0.1', self._port)
self.ready = threading.Event() self.ready = threading.Event()
self.error = None request_handler.protocol_version = "HTTP/1.0"
self.httpd = LoopbackHttpServer(('127.0.0.1', 0),
request_handler)
#print "Serving HTTP on %s port %s" % (self.httpd.server_name,
# self.httpd.server_port)
self.port = self.httpd.server_port
def stop(self): def stop(self):
"""Stops the webserver if it's currently running.""" """Stops the webserver if it's currently running."""
@ -58,24 +60,9 @@ class LoopbackHttpServerThread(threading.Thread):
self.join() self.join()
def run(self): def run(self):
protocol = "HTTP/1.0"
try:
self._RequestHandlerClass.protocol_version = protocol
httpd = LoopbackHttpServer(self._server_address,
self._RequestHandlerClass)
sa = httpd.socket.getsockname()
#print "Serving HTTP on", sa[0], "port", sa[1], "..."
except:
# Fail "gracefully" if we are unable to start.
self.ready.set()
self.error = sys.exc_info()[1]
raise
self.ready.set() self.ready.set()
while not self._stop: while not self._stop:
httpd.handle_request() self.httpd.handle_request()
# Authentication infrastructure # Authentication infrastructure
@ -232,26 +219,21 @@ class FakeProxyHandler(BaseHTTPServer.BaseHTTPRequestHandler):
class ProxyAuthTests(unittest.TestCase): class ProxyAuthTests(unittest.TestCase):
URL = "http://www.foo.com" URL = "http://www.foo.com"
PORT = 8080
USER = "tester" USER = "tester"
PASSWD = "test123" PASSWD = "test123"
REALM = "TestRealm" REALM = "TestRealm"
PROXY_URL = "http://127.0.0.1:%d" % PORT
def setUp(self): def setUp(self):
FakeProxyHandler.digest_auth_handler.set_users({ FakeProxyHandler.digest_auth_handler.set_users({
self.USER : self.PASSWD self.USER : self.PASSWD
}) })
FakeProxyHandler.digest_auth_handler.set_realm(self.REALM) FakeProxyHandler.digest_auth_handler.set_realm(self.REALM)
self.server = LoopbackHttpServerThread(self.PORT, FakeProxyHandler) self.server = LoopbackHttpServerThread(FakeProxyHandler)
self.server.start() self.server.start()
self.server.ready.wait() self.server.ready.wait()
if self.server.error: proxy_url = "http://127.0.0.1:%d" % self.server.port
raise self.server.error handler = urllib2.ProxyHandler({"http" : proxy_url})
handler = urllib2.ProxyHandler({"http" : self.PROXY_URL})
self._digest_auth_handler = urllib2.ProxyDigestAuthHandler() self._digest_auth_handler = urllib2.ProxyDigestAuthHandler()
self.opener = urllib2.build_opener(handler, self._digest_auth_handler) self.opener = urllib2.build_opener(handler, self._digest_auth_handler)