Best Python code snippet using playwright-python
_page.py
Source:_page.py
...270 else:271 asyncio.create_task(dialog.dismiss())272 def _add_event_handler(self, event: str, k: Any, v: Any) -> None:273 if event == Page.Events.FileChooser and len(self.listeners(event)) == 0:274 self._channel.send_no_reply(275 "setFileChooserInterceptedNoReply", {"intercepted": True}276 )277 super()._add_event_handler(event, k, v)278 def remove_listener(self, event: str, f: Any) -> None:279 super().remove_listener(event, f)280 if event == Page.Events.FileChooser and len(self.listeners(event)) == 0:281 self._channel.send_no_reply(282 "setFileChooserInterceptedNoReply", {"intercepted": False}283 )284 @property285 def context(self) -> "BrowserContext":286 return self._browser_context287 async def opener(self) -> Optional["Page"]:288 return from_nullable_channel(await self._channel.send("opener"))289 @property290 def main_frame(self) -> Frame:291 return self._main_frame292 def frame(self, name: str = None, url: URLMatch = None) -> Optional[Frame]:293 matcher = URLMatcher(url) if url else None294 for frame in self._frames:295 if name and frame.name == name:296 return frame297 if url and matcher and matcher.matches(frame.url):298 return frame299 return None300 @property301 def frames(self) -> List[Frame]:302 return self._frames.copy()303 def set_default_navigation_timeout(self, timeout: float) -> None:304 self._timeout_settings.set_navigation_timeout(timeout)305 self._channel.send_no_reply(306 "setDefaultNavigationTimeoutNoReply", dict(timeout=timeout)307 )308 def set_default_timeout(self, timeout: float) -> None:309 self._timeout_settings.set_timeout(timeout)310 self._channel.send_no_reply("setDefaultTimeoutNoReply", dict(timeout=timeout))311 async def query_selector(self, selector: str) -> Optional[ElementHandle]:312 return await self._main_frame.query_selector(selector)313 async def query_selector_all(self, selector: str) -> List[ElementHandle]:314 return await self._main_frame.query_selector_all(selector)315 async def wait_for_selector(316 self,317 selector: str,318 timeout: float = None,319 state: Literal["attached", "detached", "hidden", "visible"] = None,320 ) -> Optional[ElementHandle]:321 return await self._main_frame.wait_for_selector(**locals_to_params(locals()))322 async def is_checked(self, selector: str, timeout: float = None) -> bool:323 return await self._main_frame.is_checked(**locals_to_params(locals()))324 async def is_disabled(self, selector: str, timeout: float = None) -> bool:...
optomechanics.py
Source:optomechanics.py
...214 # Default to max acceleration215 time.sleep(0.1)216 self.set_acceleration_to_max()217218 def send_no_reply(self, *args):219 """Sends a command to this device, without expecting a response. """220 if len(args) == 1 and isinstance(args[0], zaber.BinaryCommand):221 command = args[0]222 elif len(args) < 4:223 command = zaber.BinaryCommand(self.number, *args)224225 command.device_number = self.number226 self.port.write(command)227228 def _distance_microsteps_from_mm(self, distance_mm):229 return int(distance_mm / self.microstep_size_mm)230231 def _speed_microsteps_from_mm_s(self, speed_mm_s):232 return int(speed_mm_s / (9.375 * self.microstep_size_mm))233234 def home(self):235 self.send_no_reply(1)236237 def get_position(self):238 """ Return position (mm). """239 return self.send(60).data240241 def move_by(self, distance):242 """ Move relative distance (mm). """243 self.send_no_reply(21, self._distance_microsteps_from_mm(distance))244 # self.move_rel(self._distance_microsteps_from_mm(distance))245246 def move_to(self, position):247 """ Move to an absolute position (only works in the stage has been homed first so it has a zero position known). """248 self.send_no_reply(20, self._distance_microsteps_from_mm(position))249 # self.move_abs(self._distance_microsteps_from_mm(position))250251 def move_continuously(self, speed):252 """ Move continuously (unless self.stop executed / buffer is hit) at speed (mm/s). """253 self.send_no_reply(22, self._speed_microsteps_from_mm_s(speed))254 # self.move_vel(self._speed_microsteps_from_mm_s(speed))255256 def set_default_move_speed(self, speed):257 """ Set default target speed, for move commands (mm/s).258 Permissible range is: 0.0009302 mm/s to 4 m /s259 """260 self.send_no_reply(42, self._speed_microsteps_from_mm_s(speed))261262 def set_acceleration_to_max(self):263 """ From manual: If acceleration is set to 0, it is as if acceleration is set to (512*R-1).264 Effectively acceleration is turned off and the device will start moving at the target speed immediately. """
...
_browser_context.py
Source:_browser_context.py
...84 return85 asyncio.create_task(binding_call.call(func))86 def set_default_navigation_timeout(self, timeout: float) -> None:87 self._timeout_settings.set_navigation_timeout(timeout)88 self._channel.send_no_reply(89 "setDefaultNavigationTimeoutNoReply", dict(timeout=timeout)90 )91 def set_default_timeout(self, timeout: float) -> None:92 self._timeout_settings.set_timeout(timeout)93 self._channel.send_no_reply("setDefaultTimeoutNoReply", dict(timeout=timeout))94 @property95 def pages(self) -> List[Page]:96 return self._pages.copy()97 @property98 def browser(self) -> Optional["Browser"]:99 return self._browser100 async def new_page(self) -> Page:101 if self._owner_page:102 raise Error("Please use browser.new_context()")103 return from_channel(await self._channel.send("newPage"))104 async def cookies(self, urls: Union[str, List[str]] = None) -> List[Cookie]:105 if urls is None:106 urls = []107 if not isinstance(urls, list):...
Controller.py
Source:Controller.py
...79 packet[Raft].sourceID = 0x080 packet[Raft].destinationID = self.nodeID81 packet[IP].srcAddr = args.source82 #packet[Raft].show()83 send_no_reply(packet)84 if (self.counter_new_request % 50) == 0 and self.nodeID == 4: # emulating failure85 import os86 print('emulating failure')87 os._exit(0)88 if packet[Raft].messageType == COMMANDS['RejectNewRequest']:89 self.counter_rejected_requests += 190 print('New request rejected; total {}'.format(self.counter_rejected_requests))91 #print('state : {}'.format(self.status))92 if packet[Raft].messageType == COMMANDS['CommitValue']:93 print('Transaction complete. time: {}'.format(time.time()))94 if packet[Raft].messageType == COMMANDS['RetrieveLog']:95 print('Retrieved Value: {} at Index: {}'.format(packet[Raft].data, packet[Raft].logIndex))96 #logger.debug(packet.sprintf())97 #packet[Raft].show()98 def raft_timeout(self):99 return random.randrange(RANDOM_TIMEOUT['min'], RANDOM_TIMEOUT['max']) / 1000100 def reset_timeout(self):101 self.election_time = time.time() + self.raft_timeout()102 def init_timeout(self):103 self.reset_timeout()104 # safety guarantee, timeout thread may expire after election105 if self.timeout_thread and self.timeout_thread.is_alive():106 return107 self.timeout_thread = threading.Thread(target=self.timeout_loop)108 self.timeout_thread.start()109 def heartbeats_loop(self):110 rate = RAFT_HEARTBEAT_RATE / 1000111 while True: # todo find a way to block this thread in a more clever way112 if self.status == STATUSES['leader']:113 #print('sending StartHeartbeat')114 self.send_heartbeat_request()115 time.sleep(rate)116 else:117 time.sleep(rate)118 return119 # the timeout function120 def timeout_loop(self):121 # only stop timeout thread when winning the election122 while self.status != STATUSES['leader']:123 delta = self.election_time - time.time()124 if delta < 0:125 self.start_election()126 self.reset_timeout()127 else:128 time.sleep(delta)129 return130 def start_election(self):131 print('starting election')132 #logger.debug("{} starting election; status: {}, term:{}".format(self.controller_ip, self.status, self.term))133 self.term += 1134 start_election_message = Raft.raft_packet(135 sourceID=0x0,136 destinationID=self.nodeID,137 data=0x0,138 logIndex=self.logIndex,139 srcIP=args.source,140 dstIP=IP_MULTICAST_ADDRESS,141 currentTerm=self.term,142 messageType=COMMANDS['Timeout']143 )144 send_no_reply(start_election_message)145 def send_heartbeat_request(self):146 #print("Sending heartbeat request")147 #logger.debug("Starting HEARTBEATS")148 heartbeat = Raft.raft_packet(149 sourceID=0x0,150 destinationID=self.nodeID,151 data=0x0,152 logIndex=self.logIndex,153 srcIP=args.source,154 dstIP=IP_MULTICAST_ADDRESS,155 currentTerm=self.term,156 messageType=COMMANDS['StartHeartbeat']157 )158 send_no_reply(heartbeat)159# def main_handle_packet(packet): 160# packet[Raft].show()161def is_ingoing_raft_packet(_packet):162 if _packet.haslayer(IP):163 if not _packet[IP].proto == 'icmp':164 if _packet.haslayer(UDP):165 if _packet[UDP].dport == RAFT_PROTOCOL_DSTPORT:166 if _packet.haslayer(Raft):167 #return True168 if _packet[Raft].sourceID != 0x0:169 return True170 return False171if __name__ == '__main__':172 parser = argparse.ArgumentParser(description='Raft Packet Sender')...
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!!