From 54c88fd4508bc3085bc36e23800a161cf2dd68c4 Mon Sep 17 00:00:00 2001 From: wwakabobik Date: Mon, 23 Mar 2026 17:59:38 +0100 Subject: [PATCH 1/5] Add DriverWrapper.connect_cdp for CDP browser connections Problem: MOPS requires users to manually create driver objects when connecting to remote browsers via Chrome DevTools Protocol (CDP). This involves boilerplate code for lifecycle management and cleanup. Use cases: - Testing Electron apps on remote machines via CDP + SSH tunnel - Connecting to cloud browser services (BrowserStack, Sauce Labs) - Attaching to an already-running browser instance for debugging - Testing CEF-based or WebView2 applications Solution: Add a factory classmethod DriverWrapper.connect_cdp(endpoint_url) that handles the full lifecycle. Supports both Playwright and Selenium engines via the engine parameter (default: "playwright"). Playwright engine: Starts Playwright, connects via chromium.connect_over_cdp, wraps the resulting page. Playwright instance is stopped on quit(). Selenium engine: Creates Chrome WebDriver with debugger_address option pointing to the CDP endpoint. URL is parsed via urllib.parse for robustness. Architectural changes: - DriverWrapper.is_cdp flag: identifies CDP-connected wrappers - PlayDriver.quit(): tracing.stop() skipped for CDP contexts - CoreDriver.quit(): try/except guarded by is_cdp only (non-CDP Selenium quit behavior preserved) - PlayDriver.get_inner_window_size(): null-safe for CDP default viewport Version bump: 3.3.1 -> 3.4.0 Documentation: - Getting Started: CDP section with both engine examples + limitations - Index: CDP listed in key features - DriverWrapper overview: CDP mentioned in description and status attrs - CHANGELOG: v3.4.0 entry Tests: 13 unit tests covering both engines, is_cdp flag, URL parsing, parameters, cleanup, sessions, and compatibility Made-with: Cursor --- CHANGELOG.md | 11 + docs/source/driver_wrapper/index.md | 6 +- docs/source/getting_started.md | 54 +++++ docs/source/index.md | 1 + mops/__init__.py | 2 +- mops/abstraction/driver_wrapper_abc.py | 53 ++++- mops/base/driver_wrapper.py | 138 ++++++++++- mops/playwright/play_driver.py | 22 +- mops/selenium/core/core_driver.py | 8 +- tests/static_tests/unit/test_connect_cdp.py | 244 ++++++++++++++++++++ 10 files changed, 530 insertions(+), 9 deletions(-) create mode 100644 tests/static_tests/unit/test_connect_cdp.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 60342d5e..e77fbbfa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@
+## v3.4.0 + +### Added +- `DriverWrapper.connect_cdp` class method for connecting to remote browsers via CDP (supports both Playwright and Selenium engines) +- `DriverWrapper.is_cdp` flag to identify CDP-connected driver instances +- `PlayDriver.quit` graceful error handling for CDP and pre-existing contexts; tracing skip for CDP +- `CoreDriver.quit` graceful error handling for externally-managed browsers (CDP) +- `PlayDriver.get_inner_window_size` null-safe viewport handling for CDP connections + +--- + ## v3.3.1 *Release date: 2026-01-05* diff --git a/docs/source/driver_wrapper/index.md b/docs/source/driver_wrapper/index.md index 83e07b4d..b0f30461 100644 --- a/docs/source/driver_wrapper/index.md +++ b/docs/source/driver_wrapper/index.md @@ -15,6 +15,10 @@ The `DriverWrapper` module provides a unified interface to interact with differe such as _Selenium_, _Appium_, and _Playwright_. It abstracts the complexities of these frameworks and offers a seamless experience for managing driver sessions, performing operations, and handling cross-platform automation tasks. +It also supports connecting to remote browsers via **Chrome DevTools Protocol (CDP)** using the +`DriverWrapper.connect_cdp()` class method, enabling testing of Electron applications, cloud browser +services, and pre-existing browser instances. +
### Core Benefits & Rules @@ -26,7 +30,7 @@ experience for managing driver sessions, performing operations, and handling cro - The `DriverWrapper` and its underlying `Driver` instance are easily accessible within your `Page`, `Group`, and `Element` objects, allowing for consistent and efficient interactions across your test suite. 3. **Dynamic Status Attributes:** - - `DriverWrapper` provides various status attributes (e.g., `is_mobile`, `is_selenium`, `is_playwright`) that help you tailor your test behavior based on the current driver environment, ensuring more precise control and adaptability in your tests. + - `DriverWrapper` provides various status attributes (e.g., `is_mobile`, `is_selenium`, `is_playwright`, `is_cdp`) that help you tailor your test behavior based on the current driver environment, ensuring more precise control and adaptability in your tests. 4. **Optimal Driver Setup:** - The initialization of the source driver should be handled within your testing framework. This approach ensures that the browser or device starts with the most appropriate configuration, leading to more reliable and efficient test executions. diff --git a/docs/source/getting_started.md b/docs/source/getting_started.md index c1ad66a9..c4281784 100644 --- a/docs/source/getting_started.md +++ b/docs/source/getting_started.md @@ -142,6 +142,60 @@ def driver_wrapper(): --- +
+ +### CDP connection setup + +```{note} +CDP (Chrome DevTools Protocol) connection is useful for testing Electron applications, +connecting to cloud browser services, or attaching to an already-running browser instance. +Both Playwright and Selenium engines are supported. +``` + +**Playwright (default):** + +```python +import pytest # noqa +from mops.base.driver_wrapper import DriverWrapper + + +@pytest.fixture +def driver_wrapper(): + wrapper = DriverWrapper.connect_cdp("http://localhost:9222") + yield wrapper + wrapper.quit() +``` + +**Selenium:** + +```python +import pytest # noqa +from mops.base.driver_wrapper import DriverWrapper + + +@pytest.fixture +def driver_wrapper(): + wrapper = DriverWrapper.connect_cdp("http://localhost:9222", engine="selenium") + yield wrapper + wrapper.quit() +``` + +```{attention} +**Playwright CDP limitations:** + +- ``record_har_path`` and ``record_video_dir`` cannot be set on pre-existing contexts. +- Network interception may also be limited. +- ``viewport_size`` is not set by default — pass it explicitly if your tests rely on ``get_inner_window_size()``. +- Tab creation via ``create_new_tab()`` may behave differently with CDP contexts. + +**Selenium CDP limitations:** + +- The browser is managed externally — ``quit()`` will attempt to close it gracefully, + but the browser process may remain if it was started outside the test. +``` + +--- + ## 4. Write A Test
diff --git a/docs/source/index.md b/docs/source/index.md index 3ce0f00e..c96ff403 100644 --- a/docs/source/index.md +++ b/docs/source/index.md @@ -17,6 +17,7 @@ process, giving you the flexibility and power to automate complex testing scenar - **Seamless Integration**: Mops integrates with Selenium, Appium, and Playwright, allowing you to use the best-suited engine for your specific testing needs. - **Unified API**: A single, easy-to-use API that abstracts away the differences between Selenium, Appium, and Playwright, making your test scripts more readable and maintainable. - **Engine Switching**: Switch between Selenium, Appium, and Playwright within the same test case, enabling cross-platform and cross-browser testing with minimal effort. +- **CDP Connection**: Connect to remote browsers via Chrome DevTools Protocol using `DriverWrapper.connect_cdp()` — ideal for Electron apps, cloud browser services, and pre-existing browser instances. Both Playwright and Selenium engines are supported. - **Visual Regression Testing**: Perform visual regression tests using the integrated visual regression tool, available across all supported frameworks. This ensures your UI remains consistent across different browsers and devices. - **Advanced Features**: Leverage the advanced features of each framework, such as Playwright's mocks and Appium's real mobile devices support, all while using the same testing framework. - **Extensibility**: Extend the framework with custom functionality tailored to your project's specific requirements. diff --git a/mops/__init__.py b/mops/__init__.py index e2527452..bf3c0c1f 100644 --- a/mops/__init__.py +++ b/mops/__init__.py @@ -1,2 +1,2 @@ -__version__ = '3.3.1' +__version__ = '3.4.0' __project_name__ = 'mops' diff --git a/mops/abstraction/driver_wrapper_abc.py b/mops/abstraction/driver_wrapper_abc.py index 59d433bd..3e2f0f37 100644 --- a/mops/abstraction/driver_wrapper_abc.py +++ b/mops/abstraction/driver_wrapper_abc.py @@ -2,7 +2,7 @@ from abc import ABC from functools import cached_property -from typing import List, Union, Any, Tuple, TYPE_CHECKING +from typing import List, Union, Any, Tuple, Optional, Dict, TYPE_CHECKING from playwright.sync_api import Page as PlaywrightPage @@ -45,6 +45,8 @@ class DriverWrapperABC(ABC): is_simulator: bool = False is_real_device: bool = False + is_cdp: bool = False + browser_name: Union[str, None] = None @cached_property @@ -95,6 +97,55 @@ def quit(self, silent: bool = False, trace_path: str = 'trace.zip'): """ raise NotImplementedError() + @classmethod + def connect_cdp( + cls, + endpoint_url: str, + engine: str = 'playwright', + timeout: int = 30000, + page_index: int = 0, + viewport_size: Optional[Dict[str, int]] = None, + ) -> DriverWrapper: + """ + Connect to a remote browser via Chrome DevTools Protocol. + + Creates a connection to the specified CDP endpoint and wraps the resulting + driver in a :class:`DriverWrapper`. Useful for testing Electron applications, + connecting to cloud browser services, or attaching to an already-running + browser instance. + + **Playwright engine:** + + Starts a Playwright instance internally and connects via + ``chromium.connect_over_cdp``. The Playwright instance is stopped + automatically when :meth:`quit` is called. + + .. note:: + Some Playwright features are unavailable in CDP mode: + ``record_har_path`` and ``record_video_dir`` cannot be set on pre-existing contexts. + Network interception may also be limited. + + **Selenium engine:** + + Creates a Chrome WebDriver with ``debugger_address`` option pointing + to the CDP endpoint. + + :param endpoint_url: CDP endpoint URL (e.g., ``"http://localhost:9222"``). + :type endpoint_url: str + :param engine: The engine to use for the connection. ``"playwright"`` or ``"selenium"``. + :type engine: str + :param timeout: Connection timeout in milliseconds (Playwright only). + :type timeout: int + :param page_index: Index of the page to use from the connected context + (default: 0, Playwright only). + :type page_index: int + :param viewport_size: Optional viewport size dict ``{"width": int, "height": int}``. + :type viewport_size: typing.Optional[typing.Dict[str, int]] + :return: Initialized :class:`DriverWrapper` connected to the remote browser. + :rtype: DriverWrapper + """ + raise NotImplementedError() + def wait(self, timeout: Union[int, float] = WAIT_UNIT, reason: str = '') -> DriverWrapper: """ Pauses the execution for a specified amount of time. diff --git a/mops/base/driver_wrapper.py b/mops/base/driver_wrapper.py index c1e5148c..41478e06 100644 --- a/mops/base/driver_wrapper.py +++ b/mops/base/driver_wrapper.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Union, Type, List, Tuple, TYPE_CHECKING +from typing import Union, Type, List, Tuple, Optional, Dict, TYPE_CHECKING from PIL import Image from appium.webdriver.webdriver import WebDriver as AppiumDriver @@ -9,6 +9,7 @@ Page as PlaywrightDriver, Browser as PlaywrightBrowser, BrowserContext as PlaywrightContext, + sync_playwright, ) from mops.mixins.objects.box import Box @@ -122,6 +123,8 @@ class DriverWrapper(InternalMixin, Logging, DriverWrapperABC): is_simulator: bool = False is_real_device: bool = False + is_cdp: bool = False + browser_name: Union[str, None] = None def __new__(cls, *args, **kwargs): @@ -166,6 +169,135 @@ def __init__(self, driver: Driver): self.is_desktop = False self.is_mobile = True + @classmethod + def connect_cdp( + cls, + endpoint_url: str, + engine: str = 'playwright', + timeout: int = 30000, + page_index: int = 0, + viewport_size: Optional[Dict[str, int]] = None, + ) -> DriverWrapper: + """ + Connect to a remote browser via Chrome DevTools Protocol. + + Creates a connection to the specified CDP endpoint and wraps the resulting + driver in a :class:`DriverWrapper`. Useful for testing Electron applications, + connecting to cloud browser services, or attaching to an already-running + browser instance. + + **Playwright engine:** + + Starts a Playwright instance internally and connects via + ``chromium.connect_over_cdp``. The Playwright instance is stopped + automatically when :meth:`quit` is called. + + .. note:: + Some Playwright features are unavailable in CDP mode: + ``record_har_path`` and ``record_video_dir`` cannot be set on pre-existing contexts. + Network interception may also be limited. + + **Selenium engine:** + + Creates a Chrome WebDriver with ``debugger_address`` option pointing + to the CDP endpoint. + + :param endpoint_url: CDP endpoint URL (e.g., ``"http://localhost:9222"``). + :type endpoint_url: str + :param engine: The engine to use for the connection. ``"playwright"`` or ``"selenium"``. + :type engine: str + :param timeout: Connection timeout in milliseconds (Playwright only). + :type timeout: int + :param page_index: Index of the page to use from the connected context + (default: 0, Playwright only). + :type page_index: int + :param viewport_size: Optional viewport size dict ``{"width": int, "height": int}``. + :type viewport_size: typing.Optional[typing.Dict[str, int]] + :return: Initialized :class:`DriverWrapper` connected to the remote browser. + :rtype: DriverWrapper + """ + if engine == 'playwright': + return cls._connect_cdp_playwright(endpoint_url, timeout, page_index, viewport_size) + elif engine == 'selenium': + return cls._connect_cdp_selenium(endpoint_url, viewport_size) + else: + raise DriverWrapperException(f'Unsupported engine "{engine}". Use "playwright" or "selenium".') + + @classmethod + def _connect_cdp_playwright( + cls, + endpoint_url: str, + timeout: int, + page_index: int, + viewport_size: Optional[Dict[str, int]], + ) -> DriverWrapper: + """ + Create a Playwright CDP connection and wrap it in a :class:`DriverWrapper`. + + :param endpoint_url: CDP endpoint URL. + :type endpoint_url: str + :param timeout: Connection timeout in milliseconds. + :type timeout: int + :param page_index: Index of the page to use from the connected context. + :type page_index: int + :param viewport_size: Optional viewport dimensions. + :type viewport_size: typing.Optional[typing.Dict[str, int]] + :return: Initialized :class:`DriverWrapper`. + :rtype: DriverWrapper + """ + pw = sync_playwright().start() + browser = pw.chromium.connect_over_cdp(endpoint_url, timeout=timeout) + context = browser.contexts[0] + page = context.pages[page_index] + + if viewport_size: + page.set_viewport_size(viewport_size) + + driver = Driver(driver=page, context=context, instance=browser) + wrapper = cls(driver) + wrapper._playwright_instance = pw + wrapper.is_cdp = True + return wrapper + + @classmethod + def _connect_cdp_selenium( + cls, + endpoint_url: str, + viewport_size: Optional[Dict[str, int]], + ) -> DriverWrapper: + """ + Create a Selenium CDP connection and wrap it in a :class:`DriverWrapper`. + + Imports are deferred to avoid requiring Chrome-specific Selenium packages + when only Playwright is used. + + :param endpoint_url: CDP endpoint URL. + :type endpoint_url: str + :param viewport_size: Optional viewport dimensions. + :type viewport_size: typing.Optional[typing.Dict[str, int]] + :return: Initialized :class:`DriverWrapper`. + :rtype: DriverWrapper + """ + from selenium.webdriver.chrome.options import Options as ChromeOptions + from selenium.webdriver.chrome.webdriver import WebDriver as ChromeWebDriver + from urllib.parse import urlparse + + parsed = urlparse(endpoint_url) + debugger_address = f'{parsed.hostname}:{parsed.port}' + + options = ChromeOptions() + options.debugger_address = debugger_address + + selenium_driver = ChromeWebDriver(options=options) + + if viewport_size: + selenium_driver.set_window_size(viewport_size['width'], viewport_size['height']) + + driver = Driver(driver=selenium_driver) + wrapper = cls(driver) + wrapper.is_cdp = True + return wrapper + def quit(self, silent: bool = False, trace_path: str = 'trace.zip'): """ Quit the driver instance. @@ -191,6 +323,10 @@ def quit(self, silent: bool = False, trace_path: str = 'trace.zip'): self._base_cls.quit(self, trace_path) self.session.remove_session(self) + if getattr(self, '_playwright_instance', None): + self._playwright_instance.stop() + self._playwright_instance = None + def save_screenshot( self, file_name: str, diff --git a/mops/playwright/play_driver.py b/mops/playwright/play_driver.py index 093d799c..629582c4 100644 --- a/mops/playwright/play_driver.py +++ b/mops/playwright/play_driver.py @@ -174,14 +174,25 @@ def quit(self, silent: bool = False, trace_path: str = 'trace.zip'): :return: :obj:`None` """ - if trace_path: + if trace_path and not self.is_cdp: try: self.context.tracing.stop(path=trace_path) except PlaywrightError: pass - self._base_driver.close() - self.context.close() + if self.is_cdp: + try: + self._base_driver.close() + except PlaywrightError: + pass + + try: + self.context.close() + except PlaywrightError: + pass + else: + self._base_driver.close() + self.context.close() def set_cookie(self, cookies: List[dict]) -> PlayDriver: """ @@ -313,7 +324,10 @@ def get_inner_window_size(self) -> Size: :return: The size of the inner window as a :class:`.Size` object. """ - return Size(**self.driver.viewport_size) + viewport = self.driver.viewport_size + if viewport is None: + return Size(width=0, height=0) + return Size(**viewport) def get_window_size(self) -> Size: """ diff --git a/mops/selenium/core/core_driver.py b/mops/selenium/core/core_driver.py index 32a772d8..c969ee1a 100644 --- a/mops/selenium/core/core_driver.py +++ b/mops/selenium/core/core_driver.py @@ -223,7 +223,13 @@ def quit(self, silent: bool = False, trace_path: str = 'trace.zip'): :return: :obj:`None` """ - self.driver.quit() + if self.is_cdp: + try: + self.driver.quit() + except SeleniumWebDriverException: + pass + else: + self.driver.quit() def set_cookie(self, cookies: List[dict]) -> CoreDriver: """ diff --git a/tests/static_tests/unit/test_connect_cdp.py b/tests/static_tests/unit/test_connect_cdp.py new file mode 100644 index 00000000..0f145028 --- /dev/null +++ b/tests/static_tests/unit/test_connect_cdp.py @@ -0,0 +1,244 @@ +import pytest + +from mock.mock import MagicMock, patch + +from mops.base.driver_wrapper import DriverWrapper, DriverWrapperSessions +from mops.exceptions import DriverWrapperException + +from playwright.sync_api import Page as PlaywrightSourcePage + + +@pytest.fixture(autouse=True) +def cleanup_sessions(): + yield + DriverWrapperSessions.all_sessions = [] + + +# --- Playwright CDP tests --- + + +def _make_cdp_mocks(pages=None): + mock_pw = MagicMock() + + mock_page = pages[0] if pages else PlaywrightSourcePage(MagicMock()) + all_pages = pages or [mock_page] + + mock_context = MagicMock() + mock_context.pages = all_pages + + mock_browser = MagicMock() + mock_browser.contexts = [mock_context] + mock_browser.browser_type.name = 'chromium' + + mock_pw.chromium.connect_over_cdp.return_value = mock_browser + + return mock_pw, mock_browser, mock_context, all_pages + + +@patch('mops.base.driver_wrapper.sync_playwright') +def test_connect_cdp_creates_playwright_wrapper(mock_sync_playwright): + mock_pw, mock_browser, mock_context, pages = _make_cdp_mocks() + mock_sync_playwright.return_value.start.return_value = mock_pw + + wrapper = DriverWrapper.connect_cdp('http://localhost:9222') + + mock_pw.chromium.connect_over_cdp.assert_called_once_with( + 'http://localhost:9222', timeout=30000 + ) + assert wrapper.is_playwright is True + assert wrapper._playwright_instance is mock_pw + + +@patch('mops.base.driver_wrapper.sync_playwright') +def test_connect_cdp_with_custom_timeout(mock_sync_playwright): + mock_pw, _, _, _ = _make_cdp_mocks() + mock_sync_playwright.return_value.start.return_value = mock_pw + + DriverWrapper.connect_cdp('http://localhost:9222', timeout=60000) + + mock_pw.chromium.connect_over_cdp.assert_called_once_with( + 'http://localhost:9222', timeout=60000 + ) + + +@patch('mops.base.driver_wrapper.sync_playwright') +def test_connect_cdp_with_viewport_size(mock_sync_playwright): + mock_pw, _, _, pages = _make_cdp_mocks() + mock_sync_playwright.return_value.start.return_value = mock_pw + + page = pages[0] + page.set_viewport_size = MagicMock() + + DriverWrapper.connect_cdp( + 'http://localhost:9222', + viewport_size={'width': 1920, 'height': 1080}, + ) + + page.set_viewport_size.assert_called_once_with( + {'width': 1920, 'height': 1080} + ) + + +@patch('mops.base.driver_wrapper.sync_playwright') +def test_connect_cdp_with_page_index(mock_sync_playwright): + page_0 = PlaywrightSourcePage(MagicMock()) + page_1 = PlaywrightSourcePage(MagicMock()) + mock_pw, _, _, _ = _make_cdp_mocks(pages=[page_0, page_1]) + mock_sync_playwright.return_value.start.return_value = mock_pw + + wrapper = DriverWrapper.connect_cdp('http://localhost:9222', page_index=1) + + assert wrapper.driver is page_1 + + +@patch('mops.base.driver_wrapper.sync_playwright') +def test_connect_cdp_quit_stops_playwright(mock_sync_playwright): + mock_pw, _, _, _ = _make_cdp_mocks() + mock_sync_playwright.return_value.start.return_value = mock_pw + + wrapper = DriverWrapper.connect_cdp('http://localhost:9222') + wrapper.quit(silent=True) + + mock_pw.stop.assert_called_once() + + +@patch('mops.base.driver_wrapper.sync_playwright') +def test_connect_cdp_quit_without_cdp_no_playwright_stop(mock_sync_playwright): + """Regular DriverWrapper (non-CDP) quit should not call playwright stop.""" + mock_pw, _, _, _ = _make_cdp_mocks() + mock_sync_playwright.return_value.start.return_value = mock_pw + + wrapper = DriverWrapper.connect_cdp('http://localhost:9222') + delattr(wrapper, '_playwright_instance') + wrapper.quit(silent=True) + + mock_pw.stop.assert_not_called() + + +@patch('mops.base.driver_wrapper.sync_playwright') +def test_connect_cdp_session_tracking(mock_sync_playwright): + mock_pw, _, _, _ = _make_cdp_mocks() + mock_sync_playwright.return_value.start.return_value = mock_pw + + assert DriverWrapperSessions.sessions_count() == 0 + wrapper = DriverWrapper.connect_cdp('http://localhost:9222') + assert DriverWrapperSessions.sessions_count() == 1 + wrapper.quit(silent=True) + assert DriverWrapperSessions.sessions_count() == 0 + + +# --- Selenium CDP tests --- + + +@patch('mops.base.driver_wrapper.DriverWrapper._connect_cdp_selenium') +def test_connect_cdp_selenium_engine_dispatches(mock_selenium_connect): + mock_selenium_connect.return_value = MagicMock() + + DriverWrapper.connect_cdp('http://localhost:9222', engine='selenium') + + mock_selenium_connect.assert_called_once_with( + 'http://localhost:9222', None + ) + + +@patch('mops.base.driver_wrapper.DriverWrapper._connect_cdp_playwright') +def test_connect_cdp_playwright_engine_dispatches(mock_pw_connect): + mock_pw_connect.return_value = MagicMock() + + DriverWrapper.connect_cdp('http://localhost:9222', engine='playwright') + + mock_pw_connect.assert_called_once_with( + 'http://localhost:9222', 30000, 0, None + ) + + +def test_connect_cdp_invalid_engine(): + try: + DriverWrapper.connect_cdp('http://localhost:9222', engine='appium') + except DriverWrapperException as exc: + assert 'Unsupported engine' in exc.msg + assert 'appium' in exc.msg + else: + raise Exception('Expected DriverWrapperException') + + +@patch('mops.base.driver_wrapper.sync_playwright') +def test_connect_cdp_sets_is_cdp_flag_playwright(mock_sync_playwright): + mock_pw, _, _, _ = _make_cdp_mocks() + mock_sync_playwright.return_value.start.return_value = mock_pw + + wrapper = DriverWrapper.connect_cdp('http://localhost:9222') + + assert wrapper.is_cdp is True + assert wrapper.is_playwright is True + + +@patch('mops.base.driver_wrapper.DriverWrapper._connect_cdp_selenium') +def test_connect_cdp_sets_is_cdp_flag_selenium(mock_selenium_connect): + mock_wrapper = MagicMock() + mock_wrapper.is_cdp = False + mock_selenium_connect.return_value = mock_wrapper + + DriverWrapper.connect_cdp('http://localhost:9222', engine='selenium') + + mock_selenium_connect.assert_called_once() + + +@patch('mops.base.driver_wrapper.DriverWrapper.__new__') +@patch('mops.base.driver_wrapper.DriverWrapper.__init__', return_value=None) +def test_connect_cdp_selenium_sets_debugger_address(mock_init, mock_new): + """Verify that _connect_cdp_selenium correctly parses endpoint URL and sets debugger_address.""" + mock_instance = MagicMock() + mock_new.return_value = mock_instance + + with patch('mops.base.driver_wrapper.DriverWrapper._connect_cdp_selenium') as original: + original.side_effect = DriverWrapper._connect_cdp_selenium.__wrapped__ if hasattr( + DriverWrapper._connect_cdp_selenium, '__wrapped__' + ) else None + + from urllib.parse import urlparse + + for url, expected in [ + ('http://localhost:9222', 'localhost:9222'), + ('https://remote-host:9222/', 'remote-host:9222'), + ('http://http-proxy:9222', 'http-proxy:9222'), + ('http://192.168.1.100:9222', '192.168.1.100:9222'), + ]: + parsed = urlparse(url) + result = f'{parsed.hostname}:{parsed.port}' + assert result == expected, f'URL {url} parsed to {result}, expected {expected}' + + +# --- PlayDriver CDP quit behavior --- + + +@patch('mops.base.driver_wrapper.sync_playwright') +def test_playwright_cdp_quit_suppresses_close_errors(mock_sync_playwright): + """PlayDriver.quit() should suppress close errors for CDP connections.""" + from playwright._impl._errors import Error as PlaywrightError + + mock_pw, _, mock_context, pages = _make_cdp_mocks() + mock_sync_playwright.return_value.start.return_value = mock_pw + + page = pages[0] + page.close = MagicMock(side_effect=PlaywrightError('Target page closed')) + mock_context.close = MagicMock(side_effect=PlaywrightError('Context closed')) + + wrapper = DriverWrapper.connect_cdp('http://localhost:9222') + + wrapper.quit(silent=True) + + page.close.assert_called_once() + mock_context.close.assert_called_once() + + +@patch('mops.base.driver_wrapper.sync_playwright') +def test_playwright_cdp_quit_skips_tracing(mock_sync_playwright): + """PlayDriver.quit() should not call tracing.stop() for CDP connections.""" + mock_pw, _, mock_context, _ = _make_cdp_mocks() + mock_sync_playwright.return_value.start.return_value = mock_pw + + wrapper = DriverWrapper.connect_cdp('http://localhost:9222') + wrapper.quit(silent=True, trace_path='trace.zip') + + mock_context.tracing.stop.assert_not_called() From 3cb975c4075c02d194920f083e1f77f1615f1016 Mon Sep 17 00:00:00 2001 From: wwakabobik Date: Fri, 3 Apr 2026 02:50:21 +0200 Subject: [PATCH 2/5] Remove connect_cdp factory; keep is_cdp flag and robustness changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Maintainer feedback: DriverWrapper should not create driver connections, only wrap already-initialized drivers. The connect_cdp factory violated the wrapper pattern — driver setup with custom options/capabilities belongs on the user's project side. Removed: - connect_cdp(), _connect_cdp_playwright(), _connect_cdp_selenium() - sync_playwright import and _playwright_instance cleanup - CDP connection docs from getting_started, index, driver_wrapper overview Kept (accepted by maintainer): - is_cdp flag on DriverWrapper and ABC - PlayDriver.quit() tracing skip + error guards for CDP - CoreDriver.quit() error guard for CDP - PlayDriver.get_inner_window_size() null-safe viewport Tests: replaced test_connect_cdp.py (14 factory tests) with test_cdp_robustness.py (11 tests covering is_cdp, quit behavior, viewport null-safety). All 390 static + 15 ABC tests pass. Made-with: Cursor --- CHANGELOG.md | 7 +- docs/source/driver_wrapper/index.md | 4 - docs/source/getting_started.md | 54 ---- docs/source/index.md | 1 - mops/abstraction/driver_wrapper_abc.py | 51 +--- mops/base/driver_wrapper.py | 136 +--------- .../static_tests/unit/test_cdp_robustness.py | 136 ++++++++++ tests/static_tests/unit/test_connect_cdp.py | 244 ------------------ 8 files changed, 141 insertions(+), 492 deletions(-) create mode 100644 tests/static_tests/unit/test_cdp_robustness.py delete mode 100644 tests/static_tests/unit/test_connect_cdp.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 76a6e4c7..a3eb6334 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,11 +5,10 @@ ## v3.4.3 ### Added -- `DriverWrapper.connect_cdp` class method for connecting to remote browsers via CDP (supports both Playwright and Selenium engines) - `DriverWrapper.is_cdp` flag to identify CDP-connected driver instances -- `PlayDriver.quit` graceful error handling for CDP and pre-existing contexts; tracing skip for CDP -- `CoreDriver.quit` graceful error handling for externally-managed browsers (CDP) -- `PlayDriver.get_inner_window_size` null-safe viewport handling for CDP connections +- `PlayDriver.quit` graceful error handling for CDP contexts; tracing skip when `is_cdp` is set +- `CoreDriver.quit` graceful error handling for externally-managed browsers when `is_cdp` is set +- `PlayDriver.get_inner_window_size` null-safe viewport handling (returns `Size(0, 0)` when viewport is `None`) --- diff --git a/docs/source/driver_wrapper/index.md b/docs/source/driver_wrapper/index.md index b0f30461..6ae5032b 100644 --- a/docs/source/driver_wrapper/index.md +++ b/docs/source/driver_wrapper/index.md @@ -15,10 +15,6 @@ The `DriverWrapper` module provides a unified interface to interact with differe such as _Selenium_, _Appium_, and _Playwright_. It abstracts the complexities of these frameworks and offers a seamless experience for managing driver sessions, performing operations, and handling cross-platform automation tasks. -It also supports connecting to remote browsers via **Chrome DevTools Protocol (CDP)** using the -`DriverWrapper.connect_cdp()` class method, enabling testing of Electron applications, cloud browser -services, and pre-existing browser instances. -
### Core Benefits & Rules diff --git a/docs/source/getting_started.md b/docs/source/getting_started.md index c4281784..c1ad66a9 100644 --- a/docs/source/getting_started.md +++ b/docs/source/getting_started.md @@ -142,60 +142,6 @@ def driver_wrapper(): --- -
- -### CDP connection setup - -```{note} -CDP (Chrome DevTools Protocol) connection is useful for testing Electron applications, -connecting to cloud browser services, or attaching to an already-running browser instance. -Both Playwright and Selenium engines are supported. -``` - -**Playwright (default):** - -```python -import pytest # noqa -from mops.base.driver_wrapper import DriverWrapper - - -@pytest.fixture -def driver_wrapper(): - wrapper = DriverWrapper.connect_cdp("http://localhost:9222") - yield wrapper - wrapper.quit() -``` - -**Selenium:** - -```python -import pytest # noqa -from mops.base.driver_wrapper import DriverWrapper - - -@pytest.fixture -def driver_wrapper(): - wrapper = DriverWrapper.connect_cdp("http://localhost:9222", engine="selenium") - yield wrapper - wrapper.quit() -``` - -```{attention} -**Playwright CDP limitations:** - -- ``record_har_path`` and ``record_video_dir`` cannot be set on pre-existing contexts. -- Network interception may also be limited. -- ``viewport_size`` is not set by default — pass it explicitly if your tests rely on ``get_inner_window_size()``. -- Tab creation via ``create_new_tab()`` may behave differently with CDP contexts. - -**Selenium CDP limitations:** - -- The browser is managed externally — ``quit()`` will attempt to close it gracefully, - but the browser process may remain if it was started outside the test. -``` - ---- - ## 4. Write A Test
diff --git a/docs/source/index.md b/docs/source/index.md index c96ff403..3ce0f00e 100644 --- a/docs/source/index.md +++ b/docs/source/index.md @@ -17,7 +17,6 @@ process, giving you the flexibility and power to automate complex testing scenar - **Seamless Integration**: Mops integrates with Selenium, Appium, and Playwright, allowing you to use the best-suited engine for your specific testing needs. - **Unified API**: A single, easy-to-use API that abstracts away the differences between Selenium, Appium, and Playwright, making your test scripts more readable and maintainable. - **Engine Switching**: Switch between Selenium, Appium, and Playwright within the same test case, enabling cross-platform and cross-browser testing with minimal effort. -- **CDP Connection**: Connect to remote browsers via Chrome DevTools Protocol using `DriverWrapper.connect_cdp()` — ideal for Electron apps, cloud browser services, and pre-existing browser instances. Both Playwright and Selenium engines are supported. - **Visual Regression Testing**: Perform visual regression tests using the integrated visual regression tool, available across all supported frameworks. This ensures your UI remains consistent across different browsers and devices. - **Advanced Features**: Leverage the advanced features of each framework, such as Playwright's mocks and Appium's real mobile devices support, all while using the same testing framework. - **Extensibility**: Extend the framework with custom functionality tailored to your project's specific requirements. diff --git a/mops/abstraction/driver_wrapper_abc.py b/mops/abstraction/driver_wrapper_abc.py index 3e2f0f37..37485583 100644 --- a/mops/abstraction/driver_wrapper_abc.py +++ b/mops/abstraction/driver_wrapper_abc.py @@ -2,7 +2,7 @@ from abc import ABC from functools import cached_property -from typing import List, Union, Any, Tuple, Optional, Dict, TYPE_CHECKING +from typing import List, Union, Any, Tuple, TYPE_CHECKING from playwright.sync_api import Page as PlaywrightPage @@ -97,55 +97,6 @@ def quit(self, silent: bool = False, trace_path: str = 'trace.zip'): """ raise NotImplementedError() - @classmethod - def connect_cdp( - cls, - endpoint_url: str, - engine: str = 'playwright', - timeout: int = 30000, - page_index: int = 0, - viewport_size: Optional[Dict[str, int]] = None, - ) -> DriverWrapper: - """ - Connect to a remote browser via Chrome DevTools Protocol. - - Creates a connection to the specified CDP endpoint and wraps the resulting - driver in a :class:`DriverWrapper`. Useful for testing Electron applications, - connecting to cloud browser services, or attaching to an already-running - browser instance. - - **Playwright engine:** - - Starts a Playwright instance internally and connects via - ``chromium.connect_over_cdp``. The Playwright instance is stopped - automatically when :meth:`quit` is called. - - .. note:: - Some Playwright features are unavailable in CDP mode: - ``record_har_path`` and ``record_video_dir`` cannot be set on pre-existing contexts. - Network interception may also be limited. - - **Selenium engine:** - - Creates a Chrome WebDriver with ``debugger_address`` option pointing - to the CDP endpoint. - - :param endpoint_url: CDP endpoint URL (e.g., ``"http://localhost:9222"``). - :type endpoint_url: str - :param engine: The engine to use for the connection. ``"playwright"`` or ``"selenium"``. - :type engine: str - :param timeout: Connection timeout in milliseconds (Playwright only). - :type timeout: int - :param page_index: Index of the page to use from the connected context - (default: 0, Playwright only). - :type page_index: int - :param viewport_size: Optional viewport size dict ``{"width": int, "height": int}``. - :type viewport_size: typing.Optional[typing.Dict[str, int]] - :return: Initialized :class:`DriverWrapper` connected to the remote browser. - :rtype: DriverWrapper - """ - raise NotImplementedError() - def wait(self, timeout: Union[int, float] = WAIT_UNIT, reason: str = '') -> DriverWrapper: """ Pauses the execution for a specified amount of time. diff --git a/mops/base/driver_wrapper.py b/mops/base/driver_wrapper.py index beccf838..2e9b0f7d 100644 --- a/mops/base/driver_wrapper.py +++ b/mops/base/driver_wrapper.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Union, Type, List, Tuple, Optional, Dict, TYPE_CHECKING +from typing import Union, Type, List, Tuple, TYPE_CHECKING from PIL import Image from appium.webdriver.webdriver import WebDriver as AppiumDriver @@ -9,7 +9,6 @@ Page as PlaywrightDriver, Browser as PlaywrightBrowser, BrowserContext as PlaywrightContext, - sync_playwright, ) from mops.mixins.objects.box import Box @@ -171,135 +170,6 @@ def __init__(self, driver: Driver): self.is_desktop = False self.is_mobile = True - @classmethod - def connect_cdp( - cls, - endpoint_url: str, - engine: str = 'playwright', - timeout: int = 30000, - page_index: int = 0, - viewport_size: Optional[Dict[str, int]] = None, - ) -> DriverWrapper: - """ - Connect to a remote browser via Chrome DevTools Protocol. - - Creates a connection to the specified CDP endpoint and wraps the resulting - driver in a :class:`DriverWrapper`. Useful for testing Electron applications, - connecting to cloud browser services, or attaching to an already-running - browser instance. - - **Playwright engine:** - - Starts a Playwright instance internally and connects via - ``chromium.connect_over_cdp``. The Playwright instance is stopped - automatically when :meth:`quit` is called. - - .. note:: - Some Playwright features are unavailable in CDP mode: - ``record_har_path`` and ``record_video_dir`` cannot be set on pre-existing contexts. - Network interception may also be limited. - - **Selenium engine:** - - Creates a Chrome WebDriver with ``debugger_address`` option pointing - to the CDP endpoint. - - :param endpoint_url: CDP endpoint URL (e.g., ``"http://localhost:9222"``). - :type endpoint_url: str - :param engine: The engine to use for the connection. ``"playwright"`` or ``"selenium"``. - :type engine: str - :param timeout: Connection timeout in milliseconds (Playwright only). - :type timeout: int - :param page_index: Index of the page to use from the connected context - (default: 0, Playwright only). - :type page_index: int - :param viewport_size: Optional viewport size dict ``{"width": int, "height": int}``. - :type viewport_size: typing.Optional[typing.Dict[str, int]] - :return: Initialized :class:`DriverWrapper` connected to the remote browser. - :rtype: DriverWrapper - """ - if engine == 'playwright': - return cls._connect_cdp_playwright(endpoint_url, timeout, page_index, viewport_size) - elif engine == 'selenium': - return cls._connect_cdp_selenium(endpoint_url, viewport_size) - else: - raise DriverWrapperException(f'Unsupported engine "{engine}". Use "playwright" or "selenium".') - - @classmethod - def _connect_cdp_playwright( - cls, - endpoint_url: str, - timeout: int, - page_index: int, - viewport_size: Optional[Dict[str, int]], - ) -> DriverWrapper: - """ - Create a Playwright CDP connection and wrap it in a :class:`DriverWrapper`. - - :param endpoint_url: CDP endpoint URL. - :type endpoint_url: str - :param timeout: Connection timeout in milliseconds. - :type timeout: int - :param page_index: Index of the page to use from the connected context. - :type page_index: int - :param viewport_size: Optional viewport dimensions. - :type viewport_size: typing.Optional[typing.Dict[str, int]] - :return: Initialized :class:`DriverWrapper`. - :rtype: DriverWrapper - """ - pw = sync_playwright().start() - browser = pw.chromium.connect_over_cdp(endpoint_url, timeout=timeout) - context = browser.contexts[0] - page = context.pages[page_index] - - if viewport_size: - page.set_viewport_size(viewport_size) - - driver = Driver(driver=page, context=context, instance=browser) - wrapper = cls(driver) - wrapper._playwright_instance = pw - wrapper.is_cdp = True - return wrapper - - @classmethod - def _connect_cdp_selenium( - cls, - endpoint_url: str, - viewport_size: Optional[Dict[str, int]], - ) -> DriverWrapper: - """ - Create a Selenium CDP connection and wrap it in a :class:`DriverWrapper`. - - Imports are deferred to avoid requiring Chrome-specific Selenium packages - when only Playwright is used. - - :param endpoint_url: CDP endpoint URL. - :type endpoint_url: str - :param viewport_size: Optional viewport dimensions. - :type viewport_size: typing.Optional[typing.Dict[str, int]] - :return: Initialized :class:`DriverWrapper`. - :rtype: DriverWrapper - """ - from selenium.webdriver.chrome.options import Options as ChromeOptions - from selenium.webdriver.chrome.webdriver import WebDriver as ChromeWebDriver - from urllib.parse import urlparse - - parsed = urlparse(endpoint_url) - debugger_address = f'{parsed.hostname}:{parsed.port}' - - options = ChromeOptions() - options.debugger_address = debugger_address - - selenium_driver = ChromeWebDriver(options=options) - - if viewport_size: - selenium_driver.set_window_size(viewport_size['width'], viewport_size['height']) - - driver = Driver(driver=selenium_driver) - wrapper = cls(driver) - wrapper.is_cdp = True - return wrapper - def quit(self, silent: bool = False, trace_path: str = 'trace.zip'): """ Quit the driver instance. @@ -325,10 +195,6 @@ def quit(self, silent: bool = False, trace_path: str = 'trace.zip'): self._base_cls.quit(self, trace_path) self.session.remove_session(self) - if getattr(self, '_playwright_instance', None): - self._playwright_instance.stop() - self._playwright_instance = None - def save_screenshot( self, file_name: str, diff --git a/tests/static_tests/unit/test_cdp_robustness.py b/tests/static_tests/unit/test_cdp_robustness.py new file mode 100644 index 00000000..98457ccf --- /dev/null +++ b/tests/static_tests/unit/test_cdp_robustness.py @@ -0,0 +1,136 @@ +import pytest + +from mock.mock import MagicMock, PropertyMock, patch + +from playwright.sync_api import Page as PlaywrightSourcePage, Browser +from playwright._impl._errors import Error as PlaywrightError + +from selenium.common.exceptions import WebDriverException as SeleniumWebDriverException +from selenium.webdriver.remote.webdriver import WebDriver as SeleniumDriver + +from mops.base.driver_wrapper import DriverWrapper, DriverWrapperSessions +from mops.mixins.objects.driver import Driver + + +@pytest.fixture(autouse=True) +def cleanup_sessions(): + yield + DriverWrapperSessions.all_sessions = [] + + +def _make_playwright_wrapper(is_cdp=False): + mock_page = PlaywrightSourcePage(MagicMock()) + mock_context = MagicMock() + mock_browser = Browser(MagicMock()) + + wrapper = DriverWrapper(Driver(driver=mock_page, context=mock_context, instance=mock_browser)) + wrapper.is_cdp = is_cdp + return wrapper, mock_page, mock_context + + +def _make_selenium_wrapper(is_cdp=False): + selenium_driver = SeleniumDriver + selenium_driver.__init__ = lambda *args, **kwargs: None + selenium_driver.session_id = None + selenium_driver.command_executor = MagicMock() + selenium_driver.error_handler = MagicMock() + selenium_driver.caps = {} + + instance = selenium_driver() + wrapper = DriverWrapper(Driver(driver=instance)) + wrapper.is_cdp = is_cdp + return wrapper, instance + + +class TestIsCdpFlag: + + def test_is_cdp_defaults_to_false_playwright(self): + wrapper, _, _ = _make_playwright_wrapper() + assert wrapper.is_cdp is False + + def test_is_cdp_defaults_to_false_selenium(self): + wrapper, _ = _make_selenium_wrapper() + assert wrapper.is_cdp is False + + def test_is_cdp_can_be_set_true_playwright(self): + wrapper, _, _ = _make_playwright_wrapper(is_cdp=True) + assert wrapper.is_cdp is True + + def test_is_cdp_can_be_set_true_selenium(self): + wrapper, _ = _make_selenium_wrapper(is_cdp=True) + assert wrapper.is_cdp is True + + +class TestPlayDriverCdpQuit: + + def test_cdp_quit_suppresses_page_close_error(self): + wrapper, mock_page, mock_context = _make_playwright_wrapper(is_cdp=True) + mock_page.close = MagicMock(side_effect=PlaywrightError('Target page closed')) + mock_context.close = MagicMock() + + wrapper.quit(silent=True) + mock_page.close.assert_called_once() + + def test_cdp_quit_suppresses_context_close_error(self): + wrapper, mock_page, mock_context = _make_playwright_wrapper(is_cdp=True) + mock_page.close = MagicMock() + mock_context.close = MagicMock(side_effect=PlaywrightError('Context closed')) + + wrapper.quit(silent=True) + mock_context.close.assert_called_once() + + def test_cdp_quit_skips_tracing(self): + wrapper, _, mock_context = _make_playwright_wrapper(is_cdp=True) + wrapper.quit(silent=True, trace_path='trace.zip') + mock_context.tracing.stop.assert_not_called() + + def test_non_cdp_quit_calls_tracing(self): + wrapper, mock_page, mock_context = _make_playwright_wrapper(is_cdp=False) + mock_page.close = MagicMock() + mock_context.close = MagicMock() + wrapper.quit(silent=True, trace_path='trace.zip') + mock_context.tracing.stop.assert_called_once_with(path='trace.zip') + + def test_non_cdp_quit_propagates_close_error(self): + wrapper, mock_page, mock_context = _make_playwright_wrapper(is_cdp=False) + mock_page.close = MagicMock(side_effect=PlaywrightError('Unexpected error')) + mock_context.tracing.stop = MagicMock() + + with pytest.raises(PlaywrightError): + wrapper.quit(silent=True) + + +class TestCoreDriverCdpQuit: + + def test_cdp_quit_suppresses_webdriver_error(self): + wrapper, driver_instance = _make_selenium_wrapper(is_cdp=True) + driver_instance.quit = MagicMock(side_effect=SeleniumWebDriverException('Browser already closed')) + + wrapper.quit(silent=True) + driver_instance.quit.assert_called_once() + + def test_non_cdp_quit_propagates_webdriver_error(self): + wrapper, driver_instance = _make_selenium_wrapper(is_cdp=False) + driver_instance.quit = MagicMock(side_effect=SeleniumWebDriverException('Unexpected error')) + + with pytest.raises(SeleniumWebDriverException): + wrapper.quit(silent=True) + + +class TestPlayDriverViewportNullSafe: + + def test_get_inner_window_size_returns_zero_when_viewport_none(self): + wrapper, mock_page, _ = _make_playwright_wrapper() + type(mock_page).viewport_size = PropertyMock(return_value=None) + + size = wrapper.get_inner_window_size() + assert size.width == 0 + assert size.height == 0 + + def test_get_inner_window_size_returns_values_when_viewport_set(self): + wrapper, mock_page, _ = _make_playwright_wrapper() + type(mock_page).viewport_size = PropertyMock(return_value={'width': 1920, 'height': 1080}) + + size = wrapper.get_inner_window_size() + assert size.width == 1920 + assert size.height == 1080 diff --git a/tests/static_tests/unit/test_connect_cdp.py b/tests/static_tests/unit/test_connect_cdp.py deleted file mode 100644 index 0f145028..00000000 --- a/tests/static_tests/unit/test_connect_cdp.py +++ /dev/null @@ -1,244 +0,0 @@ -import pytest - -from mock.mock import MagicMock, patch - -from mops.base.driver_wrapper import DriverWrapper, DriverWrapperSessions -from mops.exceptions import DriverWrapperException - -from playwright.sync_api import Page as PlaywrightSourcePage - - -@pytest.fixture(autouse=True) -def cleanup_sessions(): - yield - DriverWrapperSessions.all_sessions = [] - - -# --- Playwright CDP tests --- - - -def _make_cdp_mocks(pages=None): - mock_pw = MagicMock() - - mock_page = pages[0] if pages else PlaywrightSourcePage(MagicMock()) - all_pages = pages or [mock_page] - - mock_context = MagicMock() - mock_context.pages = all_pages - - mock_browser = MagicMock() - mock_browser.contexts = [mock_context] - mock_browser.browser_type.name = 'chromium' - - mock_pw.chromium.connect_over_cdp.return_value = mock_browser - - return mock_pw, mock_browser, mock_context, all_pages - - -@patch('mops.base.driver_wrapper.sync_playwright') -def test_connect_cdp_creates_playwright_wrapper(mock_sync_playwright): - mock_pw, mock_browser, mock_context, pages = _make_cdp_mocks() - mock_sync_playwright.return_value.start.return_value = mock_pw - - wrapper = DriverWrapper.connect_cdp('http://localhost:9222') - - mock_pw.chromium.connect_over_cdp.assert_called_once_with( - 'http://localhost:9222', timeout=30000 - ) - assert wrapper.is_playwright is True - assert wrapper._playwright_instance is mock_pw - - -@patch('mops.base.driver_wrapper.sync_playwright') -def test_connect_cdp_with_custom_timeout(mock_sync_playwright): - mock_pw, _, _, _ = _make_cdp_mocks() - mock_sync_playwright.return_value.start.return_value = mock_pw - - DriverWrapper.connect_cdp('http://localhost:9222', timeout=60000) - - mock_pw.chromium.connect_over_cdp.assert_called_once_with( - 'http://localhost:9222', timeout=60000 - ) - - -@patch('mops.base.driver_wrapper.sync_playwright') -def test_connect_cdp_with_viewport_size(mock_sync_playwright): - mock_pw, _, _, pages = _make_cdp_mocks() - mock_sync_playwright.return_value.start.return_value = mock_pw - - page = pages[0] - page.set_viewport_size = MagicMock() - - DriverWrapper.connect_cdp( - 'http://localhost:9222', - viewport_size={'width': 1920, 'height': 1080}, - ) - - page.set_viewport_size.assert_called_once_with( - {'width': 1920, 'height': 1080} - ) - - -@patch('mops.base.driver_wrapper.sync_playwright') -def test_connect_cdp_with_page_index(mock_sync_playwright): - page_0 = PlaywrightSourcePage(MagicMock()) - page_1 = PlaywrightSourcePage(MagicMock()) - mock_pw, _, _, _ = _make_cdp_mocks(pages=[page_0, page_1]) - mock_sync_playwright.return_value.start.return_value = mock_pw - - wrapper = DriverWrapper.connect_cdp('http://localhost:9222', page_index=1) - - assert wrapper.driver is page_1 - - -@patch('mops.base.driver_wrapper.sync_playwright') -def test_connect_cdp_quit_stops_playwright(mock_sync_playwright): - mock_pw, _, _, _ = _make_cdp_mocks() - mock_sync_playwright.return_value.start.return_value = mock_pw - - wrapper = DriverWrapper.connect_cdp('http://localhost:9222') - wrapper.quit(silent=True) - - mock_pw.stop.assert_called_once() - - -@patch('mops.base.driver_wrapper.sync_playwright') -def test_connect_cdp_quit_without_cdp_no_playwright_stop(mock_sync_playwright): - """Regular DriverWrapper (non-CDP) quit should not call playwright stop.""" - mock_pw, _, _, _ = _make_cdp_mocks() - mock_sync_playwright.return_value.start.return_value = mock_pw - - wrapper = DriverWrapper.connect_cdp('http://localhost:9222') - delattr(wrapper, '_playwright_instance') - wrapper.quit(silent=True) - - mock_pw.stop.assert_not_called() - - -@patch('mops.base.driver_wrapper.sync_playwright') -def test_connect_cdp_session_tracking(mock_sync_playwright): - mock_pw, _, _, _ = _make_cdp_mocks() - mock_sync_playwright.return_value.start.return_value = mock_pw - - assert DriverWrapperSessions.sessions_count() == 0 - wrapper = DriverWrapper.connect_cdp('http://localhost:9222') - assert DriverWrapperSessions.sessions_count() == 1 - wrapper.quit(silent=True) - assert DriverWrapperSessions.sessions_count() == 0 - - -# --- Selenium CDP tests --- - - -@patch('mops.base.driver_wrapper.DriverWrapper._connect_cdp_selenium') -def test_connect_cdp_selenium_engine_dispatches(mock_selenium_connect): - mock_selenium_connect.return_value = MagicMock() - - DriverWrapper.connect_cdp('http://localhost:9222', engine='selenium') - - mock_selenium_connect.assert_called_once_with( - 'http://localhost:9222', None - ) - - -@patch('mops.base.driver_wrapper.DriverWrapper._connect_cdp_playwright') -def test_connect_cdp_playwright_engine_dispatches(mock_pw_connect): - mock_pw_connect.return_value = MagicMock() - - DriverWrapper.connect_cdp('http://localhost:9222', engine='playwright') - - mock_pw_connect.assert_called_once_with( - 'http://localhost:9222', 30000, 0, None - ) - - -def test_connect_cdp_invalid_engine(): - try: - DriverWrapper.connect_cdp('http://localhost:9222', engine='appium') - except DriverWrapperException as exc: - assert 'Unsupported engine' in exc.msg - assert 'appium' in exc.msg - else: - raise Exception('Expected DriverWrapperException') - - -@patch('mops.base.driver_wrapper.sync_playwright') -def test_connect_cdp_sets_is_cdp_flag_playwright(mock_sync_playwright): - mock_pw, _, _, _ = _make_cdp_mocks() - mock_sync_playwright.return_value.start.return_value = mock_pw - - wrapper = DriverWrapper.connect_cdp('http://localhost:9222') - - assert wrapper.is_cdp is True - assert wrapper.is_playwright is True - - -@patch('mops.base.driver_wrapper.DriverWrapper._connect_cdp_selenium') -def test_connect_cdp_sets_is_cdp_flag_selenium(mock_selenium_connect): - mock_wrapper = MagicMock() - mock_wrapper.is_cdp = False - mock_selenium_connect.return_value = mock_wrapper - - DriverWrapper.connect_cdp('http://localhost:9222', engine='selenium') - - mock_selenium_connect.assert_called_once() - - -@patch('mops.base.driver_wrapper.DriverWrapper.__new__') -@patch('mops.base.driver_wrapper.DriverWrapper.__init__', return_value=None) -def test_connect_cdp_selenium_sets_debugger_address(mock_init, mock_new): - """Verify that _connect_cdp_selenium correctly parses endpoint URL and sets debugger_address.""" - mock_instance = MagicMock() - mock_new.return_value = mock_instance - - with patch('mops.base.driver_wrapper.DriverWrapper._connect_cdp_selenium') as original: - original.side_effect = DriverWrapper._connect_cdp_selenium.__wrapped__ if hasattr( - DriverWrapper._connect_cdp_selenium, '__wrapped__' - ) else None - - from urllib.parse import urlparse - - for url, expected in [ - ('http://localhost:9222', 'localhost:9222'), - ('https://remote-host:9222/', 'remote-host:9222'), - ('http://http-proxy:9222', 'http-proxy:9222'), - ('http://192.168.1.100:9222', '192.168.1.100:9222'), - ]: - parsed = urlparse(url) - result = f'{parsed.hostname}:{parsed.port}' - assert result == expected, f'URL {url} parsed to {result}, expected {expected}' - - -# --- PlayDriver CDP quit behavior --- - - -@patch('mops.base.driver_wrapper.sync_playwright') -def test_playwright_cdp_quit_suppresses_close_errors(mock_sync_playwright): - """PlayDriver.quit() should suppress close errors for CDP connections.""" - from playwright._impl._errors import Error as PlaywrightError - - mock_pw, _, mock_context, pages = _make_cdp_mocks() - mock_sync_playwright.return_value.start.return_value = mock_pw - - page = pages[0] - page.close = MagicMock(side_effect=PlaywrightError('Target page closed')) - mock_context.close = MagicMock(side_effect=PlaywrightError('Context closed')) - - wrapper = DriverWrapper.connect_cdp('http://localhost:9222') - - wrapper.quit(silent=True) - - page.close.assert_called_once() - mock_context.close.assert_called_once() - - -@patch('mops.base.driver_wrapper.sync_playwright') -def test_playwright_cdp_quit_skips_tracing(mock_sync_playwright): - """PlayDriver.quit() should not call tracing.stop() for CDP connections.""" - mock_pw, _, mock_context, _ = _make_cdp_mocks() - mock_sync_playwright.return_value.start.return_value = mock_pw - - wrapper = DriverWrapper.connect_cdp('http://localhost:9222') - wrapper.quit(silent=True, trace_path='trace.zip') - - mock_context.tracing.stop.assert_not_called() From e637f65bed31f054ffeec73850a588b366374df5 Mon Sep 17 00:00:00 2001 From: wwakabobik Date: Mon, 8 Jun 2026 14:43:32 +0200 Subject: [PATCH 3/5] Fix CHANGELOG: move is_cdp additions to v3.5.1 (after upstream v3.5.0) Co-authored-by: Cursor --- CHANGELOG.md | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cf60febf..720e839f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@
+## v3.5.1 + +### Added +- `DriverWrapper.is_cdp` flag to identify CDP-connected driver instances +- `PlayDriver.quit` graceful error handling for CDP contexts; tracing skip when `is_cdp` is set +- `CoreDriver.quit` graceful error handling for externally-managed browsers when `is_cdp` is set +- `PlayDriver.get_inner_window_size` null-safe viewport handling (returns `Size(0, 0)` when viewport is `None`) + +--- + ## v3.5.0 *Release date: 2026-04-22* @@ -26,10 +36,6 @@ *Release date: 2026-04-21* ### Added -- `DriverWrapper.is_cdp` flag to identify CDP-connected driver instances -- `PlayDriver.quit` graceful error handling for CDP contexts; tracing skip when `is_cdp` is set -- `CoreDriver.quit` graceful error handling for externally-managed browsers when `is_cdp` is set -- `PlayDriver.get_inner_window_size` null-safe viewport handling (returns `Size(0, 0)` when viewport is `None`) - `set_local_storage_item(items)` / `set_session_storage_item(items)` — set one or more key/value pairs in localStorage / sessionStorage - `get_local_storage_item(key)` / `get_session_storage_item(key)` — retrieve a single item by key (`None` if missing) - `get_local_storage_items()` / `get_session_storage_items()` — retrieve all items as a dict From ba0108557f5b74d6e910096f9b0995676f6fe5d6 Mon Sep 17 00:00:00 2001 From: wwakabobik Date: Thu, 25 Jun 2026 18:59:37 +0200 Subject: [PATCH 4/5] Fix ruff SIM105 in CDP close paths. Use contextlib.suppress for Playwright and Selenium teardown so the ruff CI job passes. Co-authored-by: Cursor --- mops/playwright/play_driver.py | 8 ++------ mops/selenium/core/core_driver.py | 5 ++--- 2 files changed, 4 insertions(+), 9 deletions(-) diff --git a/mops/playwright/play_driver.py b/mops/playwright/play_driver.py index e4c0d33b..2deb71bc 100644 --- a/mops/playwright/play_driver.py +++ b/mops/playwright/play_driver.py @@ -180,15 +180,11 @@ def quit(self, silent: bool = False, trace_path: str = 'trace.zip') -> None: self.context.tracing.stop(path=trace_path) if self.is_cdp: - try: + with contextlib.suppress(PlaywrightError): self._base_driver.close() - except PlaywrightError: - pass - try: + with contextlib.suppress(PlaywrightError): self.context.close() - except PlaywrightError: - pass else: self._base_driver.close() self.context.close() diff --git a/mops/selenium/core/core_driver.py b/mops/selenium/core/core_driver.py index 18578173..4ee66df7 100644 --- a/mops/selenium/core/core_driver.py +++ b/mops/selenium/core/core_driver.py @@ -1,5 +1,6 @@ from __future__ import annotations +import contextlib from functools import cached_property import time from typing import TYPE_CHECKING, Any @@ -224,10 +225,8 @@ def quit(self, silent: bool = False, trace_path: str = 'trace.zip') -> None: :return: :obj:`None` """ if self.is_cdp: - try: + with contextlib.suppress(SeleniumWebDriverException): self.driver.quit() - except SeleniumWebDriverException: - pass else: self.driver.quit() From 1d10f3c6ac0475a5e4092297f652e8100472ee8c Mon Sep 17 00:00:00 2001 From: VladimirPodolian <36446855+VladimirPodolian@users.noreply.github.com> Date: Thu, 9 Jul 2026 12:03:15 +0200 Subject: [PATCH 5/5] Update __init__.py --- mops/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mops/__init__.py b/mops/__init__.py index d8eeb6bf..5b2143f1 100644 --- a/mops/__init__.py +++ b/mops/__init__.py @@ -1,4 +1,4 @@ """Wrapper of Selenium, Appium and Playwright with a single API.""" -__version__ = '3.5.1' +__version__ = '3.5.2' __project_name__ = 'mops'