本文整理汇总了Python中wsgiref.validate.validator函数的典型用法代码示例。如果您正苦于以下问题:Python validator函数的具体用法?Python validator怎么用?Python validator使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了validator函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __init__
def __init__(self, app):
self.app = app
self.server = simple_server.WSGIServer(
('', 8080),
simple_server.WSGIRequestHandler,
)
self.server.set_app(validate.validator(self.app))
开发者ID:j2a,项目名称:pyo-sandbox,代码行数:7,代码来源:trivial_wsgi_server.py
示例2: start_standalone_server
def start_standalone_server(address="localhost", port=8000, app=wsgi_application):
"""Start standalone http server for processing requests"""
validator_app = validator(app)
httpd = make_server(address, port, app)
LOG.info("Serving on {0}:{1}...".format(address, port))
httpd.serve_forever()
开发者ID:jet9,项目名称:jsonrpc20,代码行数:7,代码来源:__init__.py
示例3: sync_apps
def sync_apps(profile=False, validate=False, conquer=False):
app = cherrypy.tree
if profile:
app = profiler.make_app(app, aggregate=False)
if conquer:
try:
import wsgiconq
except ImportError:
warnings.warn("Error importing wsgiconq. pyconquer will not run.")
else:
app = wsgiconq.WSGILogger(app)
if validate:
try:
from wsgiref import validate
except ImportError:
warnings.warn("Error importing wsgiref. The validator will not run.")
else:
app = validate.validator(app)
h = cherrypy.server.httpserver
if hasattr(h, 'wsgi_app'):
# CherryPy's wsgiserver
h.wsgi_app = app
elif hasattr(h, 'fcgiserver'):
# flup's WSGIServer
h.fcgiserver.application = app
elif hasattr(h, 'scgiserver'):
# flup's WSGIServer
h.scgiserver.application = app
开发者ID:Mattsface,项目名称:XenMagic,代码行数:29,代码来源:helper.py
示例4: main
def main():
from wsgiref import simple_server, validate
logging.basicConfig(level=logging.INFO)
parser = optparse.OptionParser(usage=USAGE)
parser.add_option('-p', '--port', dest='port', type='int', default=8000,
help='Port to listen on')
options, args = parser.parse_args()
if len(args) == 0:
path = os.getcwd()
elif len(args) == 1:
path = args[0]
else:
parser.print_help()
return
repo_path = mercurial.dispatch._findrepo(path)
if not repo_path:
print 'No Mercurial repository found for', path
print
parser.print_help()
return
logging.info('Mercurial repository %s', repo_path)
app = validate.validator(downy_app(repo_path))
httpd = simple_server.make_server('', options.port, app)
logging.info('Serving on port %d', options.port)
httpd.serve_forever()
开发者ID:phpmywave,项目名称:phpMyWave,代码行数:27,代码来源:main.py
示例5: main
def main():
"""Run the show"""
#Arguments parser
parser = argparse.ArgumentParser(description="Stream live TV over HTTP to multiple viewers, using Sundtek as capture card")
parser.add_argument("-p","--port",type=int,default=2000,help="server port")
parser.add_argument("-c", "--channel",type=str,default=0,help="set channel")
parser.add_argument("-D", "--dtvmode",type=str,choices=["DVBT", "DVBC", "ATSC","ISDBT"],default="ISDBT",help="set digital TV mode for device")
args=parser.parse_args()
#Setting dtv mode using mediaclient
if dtvmode(args.dtvmode):
server = Server()
#Select channel
channel = select_channel(server.channels,int(args.channel))
print("Channel selected:"+channel)
server.set_channel(channel)
feed_thread = threading.Thread(target=server.feeder)
feed_thread.daemon = True
feed_thread.start()
validator_app = validator(server.simple_app)
httpd = make_server('', args.port, validator_app,server_class=ThreadedWSGIServer)
try:
print("URL: http://%s:%s/stream" % (LOCAL_IP,args.port))
httpd.serve_forever()
finally:
server.cur_chan = None
feed_thread.join()
else:
print("Error setting the dtv mode")
开发者ID:dvidalm,项目名称:dvbbc,代码行数:33,代码来源:dvbbc.py
示例6: get_app
def get_app(self, app=None):
"""Obtain a new (decorated) WSGI app to hook into the origin server."""
if app is None:
app = cherrypy.tree
if self.conquer:
try:
import wsgiconq
except ImportError:
warnings.warn(
"Error importing wsgiconq. pyconquer will not run.")
else:
app = wsgiconq.WSGILogger(app, c_calls=True)
if self.validate:
try:
from wsgiref import validate
except ImportError:
warnings.warn(
"Error importing wsgiref. The validator will not run.")
else:
# wraps the app in the validator
app = validate.validator(app)
return app
开发者ID:ahmadfarihan,项目名称:cherrypy,代码行数:25,代码来源:helper.py
示例7: StartWebServer
def StartWebServer(self, port, application=None):
"""Start web server.
Args:
port: Port to start application on.
application: Optional WSGI function. If none provided will use
tests CreateWsgiApplication method.
Returns:
A tuple (server, application):
server: An instance of ServerThread.
application: Application that web server responds with.
"""
if not application:
application = self.CreateWsgiApplication()
validated_application = validate.validator(application)
try:
server = simple_server.make_server(
'localhost', port, validated_application)
except socket.error:
# Try IPv6
server = simple_server.make_server(
'localhost', port, validated_application, server_class=WSGIServerIPv6)
server = ServerThread(server)
server.start()
return server, application
开发者ID:craigcitro,项目名称:protorpc,代码行数:28,代码来源:webapp_test_util.py
示例8: test
def test():
from wsgiref.validate import validator
app = Cling(getcwd())
try:
print "Serving " + getcwd() + " to http://localhost:8888"
print "Serving " + getcwd() + " to http://localhost:8889"
start_server = lambda port : make_server('0.0.0.0', port, validator(app)).serve_forever()
#start_server(8888)
import threading
server_1 = threading.Thread(target=start_server, name="The 8888 server",args=[8888])
server_2 = threading.Thread(target=start_server, name="The 8889 server",args=[8889])
server_1.start()
server_2.start()
print ""
print "Point your browser at http://localhost:8888/index.html"
server_1.join()
server_2.join()
except KeyboardInterrupt, ki:
print ""
print "Ciao, baby!"
sys.exit(0)
开发者ID:svevang,项目名称:cross-domain-iframe,代码行数:26,代码来源:static.py
示例9: test_cookie_parse
def test_cookie_parse(self):
STR = 'some_string'
class RequestHeader(ComplexModel):
some_field = String
class SomeService(ServiceBase):
__in_header__ = RequestHeader
@rpc(String)
def some_call(ctx, s):
assert ctx.in_header.some_field == STR
def start_response(code, headers):
assert code == HTTP_200
c = SimpleCookie()
c['some_field'] = STR
''.join(validator(WsgiApplication(Application([SomeService], 'tns',
in_protocol=HttpRpc(parse_cookie=True), out_protocol=HttpRpc())))({
'SCRIPT_NAME': '',
'QUERY_STRING': '',
'PATH_INFO': '/some_call',
'REQUEST_METHOD': 'GET',
'SERVER_NAME': 'localhost',
'SERVER_PORT': "9999",
'HTTP_COOKIE': str(c),
'wsgi.url_scheme': 'http',
'wsgi.version': (1,0),
'wsgi.input': StringIO(),
'wsgi.errors': StringIO(),
'wsgi.multithread': False,
'wsgi.multiprocess': False,
'wsgi.run_once': True,
}, start_response))
开发者ID:warvariuc,项目名称:spyne,代码行数:35,代码来源:test_http.py
示例10: sync_apps
def sync_apps(profile=False, validate=False, conquer=False):
apps = []
for base, app in cherrypy.tree.apps.iteritems():
if base == "/":
base = ""
if profile:
app = profiler.make_app(app, aggregate=False)
if conquer:
try:
import wsgiconq
except ImportError:
warnings.warn("Error importing wsgiconq. pyconquer will not run.")
else:
app = wsgiconq.WSGILogger(app)
if validate:
try:
from wsgiref import validate
except ImportError:
warnings.warn("Error importing wsgiref. The validator will not run.")
else:
app = validate.validator(app)
apps.append((base, app))
apps.sort()
apps.reverse()
for s in cherrypy.server.httpservers:
s.mount_points = apps
开发者ID:Juanvvc,项目名称:scfs,代码行数:26,代码来源:helper.py
示例11: testGetWsdl
def testGetWsdl(self):
"""Simple test for serving of WSDL by spyne through pyramid route"""
application = PyramidApplication(
Application([self.HelloWorldService],
tns='spyne.examples.hello',
in_protocol=Soap11(validator='lxml'),
out_protocol=Soap11()))
config = Configurator(settings={'debug_all': True})
config.add_route('home', '/')
config.add_view(application, route_name='home')
wsgi_app = validator(config.make_wsgi_app())
env = {
'SCRIPT_NAME': '',
'REQUEST_METHOD': 'GET',
'PATH_INFO': '/',
'QUERY_STRING': 'wsdl',
}
setup_testing_defaults(env)
request = Request(env)
resp = request.get_response(wsgi_app)
self.assert_(resp.status.startswith("200 "))
node = etree.XML(resp.body) # will throw exception if non well formed
开发者ID:knoxsp,项目名称:spyne,代码行数:25,代码来源:test_pyramid.py
示例12: handle_connection
def handle_connection(conn, port):
request = conn.recv(1)
if not request:
print 'Error, remote client closed connection without sending anything'
return
count = 0
env = {}
while request[-4:] != '\r\n\r\n':
request += conn.recv(1)
request, data = request.split('\r\n',1)
headers = {}
for line in data.split('\r\n')[:-2]:
k, v = line.split(': ', 1)
headers[k.lower()] = v
path = urlparse(request.split(' ', 3)[1])
env['REQUEST_METHOD'] = 'GET'
env['PATH_INFO'] = path[2]
env['QUERY_STRING'] = path[4]
env['CONTENT_TYPE'] = 'text/html'
env['CONTENT_LENGTH'] = str(0)
env['SCRIPT_NAME'] = ''
env['SERVER_NAME'] = socket.getfqdn()
env['SERVER_PORT'] = str(port)
env['wsgi.version'] = (1, 0)
env['wsgi.errors'] = stderr
env['wsgi.multithread'] = False
env['wsgi.multiprocess'] = False
env['wsgi.run_once'] = False
env['wsgi.url_scheme'] = 'http'
env['HTTP_COOKIE'] = headers['cookie'] if 'cookie' in headers.keys() else ''
body = ''
if request.startswith('POST '):
env['REQUEST_METHOD'] = 'POST'
env['CONTENT_LENGTH'] = headers['content-length']
env['CONTENT_TYPE'] = headers['content-type']
while len(body) < int(headers['content-length']):
body += conn.recv(1)
def start_response(status, response_headers):
conn.send('HTTP/1.0 ')
conn.send(status)
conn.send('\r\n')
for pair in response_headers:
key, header = pair
conn.send(key + ': ' + header + '\r\n')
conn.send('\r\n')
env['wsgi.input'] = StringIO(body)
my_app = make_app()
validator_app = validator(my_app)
result = my_app(env, start_response)
for data in result:
conn.send(data)
conn.close()
开发者ID:ConnorAvery,项目名称:cse491-serverz,代码行数:59,代码来源:server.py
示例13: setUpClass
def setUpClass(cls):
cls._httpd = wsgi_boost.WsgiBoostHttp(threads=1)
app = App()
cls._httpd.set_app(validator(app))
cls._server_thread = threading.Thread(target=cls._httpd.start)
cls._server_thread.daemon = True
cls._server_thread.start()
time.sleep(0.5)
开发者ID:romanvm,项目名称:WsgiBoostServer,代码行数:8,代码来源:tests.py
示例14: test
def test():
from wsgiref.validate import validator
magics = StringMagic(title="String Test"), KidMagic(title="Kid Test")
app = Shock('testdata/pub', magics=magics)
try:
make_server('localhost', 9999, validator(app)).serve_forever()
except KeyboardInterrupt, ki:
print "Ciao, baby!"
开发者ID:bgyss,项目名称:static,代码行数:8,代码来源:static.py
示例15: main
def main():
opts, args = parse_options()
bind = parse_bind(opts.bind)
if opts.validate:
application = (Application() + Root())
app = validator(application)
httpd = make_server(bind[0], bind[1], app)
httpd.serve_forever()
raise SystemExit(0)
manager = Manager()
opts.debug and Debugger().register(manager)
Poller = select_poller(opts.poller.lower())
Poller().register(manager)
if opts.server.lower() == "base":
BaseServer(bind).register(manager)
HelloWorld().register(manager)
else:
Server(bind).register(manager)
Root().register(manager)
docroot = os.getcwd() if not args else args[0]
Static(docroot=docroot, dirlisting=True).register(manager)
opts.passwd and Authentication(passwd=opts.passwd).register(manager)
opts.logging and Logger().register(manager)
if opts.profile and hotshot:
profiler = hotshot.Profile(".profile")
profiler.start()
if opts.debug:
print(graph(manager, name="circuits.web"))
print()
print(inspect(manager))
for i in range(opts.jobs):
manager.start(process=True)
manager.run()
if opts.profile and hotshot:
profiler.stop()
profiler.close()
stats = hotshot.stats.load(".profile")
stats.strip_dirs()
stats.sort_stats("time", "calls")
stats.print_stats(20)
开发者ID:ke4roh,项目名称:circuits,代码行数:58,代码来源:main.py
示例16: test_wsgi_input
def test_wsgi_input(self):
def bad_app(e, s):
e["wsgi.input"].read()
s("200 OK", [("Content-Type", "text/plain; charset=utf-8")])
return [b"data"]
out, err = run_amock(validator(bad_app))
self.assertTrue(out.endswith(b"A server error occurred. Please contact the administrator."))
self.assertEqual(err.splitlines()[-2], "AssertionError")
开发者ID:Orav,项目名称:kbengine,代码行数:9,代码来源:test_wsgiref.py
示例17: test
def test():
from wsgiref.validate import validator
app = Cling(getcwd())
try:
print "Serving " + getcwd() + " to http://localhost:9999"
make_server('localhost', 9999, validator(app)).serve_forever()
except KeyboardInterrupt, ki:
print ""
print "Ciao, baby!"
开发者ID:GunioRobot,项目名称:gcli,代码行数:9,代码来源:static.py
示例18: StartWebServer
def StartWebServer(self, port):
"""Start web server."""
application = webapp.WSGIApplication(self.DEFAULT_MAPPING, True)
validated_application = validate.validator(application)
server = simple_server.make_server('localhost', port, validated_application)
server = ServerThread(server)
server.start()
server.wait_until_running()
return server, application
开发者ID:alimills,项目名称:stacks,代码行数:9,代码来源:webapp_test_util.py
示例19: StartServer
def StartServer(self, app):
self.validated = validate.validator(app)
self.port = test_util.pick_unused_port()
self.server = simple_server.make_server('localhost',
self.port,
self.validated)
self.server_thread = webapp_test_util.ServerThread(self.server)
self.server_thread.start()
self.server_thread.wait_until_running()
开发者ID:JoaoVasques,项目名称:MiniGoogle-PageRank,代码行数:9,代码来源:util_test.py
示例20: StartWebServer
def StartWebServer(self, port):
"""Start web server."""
application = self.CreateWsgiApplication()
validated_application = validate.validator(application)
server = simple_server.make_server('localhost', port, validated_application)
server = ServerThread(server)
server.start()
server.wait_until_running()
return server, application
开发者ID:JoaoVasques,项目名称:MiniGoogle-PageRank,代码行数:9,代码来源:webapp_test_util.py
注:本文中的wsgiref.validate.validator函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论