From 647689bc450fc2941ec7abd9efdc894f41a79905 Mon Sep 17 00:00:00 2001 From: Nasif Date: Wed, 29 Jul 2026 17:36:22 +0600 Subject: [PATCH 1/4] Unify Selenium and Playwright Chrome management - resolve Playwright Chrome launches through the shared Chrome-for-Testing browser and driver cache - centralize exact-version and Stable/Beta/Dev/Canary normalization with per-channel cache metadata - restrict Playwright browser creation to supported Chrome/Chromium variants and reject system Chrome - reuse named Selenium and Playwright sessions before resolving or launching another browser - persist the paired ChromeDriver path for exact Selenium CDP attachment - share Chromium defaults and Zeuz AI extension arguments across both launchers - use a temporary persistent Playwright context when extensions are enabled - remove the standalone Playwright browser installer and download cache path - add mocked regression coverage for shared binaries, extension loading, session reuse, and driver attachment --- .../Web/Playwright/BuiltInFunctions.py | 222 ++++++++++-------- .../Web/Playwright/utils.py | 157 ------------- .../Web/Selenium/BuiltInFunctions.py | 59 ++--- .../Built_In_Automation/Web/Selenium/utils.py | 71 ++++-- tests/test_browser_sessions.py | 85 +++++++ 5 files changed, 297 insertions(+), 297 deletions(-) delete mode 100644 Framework/Built_In_Automation/Web/Playwright/utils.py diff --git a/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py b/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py index 44255090..cf67a00e 100644 --- a/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py @@ -42,13 +42,12 @@ from Framework.Utilities import CommonUtil, ConfigModule from Framework.Utilities.decorators import logger -from settings import ZEUZ_NODE_DOWNLOADS_DIR from Framework.Built_In_Automation.Shared_Resources import ( BuiltInFunctionSharedResources as sr, ) from Framework.Built_In_Automation.Shared_Resources import LocateElement as PlaywrightLocator from Framework.Utilities.CommonUtil import failed_tag_list -from . import utils as PlaywrightUtils +from Framework.Built_In_Automation.Web.Selenium.utils import ChromeForTesting from Framework.Built_In_Automation.Web.utils import ( create_browser_session, extract_session_name, @@ -159,34 +158,6 @@ def _save_current_playwright_frame(frame_locator): sr.Set_Shared_Variables("browser_sessions", sessions) -def _get_installed_cft_chromedriver(browser_version): - system = platform.system().lower() - arch = platform.machine().lower() - - if system == "windows": - platform_key = "win64" if arch in ("amd64", "x86_64") else "win32" - executable = "chromedriver.exe" - elif system == "darwin": - platform_key = "mac-arm64" if arch == "arm64" else "mac-x64" - executable = "chromedriver" - elif system == "linux": - platform_key = "linux64" - executable = "chromedriver" - else: - return None - - driver_path = ( - ZEUZ_NODE_DOWNLOADS_DIR - / "chrome_for_testing" - / "versions" - / browser_version - / "driver" - / f"chromedriver-{platform_key}" - / executable - ) - return str(driver_path) if driver_path.exists() else None - - async def _activate_browser_session_for_action(step_data, function_name=None): """Select the requested browser session before running Playwright actions.""" @@ -216,7 +187,7 @@ async def _activate_browser_session_for_action(step_data, function_name=None): return "passed" -def connect_selenium_to_playwright(port=9222): +def connect_selenium_to_playwright(port=9222, driver_path=None): """Connect Selenium to Playwright browser via CDP""" try: from selenium import webdriver @@ -227,34 +198,39 @@ def connect_selenium_to_playwright(port=9222): options = Options() options.add_experimental_option("debuggerAddress", f"127.0.0.1:{port}") - service = None - try: - response = requests.get(f"http://127.0.0.1:{port}/json/version", timeout=5) - response.raise_for_status() - browser_version = ( - response.json() - .get("Browser", "") - .split("/", 1)[-1] - .strip() + service = Service(executable_path=str(driver_path)) if driver_path else None + if service: + CommonUtil.ExecLog( + "connect_selenium_to_playwright", + f"Using cached ChromeDriver: {driver_path}", + 1, ) - if browser_version: - driver_path = _get_installed_cft_chromedriver(browser_version) - if not driver_path: + else: + try: + response = requests.get(f"http://127.0.0.1:{port}/json/version", timeout=5) + response.raise_for_status() + browser_version = ( + response.json() + .get("Browser", "") + .split("/", 1)[-1] + .strip() + ) + if browser_version: driver_path = ChromeDriverManager( driver_version=browser_version ).install() - service = Service(executable_path=driver_path) + service = Service(executable_path=driver_path) + CommonUtil.ExecLog( + "connect_selenium_to_playwright", + f"Using ChromeDriver matching browser version {browser_version}", + 1, + ) + except Exception: CommonUtil.ExecLog( "connect_selenium_to_playwright", - f"Using ChromeDriver matching browser version {browser_version}", - 1, + "Could not resolve matching ChromeDriver for Playwright browser; falling back to Selenium Manager", + 2, ) - except Exception: - CommonUtil.ExecLog( - "connect_selenium_to_playwright", - "Could not resolve matching ChromeDriver for Playwright browser; falling back to Selenium Manager", - 2, - ) if service: driver = webdriver.Chrome(service=service, options=options) @@ -431,7 +407,7 @@ async def Open_Browser(step_data): add argument optional parameter --disable-gpu open browser playwright action open browser - Supported browsers: chrome, chromium, firefox, webkit, safari, edge + Supported browsers: chrome, chromium, chrome-beta """ sModuleInfo = inspect.currentframe().f_code.co_name + " : " + MODULE_NAME global playwright_instance, browser, context, current_page @@ -442,9 +418,13 @@ async def Open_Browser(step_data): url = None dependency = sr.Get_Shared_Variables("dependency") browser_name = "chromium" + dependency_browser = "" if isinstance(dependency, dict) and dependency.get("Browser"): - browser_name = dependency["Browser"].strip().lower().replace("headless", "").strip() or browser_name - headless = False + dependency_browser = dependency["Browser"].strip().lower() + browser_name = dependency_browser + headless = dependency_browser.replace(" ", "") == "chromeheadless" + headless_explicit = False + chrome_version = None viewport = default_viewport.copy() resolution = None args = [] @@ -476,6 +456,7 @@ async def Open_Browser(step_data): elif mid_l == "optional parameter": if left_l == "headless": + headless_explicit = True headless = right_v.lower() in ("true", "yes", "1") elif left_l == "resolution": parts = right_v.replace("x", ",").split(",") @@ -513,9 +494,50 @@ async def Open_Browser(step_data): # Handle Selenium-style capabilities where possible pass - # Ensure Playwright's managed browser is available in Zeuz's persistent cache. - success = PlaywrightUtils.ensure_playwright_browser_installed(sModuleInfo, browser_name) - if not success: + if _compact(left_l) == "chrome:version": + chrome_version = right_v + + compact_browser_name = _compact(browser_name) + if compact_browser_name == "chromeheadless": + browser_name = "chrome" + if not headless_explicit: + headless = True + elif compact_browser_name in ("chrome", "chromium", "chromebeta"): + browser_name = "chrome-beta" if compact_browser_name == "chromebeta" else compact_browser_name + else: + CommonUtil.ExecLog( + sModuleInfo, + f"Playwright only supports Chrome/Chromium; browser '{browser_name}' is not supported", + 3, + ) + return "zeuz_failed" + + if chrome_version and chrome_version.strip().lower() == "system": + CommonUtil.ExecLog( + sModuleInfo, + "Playwright requires Chrome for Testing; chrome:version = system is not supported", + 3, + ) + return "zeuz_failed" + + existing_session = get_browser_session(page_id) + if existing_session: + result = await _ensure_playwright_session(page_id, existing_session) + if result not in failed_tag_list: + if url: + await current_page.goto(url, wait_until="domcontentloaded") + CommonUtil.ExecLog(sModuleInfo, f"Using existing browser session: {page_id}", 1) + return "passed" + + chrome_channel = "Beta" if browser_name == "chrome-beta" else None + chrome_bin, driver_bin = await asyncio.to_thread( + lambda: ChromeForTesting().setup_chrome_for_testing( + chrome_version, + chrome_channel, + ) + ) + if not chrome_bin or not driver_bin: + CommonUtil.ExecLog(sModuleInfo, "Failed to setup Chrome for Testing browser and driver", 3) return "zeuz_failed" # Launch Playwright @@ -526,25 +548,34 @@ async def Open_Browser(step_data): launch_options = { "headless": headless, "slow_mo": slow_mo, + "executable_path": str(chrome_bin), } # Add remote debugging port for CDP connection with unique port per session unique_port = get_debug_port(page_id) - all_args = args + [f"--remote-debugging-port={unique_port}"] - chromium_like_browser = browser_name not in ("firefox", "webkit", "safari") - if chromium_like_browser: - if resolution and not _has_chromium_arg(all_args, ("--window-size",)): - all_args.append( - f"--window-size={resolution['width']},{resolution['height']}" - ) - elif ( - not headless - and not _has_chromium_arg( - all_args, - ("--window-size", "--start-maximized", "--kiosk"), - ) - ): - all_args.append("--start-maximized") + from Framework.Built_In_Automation.Web.Selenium import BuiltInFunctions as SeleniumBuiltInFunctions + + selenium_browser_name = "chromeheadless" if headless else "chrome" + all_args = ( + list(SeleniumBuiltInFunctions.DEFAULT_CHROMIUM_ARGUMENTS) + + args + + SeleniumBuiltInFunctions.get_zeuz_ai_extension_arguments( + selenium_browser_name + ) + + [f"--remote-debugging-port={unique_port}"] + ) + if resolution and not _has_chromium_arg(all_args, ("--window-size",)): + all_args.append( + f"--window-size={resolution['width']},{resolution['height']}" + ) + elif ( + not headless + and not _has_chromium_arg( + all_args, + ("--window-size", "--start-maximized", "--kiosk"), + ) + ): + all_args.append("--start-maximized") if devtools: all_args.append("--auto-open-devtools-for-tabs") CommonUtil.ExecLog(sModuleInfo, f"Using remote debugging port {unique_port} for session '{page_id}'", 1) @@ -552,31 +583,10 @@ async def Open_Browser(step_data): launch_options["args"] = all_args if downloads_path: launch_options["downloads_path"] = downloads_path - - selenium_cdp_supported = True - - # Select and launch browser - if browser_name in ("chrome", "chromium"): - browser = await playwright_instance.chromium.launch(**launch_options) - elif browser_name == "firefox": - selenium_cdp_supported = False - browser = await playwright_instance.firefox.launch(**launch_options) - elif browser_name in ("webkit", "safari"): - selenium_cdp_supported = False - browser = await playwright_instance.webkit.launch(**launch_options) - elif browser_name in ("edge", "msedge", "microsoft edge"): - launch_options["channel"] = "msedge" - browser = await playwright_instance.chromium.launch(**launch_options) - elif browser_name == "chrome-beta": - launch_options["channel"] = "chrome-beta" - browser = await playwright_instance.chromium.launch(**launch_options) - else: - CommonUtil.ExecLog(sModuleInfo, f"Unknown browser '{browser_name}', using chromium", 2) - browser = await playwright_instance.chromium.launch(**launch_options) # Context options. Headed Chromium sessions use the real browser window # size so attached Selenium code observes Selenium-like layout behavior. - if selenium_cdp_supported and not headless: + if not headless: context_options = {"no_viewport": True, "accept_downloads": True} else: context_options = {"viewport": viewport, "accept_downloads": True} @@ -593,10 +603,22 @@ async def Open_Browser(step_data): if color_scheme: context_options["color_scheme"] = color_scheme - # Create context and page - context = await browser.new_context(**context_options) + extension_enabled = any( + argument.startswith("--load-extension=") for argument in all_args + ) + if extension_enabled: + context = await playwright_instance.chromium.launch_persistent_context( + "", + **launch_options, + **context_options, + ) + browser = context.browser + else: + browser = await playwright_instance.chromium.launch(**launch_options) + context = await browser.new_context(**context_options) + context.set_default_timeout(timeout) - current_page = await context.new_page() + current_page = context.pages[0] if context.pages else await context.new_page() current_page_id = page_id # Store in details @@ -606,6 +628,7 @@ async def Open_Browser(step_data): "browser": browser, "playwright": playwright_instance, "remote-debugging-port": unique_port, + "driver-path": str(driver_bin), } # Navigate if URL provided @@ -634,7 +657,8 @@ async def Open_Browser(step_data): playwright_instance=playwright_instance, remote_debugging_port=unique_port, ) - session["selenium_cdp_supported"] = selenium_cdp_supported + session["selenium_cdp_supported"] = True + session["driver_path"] = str(driver_bin) sr.Set_Shared_Variables("browser_sessions", get_browser_sessions()) CommonUtil.ExecLog(sModuleInfo, f"Created browser session: {page_id}", 5) diff --git a/Framework/Built_In_Automation/Web/Playwright/utils.py b/Framework/Built_In_Automation/Web/Playwright/utils.py deleted file mode 100644 index f6e531f7..00000000 --- a/Framework/Built_In_Automation/Web/Playwright/utils.py +++ /dev/null @@ -1,157 +0,0 @@ -# -*- coding: utf-8 -*- -""" -Playwright Utility Functions for Zeuz Node - -This module provides utility functions for Playwright automation, -including browser download and setup functionality. - -Author: Zeuz/AutomationSolutionz -""" - -import os -import subprocess -import sys -from pathlib import Path - -from filelock import FileLock - -from Framework.Utilities import CommonUtil -from settings import ZEUZ_NODE_DOWNLOADS_DIR - -PLAYWRIGHT_BROWSERS_DIR = ZEUZ_NODE_DOWNLOADS_DIR / "playwright_browsers" -PLAYWRIGHT_INSTALLABLE_BROWSERS = { - "chromium": "chromium", - "chrome": "chromium", - "firefox": "firefox", - "webkit": "webkit", - "safari": "webkit", -} -PLAYWRIGHT_SYSTEM_CHANNEL_BROWSERS = { - "edge", - "msedge", - "microsoft edge", - "chrome-beta", -} - - -def _set_playwright_browsers_path(): - """Use Zeuz's persistent downloads directory for Playwright browser binaries.""" - - PLAYWRIGHT_BROWSERS_DIR.mkdir(parents=True, exist_ok=True) - os.environ["PLAYWRIGHT_BROWSERS_PATH"] = str(PLAYWRIGHT_BROWSERS_DIR) - return PLAYWRIGHT_BROWSERS_DIR - - -def _get_playwright_browser_name(browser_name): - browser_name = (browser_name or "chromium").strip().lower() - return PLAYWRIGHT_INSTALLABLE_BROWSERS.get(browser_name) - - -def _get_playwright_executable_path(browser_name): - result = subprocess.run( - [ - sys.executable, - "-c", - ( - "from playwright.sync_api import sync_playwright\n" - "with sync_playwright() as p:\n" - f" print(p.{browser_name}.executable_path)\n" - ), - ], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - env=os.environ.copy(), - ) - - if result.returncode != 0: - return None - - executable_path = result.stdout.strip().splitlines()[-1] if result.stdout.strip() else "" - return Path(executable_path) if executable_path else None - - -def _is_playwright_browser_installed(browser_name): - executable_path = _get_playwright_executable_path(browser_name) - return bool(executable_path and executable_path.exists()) - - -def ensure_playwright_browser_installed(sModuleInfo, browser_name="chromium"): - """ - Ensure Playwright's managed browser is installed in Zeuz's persistent cache. - - Args: - sModuleInfo: Module information for logging - browser_name: Requested Playwright browser/channel name - - Returns: - bool: True if the browser is ready or no managed download is required - """ - try: - browsers_dir = _set_playwright_browsers_path() - requested_browser = (browser_name or "chromium").strip().lower() - install_browser = _get_playwright_browser_name(requested_browser) - - if requested_browser in PLAYWRIGHT_SYSTEM_CHANNEL_BROWSERS: - CommonUtil.ExecLog( - sModuleInfo, - f"Using Playwright browser cache: {browsers_dir}. Browser '{browser_name}' uses a system channel.", - 1, - ) - return True - - if not install_browser: - install_browser = "chromium" - CommonUtil.ExecLog( - sModuleInfo, - f"Unknown browser '{browser_name}', preparing Playwright chromium", - 2, - ) - - CommonUtil.ExecLog( - sModuleInfo, - f"Ensuring Playwright {install_browser} browser is installed in {browsers_dir}", - 1, - ) - - if _is_playwright_browser_installed(install_browser): - CommonUtil.ExecLog( - sModuleInfo, - f"Playwright {install_browser} browser already exists in {browsers_dir}", - 1, - ) - return True - - lock_path = browsers_dir / f"{install_browser}.install.lock" - with FileLock(str(lock_path)): - if _is_playwright_browser_installed(install_browser): - CommonUtil.ExecLog( - sModuleInfo, - f"Playwright {install_browser} browser already exists in {browsers_dir}", - 1, - ) - return True - - result = subprocess.run( - [sys.executable, "-m", "playwright", "install", "--with-deps", install_browser], - env=os.environ.copy(), - ) - - if result.returncode == 0: - CommonUtil.ExecLog( - sModuleInfo, - f"Playwright {install_browser} browser is ready", - 1, - ) - return True - - CommonUtil.ExecLog( - sModuleInfo, - f"Failed to install Playwright {install_browser} browser. See terminal output for details.", - 3, - ) - return False - - except Exception as e: - CommonUtil.ExecLog(sModuleInfo, f"Error setting up Playwright browser: {str(e)}", 3) - return False diff --git a/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py b/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py index 1f9f2f91..bb952ffa 100644 --- a/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py @@ -119,6 +119,11 @@ / "AI_Recorder_2" / "public" ) +DEFAULT_CHROMIUM_ARGUMENTS = ( + "--ignore-certificate-errors", + "--ignore-ssl-errors", + "--zeuz_pid_finder", +) # Disable WebdriverManager SSL verification. os.environ["WDM_SSL_VERIFY"] = "0" @@ -199,7 +204,10 @@ def _ensure_selenium_session(session_name, existing_session): try: from Framework.Built_In_Automation.Web.Playwright import BuiltInFunctions as PlaywrightBuiltInFunctions - driver = PlaywrightBuiltInFunctions.connect_selenium_to_playwright(port=port) + driver = PlaywrightBuiltInFunctions.connect_selenium_to_playwright( + port=port, + driver_path=existing_session.get("driver_path"), + ) if driver in failed_tag_list: return "zeuz_failed" @@ -759,6 +767,26 @@ def set_extension_variables(): ) +def get_zeuz_ai_extension_arguments(browser): + """Return Zeuz AI extension arguments for supported headed debug browsers.""" + if not ( + CommonUtil.debug_status + and ConfigModule.get_config_value("Inspector", "ai_plugin").strip().lower() + in ("true", "on", "enable", "yes", "on_debug") + and browser in ("chrome", "microsoft edge chromium") + ): + return [] + + set_extension_variables() + extension_paths = f"{aiplugin_path},{ai_recorder_path}" + return [ + "--disable-features=DisableLoadExtensionCommandLineSwitch", + f"--disable-extensions-except={extension_paths}", + f"--load-extension={extension_paths}", + "--allow-running-insecure-content", + ] + + def generate_options(browser: str, browser_options: BrowserOptions): """Adds capabilities and options for Browser/WebDriver""" sModuleInfo = inspect.currentframe().f_code.co_name + " : " + MODULE_NAME @@ -837,18 +865,8 @@ def headless(): for key, value in browser_options["capabilities"].items(): options.set_capability(key, value) - # On Debug run open inspector with credentials - if ( - CommonUtil.debug_status - and ConfigModule.get_config_value("Inspector", "ai_plugin").strip().lower() - in ("true", "on", "enable", "yes", "on_debug") - and browser in ("chrome", "microsoft edge chromium") - ): - set_extension_variables() - options.add_argument("--disable-features=DisableLoadExtensionCommandLineSwitch") - options.add_argument(f"load-extension={aiplugin_path},{ai_recorder_path}") - # This is for running extension on a http server to call a https request - options.add_argument("--allow-running-insecure-content") + for argument in get_zeuz_ai_extension_arguments(browser): + options.add_argument(argument) msg += ( f"Capabilities: {json.dumps(options.capabilities, indent=2)}\n" @@ -1190,13 +1208,7 @@ async def Go_To_Link(dataset: Dataset) -> ReturnType: ) apps = "application/pdf;text/plain;application/text;text/xml;application/xml;application/xlsx;application/csv;application/zip" default_chromium_arguments = { - "add_argument": [ - "--ignore-certificate-errors", - "--ignore-ssl-errors", - "--zeuz_pid_finder", - # "--remote-debugging-port=9222", # Required for playright - # "--no-sandbox" - ], + "add_argument": list(DEFAULT_CHROMIUM_ARGUMENTS), "add_experimental_option": { "prefs": { # "profile.default_content_settings.popups": 0, @@ -1328,13 +1340,6 @@ async def Go_To_Link(dataset: Dataset) -> ReturnType: ) and not browser_options["chrome"].get("debugger_address", ""): cft = ChromeForTesting() - if chrome_version: - if chrome_version.strip().lower() in ("beta", "dev", "canary"): - chrome_channel = chrome_version.strip().capitalize() - chrome_version = None - else: - chrome_version = chrome_version.strip() - chrome_bin, driver_bin = cft.setup_chrome_for_testing( chrome_version, chrome_channel ) diff --git a/Framework/Built_In_Automation/Web/Selenium/utils.py b/Framework/Built_In_Automation/Web/Selenium/utils.py index 3c5de425..7632e987 100644 --- a/Framework/Built_In_Automation/Web/Selenium/utils.py +++ b/Framework/Built_In_Automation/Web/Selenium/utils.py @@ -33,6 +33,13 @@ class ChromeForTesting: CHROME_VERSIONS_DIR = CHROME_BASE_DIR / "versions" CHROME_INFO_FILE = CHROME_BASE_DIR / "info.json" CHROME_LINUX_UNAVAILABLE_DEPS_FILE = CHROME_BASE_DIR / "unavailable_linux_deps.txt" + CHANNELS = { + "stable": "Stable", + "beta": "Beta", + "dev": "Dev", + "canary": "Canary", + } + MIN_VERSION = (115, 0, 5763, 0) def __init__(self): self.system = platform.system().lower() @@ -251,8 +258,15 @@ def _save_info(self, info): def get_latest_version(self, channel="Stable", force_check=False): """Get the latest Chrome version with caching""" + channel = self.CHANNELS.get(str(channel).strip().lower()) + if not channel: + raise ValueError("Chrome channel must be Stable, Beta, Dev, or Canary") + info = self._load_info() - latest_info = info.get("latest", {}) + latest_by_channel = info.get("channels", {}) + latest_info = latest_by_channel.get(channel, {}) + if channel == "Stable" and not latest_info: + latest_info = info.get("latest", {}) cached_version = latest_info.get("version", "") last_check_str = latest_info.get("last_check", "") @@ -287,10 +301,13 @@ def get_latest_version(self, channel="Stable", force_check=False): new_version = data["channels"][channel]["version"] # Update info - info["latest"] = { + latest_info = { "version": new_version, "last_check": datetime.date.today().isoformat(), } + info.setdefault("channels", {})[channel] = latest_info + if channel == "Stable": + info["latest"] = latest_info self._save_info(info) return new_version @@ -534,23 +551,49 @@ def install_version(self, version): print(f"\nSuccessfully installed Chrome for Testing {version}") print(f"Installation directory: {version_dir.resolve()}") + @classmethod + def normalize_version_and_channel(cls, version=None, channel=None): + """Normalize an exact CfT version or a release channel.""" + version = str(version).strip() if version is not None else "" + channel = str(channel).strip() if channel is not None else "" + + if version.lower() in cls.CHANNELS: + channel = cls.CHANNELS[version.lower()] + version = "" + else: + channel = cls.CHANNELS.get(channel.lower() or "stable") + + if not channel: + raise ValueError("Chrome channel must be Stable, Beta, Dev, or Canary") + if not version or version.lower() == "system": + return version.lower() or None, channel + + try: + version_parts = tuple(int(part) for part in version.split(".")) + except ValueError: + raise ValueError( + "Chrome version must be an exact numeric version or Stable, Beta, Dev, or Canary" + ) from None + + if len(version_parts) != 4: + raise ValueError( + "Chrome version must be an exact four-part version, for example 138.0.7204.92" + ) + if version_parts < cls.MIN_VERSION: + raise ValueError("Chrome for Testing version must be at least 115.0.5763.0") + + return version, channel + def setup_chrome_for_testing(self, version=None, channel=None): """Setup Chrome for testing, install if necessary""" + version, channel = self.normalize_version_and_channel(version, channel) + # Clean up old versions first self.cleanup_old_versions() - if not channel: - channel = "Stable" - - if version: - if version < "115.0.5763.0": - print("Chrome for testing version must be at least: '115.0.5763.0'") - return None, None - if version.strip().lower() == "system": - print( - "Forcefully trying to use regular chrome instead of chrome for testing." - ) - return None, None + if version == "system": + print("Forcefully trying to use regular chrome instead of chrome for testing.") + return None, None # Use latest version if not specified if not version: diff --git a/tests/test_browser_sessions.py b/tests/test_browser_sessions.py index 0cd1887b..bd23d1dd 100644 --- a/tests/test_browser_sessions.py +++ b/tests/test_browser_sessions.py @@ -1,4 +1,5 @@ import asyncio +from pathlib import Path from unittest.mock import AsyncMock, MagicMock from Framework.Built_In_Automation.Shared_Resources import ( @@ -168,3 +169,87 @@ def test_playwright_switch_iframe_uses_index_parameter_after_default_reset(): frame_locator.nth.assert_called_once_with(1) assert sr.Get_Shared_Variables("playwright_frame") is indexed_frame_locator assert browser_utils.get_browser_session("default")["playwright_frame"] is indexed_frame_locator + + +def test_playwright_uses_shared_cft_pair_for_selenium_attachment(monkeypatch): + monkeypatch.setattr(playwright_bif.CommonUtil, "ExecLog", lambda *args, **kwargs: None) + monkeypatch.setattr( + playwright_bif.CommonUtil, + "set_screenshot_vars", + lambda *args, **kwargs: None, + ) + normalize = playwright_bif.ChromeForTesting.normalize_version_and_channel + assert normalize("138.0.7204.92") == ("138.0.7204.92", "Stable") + for channel in ("Stable", "Beta", "Dev", "Canary"): + assert normalize(channel.lower()) == (None, channel) + + chrome_bin = Path("/cache/chrome") + driver_bin = Path("/cache/chromedriver") + cft = MagicMock() + cft.setup_chrome_for_testing.return_value = (chrome_bin, driver_bin) + monkeypatch.setattr(playwright_bif, "ChromeForTesting", lambda: cft) + + async def run_inline(function, *args): + return function(*args) + + monkeypatch.setattr(playwright_bif.asyncio, "to_thread", run_inline) + monkeypatch.setattr(playwright_bif, "get_debug_port", lambda session_name: 9250) + + page = MagicMock() + context = MagicMock() + context.pages = [] + context.new_page = AsyncMock(return_value=page) + browser = MagicMock() + context.browser = browser + browser.new_context = AsyncMock(return_value=context) + playwright = MagicMock() + playwright.chromium.launch = AsyncMock(return_value=browser) + playwright.chromium.launch_persistent_context = AsyncMock(return_value=context) + starter = MagicMock() + starter.start = AsyncMock(return_value=playwright) + monkeypatch.setattr(playwright_bif, "async_playwright", lambda: starter) + extension_args = [ + "--disable-features=DisableLoadExtensionCommandLineSwitch", + f"--disable-extensions-except={selenium_bif.aiplugin_path},{selenium_bif.ai_recorder_path}", + f"--load-extension={selenium_bif.aiplugin_path},{selenium_bif.ai_recorder_path}", + "--allow-running-insecure-content", + ] + prepare_extensions = MagicMock() + monkeypatch.setattr(selenium_bif.CommonUtil, "debug_status", True) + monkeypatch.setattr( + selenium_bif.ConfigModule, + "get_config_value", + lambda section, key, *args: "true" + if (section, key) == ("Inspector", "ai_plugin") + else "", + ) + monkeypatch.setattr(selenium_bif, "set_extension_variables", prepare_extensions) + sr.Set_Shared_Variables("dependency", {"Browser": "Chrome"}) + + result = asyncio.run( + playwright_bif.Open_Browser( + [("chrome:version", "optional parameter", "138.0.7204.92")] + ) + ) + + assert result == "passed" + cft.setup_chrome_for_testing.assert_called_once_with("138.0.7204.92", None) + launch_options = playwright.chromium.launch_persistent_context.await_args.kwargs + assert launch_options["executable_path"] == str(chrome_bin) + assert launch_options["headless"] is False + assert set(selenium_bif.DEFAULT_CHROMIUM_ARGUMENTS + tuple(extension_args)) <= set( + launch_options["args"] + ) + prepare_extensions.assert_called_once_with() + playwright.chromium.launch.assert_not_awaited() + session = browser_utils.get_browser_session("default") + assert session["driver_path"] == str(driver_bin) + assert asyncio.run(playwright_bif.Open_Browser([])) == "passed" + cft.setup_chrome_for_testing.assert_called_once() + playwright.chromium.launch_persistent_context.assert_awaited_once() + + selenium_driver = object() + attach = MagicMock(return_value=selenium_driver) + monkeypatch.setattr(playwright_bif, "connect_selenium_to_playwright", attach) + assert selenium_bif._ensure_selenium_session("default", session) == "passed" + attach.assert_called_once_with(port=9250, driver_path=str(driver_bin)) From 5c74d8692bb72829d7285d71e35e786a33f13e82 Mon Sep 17 00:00:00 2001 From: Nasif Date: Wed, 29 Jul 2026 17:56:23 +0600 Subject: [PATCH 2/4] Add Selenium launch-field parity to Playwright - parse Selenium-compatible add argument, add extension, and add encoded extension fields for Chromium launches - cache and unpack CRX, ZIP, extension-ID downloads, and encoded extension payloads for persistent Playwright contexts - support experimental options, Chrome preferences, excluded switches, and mobile emulation - map shared capabilities for insecure certificates, proxy settings, Chrome options, and page-load strategy - support debugger-address CDP attachment without resolving or launching Chrome-for-Testing - preserve Selenium element waits and navigation strategy across Playwright sessions - create and safely clean temporary Chrome preference profiles - correct CRX2 and CRX3 header extraction and handle failed extension downloads - add regression coverage for launch-field compatibility, extension unpacking, profile preferences, session behavior, and debugger attachment --- .../Web/Playwright/BuiltInFunctions.py | 429 ++++++++++++++++-- .../Web/Selenium/BuiltInFunctions.py | 2 +- .../Built_In_Automation/Web/Selenium/utils.py | 17 +- tests/test_browser_sessions.py | 116 ++++- 4 files changed, 515 insertions(+), 49 deletions(-) diff --git a/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py b/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py index cf67a00e..83ad78b6 100644 --- a/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py @@ -21,10 +21,14 @@ """ import asyncio +import hashlib +import json +import shutil import sys import os import inspect import platform +import tempfile import time import base64 from pathlib import Path @@ -47,7 +51,10 @@ ) from Framework.Built_In_Automation.Shared_Resources import LocateElement as PlaywrightLocator from Framework.Utilities.CommonUtil import failed_tag_list -from Framework.Built_In_Automation.Web.Selenium.utils import ChromeForTesting +from Framework.Built_In_Automation.Web.Selenium.utils import ( + ChromeExtensionDownloader, + ChromeForTesting, +) from Framework.Built_In_Automation.Web.utils import ( create_browser_session, extract_session_name, @@ -82,6 +89,87 @@ def _has_chromium_arg(args, arg_names): return False +def _page_load_wait_until(strategy): + strategy = str(strategy or "eager").strip().lower() + try: + return { + "normal": "load", + "eager": "domcontentloaded", + "none": "commit", + "load": "load", + "domcontentloaded": "domcontentloaded", + "networkidle": "networkidle", + "commit": "commit", + }[strategy] + except KeyError: + raise ValueError( + "page load strategy must be normal, eager, none, load, domcontentloaded, networkidle, or commit" + ) from None + + +def _write_chrome_preferences(preferences): + """Write Selenium-style dotted Chrome preferences to a temporary profile.""" + nested_preferences = {} + for key, value in preferences.items(): + target = nested_preferences + parts = key.split(".") + for part in parts[:-1]: + target = target.setdefault(part, {}) + target[parts[-1]] = value + + user_data_dir = Path(tempfile.mkdtemp(prefix="zeuz-playwright-")) + default_profile = user_data_dir / "Default" + default_profile.mkdir() + (default_profile / "Preferences").write_text(json.dumps(nested_preferences)) + return str(user_data_dir) + + +def _cleanup_chrome_profile(user_data_dir): + if not user_data_dir: + return + profile = Path(user_data_dir) + if ( + profile.parent == Path(tempfile.gettempdir()) + and profile.name.startswith("zeuz-playwright-") + ): + shutil.rmtree(profile, ignore_errors=True) + + +def _unpack_playwright_extensions(extension_files, encoded_extensions): + """Cache CRX/base64 extension payloads and return unpacked directories.""" + payloads = [Path(path).read_bytes() for path in extension_files] + for encoded_extension in encoded_extensions: + try: + payloads.append( + base64.b64decode("".join(encoded_extension.split()), validate=True) + ) + except Exception: + raise ValueError("add encoded extension contains invalid base64 data") from None + + if not payloads: + return [] + + downloader = ChromeExtensionDownloader() + extension_dirs = [] + for payload in payloads: + cache_dir = ( + downloader.CHROME_EXTENSIONS_DIR + / "playwright" + / hashlib.sha256(payload).hexdigest() + ) + crx_path = cache_dir / "extension.crx" + unpacked_path = cache_dir / "extension" + if not (unpacked_path / "manifest.json").exists(): + cache_dir.mkdir(parents=True, exist_ok=True) + crx_path.write_bytes(payload) + downloader.extract_extension(crx_path) + if not (unpacked_path / "manifest.json").exists(): + raise ValueError("Chrome extension does not contain a manifest.json") + extension_dirs.append(str(unpacked_path)) + + return list(dict.fromkeys(extension_dirs)) + + def _set_active_playwright_session(session_name, session): """Update module globals/shared variables for a selected Playwright session.""" @@ -97,6 +185,10 @@ def _set_active_playwright_session(session_name, session): sr.Set_Shared_Variables("playwright_context", context) sr.Set_Shared_Variables("playwright_browser", browser) sr.Set_Shared_Variables("playwright_frame", session.get("playwright_frame")) + sr.Set_Shared_Variables( + "playwright_wait_until", + session.get("playwright_wait_until", "domcontentloaded"), + ) sr.Set_Shared_Variables("active_web_driver_type", "playwright") if session.get("selenium_driver"): sr.Set_Shared_Variables("selenium_driver", session["selenium_driver"]) @@ -425,6 +517,15 @@ async def Open_Browser(step_data): headless = dependency_browser.replace(" ", "") == "chromeheadless" headless_explicit = False chrome_version = None + extension_values = [] + encoded_extension_values = [] + chromium_argument_values = [] + experimental_option_values = [] + preference_values = [] + shared_capability_values = [] + debugger_address = None + page_load_strategy = "eager" + element_wait = None viewport = default_viewport.copy() resolution = None args = [] @@ -491,11 +592,33 @@ async def Open_Browser(step_data): page_id = right_v elif mid_l == "shared capability": - # Handle Selenium-style capabilities where possible - pass + shared_capability_values.append(right_v) - if _compact(left_l) == "chrome:version": + left_compact = _compact(left_l) + if left_compact == "chrome:version": chrome_version = right_v + elif mid_l in ("chromium option", "chrome option"): + if left_compact == "addargument": + chromium_argument_values.append(right_v) + elif left_compact == "addexperimentaloption": + experimental_option_values.append(right_v) + elif left_compact == "addextension": + extension_values.append(right_v) + elif left_compact == "addencodedextension": + encoded_extension_values.append(right_v) + elif left_compact == "setpreference": + preference_values.append(right_v) + elif left_compact == "pageloadstrategy": + page_load_strategy = right_v + elif left_compact == "debuggeraddress": + debugger_address = right_v + + if left_compact in ( + "waittimetoappearelement", + "waitforelement", + "elementwait", + ): + element_wait = float(right_v) compact_browser_name = _compact(browser_name) if compact_browser_name == "chromeheadless": @@ -512,7 +635,69 @@ async def Open_Browser(step_data): ) return "zeuz_failed" - if chrome_version and chrome_version.strip().lower() == "system": + from Framework.Built_In_Automation.Web.Selenium import BuiltInFunctions as SeleniumBuiltInFunctions + + capabilities = {} + for value in shared_capability_values: + parsed = CommonUtil.parse_value_into_object(value) + if not isinstance(parsed, dict): + raise ValueError("shared capability must be a dictionary") + capabilities.update(parsed) + + experimental_options = {} + for value in experimental_option_values: + experimental_options.update( + SeleniumBuiltInFunctions.parse_and_verify_datatype( + "addexperimentaloption", + value, + ) + ) + preferences = {} + for value in preference_values: + preferences.update( + SeleniumBuiltInFunctions.parse_and_verify_datatype( + "setpreference", + value, + ) + ) + for value in chromium_argument_values: + args.extend( + SeleniumBuiltInFunctions.parse_and_verify_datatype( + "addargument", + value, + ) + ) + + chrome_options = capabilities.get("goog:chromeOptions", {}) + if isinstance(chrome_options, dict): + args.extend(chrome_options.get("args", [])) + if chrome_options.get("extensions"): + encoded_extension_values.append( + repr(chrome_options["extensions"]) + ) + experimental_options.update( + { + key: value + for key, value in chrome_options.items() + if key not in ("args", "extensions") + } + ) + preferences.update(experimental_options.get("prefs", {})) + debugger_address = ( + debugger_address + or experimental_options.get("debuggerAddress") + ) + page_load_strategy = capabilities.get( + "pageLoadStrategy", + page_load_strategy, + ) + wait_until = _page_load_wait_until(page_load_strategy) + + if ( + not debugger_address + and chrome_version + and chrome_version.strip().lower() == "system" + ): CommonUtil.ExecLog( sModuleInfo, "Playwright requires Chrome for Testing; chrome:version = system is not supported", @@ -525,18 +710,22 @@ async def Open_Browser(step_data): result = await _ensure_playwright_session(page_id, existing_session) if result not in failed_tag_list: if url: - await current_page.goto(url, wait_until="domcontentloaded") + await current_page.goto(url, wait_until=wait_until) + if element_wait is not None: + sr.Set_Shared_Variables("element_wait", element_wait) CommonUtil.ExecLog(sModuleInfo, f"Using existing browser session: {page_id}", 1) return "passed" chrome_channel = "Beta" if browser_name == "chrome-beta" else None - chrome_bin, driver_bin = await asyncio.to_thread( - lambda: ChromeForTesting().setup_chrome_for_testing( - chrome_version, - chrome_channel, + chrome_bin = driver_bin = None + if not debugger_address: + chrome_bin, driver_bin = await asyncio.to_thread( + lambda: ChromeForTesting().setup_chrome_for_testing( + chrome_version, + chrome_channel, + ) ) - ) - if not chrome_bin or not driver_bin: + if not debugger_address and (not chrome_bin or not driver_bin): CommonUtil.ExecLog(sModuleInfo, "Failed to setup Chrome for Testing browser and driver", 3) return "zeuz_failed" @@ -548,39 +737,130 @@ async def Open_Browser(step_data): launch_options = { "headless": headless, "slow_mo": slow_mo, - "executable_path": str(chrome_bin), } + if chrome_bin: + launch_options["executable_path"] = str(chrome_bin) # Add remote debugging port for CDP connection with unique port per session - unique_port = get_debug_port(page_id) - from Framework.Built_In_Automation.Web.Selenium import BuiltInFunctions as SeleniumBuiltInFunctions + if debugger_address: + debugger_endpoint = ( + debugger_address + if "://" in debugger_address + else f"http://{debugger_address}" + ) + unique_port = urlparse(debugger_endpoint).port + else: + debugger_endpoint = None + unique_port = get_debug_port(page_id) + + extension_files = [] + encoded_extensions = [] + if not debugger_address: + resolved_chrome_version = next( + ( + parent.name + for parent in Path(chrome_bin).parents + if parent.parent.name == "versions" + ), + chrome_version, + ) + for value in extension_values: + extension_files.extend( + SeleniumBuiltInFunctions.parse_and_verify_datatype( + "addextension", + value, + resolved_chrome_version, + ) + ) + for value in encoded_extension_values: + encoded_extensions.extend( + SeleniumBuiltInFunctions.parse_and_verify_datatype( + "addencodedextension", + value, + ) + ) + extension_dirs = _unpack_playwright_extensions( + extension_files, + encoded_extensions, + ) selenium_browser_name = "chromeheadless" if headless else "chrome" - all_args = ( - list(SeleniumBuiltInFunctions.DEFAULT_CHROMIUM_ARGUMENTS) - + args - + SeleniumBuiltInFunctions.get_zeuz_ai_extension_arguments( - selenium_browser_name + zeuz_extension_args = [] + if not debugger_address: + zeuz_extension_args = ( + SeleniumBuiltInFunctions.get_zeuz_ai_extension_arguments( + selenium_browser_name + ) ) - + [f"--remote-debugging-port={unique_port}"] - ) - if resolution and not _has_chromium_arg(all_args, ("--window-size",)): - all_args.append( - f"--window-size={resolution['width']},{resolution['height']}" + extension_args = [] + for argument in zeuz_extension_args: + if argument.startswith("--load-extension="): + extension_dirs.extend(argument.split("=", 1)[1].split(",")) + elif not argument.startswith("--disable-extensions-except="): + extension_args.append(argument) + extension_dirs = list(dict.fromkeys(extension_dirs)) + if extension_dirs: + extension_paths = ",".join(extension_dirs) + if not any( + argument.startswith( + "--disable-features=DisableLoadExtensionCommandLineSwitch" + ) + for argument in extension_args + ): + extension_args.append( + "--disable-features=DisableLoadExtensionCommandLineSwitch" + ) + extension_args.extend( + ( + f"--disable-extensions-except={extension_paths}", + f"--load-extension={extension_paths}", + ) ) - elif ( - not headless - and not _has_chromium_arg( - all_args, - ("--window-size", "--start-maximized", "--kiosk"), + + all_args = [] + if not debugger_address: + all_args = ( + list(SeleniumBuiltInFunctions.DEFAULT_CHROMIUM_ARGUMENTS) + + args + + extension_args + + [f"--remote-debugging-port={unique_port}"] ) - ): - all_args.append("--start-maximized") - if devtools: - all_args.append("--auto-open-devtools-for-tabs") - CommonUtil.ExecLog(sModuleInfo, f"Using remote debugging port {unique_port} for session '{page_id}'", 1) + if resolution and not _has_chromium_arg(all_args, ("--window-size",)): + all_args.append( + f"--window-size={resolution['width']},{resolution['height']}" + ) + elif ( + not headless + and not _has_chromium_arg( + all_args, + ("--window-size", "--start-maximized", "--kiosk"), + ) + ): + all_args.append("--start-maximized") + if devtools: + all_args.append("--auto-open-devtools-for-tabs") + CommonUtil.ExecLog(sModuleInfo, f"Using remote debugging port {unique_port} for session '{page_id}'", 1) if all_args: launch_options["args"] = all_args + excluded_switches = experimental_options.get("excludeSwitches", []) + if excluded_switches: + launch_options["ignore_default_args"] = [ + switch if switch.startswith("--") else f"--{switch}" + for switch in excluded_switches + ] + proxy = capabilities.get("proxy") + if isinstance(proxy, dict): + proxy_server = proxy.get("server") or proxy.get("sslProxy") or proxy.get("httpProxy") + if proxy_server: + if "://" not in proxy_server: + proxy_server = f"http://{proxy_server}" + launch_options["proxy"] = {"server": proxy_server} + no_proxy = proxy.get("noProxy") + if no_proxy: + launch_options["proxy"]["bypass"] = ( + ",".join(no_proxy) if isinstance(no_proxy, list) else str(no_proxy) + ) + downloads_path = downloads_path or preferences.get("download.default_directory") if downloads_path: launch_options["downloads_path"] = downloads_path @@ -602,13 +882,62 @@ async def Open_Browser(step_data): context_options["permissions"] = permissions if color_scheme: context_options["color_scheme"] = color_scheme + if capabilities.get("acceptInsecureCerts") is not None: + context_options["ignore_https_errors"] = bool( + capabilities["acceptInsecureCerts"] + ) + + mobile_emulation = experimental_options.get("mobileEmulation") + if isinstance(mobile_emulation, dict): + device_name = mobile_emulation.get("deviceName") + if device_name: + device = playwright_instance.devices.get(device_name) + if not device: + raise ValueError(f"Unknown Playwright device: {device_name}") + context_options.pop("no_viewport", None) + context_options.update( + { + key: value + for key, value in device.items() + if key != "default_browser_type" + } + ) + device_metrics = mobile_emulation.get("deviceMetrics", {}) + if device_metrics: + context_options.pop("no_viewport", None) + context_options["viewport"] = { + "width": device_metrics["width"], + "height": device_metrics["height"], + } + context_options["device_scale_factor"] = device_metrics.get( + "pixelRatio", + 1, + ) + context_options["is_mobile"] = device_metrics.get("mobile", True) + context_options["has_touch"] = device_metrics.get("touch", True) + if mobile_emulation.get("userAgent"): + context_options["user_agent"] = mobile_emulation["userAgent"] extension_enabled = any( argument.startswith("--load-extension=") for argument in all_args ) - if extension_enabled: + user_data_dir = ( + _write_chrome_preferences(preferences) + if preferences and not debugger_address + else None + ) + if debugger_address: + browser = await playwright_instance.chromium.connect_over_cdp( + debugger_endpoint + ) + if not browser.contexts: + raise ValueError( + f"No browser context found at debugger address {debugger_address}" + ) + context = browser.contexts[0] + elif extension_enabled or user_data_dir: context = await playwright_instance.chromium.launch_persistent_context( - "", + user_data_dir or "", **launch_options, **context_options, ) @@ -628,19 +957,24 @@ async def Open_Browser(step_data): "browser": browser, "playwright": playwright_instance, "remote-debugging-port": unique_port, - "driver-path": str(driver_bin), + "driver-path": str(driver_bin) if driver_bin else None, + "user-data-dir": user_data_dir, } # Navigate if URL provided if url: - await current_page.goto(url, wait_until="domcontentloaded") + await current_page.goto(url, wait_until=wait_until) CommonUtil.ExecLog(sModuleInfo, f"Navigated to: {url}", 1) # Save to shared variables for compatibility sr.Set_Shared_Variables("playwright_page", current_page) sr.Set_Shared_Variables("playwright_context", context) sr.Set_Shared_Variables("playwright_browser", browser) - sr.Set_Shared_Variables("element_wait", timeout / 1000) # In seconds + sr.Set_Shared_Variables( + "element_wait", + element_wait if element_wait is not None else timeout / 1000, + ) + sr.Set_Shared_Variables("playwright_wait_until", wait_until) sr.Set_Shared_Variables("active_web_driver_type", "playwright") # Set screenshot variables for CommonUtil.TakeScreenShot() @@ -658,7 +992,9 @@ async def Open_Browser(step_data): remote_debugging_port=unique_port, ) session["selenium_cdp_supported"] = True - session["driver_path"] = str(driver_bin) + session["driver_path"] = str(driver_bin) if driver_bin else None + session["playwright_wait_until"] = wait_until + session["user_data_dir"] = user_data_dir sr.Set_Shared_Variables("browser_sessions", get_browser_sessions()) CommonUtil.ExecLog(sModuleInfo, f"Created browser session: {page_id}", 5) @@ -900,7 +1236,9 @@ async def Go_To_Link(step_data): return "zeuz_failed" url = None - wait_until = "domcontentloaded" + wait_until = sr.Get_Shared_Variables("playwright_wait_until") + if wait_until in failed_tag_list: + wait_until = "domcontentloaded" timeout = None element_wait_sec = None window_size_x = None @@ -937,6 +1275,11 @@ async def Go_To_Link(step_data): window_size_y = int(parts[1].strip()) except (ValueError, IndexError): pass + if ( + mid_l in ("chromium option", "chrome option") + and left_compact == "pageloadstrategy" + ): + wait_until = _page_load_wait_until(right_v) if not url: CommonUtil.ExecLog(sModuleInfo, "No URL provided", 3) @@ -1057,6 +1400,7 @@ async def Tear_Down_Playwright(step_data=None): session_selenium.quit() except Exception: pass + _cleanup_chrome_profile(existing_session.get("user_data_dir")) CommonUtil.ExecLog(sModuleInfo, f"Teared down session '{session_name}'", 1) except Exception: @@ -1117,6 +1461,7 @@ async def Tear_Down_Playwright(step_data=None): await details["browser"].close() if details.get("playwright"): await details["playwright"].stop() + _cleanup_chrome_profile(details.get("user-data-dir")) except Exception: pass diff --git a/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py b/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py index bb952ffa..9f777c1e 100644 --- a/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Web/Selenium/BuiltInFunctions.py @@ -1165,7 +1165,7 @@ def parse_and_verify_datatype(left: str, right: str, chrome_version=None): for ext_id in extension_ids: downloader = ChromeExtensionDownloader(chrome_version=chrome_version) result = downloader.setup_chrome_extension_download(extension_id=ext_id) - if result.get("crx_path"): + if result and result.get("crx_path"): extension_crxs.append(result["crx_path"]) return extension_crxs diff --git a/Framework/Built_In_Automation/Web/Selenium/utils.py b/Framework/Built_In_Automation/Web/Selenium/utils.py index 7632e987..f256127b 100644 --- a/Framework/Built_In_Automation/Web/Selenium/utils.py +++ b/Framework/Built_In_Automation/Web/Selenium/utils.py @@ -782,11 +782,18 @@ def extract_extension(self, crx_path): # Check CRX header magic = f.read(4) - if magic == b"Cr24": # CRX v3 format - # Skip header (version + header length fields) - f.read(8) - header_length = struct.unpack(" Date: Wed, 29 Jul 2026 18:09:35 +0600 Subject: [PATCH 3/4] Restore Playwright non-Chromium browser support - accept Firefox/FirefoxHeadless, WebKit/Safari, and Edge browser aliases in Playwright actions - restore Playwright-managed Firefox and WebKit installation in the persistent Zeuz browser cache - retain the Playwright system-channel launch path for Microsoft Edge - keep Chrome, Chromium, ChromeHeadless, and Chrome channels on the shared Chrome-for-Testing browser and driver installer - isolate Chromium-only arguments, extensions, preferences, and debugger attachment from Firefox and WebKit launches - mark non-CFT sessions as unsupported for lazy Selenium CDP attachment - run browser installation outside the async event loop and reuse cached binaries under an install lock - add mocked regression coverage for Firefox, WebKit, Safari, and Edge launch routing --- .../Web/Playwright/BuiltInFunctions.py | 91 +++++++++++---- .../Web/Playwright/utils.py | 109 ++++++++++++++++++ tests/test_browser_sessions.py | 71 ++++++++++++ 3 files changed, 250 insertions(+), 21 deletions(-) create mode 100644 Framework/Built_In_Automation/Web/Playwright/utils.py diff --git a/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py b/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py index 83ad78b6..b32e0a3b 100644 --- a/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py @@ -51,6 +51,7 @@ ) from Framework.Built_In_Automation.Shared_Resources import LocateElement as PlaywrightLocator from Framework.Utilities.CommonUtil import failed_tag_list +from . import utils as PlaywrightUtils from Framework.Built_In_Automation.Web.Selenium.utils import ( ChromeExtensionDownloader, ChromeForTesting, @@ -499,7 +500,7 @@ async def Open_Browser(step_data): add argument optional parameter --disable-gpu open browser playwright action open browser - Supported browsers: chrome, chromium, chrome-beta + Supported browsers: chrome, chromium, chrome-beta, firefox, webkit, safari, edge """ sModuleInfo = inspect.currentframe().f_code.co_name + " : " + MODULE_NAME global playwright_instance, browser, context, current_page @@ -621,19 +622,27 @@ async def Open_Browser(step_data): element_wait = float(right_v) compact_browser_name = _compact(browser_name) - if compact_browser_name == "chromeheadless": - browser_name = "chrome" + if compact_browser_name.endswith("headless"): + compact_browser_name = compact_browser_name.removesuffix("headless") if not headless_explicit: headless = True - elif compact_browser_name in ("chrome", "chromium", "chromebeta"): + if compact_browser_name in ("chrome", "chromium", "chromebeta"): browser_name = "chrome-beta" if compact_browser_name == "chromebeta" else compact_browser_name + elif compact_browser_name == "firefox": + browser_name = "firefox" + elif compact_browser_name in ("webkit", "safari"): + browser_name = "safari" if compact_browser_name == "safari" else "webkit" + elif compact_browser_name in ("edge", "msedge", "microsoftedge"): + browser_name = "edge" else: CommonUtil.ExecLog( sModuleInfo, - f"Playwright only supports Chrome/Chromium; browser '{browser_name}' is not supported", + f"Playwright browser '{browser_name}' is not supported", 3, ) return "zeuz_failed" + shared_cft_browser = browser_name in ("chrome", "chromium", "chrome-beta") + chromium_browser = shared_cft_browser or browser_name == "edge" from Framework.Built_In_Automation.Web.Selenium import BuiltInFunctions as SeleniumBuiltInFunctions @@ -694,7 +703,8 @@ async def Open_Browser(step_data): wait_until = _page_load_wait_until(page_load_strategy) if ( - not debugger_address + shared_cft_browser + and not debugger_address and chrome_version and chrome_version.strip().lower() == "system" ): @@ -704,6 +714,13 @@ async def Open_Browser(step_data): 3, ) return "zeuz_failed" + if not chromium_browser and debugger_address: + CommonUtil.ExecLog( + sModuleInfo, + "debugger address is only supported for Chromium browsers", + 3, + ) + return "zeuz_failed" existing_session = get_browser_session(page_id) if existing_session: @@ -716,16 +733,29 @@ async def Open_Browser(step_data): CommonUtil.ExecLog(sModuleInfo, f"Using existing browser session: {page_id}", 1) return "passed" + if not shared_cft_browser: + browser_ready = await asyncio.to_thread( + PlaywrightUtils.ensure_playwright_browser_installed, + sModuleInfo, + browser_name, + ) + if not browser_ready: + return "zeuz_failed" + chrome_channel = "Beta" if browser_name == "chrome-beta" else None chrome_bin = driver_bin = None - if not debugger_address: + if shared_cft_browser and not debugger_address: chrome_bin, driver_bin = await asyncio.to_thread( lambda: ChromeForTesting().setup_chrome_for_testing( chrome_version, chrome_channel, ) ) - if not debugger_address and (not chrome_bin or not driver_bin): + if ( + shared_cft_browser + and not debugger_address + and (not chrome_bin or not driver_bin) + ): CommonUtil.ExecLog(sModuleInfo, "Failed to setup Chrome for Testing browser and driver", 3) return "zeuz_failed" @@ -749,13 +779,16 @@ async def Open_Browser(step_data): else f"http://{debugger_address}" ) unique_port = urlparse(debugger_endpoint).port - else: + elif chromium_browser: debugger_endpoint = None unique_port = get_debug_port(page_id) + else: + debugger_endpoint = None + unique_port = None extension_files = [] encoded_extensions = [] - if not debugger_address: + if shared_cft_browser and not debugger_address: resolved_chrome_version = next( ( parent.name @@ -779,14 +812,15 @@ async def Open_Browser(step_data): value, ) ) - extension_dirs = _unpack_playwright_extensions( - extension_files, - encoded_extensions, + extension_dirs = ( + _unpack_playwright_extensions(extension_files, encoded_extensions) + if shared_cft_browser + else [] ) selenium_browser_name = "chromeheadless" if headless else "chrome" zeuz_extension_args = [] - if not debugger_address: + if shared_cft_browser and not debugger_address: zeuz_extension_args = ( SeleniumBuiltInFunctions.get_zeuz_ai_extension_arguments( selenium_browser_name @@ -818,9 +852,13 @@ async def Open_Browser(step_data): ) all_args = [] - if not debugger_address: + if chromium_browser and not debugger_address: all_args = ( - list(SeleniumBuiltInFunctions.DEFAULT_CHROMIUM_ARGUMENTS) + ( + list(SeleniumBuiltInFunctions.DEFAULT_CHROMIUM_ARGUMENTS) + if shared_cft_browser + else [] + ) + args + extension_args + [f"--remote-debugging-port={unique_port}"] @@ -840,6 +878,8 @@ async def Open_Browser(step_data): if devtools: all_args.append("--auto-open-devtools-for-tabs") CommonUtil.ExecLog(sModuleInfo, f"Using remote debugging port {unique_port} for session '{page_id}'", 1) + elif not debugger_address: + all_args = args if all_args: launch_options["args"] = all_args excluded_switches = experimental_options.get("excludeSwitches", []) @@ -866,7 +906,7 @@ async def Open_Browser(step_data): # Context options. Headed Chromium sessions use the real browser window # size so attached Selenium code observes Selenium-like layout behavior. - if not headless: + if chromium_browser and not headless: context_options = {"no_viewport": True, "accept_downloads": True} else: context_options = {"viewport": viewport, "accept_downloads": True} @@ -918,12 +958,12 @@ async def Open_Browser(step_data): if mobile_emulation.get("userAgent"): context_options["user_agent"] = mobile_emulation["userAgent"] - extension_enabled = any( + extension_enabled = shared_cft_browser and any( argument.startswith("--load-extension=") for argument in all_args ) user_data_dir = ( _write_chrome_preferences(preferences) - if preferences and not debugger_address + if shared_cft_browser and preferences and not debugger_address else None ) if debugger_address: @@ -943,7 +983,16 @@ async def Open_Browser(step_data): ) browser = context.browser else: - browser = await playwright_instance.chromium.launch(**launch_options) + if shared_cft_browser: + browser_type = playwright_instance.chromium + elif browser_name == "firefox": + browser_type = playwright_instance.firefox + elif browser_name in ("webkit", "safari"): + browser_type = playwright_instance.webkit + else: + browser_type = playwright_instance.chromium + launch_options["channel"] = "msedge" + browser = await browser_type.launch(**launch_options) context = await browser.new_context(**context_options) context.set_default_timeout(timeout) @@ -991,7 +1040,7 @@ async def Open_Browser(step_data): playwright_instance=playwright_instance, remote_debugging_port=unique_port, ) - session["selenium_cdp_supported"] = True + session["selenium_cdp_supported"] = shared_cft_browser session["driver_path"] = str(driver_bin) if driver_bin else None session["playwright_wait_until"] = wait_until session["user_data_dir"] = user_data_dir diff --git a/Framework/Built_In_Automation/Web/Playwright/utils.py b/Framework/Built_In_Automation/Web/Playwright/utils.py new file mode 100644 index 00000000..a875dcf0 --- /dev/null +++ b/Framework/Built_In_Automation/Web/Playwright/utils.py @@ -0,0 +1,109 @@ +# -*- coding: utf-8 -*- +"""Playwright-managed browser installation helpers.""" + +import os +import subprocess +import sys +from pathlib import Path + +from filelock import FileLock + +from Framework.Utilities import CommonUtil +from settings import ZEUZ_NODE_DOWNLOADS_DIR + + +PLAYWRIGHT_BROWSERS_DIR = ZEUZ_NODE_DOWNLOADS_DIR / "playwright_browsers" +PLAYWRIGHT_INSTALLABLE_BROWSERS = { + "firefox": "firefox", + "webkit": "webkit", + "safari": "webkit", +} +PLAYWRIGHT_SYSTEM_CHANNEL_BROWSERS = { + "edge", + "msedge", + "microsoft edge", +} + + +def _set_playwright_browsers_path(): + PLAYWRIGHT_BROWSERS_DIR.mkdir(parents=True, exist_ok=True) + os.environ["PLAYWRIGHT_BROWSERS_PATH"] = str(PLAYWRIGHT_BROWSERS_DIR) + return PLAYWRIGHT_BROWSERS_DIR + + +def _get_playwright_executable_path(browser_name): + result = subprocess.run( + [ + sys.executable, + "-c", + ( + "from playwright.sync_api import sync_playwright\n" + "with sync_playwright() as p:\n" + f" print(p.{browser_name}.executable_path)\n" + ), + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=os.environ.copy(), + ) + if result.returncode != 0: + return None + output = result.stdout.strip() + return Path(output.splitlines()[-1]) if output else None + + +def _is_playwright_browser_installed(browser_name): + executable_path = _get_playwright_executable_path(browser_name) + return bool(executable_path and executable_path.exists()) + + +def ensure_playwright_browser_installed(sModuleInfo, browser_name): + """Install Firefox/WebKit in Zeuz's persistent Playwright cache when needed.""" + try: + browsers_dir = _set_playwright_browsers_path() + requested_browser = (browser_name or "").strip().lower() + install_browser = PLAYWRIGHT_INSTALLABLE_BROWSERS.get(requested_browser) + + if requested_browser in PLAYWRIGHT_SYSTEM_CHANNEL_BROWSERS: + return True + if not install_browser: + CommonUtil.ExecLog( + sModuleInfo, + f"Playwright browser '{browser_name}' is not installable", + 3, + ) + return False + if _is_playwright_browser_installed(install_browser): + return True + + with FileLock(str(browsers_dir / f"{install_browser}.install.lock")): + if _is_playwright_browser_installed(install_browser): + return True + result = subprocess.run( + [ + sys.executable, + "-m", + "playwright", + "install", + "--with-deps", + install_browser, + ], + env=os.environ.copy(), + ) + + if result.returncode == 0: + return True + CommonUtil.ExecLog( + sModuleInfo, + f"Failed to install Playwright {install_browser}. See terminal output for details.", + 3, + ) + return False + except Exception as exc: + CommonUtil.ExecLog( + sModuleInfo, + f"Error setting up Playwright browser: {exc}", + 3, + ) + return False diff --git a/tests/test_browser_sessions.py b/tests/test_browser_sessions.py index 1bf337c7..57f4deed 100644 --- a/tests/test_browser_sessions.py +++ b/tests/test_browser_sessions.py @@ -367,3 +367,74 @@ def test_playwright_supports_selenium_debugger_address(monkeypatch): "http://127.0.0.1:9333" ) assert browser_utils.get_browser_session("default")["remote_debugging_port"] == 9333 + + +def test_playwright_restores_non_cft_browser_launches(monkeypatch): + monkeypatch.setattr(playwright_bif.CommonUtil, "ExecLog", lambda *args, **kwargs: None) + monkeypatch.setattr( + playwright_bif.CommonUtil, + "set_screenshot_vars", + lambda *args, **kwargs: None, + ) + installer = MagicMock(return_value=True) + monkeypatch.setattr( + playwright_bif.PlaywrightUtils, + "ensure_playwright_browser_installed", + installer, + ) + cft_factory = MagicMock(side_effect=AssertionError("CfT should not be resolved")) + monkeypatch.setattr(playwright_bif, "ChromeForTesting", cft_factory) + monkeypatch.setattr(playwright_bif, "get_debug_port", lambda session_name: 9250) + + async def run_inline(function, *args): + return function(*args) + + monkeypatch.setattr(playwright_bif.asyncio, "to_thread", run_inline) + + launches = {} + for requested_browser, browser_type_name in ( + ("FirefoxHeadless", "firefox"), + ("webkit", "webkit"), + ("safari", "webkit"), + ("edge", "chromium"), + ): + sr.shared_variables.clear() + playwright_bif.current_page = None + playwright_bif.current_page_id = None + playwright_bif.context = None + playwright_bif.browser = None + + page = MagicMock() + context = MagicMock() + context.pages = [] + context.new_page = AsyncMock(return_value=page) + browser = MagicMock() + browser.new_context = AsyncMock(return_value=context) + playwright = MagicMock() + for name in ("chromium", "firefox", "webkit"): + getattr(playwright, name).launch = AsyncMock(return_value=browser) + starter = MagicMock() + starter.start = AsyncMock(return_value=playwright) + monkeypatch.setattr(playwright_bif, "async_playwright", lambda: starter) + + result = asyncio.run( + playwright_bif.Open_Browser( + [("browser", "input parameter", requested_browser)] + ) + ) + + assert result == "passed" + launch = getattr(playwright, browser_type_name).launch + launch.assert_awaited_once() + launches[requested_browser] = launch.await_args.kwargs + session = browser_utils.get_browser_session("default") + assert session["selenium_cdp_supported"] is False + assert session["driver_path"] is None + + assert installer.call_args_list[0].args[1] == "firefox" + assert installer.call_args_list[1].args[1] == "webkit" + assert installer.call_args_list[2].args[1] == "safari" + assert installer.call_args_list[3].args[1] == "edge" + assert launches["FirefoxHeadless"]["headless"] is True + assert launches["edge"]["channel"] == "msedge" + cft_factory.assert_not_called() From 4725724395205c9cf219895c4eb02ab00b30aeac Mon Sep 17 00:00:00 2001 From: Nasif Date: Wed, 29 Jul 2026 18:41:22 +0600 Subject: [PATCH 4/4] Complete Selenium browser-option parity in Playwright - accept Selenium-style Chrome, Chromium, Edge, Firefox, and Safari option rows for the active Playwright browser - map Firefox arguments and preferences from firefox option and moz:firefoxOptions fields - map Edge arguments, preferences, extensions, and ms:edgeOptions fields through persistent Microsoft Edge contexts - pass Safari option arguments to WebKit launches - support Chromium localState profiles and browserVersion capabilities - map WebDriver implicit and page-load timeouts to Playwright context timeouts - handle accept, dismiss, notify variants, and ignore values for unhandled prompts - apply shared Chromium defaults and Zeuz extension arguments to Edge launches - add regression coverage for browser-specific options, vendor capabilities, extensions, preferences, timeouts, prompts, and local state --- .../Web/Playwright/BuiltInFunctions.py | 215 +++++++++++------- tests/test_browser_sessions.py | 93 +++++++- 2 files changed, 223 insertions(+), 85 deletions(-) diff --git a/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py b/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py index b32e0a3b..43e79bda 100644 --- a/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py +++ b/Framework/Built_In_Automation/Web/Playwright/BuiltInFunctions.py @@ -108,7 +108,7 @@ def _page_load_wait_until(strategy): ) from None -def _write_chrome_preferences(preferences): +def _write_chrome_preferences(preferences, local_state=None): """Write Selenium-style dotted Chrome preferences to a temporary profile.""" nested_preferences = {} for key, value in preferences.items(): @@ -122,6 +122,8 @@ def _write_chrome_preferences(preferences): default_profile = user_data_dir / "Default" default_profile.mkdir() (default_profile / "Preferences").write_text(json.dumps(nested_preferences)) + if local_state: + (user_data_dir / "Local State").write_text(json.dumps(local_state)) return str(user_data_dir) @@ -518,11 +520,7 @@ async def Open_Browser(step_data): headless = dependency_browser.replace(" ", "") == "chromeheadless" headless_explicit = False chrome_version = None - extension_values = [] - encoded_extension_values = [] - chromium_argument_values = [] - experimental_option_values = [] - preference_values = [] + browser_option_rows = [] shared_capability_values = [] debugger_address = None page_load_strategy = "eager" @@ -598,21 +596,14 @@ async def Open_Browser(step_data): left_compact = _compact(left_l) if left_compact == "chrome:version": chrome_version = right_v - elif mid_l in ("chromium option", "chrome option"): - if left_compact == "addargument": - chromium_argument_values.append(right_v) - elif left_compact == "addexperimentaloption": - experimental_option_values.append(right_v) - elif left_compact == "addextension": - extension_values.append(right_v) - elif left_compact == "addencodedextension": - encoded_extension_values.append(right_v) - elif left_compact == "setpreference": - preference_values.append(right_v) - elif left_compact == "pageloadstrategy": - page_load_strategy = right_v - elif left_compact == "debuggeraddress": - debugger_address = right_v + elif mid_l in ( + "chromium option", + "chrome option", + "edge option", + "firefox option", + "safari option", + ): + browser_option_rows.append((left_compact, mid_l, right_v)) if left_compact in ( "waittimetoappearelement", @@ -646,52 +637,88 @@ async def Open_Browser(step_data): from Framework.Built_In_Automation.Web.Selenium import BuiltInFunctions as SeleniumBuiltInFunctions + allowed_option_mids = { + "chrome": ("chromium option", "chrome option"), + "chromium": ("chromium option", "chrome option"), + "chrome-beta": ("chromium option", "chrome option"), + "edge": ("chromium option", "chrome option", "edge option"), + "firefox": ("firefox option",), + "webkit": ("safari option",), + "safari": ("safari option",), + }[browser_name] + extension_values = [] + encoded_extension_values = [] + experimental_options = {} + preferences = {} + firefox_preferences = {} + for left_compact, mid_l, right_v in browser_option_rows: + if mid_l not in allowed_option_mids: + continue + if left_compact == "addargument": + args.extend( + SeleniumBuiltInFunctions.parse_and_verify_datatype( + "addargument", + right_v, + ) + ) + elif left_compact == "addexperimentaloption" and chromium_browser: + experimental_options.update( + SeleniumBuiltInFunctions.parse_and_verify_datatype( + "addexperimentaloption", + right_v, + ) + ) + elif left_compact == "addextension" and chromium_browser: + extension_values.append(right_v) + elif left_compact == "addencodedextension" and chromium_browser: + encoded_extension_values.append(right_v) + elif left_compact == "setpreference": + target = firefox_preferences if browser_name == "firefox" else preferences + target.update( + SeleniumBuiltInFunctions.parse_and_verify_datatype( + "setpreference", + right_v, + ) + ) + elif left_compact == "pageloadstrategy": + page_load_strategy = right_v + elif left_compact == "debuggeraddress" and chromium_browser: + debugger_address = right_v + capabilities = {} for value in shared_capability_values: parsed = CommonUtil.parse_value_into_object(value) if not isinstance(parsed, dict): raise ValueError("shared capability must be a dictionary") capabilities.update(parsed) - - experimental_options = {} - for value in experimental_option_values: - experimental_options.update( - SeleniumBuiltInFunctions.parse_and_verify_datatype( - "addexperimentaloption", - value, - ) - ) - preferences = {} - for value in preference_values: - preferences.update( - SeleniumBuiltInFunctions.parse_and_verify_datatype( - "setpreference", - value, - ) - ) - for value in chromium_argument_values: - args.extend( - SeleniumBuiltInFunctions.parse_and_verify_datatype( - "addargument", - value, - ) - ) - - chrome_options = capabilities.get("goog:chromeOptions", {}) - if isinstance(chrome_options, dict): - args.extend(chrome_options.get("args", [])) - if chrome_options.get("extensions"): + if shared_cft_browser and not chrome_version: + capability_version = str(capabilities.get("browserVersion", "")).strip() + if capability_version: + chrome_version = capability_version + + vendor_options = capabilities.get( + "ms:edgeOptions" if browser_name == "edge" else "goog:chromeOptions", + {}, + ) + if chromium_browser and isinstance(vendor_options, dict): + args.extend(vendor_options.get("args", [])) + if vendor_options.get("extensions"): encoded_extension_values.append( - repr(chrome_options["extensions"]) + repr(vendor_options["extensions"]) ) experimental_options.update( { key: value - for key, value in chrome_options.items() + for key, value in vendor_options.items() if key not in ("args", "extensions") } ) + firefox_options = capabilities.get("moz:firefoxOptions", {}) + if browser_name == "firefox" and isinstance(firefox_options, dict): + args.extend(firefox_options.get("args", [])) + firefox_preferences.update(firefox_options.get("prefs", {})) preferences.update(experimental_options.get("prefs", {})) + local_state = experimental_options.get("localState", {}) debugger_address = ( debugger_address or experimental_options.get("debuggerAddress") @@ -770,6 +797,10 @@ async def Open_Browser(step_data): } if chrome_bin: launch_options["executable_path"] = str(chrome_bin) + elif browser_name == "edge": + launch_options["channel"] = "msedge" + if firefox_preferences: + launch_options["firefox_user_prefs"] = firefox_preferences # Add remote debugging port for CDP connection with unique port per session if debugger_address: @@ -788,15 +819,17 @@ async def Open_Browser(step_data): extension_files = [] encoded_extensions = [] - if shared_cft_browser and not debugger_address: - resolved_chrome_version = next( - ( - parent.name - for parent in Path(chrome_bin).parents - if parent.parent.name == "versions" - ), - chrome_version, - ) + if chromium_browser and not debugger_address: + resolved_chrome_version = chrome_version + if chrome_bin: + resolved_chrome_version = next( + ( + parent.name + for parent in Path(chrome_bin).parents + if parent.parent.name == "versions" + ), + chrome_version, + ) for value in extension_values: extension_files.extend( SeleniumBuiltInFunctions.parse_and_verify_datatype( @@ -814,13 +847,19 @@ async def Open_Browser(step_data): ) extension_dirs = ( _unpack_playwright_extensions(extension_files, encoded_extensions) - if shared_cft_browser + if chromium_browser else [] ) - selenium_browser_name = "chromeheadless" if headless else "chrome" + selenium_browser_name = ( + "microsoft edge chromium" + if browser_name == "edge" + else "chromeheadless" + if headless + else "chrome" + ) zeuz_extension_args = [] - if shared_cft_browser and not debugger_address: + if chromium_browser and not debugger_address: zeuz_extension_args = ( SeleniumBuiltInFunctions.get_zeuz_ai_extension_arguments( selenium_browser_name @@ -854,11 +893,7 @@ async def Open_Browser(step_data): all_args = [] if chromium_browser and not debugger_address: all_args = ( - ( - list(SeleniumBuiltInFunctions.DEFAULT_CHROMIUM_ARGUMENTS) - if shared_cft_browser - else [] - ) + list(SeleniumBuiltInFunctions.DEFAULT_CHROMIUM_ARGUMENTS) + args + extension_args + [f"--remote-debugging-port={unique_port}"] @@ -958,14 +993,12 @@ async def Open_Browser(step_data): if mobile_emulation.get("userAgent"): context_options["user_agent"] = mobile_emulation["userAgent"] - extension_enabled = shared_cft_browser and any( + extension_enabled = chromium_browser and any( argument.startswith("--load-extension=") for argument in all_args ) - user_data_dir = ( - _write_chrome_preferences(preferences) - if shared_cft_browser and preferences and not debugger_address - else None - ) + user_data_dir = None + if chromium_browser and (preferences or local_state) and not debugger_address: + user_data_dir = _write_chrome_preferences(preferences, local_state) if debugger_address: browser = await playwright_instance.chromium.connect_over_cdp( debugger_endpoint @@ -991,13 +1024,43 @@ async def Open_Browser(step_data): browser_type = playwright_instance.webkit else: browser_type = playwright_instance.chromium - launch_options["channel"] = "msedge" browser = await browser_type.launch(**launch_options) context = await browser.new_context(**context_options) context.set_default_timeout(timeout) + capability_timeouts = capabilities.get("timeouts", {}) + if isinstance(capability_timeouts, dict): + context.set_default_navigation_timeout( + capability_timeouts.get("pageLoad", timeout) + ) + if element_wait is None and capability_timeouts.get("implicit") is not None: + element_wait = float(capability_timeouts["implicit"]) / 1000 current_page = context.pages[0] if context.pages else await context.new_page() current_page_id = page_id + prompt_behavior = str( + capabilities.get("unhandledPromptBehavior", "") + ).strip().lower() + if prompt_behavior: + if prompt_behavior.startswith("accept"): + async def handle_dialog(dialog): + await dialog.accept() + elif prompt_behavior.startswith("dismiss"): + async def handle_dialog(dialog): + await dialog.dismiss() + elif prompt_behavior == "ignore": + def handle_dialog(dialog): + pass + else: + raise ValueError( + "unhandledPromptBehavior must be accept, dismiss, " + "accept and notify, dismiss and notify, or ignore" + ) + + def configure_dialog_handler(page): + page.on("dialog", handle_dialog) + + configure_dialog_handler(current_page) + context.on("page", configure_dialog_handler) # Store in details playwright_details[page_id] = { diff --git a/tests/test_browser_sessions.py b/tests/test_browser_sessions.py index 57f4deed..b6450716 100644 --- a/tests/test_browser_sessions.py +++ b/tests/test_browser_sessions.py @@ -250,7 +250,8 @@ async def run_inline(function, *args): "chromium option", "{'prefs': {'profile.test': 1}, " "'excludeSwitches': ['disable-popup-blocking'], " - "'mobileEmulation': {'deviceName': 'Test Phone'}}", + "'mobileEmulation': {'deviceName': 'Test Phone'}, " + "'localState': {'browser': {'test': True}}}", ), ( "set preference", @@ -263,7 +264,8 @@ async def run_inline(function, *args): "capabilities", "shared capability", "{'acceptInsecureCerts': True, " - "'proxy': {'httpProxy': 'proxy.test:8080'}}", + "'proxy': {'httpProxy': 'proxy.test:8080'}, " + "'unhandledPromptBehavior': 'accept'}", ), ] ) @@ -288,12 +290,17 @@ async def run_inline(function, *args): assert session["driver_path"] == str(driver_bin) assert session["playwright_wait_until"] == "load" assert sr.Get_Shared_Variables("element_wait") == 17 + page.on.assert_called_once() + assert page.on.call_args.args[0] == "dialog" user_data_dir = playwright.chromium.launch_persistent_context.await_args.args[0] preferences = json.loads( (Path(user_data_dir) / "Default" / "Preferences").read_text() ) assert preferences["profile"]["test"] == 1 assert preferences["download"]["prompt_for_download"] is False + assert json.loads((Path(user_data_dir) / "Local State").read_text())[ + "browser" + ]["test"] is True assert asyncio.run(playwright_bif.Open_Browser([])) == "passed" cft.setup_chrome_for_testing.assert_called_once() playwright.chromium.launch_persistent_context.assert_awaited_once() @@ -390,13 +397,55 @@ async def run_inline(function, *args): return function(*args) monkeypatch.setattr(playwright_bif.asyncio, "to_thread", run_inline) + unpack_extensions = MagicMock(return_value=["/extensions/custom"]) + monkeypatch.setattr( + playwright_bif, + "_unpack_playwright_extensions", + unpack_extensions, + ) launches = {} - for requested_browser, browser_type_name in ( - ("FirefoxHeadless", "firefox"), - ("webkit", "webkit"), - ("safari", "webkit"), - ("edge", "chromium"), + for requested_browser, browser_type_name, option_rows in ( + ( + "FirefoxHeadless", + "firefox", + [ + ("add argument", "firefox option", "['--safe-mode']"), + ("set preference", "firefox option", "{'browser.test': True}"), + ("page load strategy", "firefox option", "normal"), + ( + "capabilities", + "shared capability", + "{'moz:firefoxOptions': {'args': ['--private'], " + "'prefs': {'browser.shared': 1}}}", + ), + ], + ), + ( + "webkit", + "webkit", + [("add argument", "safari option", "['--webkit-test']")], + ), + ( + "safari", + "webkit", + [("add argument", "safari option", "['--safari-test']")], + ), + ( + "edge", + "chromium", + [ + ("add argument", "edge option", "['--edge-test']"), + ("set preference", "edge option", "{'profile.edge': True}"), + ("add encoded extension", "edge option", "['ZW5jb2RlZA==']"), + ( + "capabilities", + "shared capability", + "{'ms:edgeOptions': {'args': ['--edge-shared']}, " + "'timeouts': {'implicit': 9000, 'pageLoad': 45000}}", + ), + ], + ), ): sr.shared_variables.clear() playwright_bif.current_page = None @@ -413,18 +462,25 @@ async def run_inline(function, *args): playwright = MagicMock() for name in ("chromium", "firefox", "webkit"): getattr(playwright, name).launch = AsyncMock(return_value=browser) + playwright.chromium.launch_persistent_context = AsyncMock( + return_value=context + ) starter = MagicMock() starter.start = AsyncMock(return_value=playwright) monkeypatch.setattr(playwright_bif, "async_playwright", lambda: starter) result = asyncio.run( playwright_bif.Open_Browser( - [("browser", "input parameter", requested_browser)] + [("browser", "input parameter", requested_browser)] + option_rows ) ) assert result == "passed" - launch = getattr(playwright, browser_type_name).launch + launch = ( + playwright.chromium.launch_persistent_context + if requested_browser == "edge" + else getattr(playwright, browser_type_name).launch + ) launch.assert_awaited_once() launches[requested_browser] = launch.await_args.kwargs session = browser_utils.get_browser_session("default") @@ -436,5 +492,24 @@ async def run_inline(function, *args): assert installer.call_args_list[2].args[1] == "safari" assert installer.call_args_list[3].args[1] == "edge" assert launches["FirefoxHeadless"]["headless"] is True + assert launches["FirefoxHeadless"]["args"] == ["--safe-mode", "--private"] + assert launches["FirefoxHeadless"]["firefox_user_prefs"] == { + "browser.test": True, + "browser.shared": 1, + } + assert launches["webkit"]["args"] == ["--webkit-test"] + assert launches["safari"]["args"] == ["--safari-test"] assert launches["edge"]["channel"] == "msedge" + assert "--edge-test" in launches["edge"]["args"] + assert "--edge-shared" in launches["edge"]["args"] + assert "--load-extension=/extensions/custom" in launches["edge"]["args"] + edge_session = browser_utils.get_browser_session("default") + assert sr.Get_Shared_Variables("element_wait") == 9 + user_data_dir = edge_session["user_data_dir"] + assert json.loads( + (Path(user_data_dir) / "Default" / "Preferences").read_text() + )["profile"]["edge"] is True + context.set_default_navigation_timeout.assert_called_with(45000) + playwright_bif._cleanup_chrome_profile(user_data_dir) + unpack_extensions.assert_called_once_with([], ["ZW5jb2RlZA=="]) cft_factory.assert_not_called()