How to use _launch_server method in Playwright Python

Best Python code snippet using playwright-python

android.py

Source:android.py Github

copy

Full Screen

...175 def device_data(self):176 return self.__data177 178 179 def _launch_server(self, command): 180 sockobj = socket.socket(socket.AF_INET, socket.SOCK_STREAM)181 abort_event = self.__abort_event182 abort_event.clear()183 self_ip = self.__widgets['ip_list'].current_data[0]184 try:185 port = 10000186 while True:187 try:188 sockobj.bind((self_ip, port))189 except socket.error:190 port += 1191 if port > 65535:192 raise socket.error193 else:194 break195 196 self_port = port197 with self.__lock:198 self.__ip_port = (self_ip, self_port)199 self.__password = password = random.randint(0, 65535)200 self.__iv = iv = get_random_bytes(16)201 self.__key = key = get_random_bytes(32)202 qr_string = json.dumps({203 "ip": self_ip, 204 "port": self_port, 205 "password": password, 206 "aes": {207 "iv": b64encode(iv).decode(),208 "key": b64encode(key).decode()},209 "command": command})210 image = qrcode.make(qr_string).resize((self.qr_size, self.qr_size)) 211 self.__qr_image = ImageTk.PhotoImage(image=image) 212 213 if self.__qr_id is None:214 self.__qr_id = self.__qr_canvas.canvas.create_image((0, 0), image=self.__qr_image, anchor='nw')215 else:216 self.__qr_canvas.canvas.itemconfig(self.__qr_id, image=self.__qr_image) 217 self.__data_book.select(self._qr_tab)218 except Exception as err: 219 sockobj.close()220 with self.__lock:221 self.__ip_port = None222 raise err223 224 self.__enable_transfer_widgets(False)225 self.__view_model.idle.set(False)226 227 # Launch the data transfer thread 228 @self.root_node.thread_manager.new_thread_do229 def transfer():230 def show_head(datainfo, addr, read):231 mark = '=' if read else '*'232 direction = 'From' if read else 'To'233 wtext = self.__scrolled_text234 wtext.append_text(f'''235{mark*60}236{direction} Device237Device Info: 238 Manufacturer: {datainfo["manufacturer"]}239 Model: {datainfo["model"]}240IP: {addr[0]}241{datetime.datetime.now().isoformat()}242{mark*60}''') 243 244 245 def clear_qr_image():246 try:247 self.__qr_canvas.canvas.delete(self.__qr_id)248 except tk.TclError:249 pass250 self.__qr_id = None 251 252 253 def generate_plugin_link(data, type_):254 wtext = self.__scrolled_text255 256 for plugin in _plugins[type_]:257 if not plugin.test_data(data):258 continue259 260 @wtext.create_link_tag261 def link_action(dumb, plugin=plugin):262 plugin.action(data)263 264 wtext.append_text(f'[{plugin.link_text}]', link_action)265 wtext.append_text(' ') 266 267 def recv_head(ih):268 exit_flag = ih.recv(1)269 if exit_flag != b'\x00':270 # WaveSyn can abort a misson by sending a zero to itself.271 return True, None, 0272 head_len = struct.unpack("!I", ih.recv(4))[0]273 head_json = ih.recv(head_len).decode("utf-8")274 head_obj = json.loads(head_json)275 if head_obj["password"] != self.__password:276 return True, None, 0277 info_len = head_obj["info_len"]278 encrypted_info = ih.recv(info_len)279 decrypted_info = _decrypt_text(encrypted_info, key=self.__key, iv=self.__iv)280 datainfo = json.loads(decrypted_info)281 return False, datainfo, head_obj["data_len"]282 283 284 def recv_data(ih):285 data_list = []286 while True:287 data = ih.recv(8192)288 if not data:289 break290 data_list.append(data)291 return b''.join(data_list)292 293 294 try:295 sockobj.listen(2)296 sockobj.settimeout(0.5)297 abort_flag = False298 while True:299 try:300 conn, addr = sockobj.accept()301 except socket.timeout:302 if abort_event.is_set():303 abort_flag = True304 break305 else:306 break307 308 if abort_flag:309 raise AbortException310 311 ih = InterruptHandler(conn, 0.5, lambda dumb: abort_event.is_set())312 # Always manipulate the GUI components in the main thread. 313 self.root_node.thread_manager.main_thread_do(block=False)(clear_qr_image)314 315 exit_flag, datainfo, datalen = recv_head(ih)316 if exit_flag:317 return318 319 if command['action'] == 'read':320 data = None321 # Loop receiving data322 if command['source'] != 'storage':323 # For file transfer mission,324 # if the file is large, recv_data will be memory consuming. 325 data = recv_data(ih)326 # End receiving data 327 328 if command['source'] in ('clipboard', 'location_sensor'):329 # Store received data330 text = _decrypt_text(data, key=self.__key, iv=self.__iv)331 self.__data = {332 "data": text,333 "type": "text"}334 335 if command['source'] == 'location_sensor':336 pos = json.loads(text) 337 text = f'latitude={pos["latitude"]}, longitude={pos["longitude"]}'338 339 @self.root_node.thread_manager.main_thread_do(block=False)340 def show_text():341 # Always manipulate the GUI components in the main thread!342 scrolled_text = self.__scrolled_text343 show_head(datainfo, addr=addr, read=True)344 scrolled_text.append_text(f'\n\n{text}\n\n\n')345 346 generate_plugin_link(data=text, type_='text')347 if command['source'] == 'location_sensor':348 generate_plugin_link(data=pos, type_='locationsensor')349 scrolled_text.append_text('\n'*3)350 scrolled_text.see('end')351 352 elif command['source'] == 'gallery':353 bio = BytesIO(data)354 img = Image.open(bio)355 @self.root_node.thread_manager.main_thread_do(block=False)356 def show_image():357 scrolled_text = self.__scrolled_text358 show_head(datainfo, addr=addr, read=True)359 scrolled_text.append_text('\n'*2)360 pil_frame = PILImageFrame(361 scrolled_text.text, 362 pil_image=img, 363 balloon=self.root_node.gui.balloon)364 scrolled_text.text.window_create('end', window=pil_frame)365 scrolled_text.append_text('\n'*4)366 scrolled_text.see('end')367 368 elif command['source'] == 'storage':369 filename = Path(datainfo["filename"])370 directory = Path(self.__save_file_dir)371 path = directory/​filename372 if path.exists():373 stem, ext = filename.stem, filename.suffix374 for k in range(1, 10000):375 filename = f'{stem}[{k}]{ext}'376 path = directory/​filename377 if not path.exists():378 break379 with TemporaryFile() as tf:380 recvcnt = 0381 self.root_node.thread_manager.main_thread_do(block=False)(lambda: self.__view_model.transfer_progress.set(0))382 while True:383 buf = ih.recv(65536)384 if not buf:385 self.root_node.thread_manager.main_thread_do(block=False)(lambda: self.__view_model.transfer_progress.set(0))386 break387 tf.write(buf)388 recvcnt += len(buf)389 self.root_node.thread_manager.main_thread_do(block=False)(lambda: self.__view_model.transfer_progress.set(int(recvcnt/​datalen*100)))390 391 tf.seek(0, 0)392 with path.open('wb') as f:393 buflen = 65536394 aes = AES.new(self.__key, AES.MODE_CBC, iv=self.__iv)395 last_buf = b""396 while True:397 buf = tf.read(buflen)398 recvcnt += buflen399 decrypted_buf = aes.decrypt(last_buf)400 if not buf:401 decrypted_buf = Padding.unpad(decrypted_buf, block_size=16)402 f.write(decrypted_buf)403 if not buf:404 break405 else:406 last_buf = buf407 @self.root_node.thread_manager.main_thread_do(block=False)408 def show_info():409 show_head(datainfo, addr=addr, read=True)410 scrolled_text = self.__scrolled_text411 scrolled_text.append_text(f'\n\n{str(path)}\n\n')412 generate_plugin_link(data=str(path), type_='file')413 scrolled_text.append_text('\n'*3)414 scrolled_text.see('end')415 416 417 @self.root_node.thread_manager.main_thread_do(block=False)418 def on_finish():419 # Always manipulate the GUI components in the main thread.420 self.__data_book.select(self._data_tab)421 if self.__on_finish:422 self.__on_finish(self.device_data['data'])423 self.__on_finish = None 424 425 elif command['action'] == 'write': 426 if command['target'] == 'clipboard':427 text = self.root_node.interfaces.os.clipboard.read()428 data = {'data':text, 'type':'text'}429 ih.send(json.dumps(data).encode('utf-8'))430 elif command['target'].startswith('dir:'):431 if command['source'] == 'clipboard:image':432 from PIL import ImageGrab433 image = ImageGrab.grabclipboard()434 if not image:435 raise TypeError('No image in clipboard.')436 bio = BytesIO()437 image.save(bio, format='png')438 ih.send(bio.getvalue())439 bio.close()440 elif command['source'].startswith('storage'):441 filename = Path(self.__send_filename)442 filelen = filename.stat().st_size443 sendcnt = 0444 buflen = 65536445 with open(filename, 'rb') as file_send:446 while True: 447 buf = file_send.read(buflen)448 if not buf:449 self.__view_model.transfer_progress.set(0)450 break451 ih.send(buf)452 sendcnt += len(buf)453 self.__view_model.transfer_progress.set(int(sendcnt/​filelen*100))454 @self.root_node.thread_manager.main_thread_do(block=False)455 def on_finish():456 self.__data_book.select(self._data_tab)457 scrolled_text = self.__scrolled_text458 show_head(datainfo, addr=addr, read=False)459 scrolled_text.append_text('\n'*3)460 scrolled_text.see('end')461 except AbortException:462 self.root_node.thread_manager.main_thread_do(block=False)(clear_qr_image)463 finally:464 sockobj.close()465 with self.__lock:466 self.__ip_port = None 467 @self.root_node.thread_manager.main_thread_do(block=False)468 def finished():469 self.__enable_transfer_widgets(True)470 self.__view_model.idle.set(True)471 472 473 @WaveSynScriptAPI474 def read_device_clipboard(self, on_finish):475 self._launch_server(command={'action':'read', 'source':'clipboard'})476 self.__on_finish = on_finish477 478 479 @WaveSynScriptAPI480 def pick_gallery_photo(self, on_finish):481 self._launch_server(command={'action':'read', 'source':'gallery'})482 self.__on_finish = on_finish483 484 485 @WaveSynScriptAPI 486 def get_device_file(self, savein, on_finish):487 self.__save_file_dir = self.root_node.gui.dialogs.constant_handler_ASK_DIRECTORY(savein)488 self._launch_server(command={'action':'read', 'source':'storage'})489 490 491 @WaveSynScriptAPI492 def read_device_location(self, on_finish):493 self._launch_server(command={'action':'read', 'source':'location_sensor'})494 self.__on_finish = on_finish495 496 497 @WaveSynScriptAPI 498 def write_device_clipboard(self):499 self._launch_server(command={'action':'write', 'source':'', 'target':'clipboard'}) 500 501 502 @WaveSynScriptAPI 503 def send_clipboard_image_to_device(self):504 self._launch_server(command={505 'action':'write', 506 'target':'dir:Download', 507 'source':'clipboard:image', 508 'name':f'clipboard_{int(datetime.datetime.now().timestamp())}.png'})509 510 511 @WaveSynScriptAPI512 def send_image_to_device(self, filename):513 filename = Path(self.root_node.gui.dialogs.constant_handler_ASK_OPEN_FILENAME(filename))514 self.__send_filename = filename515 self._launch_server(command={516 'action':'write',517 'target':'dir:Download',518 'source':'storage:image',519 'name':f'from_pc_{filename.name}'})520 521 522 @WaveSynScriptAPI523 def send_file_to_device(self, filename):524 filename = Path(self.root_node.gui.dialogs.constant_handler_ASK_OPEN_FILENAME(filename))525 self.__send_filename = filename526 self._launch_server(command={527 'action':'write',528 'target':'dir:Download',529 'source':'storage:file',530 'name':f'from_pc_{filename.name}'})531 532 533 @WaveSynScriptAPI534 def abort(self):535 self.__abort_event.set()536 537 538 def __on_abort(self):539 with code_printer():540 self.abort()...

Full Screen

Full Screen

test_load_balance.py

Source:test_load_balance.py Github

copy

Full Screen

...116 assert lb.ttask == 0117 with raises(BalancerError) as e:118 lb._test_init_limit('ttask', 11, 1, 10)119 assert "must be lesser then" in str(e)120def test_launch_server(umax_lb, mocker):121 mocked_add_task_server = mocker.patch.object(umax_lb, "_add_task_server")122 assert umax_lb.server_id_count == 0123 assert len(umax_lb.servers_in_use) == 0124 assert umax_lb.umax == 2125 umax_lb._launch_server(1)126 assert umax_lb.server_id_count == 1127 assert len(umax_lb.servers_in_use) == 1128 assert "S-1" in umax_lb.servers_in_use129 assert mocked_add_task_server.call_count == 1130def test_launch_server_more_tasks(umax_lb, mocker):131 mocked_add_task_server = mocker.patch.object(umax_lb, "_add_task_server")132 assert umax_lb.server_id_count == 0133 assert len(umax_lb.servers_in_use) == 0134 assert umax_lb.umax == 2135 umax_lb._launch_server(2)136 assert umax_lb.server_id_count == 1137 assert len(umax_lb.servers_in_use) == 1138 assert "S-1" in umax_lb.servers_in_use139 assert mocked_add_task_server.call_count == 2140def test_launch_server_too_much_tasks(umax_lb):141 assert umax_lb.server_id_count == 0142 assert len(umax_lb.servers_in_use) == 0143 assert umax_lb.umax == 2144 with raises(BalancerError) as e:145 umax_lb._launch_server(3)146 assert "Invalid number of tasks" in str(e)147def test_add_task_server(umax_lb):148 umax_lb._launch_server(1)149 server_name = "S-1"150 assert server_name in umax_lb.servers_in_use151 assert len(umax_lb.servers_in_use[server_name]["tasks"]) == 1152 umax_lb._add_task_server(server_name)153 assert len(umax_lb.servers_in_use[server_name]["tasks"]) == 2154def test_add_task_server_invalid_server_name(umax_lb):155 assert len(umax_lb.servers_in_use) == 0156 with raises(BalancerError) as e:157 umax_lb._add_task_server("do_not_exist")158 assert "not found" in str(e)159def test_add_task_server_server_limit_reached(umax_lb):160 umax_lb._launch_server(2)161 server_name = "S-1"162 assert server_name in umax_lb.servers_in_use163 assert len(umax_lb.servers_in_use[server_name]["tasks"]) == 2164 with raises(BalancerError) as e:165 umax_lb._add_task_server(server_name)166 assert "can't start new task" in str(e)167def test_find_server_for_task(umax_lb):168 umax_lb._launch_server(1)169 assert umax_lb._find_server_for_task() == 'S-1'170def test_find_server_for_task_no_server_available(umax_lb):171 umax_lb._launch_server(2)172 assert umax_lb._find_server_for_task() is None173def test_add_new_clients_max_tasks(umax_lb, mocker):174 mocker_launch_server = mocker.patch.object(umax_lb, '_launch_server')175 mocker_find_server_for_task = mocker.patch.object(umax_lb, '_find_server_for_task')176 mocker_add_task_server = mocker.patch.object(umax_lb, '_add_task_server')177 umax_lb._add_new_clients(2)178 assert mocker_launch_server.called_once()179 assert mocker_find_server_for_task.call_count == 0180 assert mocker_add_task_server.call_count == 0181def test_add_new_clients_min_tasks_twice(umax_lb, mocker):182 mocker_launch_server = mocker.patch.object(umax_lb, '_launch_server')183 mocker_find_server_for_task = mocker.patch.object(umax_lb, '_find_server_for_task')184 mocker_add_task_server = mocker.patch.object(umax_lb, '_add_task_server')185 umax_lb._add_new_clients(1)186 assert mocker_launch_server.called_once_with(1)187 assert mocker_find_server_for_task.called_once()188 umax_lb._add_new_clients(1)189 assert mocker_add_task_server.called_once()190 assert mocker_find_server_for_task.call_count == 2191def test_remove_server(umax_lb):192 umax_lb._launch_server(1)193 server_name = "S-1"194 del umax_lb.servers_in_use[server_name]["tasks"]["T-1"]195 assert len(umax_lb.servers_in_use) == 1196 umax_lb._remove_server(server_name)197 assert len(umax_lb.servers_in_use) == 0198def test_remove_server_running_tasks(umax_lb):199 umax_lb._launch_server(1)200 with raises(BalancerError) as e:201 umax_lb._remove_server("S-1")202 assert "still has tasks running" in str(e)203def test_remove_server_unknown(umax_lb):204 with raises(BalancerError) as e:205 umax_lb._remove_server("S-1")206 assert "not found" in str(e)207def test_remove_task_server(umax_lb):208 umax_lb._launch_server(2)209 server_name = "S-1"210 assert len(umax_lb.servers_in_use[server_name]["tasks"]) == 2211 umax_lb._remove_task_server("T-1", server_name)212 assert len(umax_lb.servers_in_use[server_name]["tasks"]) == 1213def test_remove_task_server_invalid_server_name(umax_lb):214 umax_lb._launch_server(2)215 assert len(umax_lb.servers_in_use["S-1"]["tasks"]) == 2216 fake_server_name = "S-2"217 with raises(BalancerError) as e:218 umax_lb._remove_task_server("T-1", fake_server_name)219 assert f"Server {fake_server_name} not found" in str(e)220def test_remove_task_server_invalid_task_name(umax_lb):221 umax_lb._launch_server(2)222 server_name = "S-1"223 assert len(umax_lb.servers_in_use[server_name]["tasks"]) == 2224 fake_task_name = "S-2"225 with raises(BalancerError) as e:226 umax_lb._remove_task_server(fake_task_name, server_name)227 assert f"Task {fake_task_name} not found in server" in str(e)228def test_run_tick(umax_lb):229 umax_lb._add_new_clients(4)230 assert umax_lb.tick_servers_count == 0231 assert len(umax_lb.servers_in_use) == 2232 server1_name = "S-1"233 assert len(umax_lb.servers_in_use[server1_name]["tasks"]) == 2234 assert umax_lb.servers_in_use[server1_name]["tasks"]["T-1"] == umax_lb.ttask235 assert umax_lb.servers_in_use[server1_name]["tasks"]["T-2"] == umax_lb.ttask...

Full Screen

Full Screen

test_usage.py

Source:test_usage.py Github

copy

Full Screen

...20 def setUp(self):21 super().setUp()22 self.server = None23 self.context = None24 def _launch_server(self, **kwargs):25 self.server = volatildap.LdapServer(**kwargs)26 self.server.start()27 self.context = {28 'dirname': self.server._tempdir.name,29 'pid': self.server._process.pid,30 }31 def tearDown(self):32 if self.server is not None:33 self.server.stop()34 self.assertServerStopped(self.context)35 super().tearDown()36 def assertServerStopped(self, context, max_delay=5):37 now = time.time()38 # Allow some time for proper shutdown39 while time.time() < now + max_delay and os.path.exists(context['dirname']):40 time.sleep(0.2)41 self.assertFalse(os.path.exists(context['dirname']))42 # Check that the process is no longer running.43 # We cannot rely solely on "the pid is no longer running", as it may44 # have been reused by the operating system.45 # If a process by that pid still exists, we'll check that we aren't its parent.46 try:47 stats = psutil.Process(context['pid'])48 ppid = stats.ppid()49 except (psutil.NoSuchProcess, psutil.AccessDenied):50 # Process has died /​ is not in our context: all is fine.51 return52 self.assertNotEqual(ppid, os.getpid())53class ReadWriteTests(LdapServerTestCase):54 data = {55 'ou=test': {56 'objectClass': ['organizationalUnit'],57 'ou': ['test'],58 },59 }60 def test_initial_data(self):61 self._launch_server(initial_data=self.data)62 entry = self.server.get('ou=test,dc=example,dc=org')63 self.assertEqual({64 'objectClass': [b'organizationalUnit'],65 'ou': [b'test'],66 }, entry)67 def test_post_start_add(self):68 self._launch_server()69 self.server.add(self.data)70 entry = self.server.get('ou=test,dc=example,dc=org')71 self.assertEqual({72 'objectClass': [b'organizationalUnit'],73 'ou': [b'test'],74 }, entry)75 def test_get_missing_entry(self):76 self._launch_server()77 with self.assertRaises(KeyError):78 self.server.get('ou=test,dc=example,dc=org')79 def test_clear_data(self):80 self._launch_server()81 self.server.add(self.data)82 self.assertIsNotNone(self.server.get('ou=test,dc=example,dc=org'))83 self.server.start() # Actually a restart84 # Custom data has been removed85 with self.assertRaises(KeyError):86 self.server.get('ou=test,dc=example,dc=org')87 # Core data is still there88 self.assertIsNotNone(self.server.get('dc=example,dc=org'))89class ResetTests(LdapServerTestCase):90 data = {91 'ou=test': {92 'objectClass': ['organizationalUnit'],93 'ou': ['test'],94 },95 }96 def test_cleanup(self):97 self._launch_server(initial_data=self.data)98 extra = {99 'ou=subarea,ou=test': {100 'objectClass': ['organizationalUnit'],101 },102 }103 self.server.add(extra)104 entry = self.server.get('ou=subarea,ou=test,dc=example,dc=org')105 self.server.reset()106 # Extra data should have been removed107 self.assertRaises(KeyError, self.server.get, 'ou=subarea,ou=test,dc=example,dc=org')108 # Initial data should still be here109 entry = self.server.get('ou=test,dc=example,dc=org')110 self.assertEqual({111 'objectClass': [b'organizationalUnit'],112 'ou': [b'test'],113 }, entry)114class TLSTests(LdapServerTestCase):115 def test_connection(self):116 self._launch_server(tls_config=volatildap.LOCALHOST_TLS_CONFIG)117 self.assertEqual(self.server.uri[:8], 'ldaps:/​/​')118 entry = self.server.get('dc=example,dc=org')119 self.assertEqual([b'example'], entry['dc'])120class AutoCleanupTests(LdapServerTestCase):121 def test_stop(self):122 """Deleting the LdapServer object causes its cleanup."""123 self._launch_server()124class ControlTests(LdapServerTestCase):125 def setUp(self):126 super().setUp()127 self.proxy = None128 def tearDown(self):129 if self.proxy:130 self.proxy.stop()131 super().tearDown()132 def _launch_server(self, **kwargs):133 control_port = find_available_port()134 super()._launch_server(135 control_address=('localhost', control_port),136 **kwargs,137 )138 self.proxy = volatildap.ProxyServer('http:/​/​localhost:%d/​' % control_port)139 def test_launch_control(self):140 self._launch_server()141 self.assertEqual(142 self.server.uri,143 self.proxy.uri,144 )145 self.assertEqual('dc=example,dc=org', self.proxy.suffix)146 self.assertEqual(self.server.rootdn, self.proxy.rootdn)147 self.assertEqual(self.server.rootpw, self.proxy.rootpw)148 def test_tls(self):149 """The server CA should be available through the proxy."""150 self._launch_server(tls_config=volatildap.LOCALHOST_TLS_CONFIG)151 self.assertEqual(self.proxy.uri[:8], 'ldaps:/​/​')152 self.assertIsNotNone(self.proxy.tls_config)153 self.assertIsNotNone(self.proxy.tls_config.root)154 def test_get(self):155 initial = {'ou=people': {156 'objectClass': ['organizationalUnit'],157 'ou': ['people'],158 }}159 self._launch_server(160 initial_data=initial,161 )162 entry = self.proxy.get('ou=people')163 self.assertEqual(164 {165 'objectClass': [b'organizationalUnit'],166 'ou': [b'people'],167 },168 entry,169 )170 def test_add(self):171 self._launch_server()172 data = {'ou=people': {173 'objectClass': ['organizationalUnit'],174 'ou': ['people'],175 }}176 self.proxy.add(data)177 entry = self.proxy.get('ou=people')178 self.assertEqual(179 {180 'objectClass': [b'organizationalUnit'],181 'ou': [b'people'],182 },183 entry,184 )185 def test_reset(self):186 self._launch_server()187 data = {'ou=people': {188 'objectClass': ['organizationalUnit'],189 'ou': ['people'],190 }}191 self.proxy.add(data)192 # Ensure the data is visible193 self.proxy.get('ou=people')194 self.proxy.reset()195 with self.assertRaises(KeyError):...

Full Screen

Full Screen

tf_two_machines_local.py

Source:tf_two_machines_local.py Github

copy

Full Screen

...38parser.add_argument("--receiver-ip", default='127.0.0.1')39args = parser.parse_args()40cluster_spec = {'chief': [args.sender_ip + ':32300'],41 'receiver': [args.receiver_ip + ':32301']}42def _launch_server(role):43 os.environ['TF_CONFIG'] = json.dumps(44 {'cluster': cluster_spec,45 'task': {'type': role, 'index': 0}})46 config = tf.estimator.RunConfig()47 return tf.train.Server(config.cluster_spec,48 job_name=config.task_type,49 task_index=config.task_id)50def run_launcher():51 import ncluster52 ncluster.util.assert_script_in_current_directory()53 54 if args.aws:55 ncluster.set_backend('aws')56 # use 4GB instance, 0.5GB not enough57 worker = ncluster.make_task(args.name, image_name=args.image,58 instance_type='t3.medium')59 worker.upload(__file__)60 worker.upload('util.py')61 # kill python just for when tmux session reuse is on62 if not ncluster.running_locally():63 # on AWS probably running in conda DLAMI, switch into TF-enabled env64 worker._run_raw('killall python', ignore_errors=True)65 worker.run('source activate tensorflow_p36')66 ip_config = f'--sender-ip={worker.ip} --receiver-ip={worker.ip}'67 worker.run(f'python {__file__} --role=receiver {ip_config}',68 non_blocking=True)69 worker.switch_window(1) # run in new tmux window70 if not ncluster.running_locally():71 worker.run('source activate tensorflow_p36')72 worker.run(73 f'python {__file__} --role=sender {ip_config} --iters={args.iters} --size-mb={args.size_mb} --shards={args.shards}')74 print(worker.read('out'))75def run_receiver():76 server = _launch_server('receiver')77 time.sleep(365 * 24 * 3600)78 del server79def run_sender():80 param_size = 250 * 1000 * args.size_mb /​/​ args.shards # 1MB is 250k integers81 log = util.FileLogger('out')82 grads_array = []83 with tf.device('/​job:chief/​task:0'):84 # grads = tf.fill([param_size], 1.)85 for i in range(args.shards):86 grads = tf.Variable(tf.ones([param_size]))87 grads_array.append(grads)88 params_array = []89 add_op_array = []90 with tf.device('/​job:receiver/​task:0'):91 for i in range(args.shards):92 params = tf.Variable(tf.ones([param_size]))93 add_op = params.assign(grads_array[i]).op94 params_array.append(params)95 add_op_array.append(add_op)96 add_op = tf.group(*add_op_array)97 98 server = _launch_server('chief')99 sess = tf.Session(server.target)100 sess.run(tf.global_variables_initializer())101 # except Exception as e:102 # # sometimes .run fails with .UnavailableError: OS Error103 # log(f"initialization failed with {e}, retrying in 1 second")104 # time.sleep(1)105 time_list = []106 for i in range(args.iters):107 start_time = time.perf_counter()108 sess.run(add_op)109 elapsed_time_ms = (time.perf_counter() - start_time) * 1000110 time_list.append(elapsed_time_ms)111 rate = args.size_mb /​ (elapsed_time_ms /​ 1000)112 log('%03d/​%d sent %d MBs in %.1f ms: %.2f MB/​second' % (...

Full Screen

Full Screen

Playwright tutorial

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.

Chapters:

  1. What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
  2. How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
  3. Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
  4. What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
  5. Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
  6. Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
  7. Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
  8. Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
  9. Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
  10. Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
  11. Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
  12. Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.

Run Playwright Python automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful