Best Python code snippet using playwright-python
test_interception.py
Source:test_interception.py
...701 assert len(interceptedRequest) == 1702 assert interceptedRequest[0].headers["origin"] == server.PREFIX703async def test_request_fulfill_should_work_with_request_interception(page, server):704 requests = {}705 async def _handle_route(route: Route):706 requests[route.request.url.split("/").pop()] = route.request707 await route.continue_()708 await page.route("**/*", _handle_route)709 server.set_redirect("/rrredirect", "/frames/one-frame.html")710 await page.goto(server.PREFIX + "/rrredirect")711 assert requests["rrredirect"].is_navigation_request()712 assert requests["frame.html"].is_navigation_request()713 assert requests["script.js"].is_navigation_request() is False714 assert requests["style.css"].is_navigation_request() is False715async def test_Interception_should_work_with_request_interception(716 browser: Browser, https_server717):718 context = await browser.new_context(ignore_https_errors=True)719 page = await context.new_page()720 await page.route("**/*", lambda route: asyncio.ensure_future(route.continue_()))721 response = await page.goto(https_server.EMPTY_PAGE)722 assert response723 assert response.status == 200724 await context.close()725async def test_ignore_http_errors_service_worker_should_intercept_after_a_service_worker(726 page, server727):728 await page.goto(server.PREFIX + "/serviceworkers/fetchdummy/sw.html")729 await page.evaluate("() => window.activationPromise")730 # Sanity check.731 sw_response = await page.evaluate('() => fetchDummy("foo")')732 assert sw_response == "responseFromServiceWorker:foo"733 def _handle_route(route):734 asyncio.ensure_future(735 route.fulfill(736 status=200,737 content_type="text/css",738 body="responseFromInterception:" + route.request.url.split("/")[-1],739 )740 )741 await page.route("**/foo", _handle_route)742 # Page route is applied after service worker fetch event.743 sw_response2 = await page.evaluate('() => fetchDummy("foo")')744 assert sw_response2 == "responseFromServiceWorker:foo"745 # Page route is not applied to service worker initiated fetch.746 non_intercepted_response = await page.evaluate('() => fetchDummy("passthrough")')747 assert non_intercepted_response == "FAILURE: Not Found"
webhooks.py
Source:webhooks.py
...118 log.info("Received event with %s", event)119 self._validate_event(event)120 tags = set(event['tags'])121 if 'route' in tags:122 self._handle_route(event)123 return 'OK'124 elif {'swarm', 'deploy', 'done'} == tags:125 self._handle_swarm_deploy_done(event)126 return 'OK'127 elif {'scheduled', 'failed'} == tags:128 self._handle_scheduled_failed(event)129 return 'OK'130 else:131 log.info("Ignoring event")132 return 'IGNORED'133 def _handle_route(self, event):134 self._validate_event_route(event)135 swarm = event['title']136 app, sep, rest = swarm.partition('-')137 msg = "Routed {swarm}".format(**locals())138 self._broadcast(app, msg)139 def _handle_swarm_deploy_done(self, event):140 self._validate_event_swarm_deploy_done(event)141 title = event['title']142 match = self.SWARM_DEPLOY_DONE_RE.match(title)143 assert match is not None, 'Invalid title {}'.format(title)144 swarm = match.groups()[0]145 app, sep, rest = swarm.partition('-')146 msg = event['message']147 self._broadcast(app, msg)...
reader.py
Source:reader.py
1"""2Python module for reading QuickRoute files3QuickRoute data is embedded in exported JPEG files,4more specifically in its APP0 section starting with "QuickRoute" string.5All numeric values inside the data structure are little-endian6"""7from math import radians, cos, sin, asin, sqrt8from datetime import datetime, timedelta9import json10import logging11import struct12import sys13def haversine_distance(pos1, pos2, radius=6372000):14 """15 Reference http://stackoverflow.com/q/491334916 """17 lat1 = radians(pos1['lat'])18 lon1 = radians(pos1['lon'])19 lat2 = radians(pos2['lat'])20 lon2 = radians(pos2['lon'])21 dlat = lat2 - lat122 dlon = lon2 - lon123 a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**224 c = 2 * asin(sqrt(a))25 return radius * c26def format_dotnet_time(ntime):27 """28 Formats .NET style timestamp to datetime object29 Clears two most significant bits (timezone info?)30 """31 ntime &= ~(2**62 | 2**63)32 return datetime(1, 1, 1) + timedelta(microseconds=ntime/10)33def read_app_sections(stream):34 """35 Read APPn sections from JPEG file36 """37 marker = "init"38 while marker:39 marker = stream.read(2)40 if marker == '\xFF\xD8':41 # SOA, start of image. Read next chunk.42 continue43 if len(marker) < 2 or marker[0] != '\xFF':44 # somehow reading got out-of-sync45 #print "OOPS %r %r" % (marker, stream.tell())46 break47 if '\xE0' <= marker[1] <= '\xEF':48 # APPn, size includes the two bytes for len49 csize = struct.unpack('!H', stream.read(2))[0] - 250 cid = ord(marker[1]) - 0xE051 cdata = stream.read(csize)52 yield (cid, cdata)53 else:54 # Found first non-app section. so assuming real image data begins55 break56def read_quickroute_section(stream):57 """58 Read QuickRoute data section from JPEG file59 """60 for app, data in read_app_sections(stream):61 if app == 0 and data.startswith('QuickRoute'):62 return data[10:]63QR_TAGS = {64 1: 'Version',65 2: 'MapCornerPositions',66 3: 'ImageCornerPositions',67 4: 'MapLocationAndSizeInPixels',68 5: 'Sessions',69 6: 'Session',70 7: 'Route',71 8: 'Handles',72 9: 'ProjectionOrigin',73 10: 'Laps',74 11: 'SessionInfo',75 12: 'MapReadingInfo',76}77# inside Sessions -> Session -> Route78QR_ATTR_POSITION = 179QR_ATTR_TIME = 280QR_ATTR_HEARTRATE = 481QR_ATTR_ALTITUDE = 882# lap types83QR_LAP_TYPES = {84 0: 'start',85 1: 'lap',86 2: 'stop',87}88class QuickRouteData(dict):89 """90 """91 def __init__(self, jpeg=None, data=None):92 dict.__init__(self)93 if jpeg:94 with open(jpeg, "rb") as qrtfile:95 data = read_quickroute_section(qrtfile)96 if data:97 self.update(self.read(data))98 def read(self, data):99 """100 Construct internal data from binary data101 """102 ret = {}103 for key, value in self.read_data(data):104 logging.debug("%s: %.1024r", key, value)105 ret[key] = value106 return ret107 def read_data(self, data):108 """109 Reads (tag, datalen, data) blocks from the input string.110 This gets called recursively, since internal data parts111 use similar structure.112 """113 pos = 0114 while pos < len(data):115 tag, tlen = struct.unpack_from("<BI", data, pos)116 pos += 1+4117 tdata = data[pos:pos+tlen]118 pos += tlen119 tname = QR_TAGS.get(tag)120 logging.debug("tag: %s, tag name: %s, bytes: %s", tag, tname, tlen)121 func = getattr(self, '_handle_%s' % tname, None)122 if func:123 value = func(tdata)124 else:125 logging.warning("unhandled section %r %r %r", tag, tname, tlen)126 value = None127 yield tname, value128 def _handle_Version(self, data):129 """130 Read Version info131 - 4x unsigned char132 - join by dots133 """134 value = struct.unpack_from("<4B", data)135 value = ".".join(str(x) for x in value)136 return value137 def _handle_Sessions(self, data):138 """139 Reads Sessions structure:140 - number of sessions141 - tag/data pairs using generic read_data()142 """143 sessions = []144 scount = struct.unpack_from("<I", data)[0]145 logging.debug("reading %d sessions", scount)146 for key, value in self.read_data(data[4:]):147 if key != "Session":148 logging.warning("Found %r inside Sessions", key)149 continue150 sessions.append(value)151 assert len(sessions) == scount152 return sessions153 def _handle_Session(self, data):154 """155 Read (single) Session structure156 Utilizes generic read_data()157 """158 session = {}159 for key, value in self.read_data(data):160 session[key] = value161 return session162 def _handle_Laps(self, data):163 """164 Read Laps structure:165 - number of laps166 - multiple time/type pairs167 """168 laps = []169 lcount = struct.unpack_from("<I", data)[0]170 pos = 4171 for _ in range(lcount):172 ltime, ltype = struct.unpack_from("<QB", data, pos)173 pos += 9174 ltime = format_dotnet_time(ltime)175 ltype = QR_LAP_TYPES.get(ltype, ltype)176 laps.append({"time": ltime, "type": ltype})177 return laps178 def _handle_Route(self, data):179 """180 Read Route data structure181 Route can have multiple Segments182 Each segment has multiple waypoints with various attributes183 """184 route = []185 pos = 0186 attrs, extralen, segcount = struct.unpack_from("<HHI", data, pos)187 pos += 2+2+4188 logging.debug("route: attrs: %r, extralen: %r, segment count: %r", attrs, extralen, segcount)189 for i in range(segcount):190 segment = []191 wpcount = struct.unpack_from("<I", data, pos)[0]192 pos += 4193 logging.debug("segment: %d waypoints", wpcount)194 tstamp = None195 tfirst = None196 for j in range(wpcount):197 waypoint = {}198 if attrs & QR_ATTR_POSITION:199 coords = self._handle_coord(data[pos:pos+8])200 waypoint.update(coords)201 if segment:202 waypoint['distance'] = segment[-1]['distance'] + haversine_distance(segment[-1], coords)203 else:204 waypoint['distance'] = 0205 pos += 8206 if attrs & QR_ATTR_TIME:207 ttype = ord(data[pos])208 pos += 1209 if ttype == 0:210 # full date211 tstamp = format_dotnet_time(struct.unpack_from("<Q", data, pos)[0])212 pos += 8213 else:214 # diff in milisecs215 tstamp += timedelta(milliseconds=struct.unpack_from("<H", data, pos)[0])216 pos += 2217 if tfirst is None:218 tfirst = tstamp219 waypoint['time'] = tstamp220 waypoint['elapsed_time'] = tstamp - tfirst221 if attrs & QR_ATTR_HEARTRATE:222 waypoint['hr'] = struct.unpack_from("<B", data, pos)[0]223 pos += 1224 if attrs & QR_ATTR_ALTITUDE:225 waypoint['alt'] = struct.unpack_from("<H", data, pos)[0]226 pos += 2227 # extra bits for future proofing?228 pos += extralen229 logging.debug("waypoint: %r", waypoint)230 segment.append(waypoint)231 route.append(segment)232 return route233 def _handle_corners(self, data):234 """235 Read four coordinates for SW, NW, NE, SE corners236 """237 return {238 "SW": self._handle_coord(data[0:8]),239 "NW": self._handle_coord(data[8:16]),240 "NE": self._handle_coord(data[16:24]),241 "SE": self._handle_coord(data[24:32]),242 }243 _handle_ImageCornerPositions = _handle_corners244 _handle_MapCornerPositions = _handle_corners245 def _handle_coord(self, data):246 """247 Read a lon/lat coordinate pair.248 Stored as two uint32 (kind of integer milliseconds)249 """250 lon, lat = struct.unpack_from("<2I", data)251 return {'lat': lat/3600000.0, 'lon': lon/3600000.0}252 _handle_ProjectionOrigin = _handle_coord253 def _handle_Handles(self, data):254 """255 Read Handles256 These are related to adjusting route data to the bitmap image.257 See QuickRoute source code for how to actually use this data.258 """259 handles = []260 pos = 0261 hcount = struct.unpack_from("<I", data, pos)[0]262 logging.debug("reading %d handles", hcount)263 pos += 4264 for i in range(hcount):265 handle = {}266 # 3x3 doubles, not sure what would be the best data structure here, so keeping a simple flattened list for now267 # last row (or column? :) is usually 0.0, 0.0, 1.0268 tmatrix = []269 for j in range(3):270 row = struct.unpack_from("<3d", data, pos)271 tmatrix.append(row)272 pos += 8+8+8273 handle['matrix'] = tmatrix274 # ParametereizedLocation (not sure what it's used for)275 # uint32 + double, first value is "segment index"276 handle['parameterized_location'] = struct.unpack_from("<Id", data, pos)277 pos += 4+8278 # possible sub-pixels, ummhh.. ok.279 handle['pixel_location'] = struct.unpack_from("<dd", data, pos)280 pos += 8+8281 # uint32, type, usually 0?282 handle['type'] = struct.unpack_from("<H", data, pos)[0]283 pos += 2284 handles.append(handle)285 return handles286 def _handle_MapLocationAndSizeInPixels(self, data):287 """288 Read MapLocationAndSizeInPixels structure:289 - x290 - y291 - width292 - height293 """294 return struct.unpack_from("<4H", data)295 def _handle_SessionInfo(self, data):296 """297 Read SessionInfo structure:298 - name (string)299 - club (sting)300 - id (uint32)301 - description (string)302 """303 info = {}304 pos = 0305 # string length + string306 slen = struct.unpack_from("<H", data, pos)[0]307 pos += 2308 info['name'] = data[pos:pos+slen]309 pos += slen310 # string length + string311 slen = struct.unpack_from("<H", data, pos)[0]312 pos += 2313 info['club'] = data[pos:pos+slen]314 pos += slen315 # uint32, usually 0316 info['id'] = struct.unpack_from("<I", data, pos)[0]317 pos += 4318 # string length + string319 slen = struct.unpack_from("<H", data, pos)[0]320 pos += 2321 info['description'] = data[pos:pos+slen]322 pos += slen323 # in case the structure contained more data324 if len(data) > pos:325 logging.warning("%d bytes remaining", len(data)-pos)...
server.py
Source:server.py
...100 self.send_header(key, value)101 self.end_headers()102 # sending \n if no data otherwise Postman panics for some reason103 self.wfile.write(r.data or "\n")104 def _handle_route(self, method):105 # process the given url and find API and route106 api_name, path = self._parse_request_info()107 # find the handler function registered by the API and route108 route, route_handler = self._find_route_handler(api_name, path, method)109 # prepare a request obj to be passed to registered handler110 request = self._prepare_request(route, path, method)111 # if handler has uiapp in arguments, run in host api context112 if handler.RequestHandler.wants_api_context(route_handler):113 # create a handler and event object in host114 req_hndlr, event_hndlr = \115 self._prepare_host_handler(request, route_handler)116 # do the handling work117 # if request has callback url, raise the event handler and return118 # the handler, when executed, will notify the callback url119 if request.callback_url:120 self._call_host_event(req_hndlr, event_hndlr)121 # acknowledge the request is accepted and return122 self._write_response(base.Response(status=base.NO_CONTENT))123 # otherwise run the handler and wait124 else:125 self._call_host_event_sync(req_hndlr, event_hndlr)126 # prepare response127 # grab response from req_hndlr.response128 # req_hndlr.response getter is thread-safe129 self._write_response(req_hndlr.response)130 # otherwise run here131 else:132 # now run the method, and gret response133 response = \134 handler.RequestHandler.run_handler(135 handler=route_handler,136 kwargs=handler.RequestHandler.prepare_handler_kwargs(137 request=request,138 handler=route_handler139 )140 )141 # prepare response142 self._write_response(response)143 def _process_request(self, method):144 # this method is wrapping the actual handler and is145 # catching all the excp146 try:147 self._handle_route(method=method)148 except Exception as ex:149 # get exception info150 sys.exc_type, sys.exc_value, sys.exc_traceback = \151 sys.exc_info()152 # go back one frame to grab exception stack from handler153 # and grab traceback lines154 tb_report = ''.join(155 traceback.format_tb(sys.exc_traceback)[1:]156 )157 self._write_response(158 excp.ServerException(159 message=str(ex),160 exception_type=sys.exc_type,161 exception_traceback=tb_report...
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!!