# Licensed to the Software Freedom Conservancy (SFC) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The SFC licenses this file
# to you 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 typing import Dict, List, Optional, Union
from selenium.webdriver.common.bidi.common import command_builder
from .session import Session
[docs]
class ReadinessState:
"""Represents the stage of document loading at which a navigation command will return."""
NONE = "none"
INTERACTIVE = "interactive"
COMPLETE = "complete"
[docs]
class UserPromptType:
"""Represents the possible user prompt types."""
ALERT = "alert"
BEFORE_UNLOAD = "beforeunload"
CONFIRM = "confirm"
PROMPT = "prompt"
[docs]
class NavigationInfo:
"""Provides details of an ongoing navigation."""
def __init__(
self,
context: str,
navigation: Optional[str],
timestamp: int,
url: str,
):
self.context = context
self.navigation = navigation
self.timestamp = timestamp
self.url = url
[docs]
@classmethod
def from_json(cls, json: Dict) -> "NavigationInfo":
"""Creates a NavigationInfo instance from a dictionary.
Parameters:
-----------
json: A dictionary containing the navigation information.
Returns:
-------
NavigationInfo: A new instance of NavigationInfo.
"""
return cls(
context=json.get("context"),
navigation=json.get("navigation"),
timestamp=json.get("timestamp"),
url=json.get("url"),
)
[docs]
class BrowsingContextInfo:
"""Represents the properties of a navigable."""
def __init__(
self,
context: str,
url: str,
children: Optional[List["BrowsingContextInfo"]],
parent: Optional[str] = None,
user_context: Optional[str] = None,
original_opener: Optional[str] = None,
client_window: Optional[str] = None,
):
self.context = context
self.url = url
self.children = children
self.parent = parent
self.user_context = user_context
self.original_opener = original_opener
self.client_window = client_window
[docs]
@classmethod
def from_json(cls, json: Dict) -> "BrowsingContextInfo":
"""Creates a BrowsingContextInfo instance from a dictionary.
Parameters:
-----------
json: A dictionary containing the browsing context information.
Returns:
-------
BrowsingContextInfo: A new instance of BrowsingContextInfo.
"""
children = None
if json.get("children") is not None:
children = [BrowsingContextInfo.from_json(child) for child in json.get("children")]
return cls(
context=json.get("context"),
url=json.get("url"),
children=children,
parent=json.get("parent"),
user_context=json.get("userContext"),
original_opener=json.get("originalOpener"),
client_window=json.get("clientWindow"),
)
[docs]
class DownloadWillBeginParams(NavigationInfo):
"""Parameters for the downloadWillBegin event."""
def __init__(
self,
context: str,
navigation: Optional[str],
timestamp: int,
url: str,
suggested_filename: str,
):
super().__init__(context, navigation, timestamp, url)
self.suggested_filename = suggested_filename
[docs]
@classmethod
def from_json(cls, json: Dict) -> "DownloadWillBeginParams":
"""Creates a DownloadWillBeginParams instance from a dictionary.
Parameters:
-----------
json: A dictionary containing the download parameters.
Returns:
-------
DownloadWillBeginParams: A new instance of DownloadWillBeginParams.
"""
return cls(
context=json.get("context"),
navigation=json.get("navigation"),
timestamp=json.get("timestamp"),
url=json.get("url"),
suggested_filename=json.get("suggestedFilename"),
)
[docs]
class UserPromptOpenedParams:
"""Parameters for the userPromptOpened event."""
def __init__(
self,
context: str,
handler: str,
message: str,
type: str,
default_value: Optional[str] = None,
):
self.context = context
self.handler = handler
self.message = message
self.type = type
self.default_value = default_value
[docs]
@classmethod
def from_json(cls, json: Dict) -> "UserPromptOpenedParams":
"""Creates a UserPromptOpenedParams instance from a dictionary.
Parameters:
-----------
json: A dictionary containing the user prompt parameters.
Returns:
-------
UserPromptOpenedParams: A new instance of UserPromptOpenedParams.
"""
return cls(
context=json.get("context"),
handler=json.get("handler"),
message=json.get("message"),
type=json.get("type"),
default_value=json.get("defaultValue"),
)
[docs]
class UserPromptClosedParams:
"""Parameters for the userPromptClosed event."""
def __init__(
self,
context: str,
accepted: bool,
type: str,
user_text: Optional[str] = None,
):
self.context = context
self.accepted = accepted
self.type = type
self.user_text = user_text
[docs]
@classmethod
def from_json(cls, json: Dict) -> "UserPromptClosedParams":
"""Creates a UserPromptClosedParams instance from a dictionary.
Parameters:
-----------
json: A dictionary containing the user prompt closed parameters.
Returns:
-------
UserPromptClosedParams: A new instance of UserPromptClosedParams.
"""
return cls(
context=json.get("context"),
accepted=json.get("accepted"),
type=json.get("type"),
user_text=json.get("userText"),
)
[docs]
class HistoryUpdatedParams:
"""Parameters for the historyUpdated event."""
def __init__(
self,
context: str,
url: str,
):
self.context = context
self.url = url
[docs]
@classmethod
def from_json(cls, json: Dict) -> "HistoryUpdatedParams":
"""Creates a HistoryUpdatedParams instance from a dictionary.
Parameters:
-----------
json: A dictionary containing the history updated parameters.
Returns:
-------
HistoryUpdatedParams: A new instance of HistoryUpdatedParams.
"""
return cls(
context=json.get("context"),
url=json.get("url"),
)
[docs]
class BrowsingContextEvent:
"""Base class for browsing context events."""
def __init__(self, event_class: str, **kwargs):
self.event_class = event_class
self.params = kwargs
[docs]
@classmethod
def from_json(cls, json: Dict) -> "BrowsingContextEvent":
"""Creates a BrowsingContextEvent instance from a dictionary.
Parameters:
-----------
json: A dictionary containing the event information.
Returns:
-------
BrowsingContextEvent: A new instance of BrowsingContextEvent.
"""
return cls(event_class=json.get("event_class"), **json)
[docs]
class BrowsingContext:
"""BiDi implementation of the browsingContext module."""
EVENTS = {
"context_created": "browsingContext.contextCreated",
"context_destroyed": "browsingContext.contextDestroyed",
"dom_content_loaded": "browsingContext.domContentLoaded",
"download_will_begin": "browsingContext.downloadWillBegin",
"fragment_navigated": "browsingContext.fragmentNavigated",
"history_updated": "browsingContext.historyUpdated",
"load": "browsingContext.load",
"navigation_aborted": "browsingContext.navigationAborted",
"navigation_committed": "browsingContext.navigationCommitted",
"navigation_failed": "browsingContext.navigationFailed",
"navigation_started": "browsingContext.navigationStarted",
"user_prompt_closed": "browsingContext.userPromptClosed",
"user_prompt_opened": "browsingContext.userPromptOpened",
}
def __init__(self, conn):
self.conn = conn
self.subscriptions = {}
self.callbacks = {}
[docs]
def activate(self, context: str) -> None:
"""Activates and focuses the given top-level traversable.
Parameters:
-----------
context: The browsing context ID to activate.
Raises:
------
Exception: If the browsing context is not a top-level traversable.
"""
params = {"context": context}
self.conn.execute(command_builder("browsingContext.activate", params))
[docs]
def capture_screenshot(
self,
context: str,
origin: str = "viewport",
format: Optional[Dict] = None,
clip: Optional[Dict] = None,
) -> str:
"""Captures an image of the given navigable, and returns it as a Base64-encoded string.
Parameters:
-----------
context: The browsing context ID to capture.
origin: The origin of the screenshot, either "viewport" or "document".
format: The format of the screenshot.
clip: The clip rectangle of the screenshot.
Returns:
-------
str: The Base64-encoded screenshot.
"""
params = {"context": context, "origin": origin}
if format is not None:
params["format"] = format
if clip is not None:
params["clip"] = clip
result = self.conn.execute(command_builder("browsingContext.captureScreenshot", params))
return result["data"]
[docs]
def close(self, context: str, prompt_unload: bool = False) -> None:
"""Closes a top-level traversable.
Parameters:
-----------
context: The browsing context ID to close.
prompt_unload: Whether to prompt to unload.
Raises:
------
Exception: If the browsing context is not a top-level traversable.
"""
params = {"context": context, "promptUnload": prompt_unload}
self.conn.execute(command_builder("browsingContext.close", params))
[docs]
def create(
self,
type: str,
reference_context: Optional[str] = None,
background: bool = False,
user_context: Optional[str] = None,
) -> str:
"""Creates a new navigable, either in a new tab or in a new window, and returns its navigable id.
Parameters:
-----------
type: The type of the new navigable, either "tab" or "window".
reference_context: The reference browsing context ID.
background: Whether to create the new navigable in the background.
user_context: The user context ID.
Returns:
-------
str: The browsing context ID of the created navigable.
"""
params = {"type": type}
if reference_context is not None:
params["referenceContext"] = reference_context
if background is not None:
params["background"] = background
if user_context is not None:
params["userContext"] = user_context
result = self.conn.execute(command_builder("browsingContext.create", params))
return result["context"]
[docs]
def get_tree(
self,
max_depth: Optional[int] = None,
root: Optional[str] = None,
) -> List[BrowsingContextInfo]:
"""Returns a tree of all descendent navigables including the given parent itself, or all top-level contexts
when no parent is provided.
Parameters:
-----------
max_depth: The maximum depth of the tree.
root: The root browsing context ID.
Returns:
-------
List[BrowsingContextInfo]: A list of browsing context information.
"""
params = {}
if max_depth is not None:
params["maxDepth"] = max_depth
if root is not None:
params["root"] = root
result = self.conn.execute(command_builder("browsingContext.getTree", params))
return [BrowsingContextInfo.from_json(context) for context in result["contexts"]]
[docs]
def handle_user_prompt(
self,
context: str,
accept: Optional[bool] = None,
user_text: Optional[str] = None,
) -> None:
"""Allows closing an open prompt.
Parameters:
-----------
context: The browsing context ID.
accept: Whether to accept the prompt.
user_text: The text to enter in the prompt.
"""
params = {"context": context}
if accept is not None:
params["accept"] = accept
if user_text is not None:
params["userText"] = user_text
self.conn.execute(command_builder("browsingContext.handleUserPrompt", params))
[docs]
def locate_nodes(
self,
context: str,
locator: Dict,
max_node_count: Optional[int] = None,
serialization_options: Optional[Dict] = None,
start_nodes: Optional[List[Dict]] = None,
) -> List[Dict]:
"""Returns a list of all nodes matching the specified locator.
Parameters:
-----------
context: The browsing context ID.
locator: The locator to use.
max_node_count: The maximum number of nodes to return.
serialization_options: The serialization options.
start_nodes: The start nodes.
Returns:
-------
List[Dict]: A list of nodes.
"""
params = {"context": context, "locator": locator}
if max_node_count is not None:
params["maxNodeCount"] = max_node_count
if serialization_options is not None:
params["serializationOptions"] = serialization_options
if start_nodes is not None:
params["startNodes"] = start_nodes
result = self.conn.execute(command_builder("browsingContext.locateNodes", params))
return result["nodes"]
[docs]
def navigate(
self,
context: str,
url: str,
wait: Optional[str] = None,
) -> Dict:
"""Navigates a navigable to the given URL.
Parameters:
-----------
context: The browsing context ID.
url: The URL to navigate to.
wait: The readiness state to wait for.
Returns:
-------
Dict: A dictionary containing the navigation result.
"""
params = {"context": context, "url": url}
if wait is not None:
params["wait"] = wait
result = self.conn.execute(command_builder("browsingContext.navigate", params))
return result
[docs]
def print(
self,
context: str,
background: bool = False,
margin: Optional[Dict] = None,
orientation: str = "portrait",
page: Optional[Dict] = None,
page_ranges: Optional[List[Union[int, str]]] = None,
scale: float = 1.0,
shrink_to_fit: bool = True,
) -> str:
"""Creates a paginated representation of a document, and returns it as a PDF document represented as a
Base64-encoded string.
Parameters:
-----------
context: The browsing context ID.
background: Whether to include the background.
margin: The margin parameters.
orientation: The orientation, either "portrait" or "landscape".
page: The page parameters.
page_ranges: The page ranges.
scale: The scale.
shrink_to_fit: Whether to shrink to fit.
Returns:
-------
str: The Base64-encoded PDF document.
"""
params = {
"context": context,
"background": background,
"orientation": orientation,
"scale": scale,
"shrinkToFit": shrink_to_fit,
}
if margin is not None:
params["margin"] = margin
if page is not None:
params["page"] = page
if page_ranges is not None:
params["pageRanges"] = page_ranges
result = self.conn.execute(command_builder("browsingContext.print", params))
return result["data"]
[docs]
def reload(
self,
context: str,
ignore_cache: Optional[bool] = None,
wait: Optional[str] = None,
) -> Dict:
"""Reloads a navigable.
Parameters:
-----------
context: The browsing context ID.
ignore_cache: Whether to ignore the cache.
wait: The readiness state to wait for.
Returns:
-------
Dict: A dictionary containing the navigation result.
"""
params = {"context": context}
if ignore_cache is not None:
params["ignoreCache"] = ignore_cache
if wait is not None:
params["wait"] = wait
result = self.conn.execute(command_builder("browsingContext.reload", params))
return result
[docs]
def set_viewport(
self,
context: Optional[str] = None,
viewport: Optional[Dict] = None,
device_pixel_ratio: Optional[float] = None,
user_contexts: Optional[List[str]] = None,
) -> None:
"""Modifies specific viewport characteristics on the given top-level traversable.
Parameters:
-----------
context: The browsing context ID.
viewport: The viewport parameters.
device_pixel_ratio: The device pixel ratio.
user_contexts: The user context IDs.
Raises:
------
Exception: If the browsing context is not a top-level traversable.
"""
params = {}
if context is not None:
params["context"] = context
if viewport is not None:
params["viewport"] = viewport
if device_pixel_ratio is not None:
params["devicePixelRatio"] = device_pixel_ratio
if user_contexts is not None:
params["userContexts"] = user_contexts
self.conn.execute(command_builder("browsingContext.setViewport", params))
[docs]
def traverse_history(self, context: str, delta: int) -> Dict:
"""Traverses the history of a given navigable by a delta.
Parameters:
-----------
context: The browsing context ID.
delta: The delta to traverse by.
Returns:
-------
Dict: A dictionary containing the traverse history result.
"""
params = {"context": context, "delta": delta}
result = self.conn.execute(command_builder("browsingContext.traverseHistory", params))
return result
def _on_event(self, event_name: str, callback: callable) -> int:
"""Set a callback function to subscribe to a browsing context event.
Parameters:
----------
event_name: The event to subscribe to.
callback: The callback function to execute on event.
Returns:
-------
int: callback id
"""
event = BrowsingContextEvent(event_name)
def _callback(event_data):
if event_name == self.EVENTS["context_created"] or event_name == self.EVENTS["context_destroyed"]:
info = BrowsingContextInfo.from_json(event_data.params)
callback(info)
elif event_name == self.EVENTS["download_will_begin"]:
params = DownloadWillBeginParams.from_json(event_data.params)
callback(params)
elif event_name == self.EVENTS["user_prompt_opened"]:
params = UserPromptOpenedParams.from_json(event_data.params)
callback(params)
elif event_name == self.EVENTS["user_prompt_closed"]:
params = UserPromptClosedParams.from_json(event_data.params)
callback(params)
elif event_name == self.EVENTS["history_updated"]:
params = HistoryUpdatedParams.from_json(event_data.params)
callback(params)
else:
# For navigation events
info = NavigationInfo.from_json(event_data.params)
callback(info)
callback_id = self.conn.add_callback(event, _callback)
if event_name in self.callbacks:
self.callbacks[event_name].append(callback_id)
else:
self.callbacks[event_name] = [callback_id]
return callback_id
[docs]
def add_event_handler(self, event: str, callback: callable, contexts: Optional[List[str]] = None) -> int:
"""Add an event handler to the browsing context.
Parameters:
----------
event: The event to subscribe to.
callback: The callback function to execute on event.
contexts: The browsing context IDs to subscribe to.
Returns:
-------
int: callback id
"""
try:
event_name = self.EVENTS[event]
except KeyError:
raise Exception(f"Event {event} not found")
callback_id = self._on_event(event_name, callback)
if event_name in self.subscriptions:
self.subscriptions[event_name].append(callback_id)
else:
params = {"events": [event_name]}
if contexts is not None:
params["browsingContexts"] = contexts
session = Session(self.conn)
self.conn.execute(session.subscribe(**params))
self.subscriptions[event_name] = [callback_id]
return callback_id
[docs]
def remove_event_handler(self, event: str, callback_id: int) -> None:
"""Remove an event handler from the browsing context.
Parameters:
----------
event: The event to unsubscribe from.
callback_id: The callback id to remove.
"""
try:
event_name = self.EVENTS[event]
except KeyError:
raise Exception(f"Event {event} not found")
event = BrowsingContextEvent(event_name)
self.conn.remove_callback(event, callback_id)
self.subscriptions[event_name].remove(callback_id)
if len(self.subscriptions[event_name]) == 0:
params = {"events": [event_name]}
session = Session(self.conn)
self.conn.execute(session.unsubscribe(**params))
del self.subscriptions[event_name]
[docs]
def clear_event_handlers(self) -> None:
"""Clear all event handlers from the browsing context."""
for event_name in self.subscriptions:
event = BrowsingContextEvent(event_name)
for callback_id in self.subscriptions[event_name]:
self.conn.remove_callback(event, callback_id)
params = {"events": [event_name]}
session = Session(self.conn)
self.conn.execute(session.unsubscribe(**params))
self.subscriptions = {}