From 0576b49d90a8a58a9433c3517370c44e0a26c4ce Mon Sep 17 00:00:00 2001 From: alok-108 <166419621+alok-108@users.noreply.github.com> Date: Fri, 13 Nov 2026 14:00:09 +0530 Subject: [PATCH] fix(handlers): support non-attribute-writable callables in wrap_handler --- playwright/_impl/_impl_to_api_mapping.py | 39 ++++--- tests/async/python/test_handlers.py | 141 +++++++++++++++++++++++ tests/sync/python/test_handlers.py | 141 +++++++++++++++++++++++ 3 files changed, 307 insertions(+), 14 deletions(-) create mode 100644 tests/async/python/test_handlers.py create mode 100644 tests/sync/python/test_handlers.py diff --git a/playwright/_impl/_impl_to_api_mapping.py b/playwright/_impl/_impl_to_api_mapping.py index 0e2e16f5e..c1dac7500 100644 --- a/playwright/_impl/_impl_to_api_mapping.py +++ b/playwright/_impl/_impl_to_api_mapping.py @@ -13,14 +13,15 @@ # limitations under the License. import inspect +import types from typing import Any, Callable, Dict, List, Optional, Sequence, Union +from weakref import WeakValueDictionary from playwright._impl._errors import Error from playwright._impl._map import Map from playwright._impl._signature import signature API_ATTR = "_pw_api_instance_" -IMPL_ATTR = "_pw_impl_instance_" class ImplWrapper: @@ -34,6 +35,9 @@ def __repr__(self) -> str: class ImplToApiMapping: def __init__(self) -> None: self._mapping: Dict[type, type] = {} + self._wrapped_handlers: WeakValueDictionary[Any, Callable[..., None]] = ( + WeakValueDictionary() + ) def register(self, impl_class: type, api_class: type) -> None: self._mapping[impl_class] = api_class @@ -119,6 +123,23 @@ def to_impl( except RecursionError: raise Error("Maximum argument depth exceeded") + def _wrap_handler_key(self, handler: Callable[..., Any]) -> Any: + # Built-in and bound methods may return a fresh wrapper object on each + # attribute lookup, so key them by underlying receiver and function identity + # without mutating immutable, built-in, or __slots__ objects. + if inspect.ismethod(handler): + return ("method", id(handler.__self__), id(handler.__func__)) + if isinstance( + handler, (types.BuiltinFunctionType, types.MethodWrapperType) + ) and hasattr(handler, "__self__"): + return ( + "builtin_method", + id(handler.__self__), + getattr(handler, "__qualname__", None), + handler.__name__, + ) + return id(handler) + def wrap_handler(self, handler: Callable[..., Any]) -> Callable[..., None]: def wrapper_func(*args: Any) -> Any: parameters = signature(handler).parameters @@ -142,19 +163,9 @@ def wrapper_func(*args: Any) -> Any: *list(map(lambda a: self.from_maybe_impl(a), args))[:arg_count] ) - if inspect.ismethod(handler): - wrapper = getattr(handler.__self__, IMPL_ATTR + handler.__name__, None) - if not wrapper: - wrapper = wrapper_func - setattr( - handler.__self__, - IMPL_ATTR + handler.__name__, - wrapper, - ) - return wrapper - - wrapper = getattr(handler, IMPL_ATTR, None) + key = self._wrap_handler_key(handler) + wrapper = self._wrapped_handlers.get(key) if not wrapper: wrapper = wrapper_func - setattr(handler, IMPL_ATTR, wrapper) + self._wrapped_handlers[key] = wrapper return wrapper diff --git a/tests/async/python/test_handlers.py b/tests/async/python/test_handlers.py new file mode 100644 index 000000000..db0ce0618 --- /dev/null +++ b/tests/async/python/test_handlers.py @@ -0,0 +1,141 @@ +# 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. + +"""Python only: tests of behaviors that no upstream test covers.""" + +from dataclasses import dataclass, field +from typing import List +from unittest.mock import MagicMock + +from playwright.async_api import ConsoleMessage, Page, Route +from tests.server import Server + + +@dataclass(slots=True) +class SlottedCollector: + messages: List[str] = field(default_factory=list) + + def handle(self, msg: ConsoleMessage) -> None: + self.messages.append(msg.text) + + def __call__(self, msg: ConsoleMessage) -> None: + self.messages.append(msg.text) + + +@dataclass(frozen=True) +class FrozenCollector: + messages: List[str] = field(default_factory=list) + + def handle(self, msg: ConsoleMessage) -> None: + self.messages.append(msg.text) + + def __call__(self, msg: ConsoleMessage) -> None: + self.messages.append(msg.text) + + +@dataclass(slots=True) +class SlottedRouteHandler: + hits: List[str] = field(default_factory=list) + + async def fulfill(self, route: Route) -> None: + self.hits.append(route.request.url) + await route.fulfill(body="intercepted") + + +async def test_event_listener_builtin_bound_method_and_removal(page: Page) -> None: + messages: List[ConsoleMessage] = [] + page.on("console", messages.append) + await page.evaluate('() => console.log("one")') + assert [m.text for m in messages] == ["one"] + + page.remove_listener("console", messages.append) + await page.evaluate('() => console.log("two")') + assert [m.text for m in messages] == ["one"] + + +async def test_expose_function_with_builtin_callables(page: Page) -> None: + recorded: List[str] = [] + await page.expose_function("record_value", recorded.append) + await page.expose_function("get_len", len) + + length = await page.evaluate( + """async () => { + await window.record_value("hello"); + return await window.get_len("hello"); + }""" + ) + assert recorded == ["hello"] + assert length == 5 + + +async def test_event_listener_slotted_frozen_and_mock_callables(page: Page) -> None: + slotted_a = SlottedCollector() + slotted_b = SlottedCollector() + frozen_a = FrozenCollector() + frozen_b = FrozenCollector() + mock = MagicMock() + + page.on("console", slotted_a.handle) + page.on("console", slotted_b) + page.on("console", frozen_a.handle) + page.on("console", frozen_b) + page.on("console", mock) + + await page.evaluate('() => console.log("ok")') + assert slotted_a.messages == ["ok"] + assert slotted_b.messages == ["ok"] + assert frozen_a.messages == ["ok"] + assert frozen_b.messages == ["ok"] + assert mock.call_count == 1 + assert mock.call_args[0][0].text == "ok" + + page.remove_listener("console", slotted_a.handle) + page.remove_listener("console", slotted_b) + page.remove_listener("console", frozen_a.handle) + page.remove_listener("console", frozen_b) + page.remove_listener("console", mock) + + await page.evaluate('() => console.log("after")') + assert slotted_a.messages == ["ok"] + assert slotted_b.messages == ["ok"] + assert frozen_a.messages == ["ok"] + assert frozen_b.messages == ["ok"] + assert mock.call_count == 1 + + +async def test_route_and_unroute_with_builtin_method_predicate_and_slotted_handler( + page: Page, server: Server +) -> None: + allowed_urls = {server.EMPTY_PAGE} + handler = SlottedRouteHandler() + + await page.route(allowed_urls.__contains__, handler.fulfill) + await page.goto(server.EMPTY_PAGE) + assert await page.locator("body").inner_text() == "intercepted" + assert handler.hits == [server.EMPTY_PAGE] + + await page.unroute(allowed_urls.__contains__, handler.fulfill) + await page.goto(server.EMPTY_PAGE) + assert await page.locator("body").inner_text() == "" + assert handler.hits == [server.EMPTY_PAGE] + + await page.route(server.EMPTY_PAGE.__eq__, handler.fulfill) + await page.goto(server.EMPTY_PAGE) + assert await page.locator("body").inner_text() == "intercepted" + assert handler.hits == [server.EMPTY_PAGE, server.EMPTY_PAGE] + + await page.unroute(server.EMPTY_PAGE.__eq__, handler.fulfill) + await page.goto(server.EMPTY_PAGE) + assert await page.locator("body").inner_text() == "" + assert handler.hits == [server.EMPTY_PAGE, server.EMPTY_PAGE] diff --git a/tests/sync/python/test_handlers.py b/tests/sync/python/test_handlers.py new file mode 100644 index 000000000..57a99c30f --- /dev/null +++ b/tests/sync/python/test_handlers.py @@ -0,0 +1,141 @@ +# 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. + +"""Python only: tests of behaviors that no upstream test covers.""" + +from dataclasses import dataclass, field +from typing import List +from unittest.mock import MagicMock + +from playwright.sync_api import ConsoleMessage, Page, Route +from tests.server import Server + + +@dataclass(slots=True) +class SlottedCollector: + messages: List[str] = field(default_factory=list) + + def handle(self, msg: ConsoleMessage) -> None: + self.messages.append(msg.text) + + def __call__(self, msg: ConsoleMessage) -> None: + self.messages.append(msg.text) + + +@dataclass(frozen=True) +class FrozenCollector: + messages: List[str] = field(default_factory=list) + + def handle(self, msg: ConsoleMessage) -> None: + self.messages.append(msg.text) + + def __call__(self, msg: ConsoleMessage) -> None: + self.messages.append(msg.text) + + +@dataclass(slots=True) +class SlottedRouteHandler: + hits: List[str] = field(default_factory=list) + + def fulfill(self, route: Route) -> None: + self.hits.append(route.request.url) + route.fulfill(body="intercepted") + + +def test_event_listener_builtin_bound_method_and_removal(page: Page) -> None: + messages: List[ConsoleMessage] = [] + page.on("console", messages.append) + page.evaluate('() => console.log("one")') + assert [m.text for m in messages] == ["one"] + + page.remove_listener("console", messages.append) + page.evaluate('() => console.log("two")') + assert [m.text for m in messages] == ["one"] + + +def test_expose_function_with_builtin_callables(page: Page) -> None: + recorded: List[str] = [] + page.expose_function("record_value", recorded.append) + page.expose_function("get_len", len) + + length = page.evaluate( + """async () => { + await window.record_value("hello"); + return await window.get_len("hello"); + }""" + ) + assert recorded == ["hello"] + assert length == 5 + + +def test_event_listener_slotted_frozen_and_mock_callables(page: Page) -> None: + slotted_a = SlottedCollector() + slotted_b = SlottedCollector() + frozen_a = FrozenCollector() + frozen_b = FrozenCollector() + mock = MagicMock() + + page.on("console", slotted_a.handle) + page.on("console", slotted_b) + page.on("console", frozen_a.handle) + page.on("console", frozen_b) + page.on("console", mock) + + page.evaluate('() => console.log("ok")') + assert slotted_a.messages == ["ok"] + assert slotted_b.messages == ["ok"] + assert frozen_a.messages == ["ok"] + assert frozen_b.messages == ["ok"] + assert mock.call_count == 1 + assert mock.call_args[0][0].text == "ok" + + page.remove_listener("console", slotted_a.handle) + page.remove_listener("console", slotted_b) + page.remove_listener("console", frozen_a.handle) + page.remove_listener("console", frozen_b) + page.remove_listener("console", mock) + + page.evaluate('() => console.log("after")') + assert slotted_a.messages == ["ok"] + assert slotted_b.messages == ["ok"] + assert frozen_a.messages == ["ok"] + assert frozen_b.messages == ["ok"] + assert mock.call_count == 1 + + +def test_route_and_unroute_with_builtin_method_predicate_and_slotted_handler( + page: Page, server: Server +) -> None: + allowed_urls = {server.EMPTY_PAGE} + handler = SlottedRouteHandler() + + page.route(allowed_urls.__contains__, handler.fulfill) + page.goto(server.EMPTY_PAGE) + assert page.locator("body").inner_text() == "intercepted" + assert handler.hits == [server.EMPTY_PAGE] + + page.unroute(allowed_urls.__contains__, handler.fulfill) + page.goto(server.EMPTY_PAGE) + assert page.locator("body").inner_text() == "" + assert handler.hits == [server.EMPTY_PAGE] + + page.route(server.EMPTY_PAGE.__eq__, handler.fulfill) + page.goto(server.EMPTY_PAGE) + assert page.locator("body").inner_text() == "intercepted" + assert handler.hits == [server.EMPTY_PAGE, server.EMPTY_PAGE] + + page.unroute(server.EMPTY_PAGE.__eq__, handler.fulfill) + page.goto(server.EMPTY_PAGE) + assert page.locator("body").inner_text() == "" + assert handler.hits == [server.EMPTY_PAGE, server.EMPTY_PAGE]