def serve(store):
args = store.args
abe = Aml(store, args)
if args.host or args.port:
# HTTP server.
if args.host is None:
args.host = "localhost"
from wsgiref.simple_server import make_server
port = int(args.port or 80)
httpd = make_server(args.host, port, abe )
print "Listening on http://" + args.host + ":" + str(port)
try:
httpd.serve_forever()
except:
httpd.shutdown()
raise
python类make_server()的实例源码
def test_make_client(self):
test = self.Test2(value=10)
app = bottle_api.make_app(('/api', test))
server = make_server('localhost', 10001, app)
process = Thread(
target=server.serve_forever)
process.start()
time.sleep(1) # Wait for the app to run
c = client.make_client(
instance=test, base_url='http://localhost:10001/api')
data = c.endpoint(arg1='test')
expected_data = {'works': True, 'arg1': 'test', 'value': 10}
self.assertEqual(data, expected_data)
with self.assertRaises(AttributeError):
c.invalid_endpoint()
test.endpoint(arg1='test') # For 100% coverage :D
app.close()
server.shutdown()
process.join()
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
def run(self, handler):
self.certfile = 'keys/server.pem'
from wsgiref.simple_server import make_server, WSGIRequestHandler
import ssl
if self.quiet:
class QuietHandler(WSGIRequestHandler):
def log_request(*args, **kw): pass
self.options['handler_class'] = QuietHandler
srv = make_server(self.host, self.port, handler, **self.options)
try:
with open(self.certfile):
pass
srv.socket = ssl.wrap_socket(
srv.socket,
certfile=self.certfile
)
srv.serve_forever()
except IOError as e:
logging.error("Unable to open Certificate file at location {} {}".
format(self.certfile, e))
raise
def __init__(self, host="127.0.0.1", port=8545, *args, **kwargs):
if not is_testrpc_available():
raise Exception("`TestRPCProvider` requires the `eth-testrpc` package to be installed")
from testrpc.server import get_application
application = get_application()
self.server = make_server(
host,
port,
application,
)
self.thread = spawn(self.server.serve_forever)
endpoint_uri = 'http://{0}:{1}'.format(host, port)
super(TestRPCProvider, self).__init__(endpoint_uri, *args, **kwargs)
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
def run_client(queue):
try:
app = ClientApplication(
callback_url="http://127.0.0.1:15487/callback",
client_id="abc",
client_secret="xyz",
provider_url="http://127.0.0.1:15486")
httpd = make_server('', 15487, app,
handler_class=NoLoggingHandler)
queue.put({"result": 0})
httpd.serve_forever()
except Exception as e:
queue.put({"result": 1, "error_message": str(e)})
def run(self):
"""
Open WSGI server to listen to HOST_BASE address
"""
print("Running")
parts = urlparse(self.hostbase)
domain, port = parts.netloc.split(":")
print(domain, port)
self.srv = make_server(domain, int(port), self.app)
try:
self.srv.serve_forever()
except:
# We are a background thread so we have problems to interrupt tests in the case of error
import traceback
traceback.print_exc()
# Failed to start
self.srv = None
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
fixtures.py 文件源码
项目:schul_cloud_resources_server_tests
作者: schul-cloud
项目源码
文件源码
阅读 29
收藏 0
点赞 0
评论 0
def run(self, app):
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
self.srv = make_server(self.host, self.port, app, server_cls, handler_cls)
self.srv.serve_forever()
while not self.get_port(): time.sleep(0.001)
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
def __init__(self, config, port):
super(ReahlWebServer, self).__init__()
self.in_separate_thread = None
self.running = False
self.handlers = {}
self.httpd_thread = None
certfile = pkg_resources.resource_filename(__name__, 'reahl_development_cert.pem')
self.reahl_wsgi_app = WrappedApp(ReahlWSGIApplication(config))
try:
https_port = port+363
self.httpd = ReahlWSGIServer.make_server('', port, self.reahl_wsgi_app)
self.httpsd = SSLCapableWSGIServer.make_server('', https_port, certfile, self.reahl_wsgi_app)
except socket.error as ex:
message = ('Caught socket.error: %s\nThis means that another process is using one of these ports: %s, %s. ' % (ex, port, https_port)) \
+'\nIf this happens while running tests, it probably means that a browser client did not close its side of a connection to a previous server you had running - and that the server socket now sits in TIME_WAIT state. Is there perhaps a browser hanging around from a previous run? I have no idea how to fix this automatically... see http://hea-www.harvard.edu/~fine/Tech/addrinuse.html'
raise AssertionError(message)
def runServer(kodi = '', kodi_home = '', startBrowser=False):
importer = ksi.KodiScriptImporter(kodi, kodi_home)
importer.install(True)
kodiSrv = KodiServer(importer)
wsgiApp = partial(application, server=kodiSrv)
server_address = ('localhost', 5000)
httpd = make_server(
server_address[0], # The host name
server_address[1], # A port number where to wait for the request
wsgiApp # The application object name, in this case a function
)
srvThread = threading.Thread(target=httpd.serve_forever)
if startBrowser:
webbrowser.open('http://{}:{}'.format(*server_address))
srvThread.start()
return httpd
def run(self, app): # pragma: no cover
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
from wsgiref.simple_server import make_server
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
srv = make_server(self.host, self.port, app, server_cls, handler_cls)
srv.serve_forever()
def run(self, handler):
from wsgiref.simple_server import make_server, WSGIRequestHandler
import ssl
if self.quiet:
class QuietHandler(WSGIRequestHandler):
def log_request(*args, **kw):
pass
self.options['handler_class'] = QuietHandler
srv = make_server(self.host, self.port, handler, **self.options)
srv.socket = ssl.wrap_socket(
srv.socket,
keyfile=self.keyfile,
certfile=self.certfile,
ca_certs=self.cafile,
cert_reqs=ssl.CERT_REQUIRED,
server_side=True)
srv.serve_forever()
def run(self, handler):
from wsgiref.simple_server import make_server, WSGIRequestHandler
if self.quiet:
class QuietHandler(WSGIRequestHandler):
def log_request(*args, **kw):
pass
self.options['handler_class'] = QuietHandler
try:
self.server = make_server(self.host, self.port, handler,
**self.options)
self.romana_http.wsgi_server_started = True
logging.info("HTTP server: Started to listen...")
self.server.serve_forever()
except socket.error as e:
logging.fatal("HTTP server: Cannot open socket "
"(error %d: %s)... " %
(e.errno, e.strerror))
def run(self, handler):
from wsgiref.simple_server import make_server, WSGIRequestHandler
import ssl
if self.quiet:
class QuietHandler(WSGIRequestHandler):
def log_request(*args, **kw): pass
self.options['handler_class'] = QuietHandler
srv = make_server(self.host, self.port, handler, **self.options)
srv.socket = ssl.wrap_socket (
srv.socket,
keyfile='mfa_slipstream.pem',
certfile='mfa_slipstream.pem', # path to certificate
server_side=True)
srv.serve_forever()
##########################################################################
# UTILITY FUNCTIONS
##########################################################################
def run_dev_server(): # pragma: no cover
"""Usage: python -m nuka.report <type> <data_filename>"""
from wsgiref.simple_server import make_server
engine = nuka.config.get_template_engine()
type, filename = sys.argv[-2:]
def app(environ, start_response):
start_response('200 OK',
[('Content-Type', 'text/html; charset=utf8')])
template = engine.get_template('reports/{0}.html.j2'.format(type))
with open(filename) as fd:
data = json.load(fd)
html = template.render(dict(data=data, dumped_data=json.dumps(data)))
return [html.encode('utf8')]
httpd = make_server('', 8000, app)
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
def http(tasks, bind):
"""Http interface using built-in simple wsgi server"""
from wsgiref.simple_server import make_server
from .utils import load_manager
from .http import Application
host, _, port = bind.partition(':')
app = Application(load_manager(tasks))
httpd = make_server(host, int(port), app)
print('Listen on {}:{} ...'.format(host or '0.0.0.0', port), file=sys.stderr)
httpd.serve_forever()
def run(self, app): # pragma: no cover
from wsgiref.simple_server import make_server
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = self.options.get('handler_class', FixedHandler)
server_cls = self.options.get('server_class', WSGIServer)
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
self.srv = make_server(self.host, self.port, app, server_cls,
handler_cls)
self.port = self.srv.server_port # update port actual port (0 means random)
try:
self.srv.serve_forever()
except KeyboardInterrupt:
self.srv.server_close() # Prevent ResourceWarning: unclosed socket
raise
def run(self, handler):
class QuietHandler(WSGIRequestHandler):
def log_request(*args, **kw):
pass
def log_error(self, format, *args):
pass
self.options['handler_class'] = QuietHandler
srv = make_server(self.host, self.port, handler, **self.options)
srv.serve_forever()
def run(self, handler):
class QuietHandler(WSGIRequestHandler):
def log_request(*args, **kw):
pass
def log_error(self, format, *args):
pass
self.options['handler_class'] = QuietHandler
srv = make_server(self.host, self.port, handler, **self.options)
srv.socket = ssl.wrap_socket(srv.socket, certfile=CERT_PATH, server_side=True)
srv.serve_forever()
def run(self, app): # pragma: no cover
from wsgiref.simple_server import make_server
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
import socket
class FixedHandler(WSGIRequestHandler):
def address_string(self): # Prevent reverse DNS lookups please.
return self.client_address[0]
def log_request(*args, **kw):
if not self.quiet:
return WSGIRequestHandler.log_request(*args, **kw)
handler_cls = FixedHandler
server_cls = WSGIServer
if ':' in self.host: # Fix wsgiref for IPv6 addresses.
if getattr(server_cls, 'address_family') == socket.AF_INET:
class server_cls(server_cls):
address_family = socket.AF_INET6
self.srv = make_server(self.host, self.port, app, server_cls,
handler_cls)
self.port = self.srv.server_port # update port actual port (0 means random)
try:
self.srv.serve_forever()
except KeyboardInterrupt:
self.srv.server_close() # Prevent ResourceWarning: unclosed socket
raise
def run(self):
"""Run the Process"""
self.runner = TaskRunner()
self.runner_api = TaskRunnerAPI(process=self, runner=self.runner)
self.app = bottle_api.make_app(('/api', self.runner_api))
self.server = make_server(self.host, self.port, self.app)
self.runner.start()
self.server.serve_forever()
self.runner.join()