diff --git a/hyperbrowser/client/managers/async_manager/computer_action.py b/hyperbrowser/client/managers/async_manager/computer_action.py index 9929caf6..6998a27b 100644 --- a/hyperbrowser/client/managers/async_manager/computer_action.py +++ b/hyperbrowser/client/managers/async_manager/computer_action.py @@ -12,6 +12,8 @@ PressKeysActionParams, MoveMouseActionParams, ScreenshotActionParams, + CursorPositionActionParams, + ScrollAtCursorActionParams, ScrollActionParams, TypeTextActionParams, Coordinate, @@ -38,6 +40,7 @@ ComputerAction.MOVE_MOUSE.value: MoveMouseActionParams, ComputerAction.PRESS_KEYS.value: PressKeysActionParams, ComputerAction.SCREENSHOT.value: ScreenshotActionParams, + ComputerAction.CURSOR_POSITION.value: CursorPositionActionParams, ComputerAction.SCROLL.value: ScrollActionParams, ComputerAction.TYPE_TEXT.value: TypeTextActionParams, ComputerAction.GET_CLIPBOARD_TEXT.value: GetClipboardTextActionParams, @@ -47,6 +50,9 @@ def _action_param_model(params): + if isinstance(params, ScrollAtCursorActionParams): + return ScrollAtCursorActionParams + for model in _ACTION_PARAM_MODELS.values(): if isinstance(params, model): return model @@ -56,6 +62,8 @@ def _action_param_model(params): if isinstance(action, ComputerAction): action = action.value model = _ACTION_PARAM_MODELS.get(action) + if model is ScrollActionParams and "x" not in params and "y" not in params: + return ScrollAtCursorActionParams if model is not None: return model @@ -71,18 +79,13 @@ async def _execute_request( session: Union[SessionDetail, str], params: Union[ComputerActionParamsDict, ComputerActionParams], ) -> ComputerActionResponse: + payload = dump_request(params, _action_param_model(params), name="params") if isinstance(session, str): session = await self._client.sessions.get(session) if not session.computer_action_endpoint: raise ValueError("Computer action endpoint not available for this session") - payload = dump_request( - params, - _action_param_model(params), - name="params", - ) - response = await self._client.transport.post( session.computer_action_endpoint, data=payload, @@ -97,12 +100,14 @@ async def click( button: ComputerActionMouseButton = "left", num_clicks: int = 1, return_screenshot: bool = False, + keys: Optional[List[str]] = None, ) -> ComputerActionResponse: params = ClickActionParams( x=x, y=y, button=button, num_clicks=num_clicks, + keys=keys, return_screenshot=return_screenshot, ) return await self._execute_request(session, params) @@ -123,6 +128,14 @@ async def screenshot( params = ScreenshotActionParams() return await self._execute_request(session, params) + async def cursor_position( + self, + session: Union[SessionDetail, str], + return_screenshot: bool = False, + ) -> ComputerActionResponse: + params = CursorPositionActionParams(return_screenshot=return_screenshot) + return await self._execute_request(session, params) + async def press_keys( self, session: Union[SessionDetail, str], @@ -169,6 +182,7 @@ async def drag( session: Union[SessionDetail, str], path: List[Union[CoordinateDict, Coordinate]], return_screenshot: bool = False, + keys: Optional[List[str]] = None, ) -> ComputerActionResponse: params = DragActionParams( path=[ @@ -176,6 +190,7 @@ async def drag( for coordinate in path ], return_screenshot=return_screenshot, + keys=keys, ) return await self._execute_request(session, params) @@ -197,12 +212,30 @@ async def scroll( scroll_x: int, scroll_y: int, return_screenshot: bool = False, + keys: Optional[List[str]] = None, ) -> ComputerActionResponse: params = ScrollActionParams( x=x, y=y, scroll_x=scroll_x, scroll_y=scroll_y, + keys=keys, + return_screenshot=return_screenshot, + ) + return await self._execute_request(session, params) + + async def scroll_at_cursor( + self, + session: Union[SessionDetail, str], + scroll_x: int = 0, + scroll_y: int = 0, + return_screenshot: bool = False, + keys: Optional[List[str]] = None, + ) -> ComputerActionResponse: + params = ScrollAtCursorActionParams( + scroll_x=scroll_x, + scroll_y=scroll_y, + keys=keys, return_screenshot=return_screenshot, ) return await self._execute_request(session, params) diff --git a/hyperbrowser/client/managers/sync_manager/computer_action.py b/hyperbrowser/client/managers/sync_manager/computer_action.py index 121a8454..47f69508 100644 --- a/hyperbrowser/client/managers/sync_manager/computer_action.py +++ b/hyperbrowser/client/managers/sync_manager/computer_action.py @@ -12,6 +12,8 @@ PressKeysActionParams, MoveMouseActionParams, ScreenshotActionParams, + CursorPositionActionParams, + ScrollAtCursorActionParams, ScrollActionParams, TypeTextActionParams, Coordinate, @@ -38,6 +40,7 @@ ComputerAction.MOVE_MOUSE.value: MoveMouseActionParams, ComputerAction.PRESS_KEYS.value: PressKeysActionParams, ComputerAction.SCREENSHOT.value: ScreenshotActionParams, + ComputerAction.CURSOR_POSITION.value: CursorPositionActionParams, ComputerAction.SCROLL.value: ScrollActionParams, ComputerAction.TYPE_TEXT.value: TypeTextActionParams, ComputerAction.GET_CLIPBOARD_TEXT.value: GetClipboardTextActionParams, @@ -47,6 +50,9 @@ def _action_param_model(params): + if isinstance(params, ScrollAtCursorActionParams): + return ScrollAtCursorActionParams + for model in _ACTION_PARAM_MODELS.values(): if isinstance(params, model): return model @@ -56,6 +62,8 @@ def _action_param_model(params): if isinstance(action, ComputerAction): action = action.value model = _ACTION_PARAM_MODELS.get(action) + if model is ScrollActionParams and "x" not in params and "y" not in params: + return ScrollAtCursorActionParams if model is not None: return model @@ -71,18 +79,13 @@ def _execute_request( session: Union[SessionDetail, str], params: Union[ComputerActionParamsDict, ComputerActionParams], ) -> ComputerActionResponse: + payload = dump_request(params, _action_param_model(params), name="params") if isinstance(session, str): session = self._client.sessions.get(session) if not session.computer_action_endpoint: raise ValueError("Computer action endpoint not available for this session") - payload = dump_request( - params, - _action_param_model(params), - name="params", - ) - response = self._client.transport.post( session.computer_action_endpoint, data=payload, @@ -97,12 +100,14 @@ def click( button: ComputerActionMouseButton = "left", num_clicks: int = 1, return_screenshot: bool = False, + keys: Optional[List[str]] = None, ) -> ComputerActionResponse: params = ClickActionParams( x=x, y=y, button=button, num_clicks=num_clicks, + keys=keys, return_screenshot=return_screenshot, ) return self._execute_request(session, params) @@ -123,6 +128,14 @@ def screenshot( params = ScreenshotActionParams() return self._execute_request(session, params) + def cursor_position( + self, + session: Union[SessionDetail, str], + return_screenshot: bool = False, + ) -> ComputerActionResponse: + params = CursorPositionActionParams(return_screenshot=return_screenshot) + return self._execute_request(session, params) + def press_keys( self, session: Union[SessionDetail, str], @@ -169,6 +182,7 @@ def drag( session: Union[SessionDetail, str], path: List[Union[CoordinateDict, Coordinate]], return_screenshot: bool = False, + keys: Optional[List[str]] = None, ) -> ComputerActionResponse: params = DragActionParams( path=[ @@ -176,6 +190,7 @@ def drag( for coordinate in path ], return_screenshot=return_screenshot, + keys=keys, ) return self._execute_request(session, params) @@ -197,12 +212,30 @@ def scroll( scroll_x: int, scroll_y: int, return_screenshot: bool = False, + keys: Optional[List[str]] = None, ) -> ComputerActionResponse: params = ScrollActionParams( x=x, y=y, scroll_x=scroll_x, scroll_y=scroll_y, + keys=keys, + return_screenshot=return_screenshot, + ) + return self._execute_request(session, params) + + def scroll_at_cursor( + self, + session: Union[SessionDetail, str], + scroll_x: int = 0, + scroll_y: int = 0, + return_screenshot: bool = False, + keys: Optional[List[str]] = None, + ) -> ComputerActionResponse: + params = ScrollAtCursorActionParams( + scroll_x=scroll_x, + scroll_y=scroll_y, + keys=keys, return_screenshot=return_screenshot, ) return self._execute_request(session, params) diff --git a/hyperbrowser/models/__init__.py b/hyperbrowser/models/__init__.py index e99b6abf..0ef56d88 100644 --- a/hyperbrowser/models/__init__.py +++ b/hyperbrowser/models/__init__.py @@ -274,6 +274,8 @@ MoveMouseActionParams, PressKeysActionParams, ScreenshotActionParams, + CursorPositionActionParams, + ScrollAtCursorActionParams, ScrollActionParams, TypeTextActionParams, ComputerActionMouseButton, @@ -283,6 +285,7 @@ ComputerActionWindow, ComputerActionResponseDataClipboardText, ComputerActionResponseDataListWindows, + ComputerActionResponseDataCursorPosition, ComputerActionResponseData, ) from .session import ( @@ -779,6 +782,8 @@ "MoveMouseActionParams", "PressKeysActionParams", "ScreenshotActionParams", + "CursorPositionActionParams", + "ScrollAtCursorActionParams", "ScrollActionParams", "TypeTextActionParams", "ComputerActionMouseButton", @@ -788,6 +793,7 @@ "ComputerActionWindow", "ComputerActionResponseDataClipboardText", "ComputerActionResponseDataListWindows", + "ComputerActionResponseDataCursorPosition", "ComputerActionResponseData", # web "StartBatchFetchJobParams", diff --git a/hyperbrowser/models/computer_action.py b/hyperbrowser/models/computer_action.py index 79f622eb..f595f8ae 100644 --- a/hyperbrowser/models/computer_action.py +++ b/hyperbrowser/models/computer_action.py @@ -14,6 +14,7 @@ class ComputerAction(str, Enum): MOVE_MOUSE = "move_mouse" PRESS_KEYS = "press_keys" SCREENSHOT = "screenshot" + CURSOR_POSITION = "cursor_position" SCROLL = "scroll" TYPE_TEXT = "type_text" GET_CLIPBOARD_TEXT = "get_clipboard_text" @@ -43,6 +44,7 @@ class ClickActionParams(BaseModel): y: Optional[int] = Field(default=None) button: ComputerActionMouseButton = Field(default="left") num_clicks: int = Field(serialization_alias="numClicks", default=1) + keys: Optional[List[str]] = None return_screenshot: bool = Field( serialization_alias="returnScreenshot", default=False ) @@ -55,6 +57,7 @@ class DragActionParams(BaseModel): action: Literal[ComputerAction.DRAG] = ComputerAction.DRAG path: List[Coordinate] + keys: Optional[List[str]] = None return_screenshot: bool = Field( serialization_alias="returnScreenshot", default=False ) @@ -130,6 +133,17 @@ class ScreenshotActionParams(BaseModel): action: Literal[ComputerAction.SCREENSHOT] = ComputerAction.SCREENSHOT +class CursorPositionActionParams(BaseModel): + """Parameters for reading the desktop cursor position.""" + + model_config = ConfigDict(use_enum_values=True) + + action: Literal[ComputerAction.CURSOR_POSITION] = ComputerAction.CURSOR_POSITION + return_screenshot: bool = Field( + serialization_alias="returnScreenshot", default=False + ) + + class ScrollActionParams(BaseModel): """Parameters for scroll action.""" @@ -138,8 +152,23 @@ class ScrollActionParams(BaseModel): action: Literal[ComputerAction.SCROLL] = ComputerAction.SCROLL x: int y: int + keys: Optional[List[str]] = None + scroll_x: int = Field(serialization_alias="scrollX") + scroll_y: int = Field(serialization_alias="scrollY") + return_screenshot: bool = Field( + serialization_alias="returnScreenshot", default=False + ) + + +class ScrollAtCursorActionParams(BaseModel): + """Parameters for scrolling without moving the desktop cursor.""" + + model_config = ConfigDict(use_enum_values=True) + + action: Literal[ComputerAction.SCROLL] = ComputerAction.SCROLL scroll_x: int = Field(serialization_alias="scrollX") scroll_y: int = Field(serialization_alias="scrollY") + keys: Optional[List[str]] = None return_screenshot: bool = Field( serialization_alias="returnScreenshot", default=False ) @@ -201,7 +230,9 @@ class ListWindowsActionParams(BaseModel): PressKeysActionParams, MoveMouseActionParams, ScreenshotActionParams, + CursorPositionActionParams, ScrollActionParams, + ScrollAtCursorActionParams, TypeTextActionParams, HoldKeyActionParams, MouseDownActionParams, @@ -237,7 +268,15 @@ class ComputerActionResponseDataListWindows(BaseModel): windows: List[ComputerActionWindow] = Field(default_factory=list) +class ComputerActionResponseDataCursorPosition(BaseModel): + """Desktop cursor coordinates in physical screen pixels.""" + + x: int + y: int + + ComputerActionResponseData = Union[ + ComputerActionResponseDataCursorPosition, ComputerActionResponseDataListWindows, ComputerActionResponseDataClipboardText, ] diff --git a/hyperbrowser/types/__init__.py b/hyperbrowser/types/__init__.py index 4633a631..ce1204de 100644 --- a/hyperbrowser/types/__init__.py +++ b/hyperbrowser/types/__init__.py @@ -32,6 +32,8 @@ StartMetaComputerUseTaskParams, ) from .computer_action import ( + CursorPositionActionParams, + ScrollAtCursorActionParams, ClickActionParams, ComputerActionMouseButton, ComputerActionParams, @@ -212,6 +214,8 @@ "PressKeysActionParams", "PutSelectionTextActionParams", "ScreenshotActionParams", + "CursorPositionActionParams", + "ScrollAtCursorActionParams", "ScrollActionParams", "TypeTextActionParams", "GetCrawlJobParams", diff --git a/hyperbrowser/types/computer_action.py b/hyperbrowser/types/computer_action.py index e70edef5..b5f5434d 100644 --- a/hyperbrowser/types/computer_action.py +++ b/hyperbrowser/types/computer_action.py @@ -23,6 +23,7 @@ class ClickActionParams(TypedDict, total=False): y: Optional[int] button: ComputerActionMouseButton num_clicks: int + keys: List[str] return_screenshot: bool @@ -31,6 +32,7 @@ class DragActionParams(TypedDict, total=False): action: Required[Literal["drag"]] path: Required[List[Coordinate]] + keys: List[str] return_screenshot: bool @@ -82,14 +84,32 @@ class ScreenshotActionParams(TypedDict): action: Literal["screenshot"] +class CursorPositionActionParams(TypedDict, total=False): + """Parameters for reading the desktop cursor position.""" + + action: Required[Literal["cursor_position"]] + return_screenshot: bool + + class ScrollActionParams(TypedDict, total=False): """Parameters for scrolling at a screen coordinate.""" action: Required[Literal["scroll"]] x: Required[int] y: Required[int] + keys: List[str] + scroll_x: Required[int] + scroll_y: Required[int] + return_screenshot: bool + + +class ScrollAtCursorActionParams(TypedDict, total=False): + """Parameters for scrolling without moving the desktop cursor.""" + + action: Required[Literal["scroll"]] scroll_x: Required[int] scroll_y: Required[int] + keys: List[str] return_screenshot: bool @@ -129,7 +149,9 @@ class ListWindowsActionParams(TypedDict, total=False): PressKeysActionParams, MoveMouseActionParams, ScreenshotActionParams, + CursorPositionActionParams, ScrollActionParams, + ScrollAtCursorActionParams, TypeTextActionParams, HoldKeyActionParams, MouseDownActionParams, @@ -145,6 +167,7 @@ class ListWindowsActionParams(TypedDict, total=False): "ComputerActionMouseButton", "ComputerActionParams", "Coordinate", + "CursorPositionActionParams", "DragActionParams", "GetClipboardTextActionParams", "HoldKeyActionParams", @@ -156,5 +179,6 @@ class ListWindowsActionParams(TypedDict, total=False): "PutSelectionTextActionParams", "ScreenshotActionParams", "ScrollActionParams", + "ScrollAtCursorActionParams", "TypeTextActionParams", ] diff --git a/pyproject.toml b/pyproject.toml index d1f399fe..fc850c00 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "hyperbrowser" -version = "1.9.4" +version = "1.10.0" description = "Python SDK for hyperbrowser" authors = ["Nikhil Shahi "] license = "MIT" diff --git a/tests/test_computer_action_primitives.py b/tests/test_computer_action_primitives.py new file mode 100644 index 00000000..03af4876 --- /dev/null +++ b/tests/test_computer_action_primitives.py @@ -0,0 +1,333 @@ +import asyncio +import inspect +from copy import deepcopy +from types import SimpleNamespace +from typing import get_args + +import pytest +from typing_extensions import get_type_hints + +from hyperbrowser.client.managers.async_manager.computer_action import ( + ComputerActionManager as AsyncComputerActionManager, +) +from hyperbrowser.client.managers.sync_manager.computer_action import ( + ComputerActionManager, +) +from hyperbrowser.client._request import dump_request +from hyperbrowser.models import ( + ClickActionParams, + ComputerAction, + ComputerActionParams, + ComputerActionResponse, + ComputerActionResponseData, + ComputerActionResponseDataClipboardText, + ComputerActionResponseDataCursorPosition, + ComputerActionResponseDataListWindows, + CursorPositionActionParams, + DragActionParams, + ScrollActionParams, + ScrollAtCursorActionParams, +) + + +@pytest.fixture(params=[ComputerActionManager, AsyncComputerActionManager]) +def manager(request): + calls = [] + session = SimpleNamespace(computer_action_endpoint="https://session.example/action") + + def get(session_id): + calls.append(("get", session_id)) + return session + + def post(endpoint, *, data): + calls.append(("post", endpoint, deepcopy(data))) + payload = {"success": True} + if data["action"] == "cursor_position": + payload["data"] = {"x": 123, "y": 456} + return SimpleNamespace(data=payload) + + async def async_get(session_id): + return get(session_id) + + async def async_post(endpoint, *, data): + return post(endpoint, data=data) + + is_async = request.param is AsyncComputerActionManager + client = SimpleNamespace( + sessions=SimpleNamespace(get=async_get if is_async else get), + transport=SimpleNamespace(post=async_post if is_async else post), + ) + return request.param(client), session, calls + + +def run(result): + return asyncio.run(result) if inspect.isawaitable(result) else result + + +@pytest.mark.parametrize("by_id", [False, True]) +def test_cursor_position_helper(manager, by_id): + driver, session, calls = manager + result = run(driver.cursor_position("session-id" if by_id else session, True)) + assert result.success + assert isinstance(result, ComputerActionResponse) + assert ( + inspect.signature(driver.cursor_position).return_annotation + is ComputerActionResponse + ) + assert isinstance(result.data, ComputerActionResponseDataCursorPosition) + assert (result.data.x, result.data.y) == (123, 456) + assert calls[-1] == ( + "post", + session.computer_action_endpoint, + {"action": "cursor_position", "returnScreenshot": True}, + ) + assert len(calls) == (2 if by_id else 1) + + +@pytest.mark.parametrize( + "params", + [ + {"action": "cursor_position", "return_screenshot": True}, + CursorPositionActionParams(return_screenshot=True), + ], +) +def test_cursor_position_dict_and_model_requests(manager, params): + driver, session, calls = manager + payload = dump_request(params, CursorPositionActionParams, name="params") + result = run(driver._execute_request(session, params)) + assert payload == calls[-1][2] + assert isinstance(result.data, ComputerActionResponseDataCursorPosition) + assert calls[-1][2] == {"action": "cursor_position", "returnScreenshot": True} + + +@pytest.mark.parametrize("action", ["click", "drag", "scroll"]) +@pytest.mark.parametrize("keys", [None, [], ["Control_L", "Shift_L"]]) +def test_pointer_modifier_helpers_and_existing_positional_arguments( + manager, action, keys +): + driver, session, calls = manager + args = { + "click": (10, 20, "right", 2, True), + "drag": ([{"x": 10, "y": 20}, {"x": 30, "y": 40}], True), + "scroll": (10, 20, 0, -2, True), + } + result = run(getattr(driver, action)(session, *args[action], keys=keys)) + assert result.success + payload = calls[-1][2] + assert payload["action"] == action + assert payload["returnScreenshot"] is True + if keys is None: + assert "keys" not in payload + else: + assert payload["keys"] == keys + if action == "click": + assert (payload["button"], payload["numClicks"]) == ("right", 2) + if action == "scroll": + assert (payload["scrollX"], payload["scrollY"]) == (0, -2) + + +@pytest.mark.parametrize( + "params, model", + [ + ({"action": "click", "x": 10, "y": 20, "keys": ["Shift_L"]}, ClickActionParams), + ( + {"action": "drag", "path": [{"x": 10, "y": 20}], "keys": ["Shift_L"]}, + DragActionParams, + ), + ( + { + "action": "scroll", + "x": 10, + "y": 20, + "scroll_x": 0, + "scroll_y": 1, + "keys": ["Shift_L"], + }, + ScrollActionParams, + ), + ], +) +def test_modifier_dict_and_model_requests_match(manager, params, model): + driver, session, calls = manager + original = deepcopy(params) + run(driver._execute_request(session, params)) + run(driver._execute_request(session, model(**params))) + assert calls[-1] == calls[-2] + assert calls[-1][2]["keys"] == ["Shift_L"] + assert params == original + + +def test_scroll_at_current_cursor(manager): + driver, session, calls = manager + run(driver.scroll_at_cursor(session, scroll_y=2, keys=["Control_L"])) + assert calls[-1][2] == { + "action": "scroll", + "scrollX": 0, + "scrollY": 2, + "returnScreenshot": False, + "keys": ["Control_L"], + } + + +def test_scroll_at_cursor_defaults_without_moving_pointer(manager): + driver, session, calls = manager + run(driver.scroll_at_cursor(session)) + assert calls[-1][2] == { + "action": "scroll", + "scrollX": 0, + "scrollY": 0, + "returnScreenshot": False, + } + + +@pytest.mark.parametrize("action", ["click", "drag", "scroll"]) +def test_invalid_modifier_list_rejected_before_transport(manager, action): + driver, session, calls = manager + args = {"click": (), "drag": ([{"x": 10, "y": 20}],), "scroll": (10, 20, 0, 1)} + with pytest.raises(ValueError): + run(getattr(driver, action)(session, *args[action], keys="Shift_L")) + assert not calls + + +@pytest.mark.parametrize( + "data, model", + [ + ({"clipboardText": "hello"}, ComputerActionResponseDataClipboardText), + ( + { + "activeWindowId": "win", + "windows": [{"id": "win", "name": "Chrome", "active": True}], + }, + ComputerActionResponseDataListWindows, + ), + ], +) +def test_response_data_union_preserves_existing_actions(data, model): + result = ComputerActionResponse(success=True, data=data) + assert isinstance(result.data, model) + assert result.data.model_dump(by_alias=True) == data + + +def test_new_request_and_response_types_are_public(): + import hyperbrowser.models as models + import hyperbrowser.types as types + + assert "CursorPositionActionParams" in models.__all__ + assert "CursorPositionActionParams" in types.__all__ + assert "ScrollAtCursorActionParams" in models.__all__ + assert "ScrollAtCursorActionParams" in types.__all__ + assert ScrollAtCursorActionParams in get_args(ComputerActionParams) + assert types.ScrollAtCursorActionParams in get_args(types.ComputerActionParams) + assert "ComputerActionResponseDataCursorPosition" in models.__all__ + + +@pytest.mark.parametrize("keys", [None, [], ["Shift_L"]]) +def test_scroll_at_cursor_dict_and_model_wire_parity(manager, keys): + driver, session, calls = manager + params = {"action": "scroll", "scroll_x": 0, "scroll_y": -2} + if keys is not None: + params["keys"] = keys + model = ScrollAtCursorActionParams(**params) + run(driver._execute_request(session, params)) + run(driver._execute_request(session, model)) + assert calls[-1] == calls[-2] + assert calls[-1][2] == { + "action": "scroll", + "scrollX": 0, + "scrollY": -2, + "returnScreenshot": False, + **({"keys": keys} if keys is not None else {}), + } + + +@pytest.mark.parametrize("coordinates", [{"x": 10}, {"y": 20}, {"x": None, "y": None}]) +def test_scroll_requests_reject_incomplete_or_null_coordinates(manager, coordinates): + driver, session, calls = manager + with pytest.raises(ValueError): + run( + driver._execute_request( + session, + {"action": "scroll", "scroll_x": 0, "scroll_y": 1, **coordinates}, + ) + ) + assert not calls + + +def test_old_scroll_model_still_requires_coordinates(): + with pytest.raises(ValueError): + ScrollActionParams(scroll_x=0, scroll_y=1) + + +@pytest.mark.parametrize( + "manager_class", [ComputerActionManager, AsyncComputerActionManager] +) +def test_old_scroll_helper_still_requires_coordinates(manager_class): + for name in ("x", "y", "scroll_x", "scroll_y"): + param = inspect.signature(manager_class.scroll).parameters[name] + assert param.default is inspect.Parameter.empty + assert param.annotation is int + + +def test_cursor_response_preserves_failure_without_coordinates(): + response = ComputerActionResponse(success=False, error="action failed") + assert response.data is None + assert response.error == "action failed" + + +def test_action_and_response_unions_include_cursor_position(): + from hyperbrowser.types import ComputerActionParams as DictActionParams + + actions = { + "click", + "drag", + "hold_key", + "mouse_down", + "mouse_up", + "move_mouse", + "press_keys", + "screenshot", + "cursor_position", + "scroll", + "type_text", + "get_clipboard_text", + "put_selection_text", + "list_windows", + } + assert {action.value for action in ComputerAction} == actions + assert { + model.model_fields["action"].default.value + for model in get_args(ComputerActionParams) + } == actions + assert { + get_args(get_type_hints(model)["action"])[0] + for model in get_args(DictActionParams) + } == actions + assert set(get_args(ComputerActionResponseData)) == { + ComputerActionResponseDataCursorPosition, + ComputerActionResponseDataClipboardText, + ComputerActionResponseDataListWindows, + } + + +def test_cursor_helper_preserves_missing_endpoint_error(manager): + driver, _, calls = manager + with pytest.raises(ValueError, match="Computer action endpoint not available"): + run(driver.cursor_position(SimpleNamespace(computer_action_endpoint=None))) + assert not calls + + +@pytest.mark.parametrize("keys", [None, [], ["Control_L"]]) +def test_scroll_at_cursor_helper_preserves_screenshot_and_keys(manager, keys): + driver, session, calls = manager + result = run(driver.scroll_at_cursor("session-id", -1, 2, True, keys)) + expected = { + "action": "scroll", + "scrollX": -1, + "scrollY": 2, + "returnScreenshot": True, + } + if keys is not None: + expected["keys"] = keys + assert result.success + assert calls[0] == ("get", "session-id") + assert calls[1][2] == expected diff --git a/tests/test_typed_request_surface.py b/tests/test_typed_request_surface.py index 0160d6d5..64c7dfa1 100644 --- a/tests/test_typed_request_surface.py +++ b/tests/test_typed_request_surface.py @@ -11,6 +11,7 @@ ACTION_TYPES = { + "CursorPositionActionParams", "ClickActionParams", "DragActionParams", "GetClipboardTextActionParams", @@ -23,6 +24,7 @@ "PutSelectionTextActionParams", "ScreenshotActionParams", "ScrollActionParams", + "ScrollAtCursorActionParams", "TypeTextActionParams", } diff --git a/tests/typecheck/invalid_requests.py b/tests/typecheck/invalid_requests.py index f0cf6375..d59ff8a8 100644 --- a/tests/typecheck/invalid_requests.py +++ b/tests/typecheck/invalid_requests.py @@ -3,6 +3,8 @@ def invalid_sync_requests(client: Hyperbrowser) -> None: + client.computer_action.click("session-id", keys="Shift_L") # M,P + client.computer_action.scroll("session-id", keys=[123]) # M,P client.sessions.create({"made_up_option": True}) # M,P client.web.fetch({}) # M,P client.web.fetch({"url": 42}) # M,P @@ -54,6 +56,9 @@ def invalid_sync_requests(client: Hyperbrowser) -> None: async def invalid_async_requests(client: AsyncHyperbrowser) -> None: + await client.computer_action.scroll("session-id", scroll_y=1) # M,P + await client.computer_action.scroll_at_cursor("session-id", keys="Shift_L") # M,P + await client.computer_action.drag("session-id", [], keys="Shift_L") # M,P await client.sandboxes.create_image_build( { # P "image_name": "custom", diff --git a/tests/typecheck/valid_requests.py b/tests/typecheck/valid_requests.py index 47b0c47b..e70c3fc0 100644 --- a/tests/typecheck/valid_requests.py +++ b/tests/typecheck/valid_requests.py @@ -10,7 +10,14 @@ FetchParams as LegacyFetchParams, SandboxImageBuildListParams as LegacySandboxImageBuildListParams, VolumeListParams as LegacyVolumeListParams, + ScrollActionParams as LegacyScrollActionParams, + ScrollAtCursorActionParams as LegacyScrollAtCursorActionParams, + ComputerActionResponseData, + ComputerActionResponseDataListWindows, + ComputerActionResponseDataCursorPosition, + ComputerActionResponse, ) +from hyperbrowser.types import ScrollActionParams from hyperbrowser.tools import WebsiteExtractTool @@ -377,6 +384,28 @@ async def valid_async_requests(client: AsyncHyperbrowser) -> None: await client.web.fetch(LegacyFetchParams(url="https://example.com")) +def valid_sync_computer_actions(client: Hyperbrowser) -> None: + client.computer_action.cursor_position("session-id", return_screenshot=True) + client.computer_action._execute_request("session-id", {"action": "cursor_position"}) + client.computer_action.click("session-id", 10, 20, keys=["Control_L"]) + client.computer_action.drag("session-id", [{"x": 1, "y": 2}], keys=["Shift_L"]) + client.computer_action.scroll("session-id", 10, 20, 0, 1, keys=["Alt_L"]) + client.computer_action.scroll_at_cursor("session-id", scroll_y=1, keys=["Alt_L"]) + client.computer_action._execute_request( + "session-id", {"action": "scroll", "scroll_x": 0, "scroll_y": 1} + ) + client.computer_action._execute_request( + "session-id", LegacyScrollAtCursorActionParams(scroll_x=0, scroll_y=1) + ) + cursor: ComputerActionResponse = client.computer_action.cursor_position( + "session-id" + ) + if isinstance(cursor.data, ComputerActionResponseDataCursorPosition): + x: int = cursor.data.x + y: int = cursor.data.y + print(x, y) + + def valid_sync_process_collection(client: Hyperbrowser) -> None: sandbox = client.sandboxes.get("sandbox-id") sandbox.exec("echo hello", max_output_bytes=1024) @@ -385,6 +414,50 @@ def valid_sync_process_collection(client: Hyperbrowser) -> None: process.disconnect() +async def valid_async_computer_actions(client: AsyncHyperbrowser) -> None: + await client.computer_action.cursor_position("session-id") + await client.computer_action._execute_request( + "session-id", {"action": "cursor_position"} + ) + await client.computer_action.click("session-id", keys=["Control_L"]) + await client.computer_action.drag( + "session-id", [{"x": 1, "y": 2}], keys=["Shift_L"] + ) + await client.computer_action.scroll("session-id", 10, 20, 0, 1, keys=["Alt_L"]) + await client.computer_action.scroll_at_cursor( + "session-id", scroll_y=1, keys=["Alt_L"] + ) + await client.computer_action._execute_request( + "session-id", {"action": "scroll", "scroll_x": 0, "scroll_y": 1} + ) + await client.computer_action._execute_request( + "session-id", LegacyScrollAtCursorActionParams(scroll_x=0, scroll_y=1) + ) + cursor: ComputerActionResponse = await client.computer_action.cursor_position( + "session-id" + ) + if isinstance(cursor.data, ComputerActionResponseDataCursorPosition): + x: int = cursor.data.x + y: int = cursor.data.y + print(x, y) + + +def existing_scroll_model_consumer(params: LegacyScrollActionParams) -> int: + return params.x + params.y + + +def existing_scroll_dict_consumer(params: ScrollActionParams) -> int: + return params["x"] + params["y"] + + +def shared_response_consumer(data: ComputerActionResponseData) -> str: + if isinstance(data, ComputerActionResponseDataCursorPosition): + return str(data.x) + if isinstance(data, ComputerActionResponseDataListWindows): + return data.active_window_id + return data.clipboard_text or "" + + async def valid_async_process_collection(client: AsyncHyperbrowser) -> None: sandbox = await client.sandboxes.get("sandbox-id") await sandbox.exec("echo hello", max_output_bytes=1024)