def setUpClass(cls):
"""
Set up an HTTP server to serve the XML files. Set the correct port in
the IGD.xml URLBase element.
"""
# Have to chdir here because the py2 SimpleHTTPServer doesn't allow us
# to change its working directory like the py3 one does.
os.chdir(path.join(path.dirname(path.realpath(__file__)), 'xml'))
cls.httpd = sockserver.TCPServer(('127.0.0.1', 0), httpserver.SimpleHTTPRequestHandler)
cls.httpd_thread = threading.Thread(target=cls.httpd.serve_forever)
cls.httpd_thread.daemon = True
cls.httpd_thread.start()
cls.httpd_port = cls.httpd.server_address[1]
with open('upnp/IGD.xml', 'w') as out_f:
with open('upnp/IGD.xml.templ') as in_f:
out_f.write(in_f.read().format(port=cls.httpd_port))
python类SimpleHTTPRequestHandler()的实例源码
def run(self, open_immediately, port):
'''
Serves the `www` directory.
Args:
open_immediately: Whether to open the web browser immediately
port: The port at which to serve the graph
'''
os.chdir(self.directory)
handler = http.SimpleHTTPRequestHandler
handler.extensions_map.update({
'.webapp': 'application/x-web-app-manifest+json',
})
server = socketserver.TCPServer(('', port), handler)
server.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
address = 'http://localhost:{0}/graph.html'.format(port)
log.info('Serving at %s', address)
if open_immediately:
log.debug('Opening webbrowser')
webbrowser.open(address)
server.serve_forever()
def start_server():
# avoid 8080, as the router may have service on it.
# Firewall rules will need to be changed in the router
# to allow access on this port.
server_address = ('', 9001)
cs.CSClient().log(APP_NAME, "Starting Server: {}".format(server_address))
cs.CSClient().log(APP_NAME, "Web Message is: {}".format(WEB_MESSAGE))
httpd = HTTPServer(server_address, WebServerRequestHandler)
# Use the line below to serve the index.html page that is in the
# app directory.
# httpd = HTTPServer(server_address, SimpleHTTPRequestHandler)
try:
httpd.serve_forever()
except KeyboardInterrupt:
cs.CSClient().log(APP_NAME, "Stopping Server, Key Board interrupt")
return 0
def setUp(self):
super(WebSocketRequestHandlerTestCase, self).setUp()
self.stubs = stubout.StubOutForTesting()
self.tmpdir = tempfile.mkdtemp('-websockify-tests')
# Mock this out cause it screws tests up
self.stubs.Set(os, 'chdir', lambda *args, **kwargs: None)
self.stubs.Set(SimpleHTTPRequestHandler, 'send_response',
lambda *args, **kwargs: None)
def test_normal_get_with_only_upgrade_returns_error(self):
server = self._get_server(web=None)
handler = websocket.WebSocketRequestHandler(
FakeSocket('GET /tmp.txt HTTP/1.1'), '127.0.0.1', server)
def fake_send_response(self, code, message=None):
self.last_code = code
self.stubs.Set(SimpleHTTPRequestHandler, 'send_response',
fake_send_response)
handler.do_GET()
self.assertEqual(handler.last_code, 405)
def test_list_dir_with_file_only_returns_error(self):
server = self._get_server(file_only=True)
handler = websocket.WebSocketRequestHandler(
FakeSocket('GET / HTTP/1.1'), '127.0.0.1', server)
def fake_send_response(self, code, message=None):
self.last_code = code
self.stubs.Set(SimpleHTTPRequestHandler, 'send_response',
fake_send_response)
handler.path = '/'
handler.do_GET()
self.assertEqual(handler.last_code, 404)
def handle_one_request(self):
try:
httpserver.SimpleHTTPRequestHandler.handle_one_request(self)
except socket.error:
if not hasattr(self, "_host_disconnected"):
self._host_disconnected = 1
dprint("Host disconnected")
elif self._host_disconnected < MAX_DISCONNECT:
self._host_disconnected += 1
dprint("Host disconnected: %d" % self._host_disconnected)
else:
dprint("Closed connection to avoid infinite loop")
self.close_connection = True
def __init__(self, *initargs):
'''
???
'''
super(SimpleHTTPRequestHandler, self).__init__(*initargs)
def main():
SERVER_PORT = 8443
httpd = None
# Since unit tests run in parallel, the port may be in use, so
# retry creating the server while incrementing the port number
while SERVER_PORT < 8493: # Max 50 retries
try:
# SSL server copied from here:
# http://www.piware.de/2011/01/creating-an-https-server-in-python/
httpd = BaseHTTPServer.HTTPServer(
('localhost', SERVER_PORT),
SimpleHTTPServer.SimpleHTTPRequestHandler)
except socket.error as e:
if e.errno == 98: # Address in use
SERVER_PORT += 1
continue
else:
# Some other socket.error
raise
else:
# Success
break
else:
# Did not break from loop, so we ran out of retries
assert False, "Could not bind server port: all ports in use."
httpd.socket = ssl.wrap_socket(
httpd.socket, certfile=SERVER_CERT, server_side=True,
ssl_version=ssl.PROTOCOL_TLSv1,
)
def ssl_server():
httpd.serve_forever()
# Start the SSL server
thread = threading.Thread(target=ssl_server)
thread.daemon = True
thread.start()
# Wait a bit for the server to start
time.sleep(1)
# Use requests to get a page from the server
requests.get(
u"https://localhost:{}".format(SERVER_PORT),
verify=SERVER_CERT)
# requests.get("https://github.com")
def main():
SERVER_PORT = 8443
httpd = None
# Since unit tests run in parallel, the port may be in use, so
# retry creating the server while incrementing the port number
while SERVER_PORT < 8493: # Max 50 retries
try:
# SSL server copied from here:
# http://www.piware.de/2011/01/creating-an-https-server-in-python/
httpd = BaseHTTPServer.HTTPServer(
('localhost', SERVER_PORT),
SimpleHTTPServer.SimpleHTTPRequestHandler)
except socket.error as e:
if e.errno == 98: # Address in use
SERVER_PORT += 1
continue
else:
# Some other socket.error
raise
else:
# Success
break
else:
# Did not break from loop, so we ran out of retries
assert False, "Could not bind server port: all ports in use."
httpd.socket = ssl.wrap_socket(
httpd.socket, certfile=SERVER_CERT, server_side=True,
ssl_version=ssl.PROTOCOL_TLSv1,
)
def ssl_server():
httpd.serve_forever()
# Start the SSL server
thread = threading.Thread(target=ssl_server)
thread.daemon = True
thread.start()
# Wait a bit for the server to start
time.sleep(1)
# Use requests to get a page from the server
requests.get(
u"https://localhost:{}".format(SERVER_PORT),
verify=SERVER_CERT)
# requests.get("https://github.com")
