fix: make on_frame sync, use test fixtures, add DisposableStub context manager

Author: Skn0tt

playwright/_impl/_debugger.py

@@ -19,7 +19,7 @@
 
 
 class Debugger(ChannelOwner):
-    Events = {"PausedStateChanged": "pausedStateChanged"}
+    Events = {"PausedStateChanged": "pausedstatechanged"}
 
     def __init__(
         self, parent: ChannelOwner, type: str, guid: str, initializer: Dict

playwright/_impl/_disposable.py

@@ -48,6 +48,12 @@ def __init__(self, dispose_fn: Callable[[], Awaitable[None]]) -> None:
     async def dispose(self) -> None:
         await self._dispose_fn()
 
+    async def __aenter__(self) -> "DisposableStub":
+        return self
+
+    async def __aexit__(self, *args: object) -> None:
+        await self.dispose()
+
     async def close(self) -> None:
         await self.dispose()
 

playwright/_impl/_screencast.py

@@ -12,11 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import base64
 from pathlib import Path
 from typing import (
     TYPE_CHECKING,
-    Any,
-    Awaitable,
     Callable,
     Dict,
     Literal,
@@ -36,21 +35,22 @@
 class Screencast:
     def __init__(self, page: "Page") -> None:
         self._page = page
+        self._loop = page._loop
         self._started = False
         self._save_path: Optional[Union[str, Path]] = None
-        self._on_frame: Optional[Callable[[ScreencastFrame], Awaitable[Any]]] = None
+        self._on_frame: Optional[Callable[[ScreencastFrame], None]] = None
         self._artifact: Optional[Artifact] = None
         self._page._channel.on("screencastFrame", self._handle_frame)
 
     def _handle_frame(self, params: Dict) -> None:
         if self._on_frame:
-            self._on_frame({"data": params["data"]})
+            self._on_frame({"data": base64.b64decode(params["data"])})
 
     async def start(
         self,
         path: Union[str, Path] = None,
         quality: int = None,
-        onFrame: Callable[[ScreencastFrame], Awaitable[Any]] = None,
+        onFrame: Callable[[ScreencastFrame], None] = None,
     ) -> DisposableStub:
         if self._started:
             raise Exception("Screencast is already started")
@@ -66,8 +66,8 @@ async def start(
                 "record": path is not None,
             },
         )
-        if result.get("artifact"):
-            self._artifact = from_channel(result["artifact"])
+        if result:
+            self._artifact = from_channel(result)
             self._save_path = path
 
         return DisposableStub(lambda: self.stop())
@@ -100,7 +100,7 @@ async def hide_actions(self) -> None:
         await self._page._channel.send("screencastHideActions", None)
 
     async def show_overlay(self, html: str, duration: float = None) -> DisposableStub:
-        result = await self._page._channel.send(
+        overlay_id = await self._page._channel.send(
             "screencastShowOverlay",
             None,
             {"html": html, "duration": duration},
@@ -110,7 +110,7 @@ async def show_overlay(self, html: str, duration: float = None) -> DisposableStu
             lambda: self._page._channel.send(
                 "screencastRemoveOverlay",
                 None,
-                {"id": result["id"]},
+                {"id": overlay_id},
             )
         )
 

playwright/async_api/_generated.py

@@ -21598,9 +21598,7 @@ async def start(
         *,
         path: typing.Optional[typing.Union[pathlib.Path, str]] = None,
         quality: typing.Optional[int] = None,
-        on_frame: typing.Optional[
-            typing.Callable[[ScreencastFrame], typing.Awaitable[typing.Any]]
-        ] = None,
+        on_frame: typing.Optional[typing.Callable[[ScreencastFrame], None]] = None,
     ) -> "AsyncContextManager":
         """Screencast.start
 
@@ -21615,7 +21613,7 @@ async def start(
             Path where the video should be saved when the screencast is stopped. When provided, video recording is started.
         quality : Union[int, None]
             The quality of the image, between 0-100.
-        on_frame : Union[Callable[[{data: bytes}], typing.Awaitable[typing.Any]], None]
+        on_frame : Union[Callable[[{data: bytes}], None], None]
             Callback that receives JPEG-encoded frame data.
 
         Returns

playwright/sync_api/_generated.py

@@ -21688,9 +21688,7 @@ def start(
         *,
         path: typing.Optional[typing.Union[pathlib.Path, str]] = None,
         quality: typing.Optional[int] = None,
-        on_frame: typing.Optional[
-            typing.Callable[[ScreencastFrame], typing.Awaitable[typing.Any]]
-        ] = None,
+        on_frame: typing.Optional[typing.Callable[[ScreencastFrame], None]] = None,
     ) -> "SyncContextManager":
         """Screencast.start
 
@@ -21705,7 +21703,7 @@ def start(
             Path where the video should be saved when the screencast is stopped. When provided, video recording is started.
         quality : Union[int, None]
             The quality of the image, between 0-100.
-        on_frame : Union[Callable[[{data: bytes}], typing.Awaitable[typing.Any]], None]
+        on_frame : Union[Callable[[{data: bytes}], None], None]
             Callback that receives JPEG-encoded frame data.
 
         Returns

scripts/expected_api_mismatch.txt

@@ -20,3 +20,6 @@ Parameter type mismatch in BrowserContext.route_web_socket(handler=): documented
 Parameter type mismatch in Page.route_web_socket(handler=): documented as Callable[[WebSocketRoute], Union[Any, Any]], code has Callable[[WebSocketRoute], Any]
 Parameter type mismatch in WebSocketRoute.on_close(handler=): documented as Callable[[Union[int, undefined]], Union[Any, Any]], code has Callable[[Union[int, None], Union[str, None]], Any]
 Parameter type mismatch in WebSocketRoute.on_message(handler=): documented as Callable[[str], Union[Any, Any]], code has Callable[[Union[bytes, str]], Any]
+
+# Python on_frame callback is sync, not async
+Parameter type mismatch in Screencast.start(on_frame=): documented as Union[Callable[[{data: bytes}], typing.Awaitable[typing.Any]], None], code has Union[Callable[[{data: bytes}], None], None]

tests/async/test_debugger.py

@@ -0,0 +1,96 @@
+# Copyright (c) Microsoft Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+
+from playwright.async_api import BrowserContext, Page
+
+
+async def test_should_return_none_paused_details_initially(
+    context: BrowserContext,
+) -> None:
+    dbg = context.debugger
+    assert dbg.paused_details is None
+
+
+async def test_should_pause_at_next_and_resume(
+    page: Page, context: BrowserContext
+) -> None:
+    await page.set_content("<div>click me</div>")
+    dbg = context.debugger
+    assert dbg.paused_details is None
+
+    await dbg.request_pause()
+
+    paused_event = asyncio.Event()
+
+    def on_paused_state_changed() -> None:
+        if dbg.paused_details is not None:
+            assert "Click" in dbg.paused_details["title"]
+            paused_event.set()
+            asyncio.ensure_future(dbg.resume())
+
+    dbg.on("pausedstatechanged", on_paused_state_changed)
+
+    click_task = asyncio.ensure_future(page.click("div"))
+    await asyncio.wait_for(paused_event.wait(), timeout=10)
+    await asyncio.wait_for(click_task, timeout=10)
+    assert dbg.paused_details is None
+
+
+async def test_should_step_with_next(page: Page, context: BrowserContext) -> None:
+    await page.set_content("<div>click me</div>")
+    dbg = context.debugger
+    assert dbg.paused_details is None
+
+    await dbg.request_pause()
+
+    first_pause_seen = [False]
+
+    def on_paused_state_changed() -> None:
+        if dbg.paused_details is not None and not first_pause_seen[0]:
+            assert "Click" in dbg.paused_details["title"]
+            first_pause_seen[0] = True
+            asyncio.ensure_future(dbg.next())
+        elif dbg.paused_details is not None and first_pause_seen[0]:
+            asyncio.ensure_future(dbg.resume())
+
+    dbg.on("pausedstatechanged", on_paused_state_changed)
+
+    click_task = asyncio.ensure_future(page.click("div"))
+    await asyncio.wait_for(click_task, timeout=10)
+    assert dbg.paused_details is None
+
+
+async def test_should_pause_at_pause_call(page: Page, context: BrowserContext) -> None:
+    await page.set_content("<div>click me</div>")
+    dbg = context.debugger
+    assert dbg.paused_details is None
+
+    await dbg.request_pause()
+
+    paused_event = asyncio.Event()
+
+    def on_paused_state_changed() -> None:
+        if dbg.paused_details is not None:
+            assert "Pause" in dbg.paused_details["title"]
+            paused_event.set()
+            asyncio.ensure_future(dbg.resume())
+
+    dbg.on("pausedstatechanged", on_paused_state_changed)
+
+    pause_task = asyncio.ensure_future(page.pause())
+    await asyncio.wait_for(paused_event.wait(), timeout=10)
+    await asyncio.wait_for(pause_task, timeout=10)
+    assert dbg.paused_details is None

tests/async/test_screencast.py

@@ -0,0 +1,105 @@
+# Copyright (c) Microsoft Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from pathlib import Path
+from typing import List
+
+import pytest
+
+from playwright._impl._api_structures import ScreencastFrame
+from playwright.async_api import Page
+from tests.server import Server
+
+
+def _noop_frame(f: ScreencastFrame) -> None:
+    pass
+
+
+async def test_screencast_start_should_deliver_frames_via_on_frame(
+    page: Page, server: Server
+) -> None:
+    frames: List[ScreencastFrame] = []
+
+    def collect_frame(f: ScreencastFrame) -> None:
+        frames.append(f)
+
+    await page.screencast.start(on_frame=collect_frame)
+    await page.goto(server.EMPTY_PAGE)
+    await page.evaluate("() => document.body.style.backgroundColor = 'red'")
+    await page.wait_for_timeout(500)
+    await page.screencast.stop()
+    assert len(frames) > 0, "expected at least one frame"
+    for frame in frames:
+        assert frame["data"][:2] == b"\xff\xd8", "expected JPEG frame"
+
+
+async def test_screencast_start_should_throw_if_already_started(
+    page: Page,
+) -> None:
+    await page.screencast.start(on_frame=_noop_frame)
+    with pytest.raises(Exception, match="already started"):
+        await page.screencast.start(on_frame=_noop_frame)
+    await page.screencast.stop()
+
+
+async def test_screencast_start_should_record_video_to_path(
+    page: Page, server: Server, tmp_path: Path
+) -> None:
+    video_path = tmp_path / "video.webm"
+    await page.screencast.start(path=video_path)
+    await page.goto(server.EMPTY_PAGE)
+    await page.evaluate("() => document.body.style.backgroundColor = 'red'")
+    await page.wait_for_timeout(500)
+    await page.screencast.stop()
+    assert video_path.exists(), f"video file should exist: {video_path}"
+    assert video_path.stat().st_size > 0
+
+
+async def test_screencast_start_returns_disposable(page: Page) -> None:
+    disposable = await page.screencast.start(on_frame=_noop_frame)
+    async with disposable:
+        pass
+    # After dispose, starting again should succeed.
+    await page.screencast.start(on_frame=_noop_frame)
+    await page.screencast.stop()
+
+
+async def test_screencast_show_overlay(page: Page, server: Server) -> None:
+    await page.goto(server.EMPTY_PAGE)
+    disposable = await page.screencast.show_overlay("<div>Hello Overlay</div>")
+    assert disposable
+    async with disposable:
+        pass
+
+
+async def test_screencast_show_chapter(page: Page, server: Server) -> None:
+    await page.goto(server.EMPTY_PAGE)
+    await page.screencast.show_chapter("Chapter Title")
+    await page.screencast.show_chapter(
+        "With Description", description="Some details", duration=100
+    )
+
+
+async def test_screencast_hide_show_overlays(page: Page, server: Server) -> None:
+    await page.goto(server.EMPTY_PAGE)
+    await page.screencast.show_overlay("<div>visible</div>")
+    await page.screencast.hide_overlays()
+    await page.screencast.show_overlays()
+
+
+async def test_screencast_show_and_hide_actions(page: Page, server: Server) -> None:
+    await page.goto(server.EMPTY_PAGE)
+    async with await page.screencast.show_actions():
+        pass
+    await page.screencast.hide_actions()