我们从Python开源项目中，提取了以下28个代码示例，用于说明如何使用server.Server()。
def __init__(self, url, port=None, ip=None, auto_shutdown=True, wait_time=20, timeout=5, is_playing_fnc=None): self.port = port if port else random.randint(8000,8099) self.ip = ip if ip else "127.0.0.1" self.connected = False self.start_time = None self.last_connect = None self.is_playing_fnc = is_playing_fnc self.auto_shutdown = auto_shutdown self.wait_time = wait_time self.timeout = timeout self.running = False self.file = None self.files = [] self._server = Server((self.ip, self.port), Handler, client=self) self.add_url(url) self.start()
def analyze(self, number=20): self.number = number common_terms = self.terms_only() common_hashes = self.hash_only() common_mentions = self.mentions_only() common_coocurrances = self.cooccurances() self.toJSON(common_terms) #self.timeDataVisualization() #self.printInstructions() #self.server = Server() #self.server.openbrowser('http://localhost:8000/chart.html') # def printInstructions(self): # print "Setup server by going into console:" # print ">>>python -m SimpleHTTPServer" # print "Go to http://localhost:8000 on your browser" # print "View chart.html"
def testBasicSearch(self): mydb = db.DBMemory() with mydb: obj = server.Server(mydb) ctx = server.Context("<unittest>") data={} data["contentType"]="mediawiki" data["content"]="= Test =\nThis is a test page to verify that search is working\n" obj.updatePage(ctx,"testPage",data) ret = obj.search(ctx,"test") print(ret) ret = obj.search(ctx,"verify") print(ret)
def dispatch_remote_integrity_checker(args): """ Dispatch the main remote integrity tool :param args: Arguments passed to the script :return: None """ config = load_config(path=args.config) server = Server(config=config) server.connect() output = server.acquire_checksum_list() logger = Logger(config=config) integrity = Integrity(config=config) integrity.on_events_detected += logger.dispatch_syslog integrity.on_events_detected += logger.dispatch_events_mail integrity.on_events_detected += logger.dispatch_telegram_msg integrity.load_database() integrity.identify(output) integrity.print_statistics() database.commit()
def run(self): # TODO: hacer server atributo para gestionar el stop y restart server = Server(self._log, self._config) server.run() for t in server.threads: t.join()
def _wait_for_server_start(port): """Wait until server at given port is running / listening for connections.""" while(not _is_server_listening(port)): debug_print_verbose("Waiting for server start at port %d" % port) time.sleep(2) debug_print_verbose("Server started listening at port %d" % port)
def __init__(self, address, verbose = server.VERBOSE): FSProxyLocal.__init__(self) server.Server.__init__(self, address, verbose)
def _close(self): server.Server._close(self) FSProxyLocal._close(self)
def _serve(self): server.Server._serve(self) # Retreat into start directory while self._dirstack: self.back()
def start(self): self.start_time = time.time() self.running = True self._server.run() t= Thread(target=self._auto_shutdown) t.setDaemon(True) t.start() logger.info("MEGA Server Started")
def stop(self): self.running = False self._server.stop() logger.info("MEGA Server Stopped")
def main(_): model = Model() if FLAGS.train: model.train() model.save() sys.exit(0) else: model.restore() app = Flask(__name__) server = Server() server.set_model(model) app.add_url_rule('/', view_func=server.server_running) app.add_url_rule('/predict', view_func=server.predict) app.run(host= '0.0.0.0', port=80) sys.exit(0)
def __init__(self): self.log = logging.getLogger("Lambda") self.db = self.createDb() self.server = server.Server(self.db) self.server.init() self.resp = None self.ctx = None
def testRender(self): obj = server.Server(None) self.assertEquals("<body>\n<h1> Test </h1>\n</body>",obj._renderMediaWiki("= Test =\n")) self.assertEquals("<body>\n<h1> Test </h1>\n</body>",obj._renderMediaWiki("= Test ="))
def testSanitizeEvil(self): obj = server.Server(None) self.assertFalse("<script>" in obj.sanitize("<script>evil();</script>"))
def testSanitizeSafe(self): obj = server.Server(None) self.assertEquals("<h1>Welcome</h1>",obj.sanitize("<h1>Welcome</h1>"))
def testNonRenderedPage(self): mydb = db.DBMemory() obj = server.Server(mydb) ctx = server.Context("<unittest>") mydb.updatePage("testPage","<unittest>","mediawiki","= Test =",None) self.assertEquals("\n<h1> Test </h1>\n",obj.getPage(ctx,"testPage")["html"])
def testRenderedPage(self): mydb = db.DBMemory() obj = server.Server(mydb) ctx = server.Context("<unittest>") rendered = "<h1>Rendered</h1>" mydb.updatePage("testPage","<unittest>","mediawiki","= Test =",rendered) self.assertEquals(rendered,obj.getPage(ctx,"testPage")["html"])
def testNoFound(self): mydb = db.DBMemory() obj = server.Server(mydb) ctx = server.Context("<unittest>") self.assertEquals(False, mydb.doesPageExist("doesNotExist")) with self.assertRaises(custom_exceptions.NotFound): obj.getPage(ctx, "doesNotExist")
def testInvalidInput(self): mydb = db.DBMemory() obj = server.Server(mydb) ctx = server.Context("<unittest>") with self.assertRaises(custom_exceptions.InvalidInput): obj.updatePage(ctx,"Index",{}) with self.assertRaises(custom_exceptions.InvalidInput): obj.updatePage(ctx,"Index",{"contentType":"mediawiki","content":{}})
def __init__(self, args): self.args = args self.bus = Bus.Bus(args.serial_device, BAUDRATE, debug=args.bus_debug) self.bus.drain() self.sound = SoundManager({"beep":"beep_short.wav", "beep_end": "beep_end.wav", "error": "error.wav", "defused": "bomb_defused.wav", "explosion": "explosion_02.wav"}) # check for mastercontrol module if not args.ignore_master_control: control_description = self.bus.check_for_module(Bus.CONTROL_MODULE) if control_description is None: raise Exception("no mastercontrol module found!") self.bus.init_module(Bus.CONTROL_MODULE, True, "1000A", control_description["num_random"]) # check other modules self.modules = check_existing_modules(self.bus) for m in args.disable: try: self.modules.pop(m) except KeyError: print("disable module {} which was not enumerated. typo?".format(m)) if args.mode == "gui": self.server = Server("", self.modules) if args.mode == "webgui": self.server = WebServer(("0.0.0.0", 8080), module_descriptions=self.modules)
def fromDict(cls, request_dict): request = Request() request.data = request_dict["data"] request.action = request_dict["action"] request.context = request_dict["context"] if "instances" in request.data: instance = request.data["instances"][0] else: build_request = request.data["build_requests"][0] instance = build_request["nova_object.data"]["instance"] instance_data = instance["nova_object.data"] request.id = instance_data["uuid"] request.user_id = instance_data["user_id"] request.project_id = instance_data["project_id"] created_at = instance_data["created_at"] request.created_at = datetime.strptime(created_at, "%Y-%m-%dT%H:%M:%SZ") flavor_data = instance_data["flavor"]["nova_object.data"] flavor = Flavor() flavor.setId(flavor_data["flavorid"]) flavor.setName(flavor_data["name"]) flavor.setMemory(flavor_data["memory_mb"]) flavor.setVCPUs(flavor_data["vcpus"]) flavor.setStorage(flavor_data["root_gb"]) server = Server() server.setFlavor(flavor) server.setId(instance_data["uuid"]) server.setUserId(instance_data["user_id"]) server.setProjectId(instance_data["project_id"]) server.setCreatedAt(instance_data["created_at"]) server.setMetadata(instance_data["metadata"]) server.setUserData(instance_data["user_data"]) server.setKeyName(instance_data["key_name"]) server.setType() request.server = server if "filter_properties" in request.data: filter_properties = request.data["filter_properties"] request.retry = filter_properties.get("retry", {}) else: request_spec = request.data["request_specs"][0] nova_object = request_spec["nova_object.data"] request.retry = nova_object.get("retry", {}) if not request.retry: request.retry = {} return request
