feat(expect): add expect_navigation, expect_load_state, navigation tests

Author: pavelfeldman

playwright/async_api.py

@@ -50,9 +50,9 @@
 from playwright.network import Route as RouteImpl
 from playwright.page import BindingCall as BindingCallImpl
 from playwright.page import Page as PageImpl
+from playwright.page import Worker as WorkerImpl
 from playwright.playwright import Playwright as PlaywrightImpl
 from playwright.selectors import Selectors as SelectorsImpl
-from playwright.worker import Worker as WorkerImpl
 
 NoneType = type(None)
 
@@ -1622,9 +1622,9 @@ async def goto(
 
     async def waitForNavigation(
         self,
-        timeout: int = None,
-        waitUntil: Literal["load", "domcontentloaded", "networkidle"] = None,
         url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
+        waitUntil: Literal["load", "domcontentloaded", "networkidle"] = None,
+        timeout: int = None,
     ) -> typing.Union["Response", NoneType]:
         """Frame.waitForNavigation
 
@@ -1634,15 +1634,15 @@ async def waitForNavigation(
 
         Parameters
         ----------
-        timeout : Optional[int]
-            Maximum navigation time in milliseconds, defaults to 30 seconds, pass `0` to disable timeout. The default value can be changed by using the browserContext.setDefaultNavigationTimeout(timeout), browserContext.setDefaultTimeout(timeout), page.setDefaultNavigationTimeout(timeout) or page.setDefaultTimeout(timeout) methods.
+        url : Optional[str, typing.Pattern, typing.Callable[[str], bool]]
+            URL string, URL regex pattern or predicate receiving URL to match while waiting for the navigation.
         waitUntil : Optional[typing.Literal['load', 'domcontentloaded', 'networkidle']]
             When to consider navigation succeeded, defaults to `load`. Events can be either:
              - `'domcontentloaded'` - consider navigation to be finished when the `DOMContentLoaded` event is fired.
              - `'load'` - consider navigation to be finished when the `load` event is fired.
              - `'networkidle'` - consider navigation to be finished when there are no network connections for at least `500` ms.
-        url : Optional[str, typing.Pattern, typing.Callable[[str], bool]]
-            URL string, URL regex pattern or predicate receiving URL to match while waiting for the navigation.
+        timeout : Optional[int]
+            Maximum navigation time in milliseconds, defaults to 30 seconds, pass `0` to disable timeout. The default value can be changed by using the browserContext.setDefaultNavigationTimeout(timeout), browserContext.setDefaultTimeout(timeout), page.setDefaultNavigationTimeout(timeout) or page.setDefaultTimeout(timeout) methods.
 
         Returns
         -------
@@ -1651,7 +1651,7 @@ async def waitForNavigation(
         """
         return mapping.from_impl_nullable(
             await self._impl_obj.waitForNavigation(
-                timeout=timeout, waitUntil=waitUntil, url=self._wrap_handler(url)
+                url=self._wrap_handler(url), waitUntil=waitUntil, timeout=timeout
             )
         )
 
@@ -2597,6 +2597,23 @@ async def title(self) -> str:
         """
         return mapping.from_maybe_impl(await self._impl_obj.title())
 
+    def expect_load_state(
+        self,
+        state: Literal["load", "domcontentloaded", "networkidle"] = None,
+        timeout: int = None,
+    ) -> AsyncEventContextManager[typing.Union["Response", NoneType]]:
+        return AsyncEventContextManager(self._impl_obj.waitForLoadState(state, timeout))
+
+    def expect_navigation(
+        self,
+        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
+        waitUntil: Literal["load", "domcontentloaded", "networkidle"] = None,
+        timeout: int = None,
+    ) -> AsyncEventContextManager[typing.Union["Response", NoneType]]:
+        return AsyncEventContextManager(
+            self._impl_obj.waitForNavigation(url, waitUntil, timeout)
+        )
+
 
 mapping.register(FrameImpl, Frame)
 
@@ -3577,9 +3594,9 @@ async def waitForLoadState(
 
     async def waitForNavigation(
         self,
-        timeout: int = None,
+        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
         waitUntil: Literal["load", "domcontentloaded", "networkidle"] = None,
-        url: str = None,
+        timeout: int = None,
     ) -> typing.Union["Response", NoneType]:
         """Page.waitForNavigation
 
@@ -3590,15 +3607,15 @@ async def waitForNavigation(
 
         Parameters
         ----------
-        timeout : Optional[int]
-            Maximum navigation time in milliseconds, defaults to 30 seconds, pass `0` to disable timeout. The default value can be changed by using the browserContext.setDefaultNavigationTimeout(timeout), browserContext.setDefaultTimeout(timeout), page.setDefaultNavigationTimeout(timeout) or page.setDefaultTimeout(timeout) methods.
+        url : Optional[str, typing.Pattern, typing.Callable[[str], bool]]
+            A glob pattern, regex pattern or predicate receiving URL to match while waiting for the navigation.
         waitUntil : Optional[typing.Literal['load', 'domcontentloaded', 'networkidle']]
             When to consider navigation succeeded, defaults to `load`. Events can be either:
              - `'domcontentloaded'` - consider navigation to be finished when the `DOMContentLoaded` event is fired.
              - `'load'` - consider navigation to be finished when the `load` event is fired.
              - `'networkidle'` - consider navigation to be finished when there are no network connections for at least `500` ms.
-        url : Optional[str]
-            A glob pattern, regex pattern or predicate receiving URL to match while waiting for the navigation.
+        timeout : Optional[int]
+            Maximum navigation time in milliseconds, defaults to 30 seconds, pass `0` to disable timeout. The default value can be changed by using the browserContext.setDefaultNavigationTimeout(timeout), browserContext.setDefaultTimeout(timeout), page.setDefaultNavigationTimeout(timeout) or page.setDefaultTimeout(timeout) methods.
 
         Returns
         -------
@@ -3607,7 +3624,7 @@ async def waitForNavigation(
         """
         return mapping.from_impl_nullable(
             await self._impl_obj.waitForNavigation(
-                timeout=timeout, waitUntil=waitUntil, url=url
+                url=self._wrap_handler(url), waitUntil=waitUntil, timeout=timeout
             )
         )
 
@@ -4634,63 +4651,112 @@ async def pdf(
             )
         )
 
+    def expect_event(
+        self,
+        event: str,
+        predicate: typing.Union[typing.Callable[[typing.Any], bool]] = None,
+        timeout: int = None,
+    ) -> AsyncEventContextManager:
+        return AsyncEventContextManager(
+            self._impl_obj.waitForEvent(event, predicate, timeout)
+        )
+
     def expect_console_message(
         self,
         predicate: typing.Union[typing.Callable[["ConsoleMessage"], bool]] = None,
         timeout: int = None,
     ) -> AsyncEventContextManager["ConsoleMessage"]:
-        return AsyncEventContextManager(self, "console", predicate, timeout)
+        event = "console"
+        return AsyncEventContextManager(
+            self._impl_obj.waitForEvent(event, predicate, timeout)
+        )
 
     def expect_dialog(
         self,
         predicate: typing.Union[typing.Callable[["Dialog"], bool]] = None,
         timeout: int = None,
     ) -> AsyncEventContextManager["Dialog"]:
-        return AsyncEventContextManager(self, "dialog", predicate, timeout)
+        event = "dialog"
+        return AsyncEventContextManager(
+            self._impl_obj.waitForEvent(event, predicate, timeout)
+        )
 
     def expect_download(
         self,
         predicate: typing.Union[typing.Callable[["Download"], bool]] = None,
         timeout: int = None,
     ) -> AsyncEventContextManager["Download"]:
-        return AsyncEventContextManager(self, "download", predicate, timeout)
+        event = "download"
+        return AsyncEventContextManager(
+            self._impl_obj.waitForEvent(event, predicate, timeout)
+        )
 
     def expect_file_chooser(
         self,
         predicate: typing.Union[typing.Callable[["FileChooser"], bool]] = None,
         timeout: int = None,
     ) -> AsyncEventContextManager["FileChooser"]:
-        return AsyncEventContextManager(self, "filechooser", predicate, timeout)
+        event = "filechooser"
+        return AsyncEventContextManager(
+            self._impl_obj.waitForEvent(event, predicate, timeout)
+        )
 
-    def expect_request(
+    def expect_load_state(
         self,
-        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
-        predicate: typing.Union[typing.Callable[["Request"], bool]] = None,
+        state: Literal["load", "domcontentloaded", "networkidle"] = None,
         timeout: int = None,
-    ) -> AsyncEventContextManager["Request"]:
-        return AsyncEventContextManager(self, "request", predicate, timeout)
+    ) -> AsyncEventContextManager[typing.Union["Response", NoneType]]:
+        return AsyncEventContextManager(self._impl_obj.waitForLoadState(state, timeout))
 
-    def expect_response(
+    def expect_navigation(
         self,
         url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
-        predicate: typing.Union[typing.Callable[["Response"], bool]] = None,
+        waitUntil: Literal["load", "domcontentloaded", "networkidle"] = None,
         timeout: int = None,
-    ) -> AsyncEventContextManager["Response"]:
-        return AsyncEventContextManager(self, "response", predicate, timeout)
+    ) -> AsyncEventContextManager[typing.Union["Response", NoneType]]:
+        return AsyncEventContextManager(
+            self._impl_obj.waitForNavigation(url, waitUntil, timeout)
+        )
 
     def expect_popup(
         self,
         predicate: typing.Union[typing.Callable[["Page"], bool]] = None,
         timeout: int = None,
     ) -> AsyncEventContextManager["Page"]:
-        return AsyncEventContextManager(self, "popup", predicate, timeout)
+        event = "popup"
+        return AsyncEventContextManager(
+            self._impl_obj.waitForEvent(event, predicate, timeout)
+        )
+
+    def expect_request(
+        self,
+        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
+        predicate: typing.Union[typing.Callable[["Request"], bool]] = None,
+        timeout: int = None,
+    ) -> AsyncEventContextManager["Request"]:
+        return AsyncEventContextManager(
+            self._impl_obj.waitForRequest(url, predicate, timeout)
+        )
+
+    def expect_response(
+        self,
+        url: typing.Union[str, typing.Pattern, typing.Callable[[str], bool]] = None,
+        predicate: typing.Union[typing.Callable[["Request"], bool]] = None,
+        timeout: int = None,
+    ) -> AsyncEventContextManager["Response"]:
+        return AsyncEventContextManager(
+            self._impl_obj.waitForResponse(url, predicate, timeout)
+        )
 
     def expect_worker(
         self,
         predicate: typing.Union[typing.Callable[["Worker"], bool]] = None,
         timeout: int = None,
     ) -> AsyncEventContextManager["Worker"]:
-        return AsyncEventContextManager(self, "worker", predicate, timeout)
+        event = "worker"
+        return AsyncEventContextManager(
+            self._impl_obj.waitForEvent(event, predicate, timeout)
+        )
 
 
 mapping.register(PageImpl, Page)
@@ -5036,12 +5102,25 @@ async def close(self) -> NoneType:
         """
         return mapping.from_maybe_impl(await self._impl_obj.close())
 
+    def expect_event(
+        self,
+        event: str,
+        predicate: typing.Union[typing.Callable[[typing.Any], bool]] = None,
+        timeout: int = None,
+    ) -> AsyncEventContextManager:
+        return AsyncEventContextManager(
+            self._impl_obj.waitForEvent(event, predicate, timeout)
+        )
+
     def expect_page(
         self,
         predicate: typing.Union[typing.Callable[["Page"], bool]] = None,
         timeout: int = None,
     ) -> AsyncEventContextManager["Page"]:
-        return AsyncEventContextManager(self, "page", predicate, timeout)
+        event = "page"
+        return AsyncEventContextManager(
+            self._impl_obj.waitForEvent(event, predicate, timeout)
+        )
 
 
 mapping.register(BrowserContextImpl, BrowserContext)

playwright/async_base.py

@@ -13,10 +13,9 @@
 # limitations under the License.
 
 import asyncio
-from typing import Any, Callable, Generic, Optional, TypeVar, cast
+from typing import Any, Callable, Coroutine, Generic, Optional, TypeVar, cast
 
 from playwright.impl_to_api_mapping import ImplToApiMapping, ImplWrapper
-from playwright.wait_helper import WaitHelper
 
 mapping = ImplToApiMapping()
 
@@ -25,39 +24,22 @@
 
 
 class AsyncEventInfo(Generic[T]):
-    def __init__(
-        self,
-        async_base: "AsyncBase",
-        event: str,
-        predicate: Callable[[T], bool] = None,
-        timeout: int = None,
-    ) -> None:
+    def __init__(self, coroutine: Coroutine) -> None:
         self._value: Optional[T] = None
-
-        wait_helper = WaitHelper(async_base._loop)
-        wait_helper.reject_on_timeout(
-            timeout or 30000, f'Timeout while waiting for event "${event}"'
-        )
-        self._future = asyncio.get_event_loop().create_task(
-            wait_helper.wait_for_event(async_base._impl_obj, event, predicate)
-        )
+        self._future = asyncio.get_event_loop().create_task(coroutine)
+        self._done = False
 
     @property
     async def value(self) -> T:
-        if not self._value:
+        if not self._done:
             self._value = mapping.from_maybe_impl(await self._future)
+            self._done = True
         return cast(T, self._value)
 
 
 class AsyncEventContextManager(Generic[T]):
-    def __init__(
-        self,
-        async_base: "AsyncBase",
-        event: str,
-        predicate: Callable[[T], bool] = None,
-        timeout: int = None,
-    ) -> None:
-        self._event = AsyncEventInfo(async_base, event, predicate, timeout)
+    def __init__(self, coroutine: Coroutine) -> None:
+        self._event: AsyncEventInfo = AsyncEventInfo(coroutine)
 
     async def __aenter__(self) -> AsyncEventInfo[T]:
         return self._event
@@ -90,8 +72,3 @@ def once(self, event_name: str, handler: Any) -> None:
 
     def remove_listener(self, event_name: str, handler: Any) -> None:
         self._impl_obj.remove_listener(event_name, handler)
-
-    def expect_event(
-        self, event: str, predicate: Callable[[Any], bool] = None, timeout: int = None,
-    ) -> AsyncEventContextManager:
-        return AsyncEventContextManager(self, event, predicate, timeout)

playwright/browser_context.py

@@ -216,9 +216,9 @@ async def close(self) -> None:
     def expect_event(
         self, event: str, predicate: Callable[[Any], bool] = None, timeout: int = None,
     ) -> EventContextManagerImpl:
-        return EventContextManagerImpl(self, event, predicate, timeout)
+        return EventContextManagerImpl(self.waitForEvent(event, predicate, timeout))
 
     def expect_page(
         self, predicate: Callable[[Page], bool] = None, timeout: int = None,
     ) -> EventContextManagerImpl[Page]:
-        return EventContextManagerImpl(self, "page", predicate, timeout)
+        return EventContextManagerImpl(self.waitForEvent("page", predicate, timeout))