Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use _handle_route method in Playwright Python

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...

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:

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.
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.
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.
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.
Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
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.
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.
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.
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.
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.
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.
Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.