Best Python code snippet using playwright-python
web_socket.py
Source:web_socket.py
...13 print("WebSocket opened")14 15 def on_close(self):16 print("WebSocket closed")17 self.application.outq.remove_listener(self.sec_num)18 # self.application.pc.remove_event_listener(self)19 def on_message(self, message):20 #"op_type;amount;acc_local;acc_remote"21 # print message22 self.sec_num = self.application.outq.add_listener(self)23 info = json.loads(message)24 # {âestadoâ: <âcomienzoâ|âconfirmacionâ|âerrorâ>, 25 # âidâ: nId, âtipoâ: <âconsignarâ|âretirarâ>, 26 # âmontoâ: nMonto, 27 # âcuentaDestinoâ: nCDestino, âcuentaOrigenâ: nCOrigen}28 # acc_num = 129 header = 'comienzo'30 # op_type = 'transaccion'31 # amount = 400032 # code = self.sec_num33 message = {u'id':self.sec_num,34 u'estado':header, 35 u'tipo':info['op_type'],36 u'monto':info['amount'],37 u'cuentaOrigen':info['acc_local'],38 u'cuentaDestino':info['acc_remote'],39 u'idCliente':int(self.get_cookie("authcookie").split('-')[0])}40 print message41 self.application.outq.init_transaction(message)42 # self.application.pc.publish_message(message)asdsdfsdfsdfsdfs.jhdkhgdhhkghjghghjf43 def notify_client(self, msg):44 self.write_message(msg)45 self.application.outq.remove_listener(self.sec_num)46class WebSocketHandlerAssociate(tornado.websocket.WebSocketHandler):47 def open(self, *args, **kwargs):48 # self.application.pc.add_event_listener(self)49 print("WebSocket opened")50 51 def on_close(self):52 print("WebSocket closed")53 self.application.outq.remove_listener(self.sec_num)54 # self.application.pc.remove_event_listener(self)55 def on_message(self, message):56 #"op_type;amount;acc_local;acc_remote"57 # print message58 self.sec_num = self.application.outq.add_listener(self)59 info = json.loads(message)60 #{âestadoâ: <âcomienzoâ|âconfirmacionâ|âerrorâ>âidâ: nId, âtipoâ: âasociarâ, âcuentaOrigenâ: nCOrigen, 61 #âbancoOrigenâ: <âbancandesâ|âllamabankâ> âcuentaDestinoâ: nCDestino, âbancoDestinoâ: <âbancandesâ|âllamabankâ>, 62 #ânombreEmpleadoâ:nNombre, âvalorâ: nValor, âfrecuenciaâ:<âmensualâ|âquincenalâ>}63 # acc_num = 164 header = 'comienzo'65 # op_type = 'transaccion'66 # amount = 400067 # code = self.sec_num68 if info['frec'] is 1:69 frecuencia = 'mensual'70 else:71 frecuencia = 'quincenal'72 message = {u'id':self.sec_num,73 u'estado':header, 74 u'tipo':'asociar',75 u'valor':info['salario'],76 u'frecuencia':frecuencia,77 u'cuentaOrigen':info['cuenta'],78 'bancoOrigen':"llamabank",79 'bancoDestino':'bancandes',80 u'cuentaDestino':info['cuenta_empl'],81 u'idCliente':int(self.get_cookie("authcookie").split('-')[0])}82 print message83 self.application.outq.init_associate(message)84 # self.application.pc.publish_message(message)asdsdfsdfsdfsdfs.jhdkhgdhhkghjghghjf85 def notify_client(self, msg):86 self.write_message(msg)87 self.application.outq.remove_listener(self.sec_num)88class WebSocketHandlerPay(tornado.websocket.WebSocketHandler):89 def open(self, *args, **kwargs):90 # self.application.pc.add_event_listener(self)91 print("WebSocket opened")92 93 def on_close(self):94 print("WebSocket closed")95 self.application.outq.remove_listener(self.sec_num)96 # self.application.pc.remove_event_listener(self)97 def on_message(self, message):98 #"op_type;amount;acc_local;acc_remote"99 # print message100 self.sec_num = self.application.outq.add_listener(self)101 info = json.loads(message)102 #{âestadoâ: <âcomienzoâ|âconfirmacionâ|âerrorâ>âidâ: nId, âtipoâ: âasociarâ, âcuentaOrigenâ: nCOrigen, 103 #âbancoOrigenâ: <âbancandesâ|âllamabankâ> âcuentaDestinoâ: nCDestino, âbancoDestinoâ: <âbancandesâ|âllamabankâ>, 104 #ânombreEmpleadoâ:nNombre, âvalorâ: nValor, âfrecuenciaâ:<âmensualâ|âquincenalâ>}105 # acc_num = 1106 header = 'comienzo'107 # op_type = 'transaccion'108 # amount = 4000109 # code = self.sec_num110 # {u'numCuenta': 123, u'saldo': 6666666, u'estado': u'comienzo', u'id': u'fghiop', u'tipo': u'pagar'111 message = {u'id':self.sec_num,112 u'estado':header, 113 u'tipo':'pagar',114 u'numCuenta':info['acc'],115 u'idCliente':int(self.get_cookie("authcookie").split('-')[0])}116 print message117 self.application.outq.init_pay(message)118 # self.application.pc.publish_message(message)asdsdfsdfsdfsdfs.jhdkhgdhhkghjghghjf119 def notify_client(self, msg):120 self.write_message(msg)121 self.application.outq.remove_listener(self.sec_num)122class WebSocketHandlerOperations(tornado.websocket.WebSocketHandler):123 def open(self, *args, **kwargs):124 # self.application.pc.add_event_listener(self)125 print("WebSocket opened")126 self.sec_num = self.application.outq.add_listener(self)127 def on_close(self):128 print("WebSocket closed")129 self.application.outq.remove_listener(self.sec_num)130 def on_message(self, message):131 # print(message)132 # [{"name":"draw","value":1},133 # {"name":"columns","value":134 # [{"data":"fecha","name":"","searchable":true,"orderable":true,"search":{"value":"","regex":false}},{"data":"tipo","name":"","searchable":true,"orderable":true,"search":{"value":"","regex":false}},{"data":"id_cliente","name":"","searchable":true,"orderable":true,"search":{"value":"","regex":false}},{"data":"valor","name":"","searchable":true,"orderable":true,"search":{"value":"","regex":false}}]},{"name":"order","value":[{"column":0,"dir":"asc"}]},{"name":"start","value":0},{"name":"length","value":10},135 # {"name":"search","value":{"value":"","regex":false}}]136 full = json.loads(message)137 data = full['data']138 draw = data[0]139 columns = data[1]140 order = data[2]141 start = data[3]142 length = data[4]143 search = data[5]...
accessories.py
Source:accessories.py
...36 # pylint: disable=not-callable37 hass = args[0].hass38 nonlocal lastargs, remove_listener39 if remove_listener:40 remove_listener()41 lastargs = remove_listener = None42 lastargs = {'hass': hass, 'args': [*args]}43 remove_listener = track_point_in_utc_time(44 hass, call_later_listener,45 dt_util.utcnow() + timedelta(seconds=DEBOUNCE_TIMEOUT))46 logger.debug('%s: Start %s timeout', args[0].entity_id,47 func.__name__.replace('set_', ''))48 remove_listener = None49 lastargs = None50 name = getmodule(func).__name__51 logger = logging.getLogger(name)52 return wrapper53class HomeAccessory(Accessory):54 """Adapter class for Accessory."""...
_inputsensor.py
Source:_inputsensor.py
...11"""12class keyboardsensor_base(bee.worker):13 def set_add_listener(self, add_listener):14 self.add_listener = add_listener15 def set_remove_listener(self, remove_listener):16 self.remove_listener = remove_listener17 def place(self):18 libcontext.socket(("evin", ("input", "keyboard")), socket_flag())19 libcontext.socket(("evin", "add_listener"), socket_single_required(self.set_add_listener))20 libcontext.socket(("evin", "remove_listener"), socket_single_required(self.set_remove_listener))21 libcontext.plugin(("bee", "init"), plugin_single_required(self.enable))22 def enable(self):23 pass24 def disable(self):25 pass26""" 27class keyboardsensor_pulse(keyboardsensor_base):28 keycode = variable("str")29 parameter(keycode, "SPACE") 30 def __init__(self, keycode):31 self.pulses = []32 keyboardsensor_base.__init__(self, keycode)33 self.pressed = False34 def _pulse(self, channel, message):35 count = 136 if channel[1] == "keyreleased" and self.pressed == False: count = 237 if channel[1] == "keypressed" and self.pressed == True: count = 238 for n in range(count):39 for p in self.pulses: p() 40 self.pressed = not self.pressed41 def add_pulse(self, pulse):42 assert callable(pulse)43 self.pulses.append(pulse)44 def place(self):45 libcontext.socket("eventhandler", socket_single_required(self.set_eventhandler)) 46 libcontext.socket(("input", "keyboard", "keyreleased"), socket_flag())47 if self.keycode not in asciilist:48 libcontext.socket(("input", "keyboard", "extended"), socket_flag())49 libcontext.socket("pulse", socket_supplier(self.add_pulse))50 keyboardsensor_base.place(self) 51 def enable(self):52 self.eventhandler.new_processor(("keyboard", "keypressed", self.keycode), self._pulse)53 self.eventhandler.new_processor(("keyboard", "keyreleased", self.keycode), self._pulse)54 def disable(self):55 self.eventhandler.remove_processor(("keyboard", "keypressed", self.keycode), self._pulse)56 self.eventhandler.remove_processor(("keyboard", "keyreleased", self.keycode), self._pulse)57"""58class mousesensor_base(bee.worker):59 def set_add_listener(self, add_listener):60 self.add_listener = add_listener61 def set_remove_listener(self, remove_listener):62 self.remove_listener = remove_listener63 def place(self):64 libcontext.socket(("evin", ("input", "mouse")), socket_flag())65 libcontext.socket(("evin", "add_listener"), socket_single_required(self.set_add_listener))66 libcontext.socket(("evin", "remove_listener"), socket_single_required(self.set_remove_listener))67 libcontext.plugin(("bee", "init"), plugin_single_required(self.enable))68 def enable(self):69 pass70 def disable(self):71 pass72 def __del__(self):73 self.disable()74"""75class mousebuttonsensor_pulse(mousebuttonsensor_base):...
threaded_event_emitter.py
Source:threaded_event_emitter.py
...37 return self.pool.apply_async(f, args, kwargs)38 wrapped = super().once(event, wrapped)39 self.wrappers[event].append((f, wrapped))40 return wrapped41 def remove_listener(self, event_name, func):42 """ Wrap the remove to translate from function to wrapped43 function.44 """45 for w in self.wrappers[event_name]:46 if w[0] == func:47 self.wrappers[event_name].remove(w)48 return super().remove_listener(event_name, w[1])49 # if no wrapper exists try removing the function...
LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Get 100 minutes of automation test minutes FREE!!