Best Python code snippet using playwright-python
test_reader_text.py
Source: test_reader_text.py
...1105 event_pairs=start + mid + end,1106 )1107 if not is_delegate:1108 break1109def _expect_event(expected_event, data, events, delimiter):1110 """Generates event pairs for a stream that ends in an expected event (or exception), given the text and the output1111 events preceding the expected event.1112 """1113 events += (expected_event,)1114 outputs = events[1:]1115 event_pairs = [(e_read(data + delimiter), events[0])] + list(zip([NEXT] * len(outputs), outputs))1116 return event_pairs1117@coroutine1118def _basic_params(event_func, desc, delimiter, data_event_pairs, is_delegate=False, top_level=True):1119 """Generates parameters from a sequence whose first element is the raw data and the following1120 elements are the expected output events.1121 """1122 while True:1123 yield...
_page.py
Source: _page.py
...809 event: str,810 predicate: Callable = None,811 timeout: float = None,812 ) -> EventContextManagerImpl:813 return self._expect_event(814 event, predicate, timeout, f'waiting for event "{event}"'815 )816 def _expect_event(817 self,818 event: str,819 predicate: Callable = None,820 timeout: float = None,821 log_line: str = None,822 ) -> EventContextManagerImpl:823 if timeout is None:824 timeout = self._timeout_settings.timeout()825 wait_helper = WaitHelper(self, f"page.expect_event({event})")826 wait_helper.reject_on_timeout(827 timeout, f'Timeout {timeout}ms exceeded while waiting for event "{event}"'828 )829 if log_line:830 wait_helper.log(log_line)831 if event != Page.Events.Crash:832 wait_helper.reject_on_event(self, Page.Events.Crash, Error("Page crashed"))833 if event != Page.Events.Close:834 wait_helper.reject_on_event(self, Page.Events.Close, Error("Page closed"))835 wait_helper.wait_for_event(self, event, predicate)836 return EventContextManagerImpl(wait_helper.result())837 def expect_console_message(838 self,839 predicate: Callable[[ConsoleMessage], bool] = None,840 timeout: float = None,841 ) -> EventContextManagerImpl[ConsoleMessage]:842 return self.expect_event(Page.Events.Console, predicate, timeout)843 def expect_download(844 self,845 predicate: Callable[[Download], bool] = None,846 timeout: float = None,847 ) -> EventContextManagerImpl[Download]:848 return self.expect_event(Page.Events.Download, predicate, timeout)849 def expect_file_chooser(850 self,851 predicate: Callable[[FileChooser], bool] = None,852 timeout: float = None,853 ) -> EventContextManagerImpl[FileChooser]:854 return self.expect_event(Page.Events.FileChooser, predicate, timeout)855 def expect_navigation(856 self,857 url: URLMatch = None,858 wait_until: DocumentLoadState = None,859 timeout: float = None,860 ) -> EventContextManagerImpl[Response]:861 return self.main_frame.expect_navigation(url, wait_until, timeout)862 def expect_popup(863 self,864 predicate: Callable[["Page"], bool] = None,865 timeout: float = None,866 ) -> EventContextManagerImpl["Page"]:867 return self.expect_event(Page.Events.Popup, predicate, timeout)868 def expect_request(869 self,870 url_or_predicate: URLMatchRequest,871 timeout: float = None,872 ) -> EventContextManagerImpl[Request]:873 matcher = (874 None875 if callable(url_or_predicate)876 else URLMatcher(877 self._browser_context._options.get("baseURL"), url_or_predicate878 )879 )880 predicate = url_or_predicate if callable(url_or_predicate) else None881 def my_predicate(request: Request) -> bool:882 if matcher:883 return matcher.matches(request.url)884 if predicate:885 return predicate(request)886 return True887 trimmed_url = trim_url(url_or_predicate)888 log_line = f"waiting for request {trimmed_url}" if trimmed_url else None889 return self._expect_event(890 Page.Events.Request,891 predicate=my_predicate,892 timeout=timeout,893 log_line=log_line,894 )895 def expect_request_finished(896 self,897 predicate: Callable[["Request"], bool] = None,898 timeout: float = None,899 ) -> EventContextManagerImpl[Request]:900 return self.expect_event(901 Page.Events.RequestFinished, predicate=predicate, timeout=timeout902 )903 def expect_response(904 self,905 url_or_predicate: URLMatchResponse,906 timeout: float = None,907 ) -> EventContextManagerImpl[Response]:908 matcher = (909 None910 if callable(url_or_predicate)911 else URLMatcher(912 self._browser_context._options.get("baseURL"), url_or_predicate913 )914 )915 predicate = url_or_predicate if callable(url_or_predicate) else None916 def my_predicate(response: Response) -> bool:917 if matcher:918 return matcher.matches(response.url)919 if predicate:920 return predicate(response)921 return True922 trimmed_url = trim_url(url_or_predicate)923 log_line = f"waiting for response {trimmed_url}" if trimmed_url else None924 return self._expect_event(925 Page.Events.Response,926 predicate=my_predicate,927 timeout=timeout,928 log_line=log_line,929 )930 def expect_websocket(931 self,932 predicate: Callable[["WebSocket"], bool] = None,933 timeout: float = None,934 ) -> EventContextManagerImpl["WebSocket"]:935 return self.expect_event("websocket", predicate, timeout)936 def expect_worker(937 self,938 predicate: Callable[["Worker"], bool] = None,...
rlib_popen4.py
Source: rlib_popen4.py
1import subprocess2import thread3import threading4import StringIO5import re6from org.gvsig.tools import ToolsLocator7import rlib_base8reload(rlib_base)9RE_ERROR = re.compile("Error[^:]*:")10 11def debug(otype, *values):12 #return13 print "::%-7.7s:" % otype,14 for value in values:15 print value,16 print17class RValue:18 def __init__(self, value):19 self._value = value20 def __str__(self):21 s = self._value22 if s==None:23 return "None"24 if s.startswith("[1] "):25 s = s[4:].strip()26 if s[0]=='"':27 s = s[1:-1]28 return s29 def __repr__(self):30 return repr(str(self))31 def __int__(self):32 return int(str(self))33 def __long__(self):34 return long(str(self))35 def __float__(self):36 return float(str(self))37 def asint(self):38 return int(str(self))39 def asstr(self):40 return str(self)41 def asfloat(self):42 return float(str(self))43 def aslist(self):44 return None45class RFunction:46 def __init__(self,rengine,funcname):47 self._rengine = rengine48 self._funcname = funcname49 def __call__(self,*args):50 cmd = StringIO.StringIO()51 cmd.write(self._funcname)52 cmd.write("(")53 n = 054 for arg in args:55 if n>0 and n<len(args):56 cmd.write(", ")57 if isinstance(arg,str) or isinstance(arg,unicode) :58 arg = '"' + repr(arg)[1:-1] + '"'59 else:60 arg = str(arg)61 cmd.write(arg)62 n+=163 cmd.write(")")64 cmd = cmd.getvalue()65 return self._rengine.eval(cmd)66class ProcessRStderr(threading.Thread):67 def __init__(self, stderr, console_output):68 threading.Thread.__init__(self)69 self._console_output = console_output70 self._stderr = stderr71 self._last_error = None72 def run(self):73 line = StringIO.StringIO()74 while True:75 try:76 c = self._stderr.read(1)77 except ValueError:78 break79 if c == "\r":80 continue81 #print repr(c),82 line.write(c)83 if c == "":84 break85 if c == "\n":86 s = line.getvalue()87 if s!=None and RE_ERROR.match(s):88 debug("ERROR1",repr(s))89 self._last_error = s90 debug("STDERR",repr(s))91 self._console_output(s,1)92 line = StringIO.StringIO()93 debug("CLOSE","child_stderr")94 def resetLastError(self):95 self._last_error = None96 def getLastError(self):97 return self._last_error98class ProcessRStdout(threading.Thread):99 def __init__(self, stdout, console_output):100 threading.Thread.__init__(self)101 self._stdout = stdout102 self._console_output = console_output103 self._last_value = None104 self._expect_value = None105 self._expect_event = threading.Event()106 self._block_semaphore = threading.Semaphore()107 self.remove_values_from_output = False108 def begin(self):109 self._block_semaphore.acquire()110 def end(self):111 self._block_semaphore.release()112 def run(self):113 self.process_stdout()114 def process_stdout(self):115 line = StringIO.StringIO()116 while True:117 try:118 c = self._stdout.read(1)119 except ValueError:120 break121 #print repr(c),122 if c == "\r":123 continue124 self.begin()125 line.write(c)126 if c == "":127 break128 if self._expect_value != None:129 s = line.getvalue()130 n = s.find(self._expect_value)131 if n >= 0:132 l = len(self._expect_value)133 s1 = s[:n]134 s2 = s[n+l:]135 line = StringIO.StringIO(s2)136 line.seek(0,2)137 debug("NOTIFY", repr(self._expect_value))138 self._expect_value = None139 self._expect_event.set()140 if c == "\n":141 s = line.getvalue()142 if s!=None and s.startswith("[1] "):143 self._last_value = s144 if not self.remove_values_from_output:145 debug("STDOUT",repr(s))146 self._console_output(s,0)147 else:148 debug("STDOUT",repr(s))149 self._console_output(s,0)150 line = StringIO.StringIO()151 self.end()152 debug("CLOSE","stdout")153 self._expect_event.set()154 def wait(self):155 debug("WAIT", "")156 self._expect_event.wait()157 debug("WAITOK", "")158 def resetLastValue(self):159 self._last_value = None160 def getLastValue(self):161 if self._last_value == None:162 return None163 return RValue(self._last_value)164 def expect(self, value):165 debug("EXPECT", repr(value))166 self._expect_value = value167 self._expect_event.clear()168class REngine_popen(rlib_base.REngine_base):169 def __init__(self, consoleListener=None):170 rlib_base.REngine_base.__init__(self,consoleListener)171 self.__dict__["_prompt"] = None172 self.__dict__["_child"] = None173 self.__dict__["_processRStdout"] = None174 self.__dict__["_processRStderr"] = None175 self.run()176 def run(self):177 if self._child!=None:178 if self._child.poll()==None:179 raise RuntimeError("R is already running")180 self._prompt="@x#x@>"181 if self.getOperatingSystem() == "win": 182 cmd = (self.getRExecPathname(),183 "--ess", # Fuerza sesion interactiva en Windows.184 "--no-restore",185 "--no-save"186 )187 else:188 cmd = (self.getRExecPathname(),189 "--interactive",190 "--no-readline",191 "--no-restore",192 "--no-save"193 )194 195 # Si no forzamos la sesion interactiva, y R arranca en modo batch,196 # ante cualquier error, R se muere.197 debug("CMD",cmd)198 self._child = subprocess.Popen(cmd,199 stdin=subprocess.PIPE,200 stdout=subprocess.PIPE,201 stderr=subprocess.PIPE202 )203 self._processRStdout = ProcessRStdout(self._child.stdout, self.console_output)204 self._processRStderr = ProcessRStderr(self._child.stderr, self.console_output)205 self._processRStdout.start()206 self._processRStderr.start()207 #208 # wait initial prompt209 self._processRStdout.expect("> ")210 self._processRStdout.wait()211 #212 # Change prompt213 self.eval('options(prompt=paste("@x#x","@>",sep=""), continue=" ")')214 return True215 def eval(self,expr):216 if self._child==None or self._child.poll()!=None:217 raise RuntimeError("R process is dead")218 expr += "\n"219 #220 # Send comman and wait the echo221 self._processRStdout.begin()222 self._processRStdout.resetLastValue()223 self._processRStderr.resetLastError()224 self._processRStdout.expect(self._prompt)225 self._child.stdin.write(expr)226 self._child.stdin.flush()227 self._processRStdout.end()228 self._processRStdout.wait()229 if self._child.poll()!=None:230 raise RuntimeError("R process is dead")231 value = self._processRStdout.getLastValue()232 debug("EVAL",repr(expr))233 debug("VALUE",str(value))234 if self._processRStderr.getLastError() != None:235 debug("ERROR2",str(value))236 raise ValueError(self._processRStderr.getLastError())237 return value238 def get(self, name):239 theclass = str(self.getclass(name))240 if theclass == "function":241 return RFunction(self, name)242 else:243 return self.eval(name)244 def set(self, name, value):245 if isinstance(value,str) or isinstance(value,unicode) :246 value = '"' + repr(value)[1:-1] + '"'247 else:248 value = str(value)249 self.eval("%s <- %s" % (name,value))250 def getclass(self, name):251 x = self.eval('class(%s)' % name)252 if x == None:253 return None254 return str(x)255 def source(self, pathname):256 return self.eval('source("%s")' % self.getPathName(pathname))257 def end(self):258 self._child.stdin.close()259 self._child.stdout.close()260 self._child.stderr.close()261 def __getattr__(self,name):262 try:263 return self.get(name)264 except ValueError as e:265 raise AttributeError(str(e))266 def __setattr__(self, name, value):267 if name in self.__dict__.keys():268 self.__dict__[name] = value269 else:270 self.set(name, value)271 def isdead(self):272 return self._child.poll()!=None273def getREngine(consoleListener=None):...
Why are the values yielded by a pytest fixture and a function called directly different?
Refreshing cookie using Playwright for Python
Playwright: click on element within one/multiple elements using Python
How to quickly find out if an element exists in a page or not using playwright
Passing cmd line args to a class
Is there a way to list all video URLs of YouTube search results in Python?
Running PlayWright for python running on google cloud app engine
fixture 'page' not found - pytest playwright
Trying to select the option
Get element text behind shadow DOM element using Playwright
For the 1st form:
def get_playwright():
with sync_playwright() as playwright:
yield playwright
print(get_playwright()) # <generator object get_playwright at 0x108aac580>
That is expected because get_playwright
is a generator, which returns a generator iterator, which you have to call next(...)
on to get each yielded value from the iterator.
Consider a simpler, non-playwright example:
In [14]: def generate_nums():
...: for num in range(10):
...: yield num
...:
In [15]: nums = generate_nums()
In [16]: nums
Out[16]: <generator object generate_nums at 0x11115e6d0>
In [17]: next(nums)
Out[17]: 0
In [18]: next(nums)
Out[18]: 1
In [19]: next(nums)
Out[19]: 2
For more examples, see Understanding generators in Python.
Since your get_playwright
returns an iterator, you need to call next()
once to get the actual object:
from playwright.sync_api import sync_playwright
def get_playwright():
with sync_playwright() as playwright:
yield playwright
playwright_generator = get_playwright()
print(playwright_generator) # <generator object get_playwright at 0x104031580>
playwright = next(playwright_generator)
print(playwright) # <playwright._impl._playwright.Playwright object at 0x1041aabb0>
For the 2nd form:
@pytest.fixture()
def get_playwright():
with sync_playwright() as playwright:
yield playwright
def test(get_playwright):
print(get_playwright)
It should be the same case, but it's just that pytest automatically calls next()
on the fixture value if it's a generator. I could not find documentation for this behavior from the pytest docs, but it was mentioned by one of the pytest author's/maintainer's in a different answer:
Here's roughly the execution here
- pytest notices your fixture is used for the test function
- pytest calls the fixture function
- since it is a generator, it returns immediately without executing code
- pytest notices it is a generator, calls
next(...)
on it
- this causes the code to execute until the
yield
and then "pausing". you can think of it kind of as a co-routine ...- pytest then executes your test function
...which is probably why the value passed to your test function is already the next
-ed value, the playwright object.
Check out the latest blogs from LambdaTest on this topic:
With the rapid evolution in technology and a massive increase of businesses going online after the Covid-19 outbreak, web applications have become more important for organizations. For any organization to grow, the web application interface must be smooth, user-friendly, and cross browser compatible with various Internet browsers.
A good User Interface (UI) is essential to the quality of software or application. A well-designed, sleek, and modern UI goes a long way towards providing a high-quality product for your customers − something that will turn them on.
In today’s data-driven world, the ability to access and analyze large amounts of data can give researchers, businesses & organizations a competitive edge. One of the most important & free sources of this data is the Internet, which can be accessed and mined through web scraping.
The year 2021 can be encapsulated as one major transition. In 2022, the current breakthroughs in the elusive fight to eliminate the COVID-19 pandemic are top of mind for enterprises globally. At the same time, we are witnessing recent strides in technological advancements as the world gets digitized. As a result, the year 2022 will see the resumption of massive changes in technology and digital transformation, driving firms to adapt and transform themselves perpetually.
Selenium, a project hosted by the Apache Software Foundation, is an umbrella open-source project comprising a variety of tools and libraries for test automation. Selenium automation framework enables QA engineers to perform automated web application testing using popular programming languages like Python, Java, JavaScript, C#, Ruby, and PHP.
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!!