feat(tab): add TabProtocol; converge execute_script to Python values

Add TabProtocol (portable tab contract) + a mypy conformance harness (browser/_conformance.py) asserting both Tab (CDP) and BiDiTab (BiDi) satisfy it.

BREAKING (CDP): Tab.execute_script now returns the script's deserialized Python value (was the raw CDP response) and takes (script, *args); the CDP power kwargs move to the escape hatch (execute_protocol_command(RuntimeCommands.evaluate(...))). Both protocols now behave identically: same call -> same Python value -> same ScriptExecutionError on throw. BiDiTab.execute_script gains *args support.

Also: Tab.get_cookies/set_cookies use the generic Cookie/CookieParam types; BiDiTab.take_screenshot/print_to_pdf gain quality/beyond_viewport and display_header_footer/print_background for CDP parity; new ScriptExecutionError raised by both backends. Remove the now-dead _get_evaluate_command and _validate_argument_error.

Author: thalissonvs

pydoll/browser/_conformance.py

@@ -0,0 +1,23 @@
+"""Static conformance checks (mypy-only): concrete classes satisfy the portable protocols.
+
+This module has no runtime effect. The assignments exist purely so mypy verifies that the
+concrete CDP and BiDi tab classes implement the portable protocol surface, keeping the two
+backends from drifting apart. Check with::
+
+    mypy --follow-imports=silent pydoll/browser/_conformance.py
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from pydoll.browser.chromium.tab import Tab
+    from pydoll.browser.firefox.tab import BiDiTab
+    from pydoll.browser.protocols import TabProtocol
+
+    def _cdp_tab_conforms(tab: Tab) -> TabProtocol:
+        return tab
+
+    def _bidi_tab_conforms(tab: BiDiTab) -> TabProtocol:
+        return tab

pydoll/browser/chromium/tab.py

@@ -4,6 +4,7 @@
 import base64 as _b64
 import contextlib
 import io
+import json
 import logging
 import shutil
 import warnings
@@ -49,14 +50,14 @@
     IFrameNotFound,
     InvalidFileExtension,
     InvalidIFrame,
-    InvalidScriptWithElement,
     InvalidTabInitialization,
     MissingScreenshotPath,
     NavigationError,
     NetworkEventsNotEnabled,
     NoDialogPresent,
     NotAnIFrame,
     PageLoadTimeout,
+    ScriptExecutionError,
     TopLevelTargetRequired,
     WaitElementTimeout,
     WebSocketConnectionClosed,
@@ -70,11 +71,8 @@
 from pydoll.protocol.cdp.page.events import PageEvent
 from pydoll.protocol.cdp.page.types import FrameResourceTree, ScreenshotFormat
 from pydoll.protocol.cdp.runtime.methods import (
-    CallFunctionOnResponse,
     EvaluateResponse,
-    SerializationOptions,
 )
-from pydoll.protocol.cdp.runtime.types import CallArgument
 from pydoll.protocol.cdp.target.types import TargetInfo
 from pydoll.protocol.events import CDP_DOMAIN_MAP, CDP_EVENT_MAP, Event
 from pydoll.utils import (
@@ -113,8 +111,6 @@
     from pydoll.protocol.cdp.network.methods import GetCookiesResponse as NetworkGetCookiesResponse
     from pydoll.protocol.cdp.network.methods import GetResponseBodyResponse
     from pydoll.protocol.cdp.network.types import (
-        Cookie,
-        CookieParam,
         ErrorReason,
         RequestMethod,
     )
@@ -126,9 +122,10 @@
         NavigateResponse,
         PrintToPDFResponse,
     )
-    from pydoll.protocol.cdp.runtime.methods import CallFunctionOnResponse, EvaluateResponse
+    from pydoll.protocol.cdp.runtime.methods import EvaluateResponse
     from pydoll.protocol.cdp.storage.methods import GetCookiesResponse as StorageGetCookiesResponse
     from pydoll.protocol.cdp.target.methods import AttachToTargetResponse, GetTargetsResponse
+    from pydoll.protocol.types import Cookie, CookieParam
 
 logger = logging.getLogger(__name__)
 
@@ -1176,8 +1173,8 @@ async def _fetch_document_html(self, frame_tree: FrameResourceTree) -> str:
             return html
         except Exception:
             logger.debug('getResourceContent failed for document, falling back to JS')
-            response = await self.execute_script('return document.documentElement.outerHTML')
-            return cast(str, response['result']['result']['value'])
+            html = await self.execute_script('return document.documentElement.outerHTML')
+            return cast(str, html)
 
     async def _fetch_bundle_assets(
         self,
@@ -1258,175 +1255,53 @@ async def handle_dialog(self, accept: bool, prompt_text: Optional[str] = None):
             PageCommands.handle_javascript_dialog(accept=accept, prompt_text=prompt_text)
         )
 
-    @overload
-    async def execute_script(
-        self,
-        script: str,
-        *,
-        object_group: Optional[str] = None,
-        include_command_line_api: Optional[bool] = None,
-        silent: Optional[bool] = None,
-        context_id: Optional[int] = None,
-        return_by_value: Optional[bool] = None,
-        generate_preview: Optional[bool] = None,
-        user_gesture: Optional[bool] = None,
-        await_promise: Optional[bool] = None,
-        throw_on_side_effect: Optional[bool] = None,
-        timeout: Optional[float] = None,
-        disable_breaks: Optional[bool] = None,
-        repl_mode: Optional[bool] = None,
-        allow_unsafe_eval_blocked_by_csp: Optional[bool] = None,
-        unique_context_id: Optional[str] = None,
-        serialization_options: Optional[SerializationOptions] = None,
-    ) -> EvaluateResponse: ...
-
-    @overload
-    async def execute_script(
-        self,
-        script: str,
-        element: WebElement,
-        *,
-        arguments: Optional[list[CallArgument]] = None,
-        silent: Optional[bool] = None,
-        return_by_value: Optional[bool] = None,
-        generate_preview: Optional[bool] = None,
-        user_gesture: Optional[bool] = None,
-        await_promise: Optional[bool] = None,
-        execution_context_id: Optional[int] = None,
-        object_group: Optional[str] = None,
-        throw_on_side_effect: Optional[bool] = None,
-        unique_context_id: Optional[str] = None,
-        serialization_options: Optional[SerializationOptions] = None,
-    ) -> CallFunctionOnResponse: ...
-
-    async def execute_script(
-        self,
-        script: str,
-        element: Optional[WebElement] = None,
-        *,
-        arguments: Optional[list[CallArgument]] = None,
-        object_group: Optional[str] = None,
-        include_command_line_api: Optional[bool] = None,
-        silent: Optional[bool] = None,
-        context_id: Optional[int] = None,
-        return_by_value: Optional[bool] = None,
-        generate_preview: Optional[bool] = None,
-        user_gesture: Optional[bool] = None,
-        await_promise: Optional[bool] = None,
-        execution_context_id: Optional[int] = None,
-        throw_on_side_effect: Optional[bool] = None,
-        timeout: Optional[float] = None,
-        disable_breaks: Optional[bool] = None,
-        repl_mode: Optional[bool] = None,
-        allow_unsafe_eval_blocked_by_csp: Optional[bool] = None,
-        unique_context_id: Optional[str] = None,
-        serialization_options: Optional[SerializationOptions] = None,
-    ) -> Union[EvaluateResponse, CallFunctionOnResponse]:
-        """
-        Execute JavaScript in page context.
+    async def execute_script(self, script: str, *args: object) -> object:
+        """Execute JavaScript in the page and return its deserialized result.
 
         Args:
-            script (str): JavaScript code to execute.
-            element (Optional[WebElement]): Optional WebElement to execute script on.
-            arguments (Optional[list[CallArgument]]): Arguments to pass to the function.
-            object_group (Optional[str]): Symbolic group name for the result (Runtime.evaluate).
-            include_command_line_api (Optional[bool]): Whether to include command line API
-                (Runtime.evaluate).
-            silent (Optional[bool]): Whether to silence exceptions (Runtime.evaluate).
-            context_id (Optional[int]): ID of the execution context to evaluate in
-                (Runtime.evaluate).
-            return_by_value (Optional[bool]): Whether to return the result by value instead of
-                reference (Runtime.evaluate).
-            generate_preview (Optional[bool]): Whether to generate a preview for the result
-                (Runtime.evaluate).
-            user_gesture (Optional[bool]): Whether to treat evaluation as initiated by user
-                gesture (Runtime.evaluate).
-            await_promise (Optional[bool]): Whether to await promise result (Runtime.evaluate).
-            execution_context_id (Optional[int]): ID of the execution context to call the
-                function in.
-            throw_on_side_effect (Optional[bool]): Whether to throw if side effect cannot be
-                ruled out (Runtime.evaluate).
-            timeout (Optional[float]): Timeout in milliseconds (Runtime.evaluate).
-            disable_breaks (Optional[bool]): Whether to disable breakpoints during evaluation
-                (Runtime.evaluate).
-            repl_mode (Optional[bool]): Whether to execute in REPL mode (Runtime.evaluate).
-            allow_unsafe_eval_blocked_by_csp (Optional[bool]): Allow unsafe evaluation
-                (Runtime.evaluate).
-            unique_context_id (Optional[str]): Unique context ID for evaluation
-                (Runtime.evaluate).
-            serialization_options (Optional[SerializationOptions]): Serialization options for
-                the result (Runtime.evaluate).
+            script: JavaScript to run. With no args, a bare ``return`` statement is
+                wrapped automatically. With args, pass a function expression
+                (e.g. ``(a, b) => a + b``) — the args are forwarded to it.
+            *args: JSON-serializable values forwarded to the script function.
 
         Returns:
-            Union[EvaluateResponse, CallFunctionOnResponse]: The result of the script execution.
+            The script's return value as a Python object (None when it returns nothing).
 
         Raises:
-            InvalidScriptWithElement: If script uses 'argument' keyword but no element is provided.
+            ScriptExecutionError: If the script throws.
 
-        Examples:
-            # Execute a simple script to log a message
-            await page.execute_script('console.log("Hello World")')
-
-            # Execute a script that returns the page title
-            await page.execute_script('return document.title')
-
-            # Execute a script on an element to click it
-            await page.execute_script('argument.click()', element)
-
-            # Execute a script on an element to set its value
-            await page.execute_script('argument.value = "Hello"', element)
-        """
-        logger.debug(f'Executing script: with_element={bool(element)}, length={len(script)}')
-        if element is not None:
-            warnings.warn(
-                'Passing a WebElement to Tab.execute_script() is deprecated. '
-                'Use WebElement.execute_script() instead.',
-                DeprecationWarning,
-                stacklevel=2,
-            )
+        Note:
+            For fine-grained CDP control (silent, contextId, raw RemoteObject, etc.)
+            use ``execute_protocol_command(RuntimeCommands.evaluate(...))``.
 
-            return await element.execute_script(
-                script,
-                arguments=arguments,
-                silent=silent,
-                return_by_value=return_by_value,
-                generate_preview=generate_preview,
-                user_gesture=user_gesture,
-                await_promise=await_promise,
-                execution_context_id=execution_context_id,
-                object_group=object_group,
-                throw_on_side_effect=throw_on_side_effect,
-                unique_context_id=unique_context_id,
-                serialization_options=serialization_options,
-            )
+        Examples:
+            await tab.execute_script('return document.title')
+            await tab.execute_script('(a, b) => a + b', 2, 3)
+        """
+        if args:
+            serialized = ', '.join(json.dumps(arg) for arg in args)
+            expression = f'({script})({serialized})'
+        elif has_return_outside_function(script):
+            expression = f'(function(){{ {script} }})()'
+        else:
+            expression = script
 
-        if has_return_outside_function(script):
-            script = f'(function(){{ {script} }})()'
-
-        command = self._get_evaluate_command(
-            script,
-            object_group=object_group,
-            include_command_line_api=include_command_line_api,
-            silent=silent,
-            context_id=context_id,
-            return_by_value=return_by_value,
-            generate_preview=generate_preview,
-            user_gesture=user_gesture,
-            await_promise=await_promise,
-            throw_on_side_effect=throw_on_side_effect,
-            timeout=timeout,
-            disable_breaks=disable_breaks,
-            repl_mode=repl_mode,
-            allow_unsafe_eval_blocked_by_csp=allow_unsafe_eval_blocked_by_csp,
-            unique_context_id=unique_context_id,
-            serialization_options=serialization_options,
-        )
-        logger.debug(f'Executing script without element: length={len(script)}')
-        result: Union[EvaluateResponse, CallFunctionOnResponse] = await self._execute_command(
-            command
+        logger.debug(f'Executing script: length={len(script)}, args={len(args)}')
+        response: EvaluateResponse = await self._execute_command(
+            RuntimeCommands.evaluate(expression, return_by_value=True, await_promise=True)
         )
-        self._validate_argument_error(result)
-        return result
+        return self._extract_script_value(response)
+
+    @staticmethod
+    def _extract_script_value(response: EvaluateResponse) -> object:
+        """Extract the Python value from a Runtime.evaluate response, raising on JS errors."""
+        evaluate_result = response.get('result', {})
+        exception_details = evaluate_result.get('exceptionDetails')
+        if exception_details:
+            exception = exception_details.get('exception', {})
+            message = exception.get('description') or exception_details.get('text', 'Script error')
+            raise ScriptExecutionError(message)
+        return evaluate_result.get('result', {}).get('value')
 
     # TODO: think about how to remove these duplications with the base class
     async def continue_request(
@@ -1849,73 +1724,6 @@ def _get_connection_handler(self) -> ConnectionHandler:
         )
         return ConnectionHandler(self._connection_port, self._target_id)
 
-    @staticmethod
-    def _get_evaluate_command(
-        script: str,
-        *,
-        object_group: Optional[str] = None,
-        include_command_line_api: Optional[bool] = None,
-        silent: Optional[bool] = None,
-        context_id: Optional[int] = None,
-        return_by_value: Optional[bool] = None,
-        generate_preview: Optional[bool] = None,
-        user_gesture: Optional[bool] = None,
-        await_promise: Optional[bool] = None,
-        throw_on_side_effect: Optional[bool] = None,
-        timeout: Optional[float] = None,
-        disable_breaks: Optional[bool] = None,
-        repl_mode: Optional[bool] = None,
-        allow_unsafe_eval_blocked_by_csp: Optional[bool] = None,
-        unique_context_id: Optional[str] = None,
-        serialization_options: Optional[SerializationOptions] = None,
-    ):
-        """Create an evaluate command with the given parameters."""
-        return RuntimeCommands.evaluate(
-            expression=script,
-            object_group=object_group,
-            include_command_line_api=include_command_line_api,
-            silent=silent,
-            context_id=context_id,
-            return_by_value=return_by_value,
-            generate_preview=generate_preview,
-            user_gesture=user_gesture,
-            await_promise=await_promise,
-            throw_on_side_effect=throw_on_side_effect,
-            timeout=timeout,
-            disable_breaks=disable_breaks,
-            repl_mode=repl_mode,
-            allow_unsafe_eval_blocked_by_csp=allow_unsafe_eval_blocked_by_csp,
-            unique_context_id=unique_context_id,
-            serialization_options=serialization_options,
-        )
-
-    @staticmethod
-    def _validate_argument_error(response: EvaluateResponse) -> None:
-        """
-        Validate that script didn't fail with ReferenceError about 'argument' being undefined.
-
-        Raises:
-            InvalidScriptWithElement: If script uses 'argument' keyword but no element was provided.
-        """
-        evaluate_result = response.get('result')
-        if not isinstance(evaluate_result, dict):
-            return
-
-        remote_object = evaluate_result.get('result')
-        if not isinstance(remote_object, dict):
-            return
-
-        if not (
-            remote_object.get('type') == 'object'
-            and remote_object.get('subtype') == 'error'
-            and remote_object.get('className') == 'ReferenceError'
-        ):
-            return
-
-        description = remote_object.get('description', '')
-        if 'argument is not defined' in description:
-            raise InvalidScriptWithElement('Script contains "argument" but no element was provided')
-
     _PAGE_LOAD_EVENT_MAP = {
         PageLoadState.INTERACTIVE: PageEvent.DOM_CONTENT_EVENT_FIRED,
         PageLoadState.COMPLETE: PageEvent.LOAD_EVENT_FIRED,