def tcp_keepalive(server, transport):
sock = transport.get_extra_info('socket')
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
python类server()的实例源码
def tcp_keepalive(server, transport): # pragma: no cover
pass
def test_simple_HTTP(self) :
# ============== Setting up an HTTP server at 'http://localhost:8001/' in current directory
try :
PORT = int(sys.argv[1])
except :
PORT = randint(1025, 65535)
try :
Handler = SimpleHTTPServer.SimpleHTTPRequestHandler
httpd = SocketServer.TCPServer(("", PORT), Handler)
except :
Handler = http.server.SimpleHTTPRequestHandler
httpd = socketserver.TCPServer(("", PORT), Handler)
print ("Serving at port %d" % PORT)
http_thread = Thread( target = httpd.serve_forever, )
http_thread.daemon = True
# ============== Starting the HTTP server
http_thread.start()
# ============== Wait until HTTP server is ready
sleep(1)
with remote_repo(['test_package'], base_url = 'http://localhost:%d/' % PORT) :
from test_package import module1
self.assertTrue(module1.dummy_str) # If this point is reached then the module1 is imported succesfully!
def __init__(self, config, handler_class):
super().__init__(name='Webhook', daemon=False)
self.config = config
if self.config.use_https:
self.server = WebhookHTTPServer(('', self.config.port), handler_class)
self.server.socket = ssl.wrap_socket(self.server.socket.socket,
keyfile=config.https_keyfile,
certfile=config.https_certfile,
server_side=True)
else:
self.server = WebhookHTTPServer(('', self.config.port), handler_class)
self.server.init_props()
def webhook_url(self):
if not hasattr(self, '_webhook_url'):
self.hostname = self.config.host if self.config.host != '' else resolve_public_ip()
self._webhook_url = '%s://%s:%d/%s' % (
'https' if self.config.use_https else 'http',
self.hostname, self.server.server_port, self.server.session_token)
return self._webhook_url
def set_worker(self, worker):
self.server.worker_thread = worker
def stop(self):
self.server.shutdown()
def run(self):
logging.info('Webhook server listening on %s.', self.webhook_url)
self.server.serve_forever()
logging.info('Webhook server stopped.')
def get_voters(amount: hug.types.number = 20):
try:
voters = sorted(json.loads(
linecache.getline("/root/workspace/Chaos/server/voters.json", 1)).items(),
key=lambda x: x[1], reverse=True)
if amount > 0:
voters = voters[:amount]
return {x[0]: x[1] for x in voters}
except:
__log.exception("Failed to read voters!")
def get_meritocracy(amount: hug.types.number = 20):
try:
meritocracy = json.loads(
linecache.getline("/root/workspace/Chaos/server/meritocracy.json", 1))
if amount > 0:
meritocracy = meritocracy[:amount]
return meritocracy
except:
__log.exception("Failed to read meritocracy!")
def server(username, filename, data, parsedQuery): # Ignore filename
if data:
if data == "sfs_version":
return Responder.writeResult(VERSION)
elif data == "python_version":
return Responder.writeResult(sys.version)
else:
return Responder.error("invalid server information request")
else:
return Responder.error("no server information requested")
def test_can_import_several(self):
"""
This test failed in v0.12-pre if e.g.
future/standard_library/email/header.py contained:
from future import standard_library
standard_library.remove_hooks()
"""
import future.moves.urllib.parse as urllib_parse
import future.moves.urllib.request as urllib_request
import http.server
for m in [urllib_parse, urllib_request, http.server]:
self.assertTrue(m is not None)
def test_other_http_imports(self):
import http
import http.server
import http.cookies
import http.cookiejar
self.assertTrue(True)
def from_import(module_name, *symbol_names, **kwargs):
"""
Example use:
>>> HTTPConnection = from_import('http.client', 'HTTPConnection')
>>> HTTPServer = from_import('http.server', 'HTTPServer')
>>> urlopen, urlparse = from_import('urllib.request', 'urlopen', 'urlparse')
Equivalent to this on Py3:
>>> from module_name import symbol_names[0], symbol_names[1], ...
and this on Py2:
>>> from future.moves.module_name import symbol_names[0], ...
or:
>>> from future.backports.module_name import symbol_names[0], ...
except that it also handles dotted module names such as ``http.client``.
"""
if PY3:
return __import__(module_name)
else:
if 'backport' in kwargs and bool(kwargs['backport']):
prefix = 'future.backports'
else:
prefix = 'future.moves'
parts = prefix.split('.') + module_name.split('.')
module = importlib.import_module(prefix + '.' + module_name)
output = [getattr(module, name) for name in symbol_names]
if len(output) == 1:
return output[0]
else:
return output
def __init__(self, *args, **kwargs):
self.handlers = {
'__cobra_hello' : self.handleHello,
'__cobra_getattr' : self.handleGetAttr,
'__cobra_setattr' : self.handleSetAttr,
'__cobra_login' : self.handleLogin
}
http.server.BaseHTTPRequestHandler.__init__(self, *args, **kwargs)
def do_POST(self):
# this is a json posted data list (args, kwargs)
body = json.loads(urllib.parse.unquote(self.rfile.read( int(self.headers['Content-Length'])) ))
#self.handleClient(s.server, s, s.path, body=body)
self.handleClient(body=body)
def handleHello(self, objname):
'''
Hello messages are used to get the initial cache of
method names for the newly connected object.
'''
if verbose: print("GOT A HELLO")
obj = self.server.getSharedObject(objname)
ret = {}
for name in dir(obj):
if type(getattr(obj,name)) in (types.MethodType, types.BuiltinMethodType):
ret[name] = True
self.send_response(http.client.OK)
self.end_headers()
self.wfile.write(json.dumps(ret))
return
def handleSetAttr(self, objname, name, value):
if not self.server.attr:
self.send_response(http.client.FORBIDDEN)
self.end_headers()
excinfo = "__setattr__ disabled"
self.wfile.write(json.dumps(excinfo))
return
obj = self.server.getSharedObject(objname)
if verbose: print("SETTING ATTRIBUTE:", urlparams['args'])
setattr(obj, name, value)
self.send_response(http.client.OK)
self.end_headers()
def setAuthModule(self, authmod):
'''
Enable an authentication module for this server
( all connections *must* be authenticated through the authmod )
NOTE: See cobra.auth.* for various auth module implementations
Example:
import cobra.auth.shadow as c_a_shadow
authmod = c_a_shadow.ShadowFileAuth('passwdfile.txt')
cdaemon = CobraDaemon()
cdaemon.setAuthModule()
'''
self.authmod = authmod
def from_import(module_name, *symbol_names, **kwargs):
"""
Example use:
>>> HTTPConnection = from_import('http.client', 'HTTPConnection')
>>> HTTPServer = from_import('http.server', 'HTTPServer')
>>> urlopen, urlparse = from_import('urllib.request', 'urlopen', 'urlparse')
Equivalent to this on Py3:
>>> from module_name import symbol_names[0], symbol_names[1], ...
and this on Py2:
>>> from future.moves.module_name import symbol_names[0], ...
or:
>>> from future.backports.module_name import symbol_names[0], ...
except that it also handles dotted module names such as ``http.client``.
"""
if PY3:
return __import__(module_name)
else:
if 'backport' in kwargs and bool(kwargs['backport']):
prefix = 'future.backports'
else:
prefix = 'future.moves'
parts = prefix.split('.') + module_name.split('.')
module = importlib.import_module(prefix + '.' + module_name)
output = [getattr(module, name) for name in symbol_names]
if len(output) == 1:
return output[0]
else:
return output
