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...
How to take a screenshot of a reddit post using playwright?
How can I select a button contained within an iFrame in Playwright (python) by index?
how to locate the elements which only appear when mouse move to the specific spot in playwright?
How to try clicking on elements in Playwright without try except block?
Error message saying "AttributeError: 'NoneType' object has no attribute 'text'"
Why does Playwright not change url once button is clicked on Uber Eats?
Exit an async with in Python
sync_playwright().start() seems to hang
How to handle comfirm alert in Playwright python
Where can I configure the browser behavior of Playwright in VSCode?
This should get you started, because you don't really need to do fancy image trickery.
All reddit posts have data-testid=post-container
selectors which let you get the top level post. All comments have a .Comment
class you can use, so you can enumerate all the comments on any given reddit post and capture them separately if that's what you want.
Some major caveats of this code:
It does not respect comment nesting - I only took a brief glance at Reddit's page source but they don't seem to have easy ways of getting comment chains.
Reddit could change their selector names/CSS classes any time, and they do have some inconsistencies in how they seem to name things anyway.
You specify you want to capture AskReddit threads, which can be quite large, so you could easily end up with, say, 1000+ comment screenshots on popular threads, or even 10K+ screenshots if you just let it run with abandon. I set a limit of 10, modify code as needed.
I wrote this as a demonstration, I have no idea how it will perform on giant posts or such, but hopefully it shows you how to actually pull the information you want. If you need to do more work, you will likely have to inspect Reddit page layouts with your browser's developer tools.
#!/usr/bin/env python3
import asyncio
from playwright.async_api import async_playwright, Browser
MAX_COMMENT_LIMIT = 10
async def capture(browser: Browser, url: str) -> None:
page = await browser.new_page()
await page.goto(url)
posts = page.locator("data-testid=post-container")
# reddit loads differently sometimes, so you need to
# find the first post on the page
await posts.first.screenshot(path="post.png")
# this does not respect comment-nesting
comments = page.locator(".Comment")
count = await comments.count()
print(f"Found {count} comments")
for i in range(count):
await comments.nth(i).screenshot(path=f"comment-{i}.png")
if i+1 == MAX_COMMENT_LIMIT:
print(f"Screened {MAX_COMMENT_LIMIT} images, stopping.")
break
async def main() -> None:
reddit_url = "https://www.reddit.com/r/cpp/comments/w9xkmf/c_for_scientific_programming/"
async with async_playwright() as p:
browser = await p.chromium.launch()
await capture(browser, reddit_url)
await browser.close()
asyncio.run(main())
The thread I picked generated some images like this...
Top-level post text
A comment
Check out the latest blogs from LambdaTest on this topic:
It’s essential to test all components of your website to see if they work as expected. Playwright’s end to end testing capability helps you achieve this easily. However, if you’re comfortable using Python, you can pair it with the Playwright testing framework to run Python end to end testing on your website.
Open MCT is a next-generation mission control framework for data visualization on desktop and mobile devices. It was created at NASA’s Ames Research Center, and NASA uses it to analyze spacecraft mission data.
Selenium is still the most influential and well-developed framework for web automation testing. Being one of the best automation frameworks with constantly evolving features, it is poised to lead the industry in all aspects as compared to other trending frameworks like Cypress, Puppeteer, PlayWright, etc. Furthermore, using Selenium gives you the flexibility to use different programming languages like C#, Ruby, Perl, Java, Python, etc., and also accommodate different operating systems and web browsers for Selenium automation testing.
One of the biggest problems I’ve faced when building a test suite is not the writing of the tests but the execution. How can I execute 100s or 1000s of tests in parallel?If I try that on my local machine, it would probably catch fire – so we need a remote environment to send these to.
To decide what automation technology to use, we brought together Joe Colantonio, Founder of TestGuild, Sneha. V, Director of Quality Engineering, Everfi, and Carlos Kidman, Director of Engineering, Stealth Startup. The panel discussion was hosted by Mudit Singh, Marketing head at LambdaTest. Mudit decided to take a step backwards and let the panel discussion happen.
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!!