KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _expect_event method in Playwright Python

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

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.