Conversation
9947809 to
38e222b
Compare
38e222b to
79aa226
Compare
There was a problem hiding this comment.
Pull request overview
This PR introduces a CDDL (Concise Data Definition Language) to Python generator for WebDriver BiDi modules. It generates 9 BiDi protocol modules from the W3C specification, replacing hand-written implementations with auto-generated code.
Changes:
- Adds
py/generate_bidi.py- CDDL parser and Python code generator (623 lines) - Adds Bazel build integration for code generation
- Generates 9 BiDi modules (browser, browsing_context, emulation, input, network, script, session, storage, webextension) with 146 type definitions and 52 commands
- Adds validation tooling and documentation
Reviewed changes
Copilot reviewed 21 out of 21 changed files in this pull request and generated 13 comments.
Show a summary per file
| File | Description |
|---|---|
| py/generate_bidi.py | Core CDDL parser and Python code generator |
| py/private/generate_bidi.bzl | Bazel rule for BiDi code generation |
| py/BUILD.bazel | Integration of generation target |
| py/requirements.txt | Added pycddl dependency |
| py/selenium/webdriver/common/bidi/*.py | Generated BiDi module replacements |
| py/validate_bidi_modules.py | Validation tooling for comparing generated vs hand-written code |
| common/bidi/spec/local.cddl | CDDL specification (1331 lines) |
| Various .md files | Documentation and findings |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 23 out of 24 changed files in this pull request and generated 3 comments.
Comments suppressed due to low confidence (1)
py/validate_bidi_modules.py:1
- Corrected spelling of 'Analyze' to match class name convention.
#!/usr/bin/env python3
c7311e8 to
8ea2d91
Compare
| @dataclass | ||
| class disownDataParameters: | ||
| """disownDataParameters type type.""" | ||
|
|
||
| data_type: Any | None = None | ||
| collector: Any | None = None | ||
| request: Any | None = None |
There was a problem hiding this comment.
Generated dataclass/type names like disownDataParameters start with a lowercase letter, which is inconsistent with Python class naming and the surrounding generated types. This is likely a generator bug in the CDDL-to-Python name mapping; please ensure all generated class/type names use PascalCase (e.g., DisownDataParameters).
| def script(self) -> Script: | ||
| if not self._websocket_connection: | ||
| self._start_bidi() | ||
|
|
||
| if not self._script: | ||
| self._script = Script(self._websocket_connection, self) | ||
|
|
||
| return self._script | ||
|
|
||
| def _start_bidi(self) -> None: | ||
| if self.caps.get("webSocketUrl"): | ||
| ws_url = self.caps.get("webSocketUrl") | ||
| else: | ||
| raise WebDriverException("Unable to find url to connect to from capabilities") | ||
| raise WebDriverException( | ||
| "Unable to find url to connect to from capabilities" | ||
| ) | ||
|
|
||
| if not isinstance(self.command_executor, RemoteConnection): | ||
| raise WebDriverException("command_executor must be a RemoteConnection instance for BiDi support") | ||
| raise WebDriverException( | ||
| "command_executor must be a RemoteConnection instance for BiDi support" | ||
| ) | ||
|
|
||
| self._websocket_connection = WebSocketConnection( | ||
| ws_url, | ||
| self.command_executor.client_config.websocket_timeout, | ||
| self.command_executor.client_config.websocket_interval, | ||
| ) | ||
|
|
||
| @property | ||
| def network(self) -> Network: | ||
| if not self._websocket_connection: | ||
| self._start_bidi() | ||
|
|
||
| assert self._websocket_connection is not None | ||
| if not hasattr(self, "_network") or self._network is None: | ||
| assert self._websocket_connection is not None | ||
| self._network = Network(self._websocket_connection) | ||
|
|
||
| return self._network |
There was a problem hiding this comment.
The BiDi module properties still instantiate modules with the old constructor signatures (e.g., Script(self._websocket_connection, self)), but the updated/generated BiDi modules now take a single driver argument. This will raise TypeError when accessing driver.script/driver.network/driver.browser/etc. Either keep the BiDi module constructors compatible with WebSocketConnection, or update these property implementations to pass the expected driver and have the modules call driver._websocket_connection.execute(...) internally.
| @dataclass | ||
| class setNetworkConditionsParameters: | ||
| """setNetworkConditionsParameters type type.""" | ||
|
|
||
| network_conditions: Any | None = None | ||
| contexts: list[Any | None] | None = field(default_factory=list) | ||
| user_contexts: list[Any | None] | None = field(default_factory=list) |
There was a problem hiding this comment.
Generated dataclass/type names like setNetworkConditionsParameters start with a lowercase letter, which is inconsistent with Python class naming and the surrounding generated types. This is likely a generator bug in the CDDL-to-Python name mapping; please ensure all generated class/type names use PascalCase (e.g., SetNetworkConditionsParameters).
| import argparse | ||
| import importlib.util | ||
| import logging | ||
| import re | ||
| import sys | ||
| from collections import defaultdict | ||
| from dataclasses import dataclass, field | ||
| from enum import Enum | ||
| from pathlib import Path | ||
| from textwrap import dedent, indent as tw_indent | ||
| from typing import Any, Dict, List, Optional, Set, Tuple | ||
|
|
There was a problem hiding this comment.
New code in the generator uses Optional[...]/Union[...] imports and annotations (e.g., Optional[str]), but the repo’s conventions in py/AGENTS.md prefer X | None and avoiding Optional. Please update the generator to follow the same typing style to keep generated code and tooling consistent.
| def execute( | ||
| self, driver_command: str, params: dict[str, Any] | None = None | ||
| ) -> dict[str, Any]: | ||
| """Sends a command to be executed by a command.CommandExecutor. | ||
|
|
||
| Args: | ||
| driver_command: The name of the command to execute as a string. | ||
| driver_command: The name of the command to execute as a string. Can also be a generator | ||
| for BiDi protocol commands. | ||
| params: A dictionary of named parameters to send with the command. | ||
|
|
||
| Returns: | ||
| The command's JSON response loaded into a dictionary object. | ||
| """ | ||
| # Handle BiDi generator commands | ||
| if inspect.isgenerator(driver_command): | ||
| # BiDi command: use WebSocketConnection directly | ||
| return self.command_executor.execute(driver_command) | ||
|
|
There was a problem hiding this comment.
execute() now treats a generator driver_command as a BiDi command and calls self.command_executor.execute(driver_command). In this class, command_executor is typically a RemoteConnection whose execute() requires (command, params), so this will raise TypeError at runtime. BiDi commands should be routed through self._websocket_connection.execute(...) (after ensuring _start_bidi() has initialized it), and the execute() signature/type should be updated accordingly (e.g., accept Generator[dict, dict, dict] in addition to str).
8ea2d91 to
1fed8ae
Compare
| class setNetworkConditionsParameters: | ||
| """setNetworkConditionsParameters.""" |
There was a problem hiding this comment.
The class name setNetworkConditionsParameters violates Python naming conventions. Class names should use PascalCase, so this should be SetNetworkConditionsParameters.
| class setNetworkConditionsParameters: | |
| """setNetworkConditionsParameters.""" | |
| class SetNetworkConditionsParameters: | |
| """SetNetworkConditionsParameters.""" |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 34 out of 35 changed files in this pull request and generated 4 comments.
Comments suppressed due to low confidence (1)
py/selenium/webdriver/remote/webdriver.py:1107
start_devtools()/bidi_connection()callimport_cdp()which importsselenium.webdriver.common.bidi.cdp, but this PR deletespy/selenium/webdriver/common/bidi/cdp.py. That will cause aModuleNotFoundErrorthe first time devtools/BiDi connection code runs. Either keep/replace thecdpmodule (and updateimport_cdp()accordingly) or remove these code paths.
| def get_cookies(self, filter=None, partition=None): | ||
| """Execute storage.getCookies and return a GetCookiesResult.""" | ||
| if filter and hasattr(filter, "to_bidi_dict"): | ||
| filter = filter.to_bidi_dict() | ||
| if partition and hasattr(partition, "to_bidi_dict"): | ||
| partition = partition.to_bidi_dict() |
There was a problem hiding this comment.
Storage.get_cookies/set_cookie/delete_cookies are each defined twice in the same class; the later definitions overwrite the earlier ones. This should be consolidated to a single implementation per method (and ideally keep the type-annotated signatures).
769bc91 to
3ee0804
Compare
| """Test ClientWindowNamedState constants.""" | ||
| assert ClientWindowNamedState.MAXIMIZED == "maximized" | ||
| assert ClientWindowNamedState.MINIMIZED == "minimized" |
There was a problem hiding this comment.
The test removed the assertion for ClientWindowState.FULLSCREEN and ClientWindowState.NORMAL. Since the class was renamed to ClientWindowNamedState, these constants should still be tested if they still exist in the new class. The generated class ClientWindowNamedState only defines FULLSCREEN, MAXIMIZED, and MINIMIZED but not NORMAL. If NORMAL is a valid window state in the BiDi spec, it should be added to the generated class definition.
| type: str = field(default="none", init=False) | ||
| id: str | None = None | ||
| actions: list[Any | None] | None = field(default_factory=list) |
There was a problem hiding this comment.
The dataclass fields use field(default="...", init=False) pattern for discriminator fields. This should use field(default_factory=lambda: "...") when the default is mutable, though strings are immutable so this is acceptable. However, be aware that init=False means these fields cannot be set during initialization, which may not be the intended behavior for all use cases.
| def default(self, o): | ||
| if dataclasses.is_dataclass(o) and not isinstance(o, type): | ||
| result = {} | ||
| for f in dataclasses.fields(o): | ||
| value = getattr(o, f.name) | ||
| if value is None: | ||
| continue | ||
| camel_key = _snake_to_camel(f.name) | ||
| # Flatten PointerCommonProperties fields inline into the parent | ||
| if camel_key == "properties" and dataclasses.is_dataclass(value): | ||
| for pf in dataclasses.fields(value): | ||
| pv = getattr(value, pf.name) | ||
| if pv is not None: | ||
| result[_snake_to_camel(pf.name)] = pv | ||
| else: | ||
| result[camel_key] = value | ||
| return result | ||
| return super().default(o) |
There was a problem hiding this comment.
The _BiDiEncoder.default() method recursively encodes dataclass values, but it doesn't handle the case where nested values might also need encoding (e.g., lists of dataclasses, dicts containing dataclasses). The encoder should recursively process list and dict values to ensure all nested dataclasses are properly converted.
| def command_builder( | ||
| method: str, params: dict[str, Any] | ||
| ) -> Generator[dict[str, Any], Any, Any]: | ||
| """Build a BiDi command generator. | ||
|
|
||
| Args: | ||
| method: The method to execute. | ||
| params: The parameters to pass to the method. Default is None. | ||
| method: The BiDi method name (e.g., "session.status", "browser.close") | ||
| params: The parameters for the command | ||
|
|
||
| Yields: | ||
| A dictionary representing the BiDi command | ||
|
|
||
| Returns: | ||
| The response from the command execution. | ||
| The result from the BiDi command execution | ||
| """ | ||
| if params is None: | ||
| params = {} | ||
|
|
||
| command = {"method": method, "params": params} | ||
| cmd = yield command | ||
| return cmd | ||
| result = yield {"method": method, "params": params} | ||
| return result |
There was a problem hiding this comment.
The command_builder function signature changed from params: dict | None = None to params: dict[str, Any] (required). This is a breaking API change that may affect existing code that calls command_builder(method) without params. The old behavior allowed None as default, which was converted to an empty dict. Consider making params optional with a default of empty dict: params: dict[str, Any] | None = None and adding if params is None: params = {} in the function body to maintain backward compatibility.
| class _EventWrapper: | ||
| """Wrapper to provide event_class attribute for WebSocketConnection callbacks.""" | ||
| def __init__(self, bidi_event: str, event_class: type): | ||
| self.event_class = bidi_event # WebSocket expects the BiDi event name as event_class | ||
| self._python_class = event_class # Keep reference to Python dataclass for deserialization | ||
|
|
||
| def from_json(self, params: dict) -> Any: | ||
| """Deserialize event params into the wrapped Python dataclass. | ||
|
|
||
| Args: | ||
| params: Raw BiDi event params with camelCase keys. | ||
|
|
||
| Returns: | ||
| An instance of the dataclass, or the raw dict on failure. | ||
| """ | ||
| if self._python_class is None or self._python_class is dict: | ||
| return params | ||
| try: | ||
| # Delegate to a classmethod from_json if the class defines one | ||
| if hasattr(self._python_class, "from_json") and callable( | ||
| self._python_class.from_json | ||
| ): | ||
| return self._python_class.from_json(params) | ||
| import dataclasses as dc | ||
|
|
||
| snake_params = {self._camel_to_snake(k): v for k, v in params.items()} | ||
| if dc.is_dataclass(self._python_class): | ||
| valid_fields = {f.name for f in dc.fields(self._python_class)} | ||
| filtered = {k: v for k, v in snake_params.items() if k in valid_fields} | ||
| return self._python_class(**filtered) | ||
| return self._python_class(**snake_params) | ||
| except Exception: | ||
| return params | ||
|
|
||
| @staticmethod | ||
| def _camel_to_snake(name: str) -> str: | ||
| result = [name[0].lower()] | ||
| for char in name[1:]: | ||
| if char.isupper(): | ||
| result.extend(["_", char.lower()]) | ||
| else: | ||
| result.append(char) | ||
| return "".join(result) | ||
|
|
||
|
|
||
| class _EventManager: | ||
| """Manages event subscriptions and callbacks.""" | ||
|
|
||
| def __init__(self, conn, event_configs: dict[str, EventConfig]): | ||
| self.conn = conn | ||
| self.event_configs = event_configs | ||
| self.subscriptions: dict = {} | ||
| self._event_wrappers = {} # Cache of _EventWrapper objects | ||
| self._bidi_to_class = {config.bidi_event: config.event_class for config in event_configs.values()} | ||
| self._available_events = ", ".join(sorted(event_configs.keys())) | ||
| self._subscription_lock = threading.Lock() | ||
|
|
||
| # Create event wrappers for each event | ||
| for config in event_configs.values(): | ||
| wrapper = _EventWrapper(config.bidi_event, config.event_class) | ||
| self._event_wrappers[config.bidi_event] = wrapper | ||
|
|
||
| def validate_event(self, event: str) -> EventConfig: | ||
| event_config = self.event_configs.get(event) | ||
| if not event_config: | ||
| raise ValueError(f"Event '{event}' not found. Available events: {self._available_events}") | ||
| return event_config | ||
|
|
||
| def subscribe_to_event(self, bidi_event: str, contexts: list[str] | None = None) -> None: | ||
| """Subscribe to a BiDi event if not already subscribed.""" | ||
| with self._subscription_lock: | ||
| if bidi_event not in self.subscriptions: | ||
| session = Session(self.conn) | ||
| result = session.subscribe([bidi_event], contexts=contexts) | ||
| sub_id = ( | ||
| result.get("subscription") if isinstance(result, dict) else None | ||
| ) | ||
| self.subscriptions[bidi_event] = { | ||
| "callbacks": [], | ||
| "subscription_id": sub_id, | ||
| } | ||
|
|
||
| def unsubscribe_from_event(self, bidi_event: str) -> None: | ||
| """Unsubscribe from a BiDi event if no more callbacks exist.""" | ||
| with self._subscription_lock: | ||
| entry = self.subscriptions.get(bidi_event) | ||
| if entry is not None and not entry["callbacks"]: | ||
| session = Session(self.conn) | ||
| sub_id = entry.get("subscription_id") | ||
| if sub_id: | ||
| session.unsubscribe(subscriptions=[sub_id]) | ||
| else: | ||
| session.unsubscribe(events=[bidi_event]) | ||
| del self.subscriptions[bidi_event] | ||
|
|
||
| def add_callback_to_tracking(self, bidi_event: str, callback_id: int) -> None: | ||
| with self._subscription_lock: | ||
| self.subscriptions[bidi_event]["callbacks"].append(callback_id) | ||
|
|
||
| def remove_callback_from_tracking(self, bidi_event: str, callback_id: int) -> None: | ||
| with self._subscription_lock: | ||
| entry = self.subscriptions.get(bidi_event) | ||
| if entry and callback_id in entry["callbacks"]: | ||
| entry["callbacks"].remove(callback_id) | ||
|
|
||
| def add_event_handler(self, event: str, callback: Callable, contexts: list[str] | None = None) -> int: | ||
| event_config = self.validate_event(event) | ||
| # Use the event wrapper for add_callback | ||
| event_wrapper = self._event_wrappers.get(event_config.bidi_event) | ||
| callback_id = self.conn.add_callback(event_wrapper, callback) | ||
| self.subscribe_to_event(event_config.bidi_event, contexts) | ||
| self.add_callback_to_tracking(event_config.bidi_event, callback_id) | ||
| return callback_id | ||
|
|
||
| def remove_event_handler(self, event: str, callback_id: int) -> None: | ||
| event_config = self.validate_event(event) | ||
| event_wrapper = self._event_wrappers.get(event_config.bidi_event) | ||
| self.conn.remove_callback(event_wrapper, callback_id) | ||
| self.remove_callback_from_tracking(event_config.bidi_event, callback_id) | ||
| self.unsubscribe_from_event(event_config.bidi_event) | ||
|
|
||
| def clear_event_handlers(self) -> None: | ||
| """Clear all event handlers.""" | ||
| with self._subscription_lock: | ||
| if not self.subscriptions: | ||
| return | ||
| session = Session(self.conn) | ||
| for bidi_event, entry in list(self.subscriptions.items()): | ||
| event_wrapper = self._event_wrappers.get(bidi_event) | ||
| callbacks = entry["callbacks"] if isinstance(entry, dict) else entry | ||
| if event_wrapper: | ||
| for callback_id in callbacks: | ||
| self.conn.remove_callback(event_wrapper, callback_id) | ||
| sub_id = ( | ||
| entry.get("subscription_id") if isinstance(entry, dict) else None | ||
| ) | ||
| if sub_id: | ||
| session.unsubscribe(subscriptions=[sub_id]) | ||
| else: | ||
| session.unsubscribe(events=[bidi_event]) | ||
| self.subscriptions.clear() | ||
|
|
There was a problem hiding this comment.
The _EventManager and _EventWrapper classes are duplicated across multiple generated modules (log.py, network.py, input.py). This is significant code duplication (~200+ lines per module). These classes should be extracted to a shared module (e.g., common.py or a new event_manager.py) and imported by the generated modules to follow the DRY principle and reduce maintenance burden.
…le (#17183) * make the signature change in `ExecuteMethod` backward compatible ... to avoid too many changes e.g. in Appium project (its has own implementation `AppiumExecutionMethod`). In PR #17152, I changed signature of `ExecuteMethod.execute()` - and later realized that it breaks backward compatibility in Appium. * rename the additional helper methods `ExecuteMethod` * rename the additional helper methods `ExecuteMethod`
Update pinned browser versions Co-authored-by: Selenium CI Bot <selenium-ci@users.noreply.github.com>
Update pinned browser versions Co-authored-by: Selenium CI Bot <selenium-ci@users.noreply.github.com>
Update pinned browser versions Co-authored-by: Selenium CI Bot <selenium-ci@users.noreply.github.com>
Update pinned browser versions Co-authored-by: Selenium CI Bot <selenium-ci@users.noreply.github.com>
…, high-latency proxying (#17197) Signed-off-by: Viet Nguyen Duc <nguyenducviet4496@gmail.com>
Signed-off-by: Viet Nguyen Duc <nguyenducviet4496@gmail.com>
dotnet/src/webdriver/assets/nuget/buildTransitive/Selenium.WebDriver.props
Show resolved
Hide resolved
| def test_client_window_state_constants(driver): | ||
| assert ClientWindowState.FULLSCREEN == "fullscreen" | ||
| assert ClientWindowState.MAXIMIZED == "maximized" | ||
| assert ClientWindowState.MINIMIZED == "minimized" | ||
| assert ClientWindowState.NORMAL == "normal" | ||
| """Test ClientWindowNamedState constants.""" | ||
| assert ClientWindowNamedState.MAXIMIZED == "maximized" | ||
| assert ClientWindowNamedState.MINIMIZED == "minimized" |
There was a problem hiding this comment.
ClientWindowNamedState also defines FULLSCREEN and NORMAL, but this test now only asserts MAXIMIZED/MINIMIZED. Please add assertions for the remaining constants so regressions are caught.
| def remove_user_context(self, user_context: Any | None = None): | ||
| """Execute browser.removeUserContext.""" | ||
| if user_context is None: | ||
| raise TypeError("remove_user_context() missing required argument: {{snake_param!r}}") |
There was a problem hiding this comment.
This TypeError message contains an unexpanded template placeholder ({{snake_param!r}}), which will be exposed to users. The generator should emit the actual missing parameter name (or drop the manual check and let Python’s own missing-arg TypeError fire).
| raise TypeError("remove_user_context() missing required argument: {{snake_param!r}}") | |
| raise TypeError("remove_user_context() missing required argument: 'user_context'") |
| """Execute session.subscribe.""" | ||
| if events is None: | ||
| raise TypeError("subscribe() missing required argument: {{snake_param!r}}") | ||
|
|
There was a problem hiding this comment.
This TypeError message contains an unexpanded template placeholder ({{snake_param!r}}). The generator should substitute the actual parameter name (or avoid generating this manual required-arg check).
| def to_bidi_dict(self) -> dict: | ||
| """Serialize to the BiDi wire-protocol dict.""" | ||
| result: dict = {} | ||
| if self.name is not None: | ||
| result["name"] = self.name | ||
| if self.value is not None: | ||
| result["value"] = self.value.to_dict() | ||
| result["value"] = self.value.to_bidi_dict() if hasattr(self.value, "to_bidi_dict") else self.value |
There was a problem hiding this comment.
CookieFilter/PartialCookie/BytesValue switched to to_bidi_dict(); if callers were using the previous to_dict() method, this is a breaking change. Consider keeping to_dict() as a deprecated alias to preserve backward compatibility.
| def execute( | ||
| self, driver_command: str, params: dict[str, Any] | None = None | ||
| ) -> dict[str, Any]: | ||
| """Sends a command to be executed by a command.CommandExecutor. | ||
|
|
||
| Args: | ||
| driver_command: The name of the command to execute as a string. | ||
| driver_command: The name of the command to execute as a string. Can also be a generator | ||
| for BiDi protocol commands. | ||
| params: A dictionary of named parameters to send with the command. | ||
|
|
||
| Returns: | ||
| The command's JSON response loaded into a dictionary object. | ||
| """ | ||
| # Handle BiDi generator commands | ||
| if inspect.isgenerator(driver_command): | ||
| # BiDi command: route through the WebSocket connection, not the | ||
| # HTTP RemoteConnection which only accepts (command, params) pairs. | ||
| if not self._websocket_connection: | ||
| self._start_bidi() | ||
| assert self._websocket_connection is not None | ||
| return self._websocket_connection.execute(driver_command) |
There was a problem hiding this comment.
execute() now supports BiDi generator commands, but the type annotation still declares driver_command as str. Please update the signature (and docstring) to accept a generator type as well, so type checkers/IDEs match the actual supported API.
| @dataclass | ||
| class EventConfig: | ||
| """Configuration for a BiDi event.""" | ||
| event_key: str | ||
| bidi_event: str | ||
| event_class: type | ||
|
|
There was a problem hiding this comment.
This module duplicates the EventConfig/_EventWrapper/_EventManager implementation even though the PR adds a shared selenium.webdriver.common.bidi._event_manager module. Consider importing and reusing the shared helpers to avoid copying ~160 LOC into every generated module.
| def new(self, capabilities: Any | None = None): | ||
| """Execute session.new.""" | ||
| if capabilities is None: | ||
| raise TypeError("new() missing required argument: {{snake_param!r}}") | ||
|
|
There was a problem hiding this comment.
This TypeError message contains an unexpanded template placeholder ({{snake_param!r}}). The generator should substitute the actual parameter name (or avoid generating this manual required-arg check).
| if self.prompt is not None: | ||
| result["prompt"] = self.prompt | ||
| return result | ||
|
|
There was a problem hiding this comment.
Renaming UserPromptHandler.to_dict() to to_bidi_dict() is a breaking change for existing callers. Consider keeping to_dict as a backward-compatible (possibly deprecated) alias that forwards to to_bidi_dict().
| def to_dict(self) -> dict: | |
| """Backward-compatible alias for to_bidi_dict().""" | |
| return self.to_bidi_dict() |
| class disownDataParameters: | ||
| """disownDataParameters.""" |
There was a problem hiding this comment.
Class name disownDataParameters is not PEP8 CapWords and is inconsistent with the other generated *Parameters classes. Please fix the generator naming so this becomes something like DisownDataParameters.
| class disownDataParameters: | |
| """disownDataParameters.""" | |
| class DisownDataParameters: | |
| """DisownDataParameters.""" |
| # DO NOT EDIT THIS FILE! | ||
| # | ||
| # http://www.apache.org/licenses/LICENSE-2.0 | ||
| # This file is generated from the WebDriver BiDi specification. If you need to make | ||
| # changes, edit the generator and regenerate all of the modules. |
There was a problem hiding this comment.
The generated BiDi module header removed the standard Selenium Apache 2.0 / SFC license header and replaced it with a short “DO NOT EDIT” banner. Most Python sources in this repo keep the full header (e.g., py/selenium/webdriver/common/action_chains.py:1-16). Please ensure the generator preserves the license header (you can keep the “DO NOT EDIT” notice below it if desired).
shbenzer
left a comment
There was a problem hiding this comment.
This is genuinely impressive - awesome work @AutomatedTester!
| class setNetworkConditionsParameters: | ||
| """setNetworkConditionsParameters.""" |
| def subscribe( | ||
| self, | ||
| events: list[Any] | None = None, | ||
| contexts: list[Any] | None = None, | ||
| user_contexts: list[Any] | None = None, | ||
| ): | ||
| """Execute session.subscribe.""" | ||
| if events is None: | ||
| raise TypeError("subscribe() missing required argument: {{snake_param!r}}") | ||
|
|
||
| Returns information about the remote end's readiness to create new sessions | ||
| and may include implementation-specific metadata. | ||
| params = { | ||
| "events": events, | ||
| "contexts": contexts, | ||
| "userContexts": user_contexts, | ||
| } | ||
| params = {k: v for k, v in params.items() if v is not None} | ||
| cmd = command_builder("session.subscribe", params) | ||
| result = self._conn.execute(cmd) | ||
| return result |
| def uninstall(self, extension: str | dict): | ||
| """Uninstall a web extension. | ||
|
|
| def command_builder( | ||
| method: str, params: dict[str, Any] | ||
| ) -> Generator[dict[str, Any], Any, Any]: | ||
| """Build a BiDi command generator. | ||
| Args: | ||
| method: The method to execute. | ||
| params: The parameters to pass to the method. Default is None. | ||
| method: The BiDi method name (e.g., "session.status", "browser.close") | ||
| params: The parameters for the command | ||
| Yields: | ||
| A dictionary representing the BiDi command | ||
| Returns: | ||
| The response from the command execution. | ||
| The result from the BiDi command execution | ||
| """ | ||
| if params is None: | ||
| params = {} | ||
|
|
||
| command = {"method": method, "params": params} | ||
| cmd = yield command | ||
| return cmd | ||
| result = yield {"method": method, "params": params} | ||
| return result |
| def new(self, capabilities: Any | None = None): | ||
| """Execute session.new.""" | ||
| if capabilities is None: | ||
| raise TypeError("new() missing required argument: {{snake_param!r}}") |
| @dataclass | ||
| class EventConfig: | ||
| """Configuration for a BiDi event.""" | ||
| event_key: str | ||
| bidi_event: str | ||
| event_class: type | ||
|
|
||
|
|
||
| class _EventWrapper: | ||
| """Wrapper to provide event_class attribute for WebSocketConnection callbacks.""" | ||
| def __init__(self, bidi_event: str, event_class: type): | ||
| self.event_class = bidi_event # WebSocket expects the BiDi event name as event_class | ||
| self._python_class = event_class # Keep reference to Python dataclass for deserialization | ||
|
|
||
| def from_json(self, params: dict) -> Any: | ||
| """Deserialize event params into the wrapped Python dataclass. | ||
|
|
||
| Args: | ||
| params: Raw BiDi event params with camelCase keys. | ||
|
|
||
| Returns: | ||
| An instance of the dataclass, or the raw dict on failure. | ||
| """ | ||
| if self._python_class is None or self._python_class is dict: | ||
| return params | ||
| try: | ||
| # Delegate to a classmethod from_json if the class defines one | ||
| if hasattr(self._python_class, "from_json") and callable( | ||
| self._python_class.from_json | ||
| ): | ||
| return self._python_class.from_json(params) | ||
| import dataclasses as dc | ||
|
|
||
| snake_params = {self._camel_to_snake(k): v for k, v in params.items()} | ||
| if dc.is_dataclass(self._python_class): | ||
| valid_fields = {f.name for f in dc.fields(self._python_class)} | ||
| filtered = {k: v for k, v in snake_params.items() if k in valid_fields} | ||
| return self._python_class(**filtered) | ||
| return self._python_class(**snake_params) | ||
| except Exception: | ||
| return params | ||
|
|
||
| @staticmethod | ||
| def _camel_to_snake(name: str) -> str: | ||
| result = [name[0].lower()] | ||
| for char in name[1:]: | ||
| if char.isupper(): | ||
| result.extend(["_", char.lower()]) | ||
| else: | ||
| result.append(char) | ||
| return "".join(result) | ||
|
|
||
|
|
||
| class _EventManager: | ||
| """Manages event subscriptions and callbacks.""" | ||
|
|
||
| def __init__(self, conn, event_configs: dict[str, EventConfig]): | ||
| self.conn = conn | ||
| self.event_configs = event_configs | ||
| self.subscriptions: dict = {} | ||
| self._event_wrappers = {} # Cache of _EventWrapper objects | ||
| self._bidi_to_class = {config.bidi_event: config.event_class for config in event_configs.values()} | ||
| self._available_events = ", ".join(sorted(event_configs.keys())) | ||
| self._subscription_lock = threading.Lock() | ||
|
|
||
| # Create event wrappers for each event | ||
| for config in event_configs.values(): | ||
| wrapper = _EventWrapper(config.bidi_event, config.event_class) | ||
| self._event_wrappers[config.bidi_event] = wrapper | ||
|
|
||
| def validate_event(self, event: str) -> EventConfig: | ||
| event_config = self.event_configs.get(event) | ||
| if not event_config: | ||
| raise ValueError(f"Event '{event}' not found. Available events: {self._available_events}") | ||
| return event_config | ||
|
|
||
| def subscribe_to_event(self, bidi_event: str, contexts: list[str] | None = None) -> None: | ||
| """Subscribe to a BiDi event if not already subscribed.""" | ||
| with self._subscription_lock: | ||
| if bidi_event not in self.subscriptions: | ||
| session = Session(self.conn) | ||
| result = session.subscribe([bidi_event], contexts=contexts) | ||
| sub_id = ( | ||
| result.get("subscription") if isinstance(result, dict) else None | ||
| ) | ||
| self.subscriptions[bidi_event] = { | ||
| "callbacks": [], | ||
| "subscription_id": sub_id, | ||
| } | ||
|
|
||
| def unsubscribe_from_event(self, bidi_event: str) -> None: | ||
| """Unsubscribe from a BiDi event if no more callbacks exist.""" | ||
| with self._subscription_lock: | ||
| entry = self.subscriptions.get(bidi_event) | ||
| if entry is not None and not entry["callbacks"]: | ||
| session = Session(self.conn) | ||
| sub_id = entry.get("subscription_id") | ||
| if sub_id: | ||
| session.unsubscribe(subscriptions=[sub_id]) | ||
| else: | ||
| session.unsubscribe(events=[bidi_event]) | ||
| del self.subscriptions[bidi_event] | ||
|
|
||
| def add_callback_to_tracking(self, bidi_event: str, callback_id: int) -> None: | ||
| with self._subscription_lock: | ||
| self.subscriptions[bidi_event]["callbacks"].append(callback_id) | ||
|
|
||
| def remove_callback_from_tracking(self, bidi_event: str, callback_id: int) -> None: | ||
| with self._subscription_lock: | ||
| entry = self.subscriptions.get(bidi_event) | ||
| if entry and callback_id in entry["callbacks"]: | ||
| entry["callbacks"].remove(callback_id) | ||
|
|
||
| def add_event_handler(self, event: str, callback: Callable, contexts: list[str] | None = None) -> int: | ||
| event_config = self.validate_event(event) | ||
| # Use the event wrapper for add_callback | ||
| event_wrapper = self._event_wrappers.get(event_config.bidi_event) | ||
| callback_id = self.conn.add_callback(event_wrapper, callback) | ||
| self.subscribe_to_event(event_config.bidi_event, contexts) | ||
| self.add_callback_to_tracking(event_config.bidi_event, callback_id) | ||
| return callback_id | ||
|
|
||
| def remove_event_handler(self, event: str, callback_id: int) -> None: | ||
| event_config = self.validate_event(event) | ||
| event_wrapper = self._event_wrappers.get(event_config.bidi_event) | ||
| self.conn.remove_callback(event_wrapper, callback_id) | ||
| self.remove_callback_from_tracking(event_config.bidi_event, callback_id) | ||
| self.unsubscribe_from_event(event_config.bidi_event) | ||
|
|
||
| def clear_event_handlers(self) -> None: | ||
| """Clear all event handlers.""" | ||
| with self._subscription_lock: | ||
| if not self.subscriptions: | ||
| return | ||
| session = Session(self.conn) | ||
| for bidi_event, entry in list(self.subscriptions.items()): | ||
| event_wrapper = self._event_wrappers.get(bidi_event) | ||
| callbacks = entry["callbacks"] if isinstance(entry, dict) else entry | ||
| if event_wrapper: | ||
| for callback_id in callbacks: | ||
| self.conn.remove_callback(event_wrapper, callback_id) | ||
| sub_id = ( | ||
| entry.get("subscription_id") if isinstance(entry, dict) else None | ||
| ) | ||
| if sub_id: | ||
| session.unsubscribe(subscriptions=[sub_id]) | ||
| else: | ||
| session.unsubscribe(events=[bidi_event]) | ||
| self.subscriptions.clear() | ||
|
|
||
|
|
||
|
|
||
|
|
||
| class Log: | ||
| """WebDriver BiDi log module.""" | ||
|
|
||
| EVENT_CONFIGS: dict[str, EventConfig] = {} | ||
| def __init__(self, conn) -> None: | ||
| self._conn = conn | ||
| self._event_manager = _EventManager(conn, self.EVENT_CONFIGS) |
| class disownDataParameters: | ||
| """disownDataParameters.""" |
This generates bidi code based off of the CDDL that we can update from the specification. I expect over time the generation will need other features added.