fix(typing): typed overloads for expect_event and wait_for_event (#3061)

Author: dgozman

playwright/async_api/_generated.py

@@ -243,6 +243,55 @@ def print_events(self, class_name: str) -> None:
                 doc.append(f"        return super().{event_type}(event=event,f=f)")
             print("\n".join(doc))
 
+    def print_event_overloads(self, class_name: str, method_name: str) -> None:
+        """Emit ``@typing.overload`` stubs for ``expect_event`` / ``wait_for_event``
+        keyed on ``Literal`` event names with their payload types from api.json,
+        so pyright/mypy can narrow the return type at call sites.
+        Must be called right before the implementation signature is emitted.
+        """
+        if class_name not in self.classes:
+            return
+        events = self.classes[class_name].get("events") or []
+        if not events:
+            return
+        is_expect = method_name == "expect_event"
+        async_prefix = "async " if not is_expect and self.is_async else ""
+        if is_expect:
+            ctx_mgr = (
+                "AsyncEventContextManager" if self.is_async else "EventContextManager"
+            )
+        for event in events:
+            payload = self.serialize_doc_type(event["type"], "")
+            if payload.startswith("{"):
+                payload = "typing.Dict"
+            if "Union[" in payload:
+                payload = payload.replace("Union[", "typing.Union[")
+            return_type = f'{ctx_mgr}["{payload}"]' if is_expect else f'"{payload}"'
+            event_literal = event["name"].lower()
+            print("    @typing.overload")
+            print(f"    {async_prefix}def {method_name}(")
+            print("        self,")
+            print(f'        event: typing.Literal["{event_literal}"],')
+            print(
+                f'        predicate: typing.Optional[typing.Callable[["{payload}"], bool]] = None,'
+            )
+            print("        *,")
+            print("        timeout: typing.Optional[float] = None,")
+            print(f"    ) -> {return_type}: ...")
+            print("")
+        # Catch-all overload for non-literal event names — keeps pyright happy
+        # with `event: str` callers without falling through to `Unknown`.
+        catchall_return = f"{ctx_mgr}[typing.Any]" if is_expect else "typing.Any"
+        print("    @typing.overload")
+        print(f"    {async_prefix}def {method_name}(")
+        print("        self,")
+        print("        event: str,")
+        print("        predicate: typing.Optional[typing.Callable[..., bool]] = None,")
+        print("        *,")
+        print("        timeout: typing.Optional[float] = None,")
+        print(f"    ) -> {catchall_return}: ...")
+        print("")
+
     def indent_paragraph(self, p: str, indent: str) -> str:
         lines = p.split("\n")
         result = [lines[0]]

playwright/sync_api/_generated.py

@@ -92,6 +92,8 @@ def generate(t: Any) -> None:
                 '"Disposable"', '"AsyncContextManager"'
             ).replace('"DisposableStub"', '"AsyncContextManager"')
             print("")
+            if name in ("expect_event", "wait_for_event"):
+                documentation_provider.print_event_overloads(class_name, name)
             async_prefix = "async " if is_async else ""
             print(
                 f"    {async_prefix}def {name}({signature(value, len(name) + 9)}) -> {return_type_value}:"

scripts/documentation_provider.py

@@ -93,6 +93,8 @@ def generate(t: Any) -> None:
                 '"Disposable"', '"SyncContextManager"'
             ).replace('"DisposableStub"', '"SyncContextManager"')
             print("")
+            if name in ("expect_event", "wait_for_event"):
+                documentation_provider.print_event_overloads(class_name, name)
             print(
                 f"    def {name}({signature(value, len(name) + 9)}) -> {return_type_value}:"
             )

scripts/generate_async_api.py

@@ -197,7 +197,7 @@ async def test_page_error_event_should_work(page: Page) -> None:
         await page.set_content('<script>throw new Error("boom")</script>')
     page_error = await page_error_info.value
     assert page_error.page == page
-    assert "boom" in page_error.error.stack
+    assert page_error.error.stack and "boom" in page_error.error.stack
 
 
 async def test_weberror_event_should_work(context: BrowserContext, page: Page) -> None:

scripts/generate_sync_api.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 
 import asyncio
-from typing import cast
 
 import pytest
 
@@ -98,4 +97,4 @@ async def route_handler(route: Route) -> None:
     [popup, _] = await asyncio.gather(
         page.wait_for_event("popup"), page.get_by_text("click me").click()
     )
-    await expect(cast(Page, popup).locator("body")).to_have_text("hello")
+    await expect(popup.locator("body")).to_have_text("hello")

tests/async/test_browsercontext_events.py

@@ -12,11 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Dict
 
 import pytest
 
-from playwright.async_api import Browser, Page
+from playwright.async_api import Browser, Page, ResourceTiming
 from tests.server import Server
 
 
@@ -95,7 +94,7 @@ def verify_timing_value(value: float, previous: float) -> None:
     assert value == -1 or value > 0 and value >= previous
 
 
-def verify_connections_timing_consistency(timing: Dict) -> None:
+def verify_connections_timing_consistency(timing: ResourceTiming) -> None:
     verify_timing_value(timing["domainLookupStart"], -1)
     verify_timing_value(timing["domainLookupEnd"], timing["domainLookupStart"])
     verify_timing_value(timing["connectStart"], timing["domainLookupEnd"])

tests/async/test_page_request_intercept.py

@@ -12,11 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Dict
-
 import pytest
 
-from playwright.sync_api import Browser, Page
+from playwright.sync_api import Browser, Page, ResourceTiming
 from tests.server import Server
 
 
@@ -99,7 +97,7 @@ def verify_timing_value(value: float, previous: float) -> None:
     assert value == -1 or value > 0 and value >= previous
 
 
-def verify_connections_timing_consistency(timing: Dict) -> None:
+def verify_connections_timing_consistency(timing: ResourceTiming) -> None:
     verify_timing_value(timing["domainLookupStart"], -1)
     verify_timing_value(timing["domainLookupEnd"], timing["domainLookupStart"])
     verify_timing_value(timing["connectStart"], timing["domainLookupEnd"])