Add minimal SeleniumBase CDP connection test, Selenium debugger_address connection test, and update SeleniumBase CDP POC test. Refactor browser client and actions to use Selenium WebDriver instead of Playwright, ensuring no page refresh on connection. Enhance diagnostics and waiting utilities for Selenium compatibility.

This commit is contained in:
2025-11-23 07:19:23 +00:00
parent 1aac3a9a3e
commit 741059009f
16 changed files with 586 additions and 388 deletions

View File

@@ -1,7 +1,7 @@
from playwright.async_api import Page
from typing import ClassVar, override
from selenium.webdriver.remote.webdriver import WebDriver
from guide.app.actions.base import DemoAction, register_action
from guide.app.auth import DummyMfaCodeProvider, ensure_persona
from guide.app import errors
@@ -25,11 +25,11 @@ class LoginAsPersonaAction(DemoAction):
self._login_url = login_url
@override
async def run(self, page: Page, context: ActionContext) -> ActionResult:
async def run(self, driver: WebDriver, context: ActionContext) -> ActionResult:
if context.persona_id is None:
raise errors.PersonaError("persona_id is required for login action")
persona = self._personas.get(context.persona_id)
await ensure_persona(
page, persona, self._mfa_provider, login_url=self._login_url
ensure_persona(
driver, persona, self._mfa_provider, login_url=self._login_url
)
return ActionResult(details={"persona_id": persona.id, "status": "logged_in"})

View File

@@ -3,7 +3,7 @@ from collections.abc import Callable, Iterable, Mapping
from inspect import Parameter, signature
from typing import ClassVar, override, cast
from playwright.async_api import Page
from selenium.webdriver.remote.webdriver import WebDriver
from guide.app import errors
from guide.app.models.domain import ActionContext, ActionMetadata, ActionResult
@@ -21,7 +21,7 @@ class DemoAction(ABC):
category: ClassVar[str]
@abstractmethod
async def run(self, page: Page, context: ActionContext) -> ActionResult:
async def run(self, driver: WebDriver, context: ActionContext) -> ActionResult:
"""Execute the action and return a result."""
...
@@ -84,11 +84,11 @@ class CompositeAction(DemoAction):
self.context: ActionContext | None = None
@override
async def run(self, page: Page, context: ActionContext) -> ActionResult:
async def run(self, driver: WebDriver, context: ActionContext) -> ActionResult:
"""Execute all child actions in sequence.
Args:
page: The Playwright page instance
driver: The Selenium WebDriver instance
context: The action context (shared across all steps)
Returns:
@@ -101,7 +101,7 @@ class CompositeAction(DemoAction):
for step_id in self.child_actions:
try:
action = self.registry.get(step_id)
result = await action.run(page, context)
result = await action.run(driver, context)
results[step_id] = result
details[step_id] = result.details

View File

@@ -6,7 +6,7 @@ browser interactions with minimal imports.
from typing import ClassVar, override
from playwright.async_api import Page
from selenium.webdriver.remote.webdriver import WebDriver
from guide.app.actions.base import DemoAction, register_action
from guide.app.browser.helpers import PageHelpers, AccordionCollapseResult
@@ -30,7 +30,7 @@ class CollapseAccordionsDemoAction(DemoAction):
category: ClassVar[str] = "demo"
@override
async def run(self, page: Page, context: ActionContext) -> ActionResult:
async def run(self, driver: WebDriver, context: ActionContext) -> ActionResult:
"""Collapse accordions on the current page.
Parameters:
@@ -60,7 +60,7 @@ class CollapseAccordionsDemoAction(DemoAction):
timeout_ms: int = _coerce_to_int(context.params.get("timeout_ms"), 5000)
# Use PageHelpers for the interaction (single import!)
helpers = PageHelpers(page)
helpers = PageHelpers(driver)
result: AccordionCollapseResult = await helpers.collapse_accordions(
selector, timeout_ms
)

View File

@@ -1,12 +1,20 @@
from playwright.async_api import Page
import logging
import time
from typing import ClassVar, override
from selenium.webdriver.common.by import By
from selenium.webdriver.remote.webdriver import WebDriver
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
from guide.app.actions.base import DemoAction, register_action
from guide.app.models.domain import ActionContext, ActionResult
from guide.app.strings.registry import app_strings
_logger = logging.getLogger(__name__)
@register_action
class FillIntakeBasicAction(DemoAction):
id: ClassVar[str] = "fill-intake-basic"
@@ -16,10 +24,10 @@ class FillIntakeBasicAction(DemoAction):
category: ClassVar[str] = "intake"
@override
async def run(self, page: Page, context: ActionContext) -> ActionResult:
async def run(self, driver: WebDriver, context: ActionContext) -> ActionResult:
description_val = app_strings.intake.conveyor_belt_request
await page.fill(app_strings.intake.description_field, description_val)
await page.click(app_strings.intake.next_button)
driver.find_element(By.CSS_SELECTOR, app_strings.intake.description_field).send_keys(description_val)
driver.find_element(By.CSS_SELECTOR, app_strings.intake.next_button).click()
return ActionResult(details={"message": "Intake filled"})
@@ -32,97 +40,137 @@ class FillSourcingRequestAction(DemoAction):
category: ClassVar[str] = "intake"
@override
async def run(self, page: Page, context: ActionContext) -> ActionResult:
async def run(self, driver: WebDriver, context: ActionContext) -> ActionResult:
"""Fill the sourcing request form with demo data.
Handles multi-select commodities, regions, and all other form fields.
"""
import logging
_logger = logging.getLogger(__name__)
_logger.info(f"[FORM-FILL] Starting form fill action")
_logger.info("[FORM-FILL] Starting form fill action")
# Fill commodities (multi-select autocomplete)
commodity_input = page.locator(app_strings.intake.commodity_field).locator("input")
commodity_container = driver.find_element(By.CSS_SELECTOR, app_strings.intake.commodity_field)
commodity_input = commodity_container.find_element(By.CSS_SELECTOR, "input")
for commodity in app_strings.intake.commodity_request:
await commodity_input.click()
await commodity_input.type(commodity) # Use type() to avoid clearing existing
await page.wait_for_timeout(800) # Wait for autocomplete dropdown
# Click the matching option from dropdown
option = page.locator(f'li[role="option"]:has-text("{commodity}")').first
await option.wait_for(state="visible", timeout=5000)
await option.click()
await page.wait_for_timeout(500) # Allow UI to update
commodity_input.click()
commodity_input.send_keys(commodity)
time.sleep(0.8) # Wait for autocomplete dropdown
# Wait for and click the matching option from dropdown
wait = WebDriverWait(driver, 5)
option = wait.until(
EC.visibility_of_element_located(
(By.XPATH, f'//li[@role="option"][contains(text(), "{commodity}")]')
)
)
option.click()
time.sleep(0.5) # Allow UI to update
# Fill planned (single-select dropdown)
await page.locator(app_strings.intake.planned_field).click()
driver.find_element(By.CSS_SELECTOR, app_strings.intake.planned_field).click()
# Wait for the dropdown menu to appear
dropdown_menu = page.locator('[id^="menu-"][role="presentation"]').first
await dropdown_menu.wait_for(state="visible", timeout=5000)
await page.wait_for_timeout(300) # Additional wait for options to render
planned_option = page.locator(f'li[role="option"]:has-text("{app_strings.intake.planned_request}")').first
await planned_option.wait_for(state="visible", timeout=5000)
await planned_option.click()
await page.wait_for_timeout(500) # Wait for auto-close
wait = WebDriverWait(driver, 5)
wait.until(
EC.visibility_of_element_located(
(By.CSS_SELECTOR, '[id^="menu-"][role="presentation"]')
)
)
time.sleep(0.3) # Additional wait for options to render
planned_option = wait.until(
EC.visibility_of_element_located(
(By.XPATH, f'//li[@role="option"][contains(text(), "{app_strings.intake.planned_request}")]')
)
)
planned_option.click()
time.sleep(0.5) # Wait for auto-close
# Explicitly close dropdown by clicking elsewhere
await page.locator(app_strings.intake.page_header_title).click()
await page.wait_for_timeout(500)
driver.find_element(By.CSS_SELECTOR, app_strings.intake.page_header_title).click()
time.sleep(0.5)
# Fill regions (multi-select dropdown)
for region in app_strings.intake.regions_request:
await page.locator(app_strings.intake.regions_field).click()
driver.find_element(By.CSS_SELECTOR, app_strings.intake.regions_field).click()
# Wait for the dropdown menu to appear
dropdown_menu = page.locator('[id^="menu-"][role="presentation"]').first
await dropdown_menu.wait_for(state="visible", timeout=5000)
await page.wait_for_timeout(300) # Additional wait for options to render
region_option = page.locator(f'li[role="option"]:has-text("{region}")').first
await region_option.wait_for(state="visible", timeout=5000)
await region_option.click()
await page.wait_for_timeout(300)
wait = WebDriverWait(driver, 5)
wait.until(
EC.visibility_of_element_located(
(By.CSS_SELECTOR, '[id^="menu-"][role="presentation"]')
)
)
time.sleep(0.3) # Additional wait for options to render
region_option = wait.until(
EC.visibility_of_element_located(
(By.XPATH, f'//li[@role="option"][contains(text(), "{region}")]')
)
)
region_option.click()
time.sleep(0.3)
# Click elsewhere to close the regions dropdown (click on page title)
await page.locator(app_strings.intake.page_header_title).click()
await page.wait_for_timeout(500)
driver.find_element(By.CSS_SELECTOR, app_strings.intake.page_header_title).click()
time.sleep(0.5)
# Fill OpEx/CapEx (single-select dropdown)
await page.locator(app_strings.intake.opex_capex_field).click()
driver.find_element(By.CSS_SELECTOR, app_strings.intake.opex_capex_field).click()
# Wait for the dropdown menu to appear
dropdown_menu = page.locator('[id^="menu-"][role="presentation"]').first
await dropdown_menu.wait_for(state="visible", timeout=5000)
await page.wait_for_timeout(300) # Additional wait for options to render
opex_option = page.locator(f'li[role="option"]:has-text("{app_strings.intake.opex_capex_request}")').first
await opex_option.wait_for(state="visible", timeout=5000)
await opex_option.click()
wait = WebDriverWait(driver, 5)
wait.until(
EC.visibility_of_element_located(
(By.CSS_SELECTOR, '[id^="menu-"][role="presentation"]')
)
)
time.sleep(0.3) # Additional wait for options to render
opex_option = wait.until(
EC.visibility_of_element_located(
(By.XPATH, f'//li[@role="option"][contains(text(), "{app_strings.intake.opex_capex_request}")]')
)
)
opex_option.click()
# Wait for dropdown to close automatically after selection
await page.wait_for_timeout(500)
time.sleep(0.5)
# Fill description (required textarea)
await page.locator(app_strings.intake.description_textarea).fill(
app_strings.intake.description_request
)
description_field = driver.find_element(By.CSS_SELECTOR, app_strings.intake.description_textarea)
description_field.clear()
description_field.send_keys(app_strings.intake.description_request)
# Fill desired supplier name (textarea)
await page.locator(app_strings.intake.desired_supplier_name_textarea).fill(
app_strings.intake.desired_supplier_name_request
)
supplier_name_field = driver.find_element(By.CSS_SELECTOR, app_strings.intake.desired_supplier_name_textarea)
supplier_name_field.clear()
supplier_name_field.send_keys(app_strings.intake.desired_supplier_name_request)
# Fill desired supplier contact (textarea)
await page.locator(app_strings.intake.desired_supplier_contact_textarea).fill(
app_strings.intake.desired_supplier_contact_request
)
supplier_contact_field = driver.find_element(By.CSS_SELECTOR, app_strings.intake.desired_supplier_contact_textarea)
supplier_contact_field.clear()
supplier_contact_field.send_keys(app_strings.intake.desired_supplier_contact_request)
# Fill reseller (textarea)
await page.locator(app_strings.intake.reseller_textarea).fill(
app_strings.intake.reseller_request
)
reseller_field = driver.find_element(By.CSS_SELECTOR, app_strings.intake.reseller_textarea)
reseller_field.clear()
reseller_field.send_keys(app_strings.intake.reseller_request)
# Fill entity (autocomplete)
entity_input = page.locator(app_strings.intake.entity_field).locator("input")
await entity_input.click()
await entity_input.type(app_strings.intake.entity_request)
await page.wait_for_timeout(800) # Wait for autocomplete dropdown
entity_container = driver.find_element(By.CSS_SELECTOR, app_strings.intake.entity_field)
entity_input = entity_container.find_element(By.CSS_SELECTOR, "input")
entity_input.click()
entity_input.send_keys(app_strings.intake.entity_request)
time.sleep(0.8) # Wait for autocomplete dropdown
# Click the matching option from dropdown
entity_option = page.locator(f'li[role="option"]:has-text("{app_strings.intake.entity_request}")').first
await entity_option.wait_for(state="visible", timeout=5000)
await entity_option.click()
wait = WebDriverWait(driver, 5)
entity_option = wait.until(
EC.visibility_of_element_located(
(By.XPATH, f'//li[@role="option"][contains(text(), "{app_strings.intake.entity_request}")]')
)
)
entity_option.click()
return ActionResult(
details={

View File

@@ -1,7 +1,8 @@
from playwright.async_api import Page
from typing import ClassVar, override
from selenium.webdriver.common.by import By
from selenium.webdriver.remote.webdriver import WebDriver
from guide.app.actions.base import DemoAction, register_action
from guide.app.models.domain import ActionContext, ActionResult
from guide.app.strings.registry import app_strings
@@ -14,9 +15,11 @@ class AddThreeSuppliersAction(DemoAction):
category: ClassVar[str] = "sourcing"
@override
async def run(self, page: Page, context: ActionContext) -> ActionResult:
async def run(self, driver: WebDriver, context: ActionContext) -> ActionResult:
suppliers = app_strings.sourcing.default_trio
for supplier in suppliers:
await page.fill(app_strings.sourcing.supplier_search_input, supplier)
await page.click(app_strings.sourcing.add_supplier_button)
search_input = driver.find_element(By.CSS_SELECTOR, app_strings.sourcing.supplier_search_input)
search_input.clear()
search_input.send_keys(supplier)
driver.find_element(By.CSS_SELECTOR, app_strings.sourcing.add_supplier_button).click()
return ActionResult(details={"added_suppliers": list(suppliers)})

View File

@@ -94,12 +94,12 @@ async def execute_action(
mfa_provider = DummyMfaCodeProvider()
try:
async with browser_client.open_page(target_host_id) as page:
async with browser_client.open_page(target_host_id) as driver:
if persona:
await ensure_persona(
page, persona, mfa_provider, login_url=settings.raindrop_base_url
ensure_persona(
driver, persona, mfa_provider, login_url=settings.raindrop_base_url
)
result = await action.run(page, context)
result = await action.run(driver, context)
except errors.GuideError as exc:
return ActionEnvelope(
status=ActionStatus.ERROR,

View File

@@ -1,65 +1,90 @@
from playwright.async_api import Page
from selenium.webdriver.common.by import By
from selenium.webdriver.remote.webdriver import WebDriver
from selenium.common.exceptions import NoSuchElementException
from guide.app.auth.mfa import MfaCodeProvider
from guide.app.models.personas.models import DemoPersona
from guide.app.strings.registry import app_strings
async def detect_current_persona(page: Page) -> str | None:
def detect_current_persona(driver: WebDriver) -> str | None:
"""Return the email/identifier of the currently signed-in user, if visible."""
element = page.locator(app_strings.auth.current_user_display)
if await element.count() == 0:
try:
element = driver.find_element(By.CSS_SELECTOR, app_strings.auth.current_user_display)
text = element.text
if not text:
return None
prefix = app_strings.auth.current_user_display_prefix
if prefix and text.startswith(prefix):
return text.removeprefix(prefix).strip()
return text.strip()
except NoSuchElementException:
return None
text = await element.first.text_content()
if text is None:
return None
prefix = app_strings.auth.current_user_display_prefix
if prefix and text.startswith(prefix):
return text.removeprefix(prefix).strip()
return text.strip()
async def login_with_mfa(
page: Page, email: str, mfa_provider: MfaCodeProvider, login_url: str | None = None
def login_with_mfa(
driver: WebDriver, email: str, mfa_provider: MfaCodeProvider, login_url: str | None = None
) -> None:
"""Log in with MFA. Only proceeds if email input exists after navigation."""
email_input = page.locator(app_strings.auth.email_input)
# Check if email input exists
email_input_exists = _element_exists(driver, app_strings.auth.email_input)
# Check if we need to navigate to the login page
if await email_input.count() == 0:
if not email_input_exists:
if login_url:
_response = await page.goto(login_url)
del _response
driver.get(login_url)
# Check again after navigation - user might already be logged in
if await email_input.count() == 0:
email_input_exists = _element_exists(driver, app_strings.auth.email_input)
if not email_input_exists:
# No email input after navigation - already logged in
return
else:
# No login URL and no email input - already logged in
return
await page.fill(app_strings.auth.email_input, email)
await page.click(app_strings.auth.send_code_button)
email_input = driver.find_element(By.CSS_SELECTOR, app_strings.auth.email_input)
email_input.clear()
email_input.send_keys(email)
driver.find_element(By.CSS_SELECTOR, app_strings.auth.send_code_button).click()
code = mfa_provider.get_code(email)
await page.fill(app_strings.auth.code_input, code)
await page.click(app_strings.auth.submit_button)
code_input = driver.find_element(By.CSS_SELECTOR, app_strings.auth.code_input)
code_input.clear()
code_input.send_keys(code)
driver.find_element(By.CSS_SELECTOR, app_strings.auth.submit_button).click()
async def logout(page: Page) -> None:
def logout(driver: WebDriver) -> None:
"""Log out if the logout button exists."""
logout_btn = page.locator(app_strings.auth.logout_button)
if await logout_btn.count() > 0:
await logout_btn.click()
if _element_exists(driver, app_strings.auth.logout_button):
driver.find_element(By.CSS_SELECTOR, app_strings.auth.logout_button).click()
async def ensure_persona(
page: Page,
def ensure_persona(
driver: WebDriver,
persona: DemoPersona,
mfa_provider: MfaCodeProvider,
login_url: str | None = None,
) -> None:
current = await detect_current_persona(page)
"""Ensure the specified persona is logged in.
If already logged in as the persona, does nothing.
Otherwise, logs out and logs in as the persona.
"""
current = detect_current_persona(driver)
if current and current.lower() == persona.email.lower():
return
await logout(page)
await login_with_mfa(page, persona.email, mfa_provider, login_url=login_url)
logout(driver)
login_with_mfa(driver, persona.email, mfa_provider, login_url=login_url)
def _element_exists(driver: WebDriver, selector: str) -> bool:
"""Check if element matching selector exists in DOM."""
try:
driver.find_element(By.CSS_SELECTOR, selector)
return True
except NoSuchElementException:
return False

View File

@@ -1,22 +1,20 @@
import contextlib
from collections.abc import AsyncIterator
from playwright.async_api import Page
from selenium.webdriver.remote.webdriver import WebDriver
from guide.app.browser.pool import BrowserPool
class BrowserClient:
"""Provides page access via a persistent browser pool with context isolation.
"""Provides WebDriver access via a persistent browser pool.
This client uses the BrowserPool to obtain fresh browser contexts for each
request. Each context is isolated and closed after use to prevent state
pollution between actions.
This client uses the BrowserPool to obtain WebDriver instances for each
request. Selenium doesn't have separate contexts like Playwright, so we
directly return the WebDriver.
Context lifecycle:
- Creation: Fresh context allocated from pool on request
- Usage: Exclusive use during action execution
- Cleanup: Context closed immediately after use
For CDP mode: Returns persistent WebDriver connected to existing browser.
For headless mode: Returns WebDriver from pool.
"""
def __init__(self, pool: BrowserPool) -> None:
@@ -28,17 +26,14 @@ class BrowserClient:
self.pool: BrowserPool = pool
@contextlib.asynccontextmanager
async def open_page(self, host_id: str | None = None) -> AsyncIterator[Page]:
"""Get a fresh page from the pool with guaranteed isolation.
For headless mode: allocates a new context and page, closes context after use.
For CDP mode: uses existing context and page, does not close context.
async def open_page(self, host_id: str | None = None) -> AsyncIterator[WebDriver]:
"""Get a WebDriver instance from the pool.
Args:
host_id: The host identifier, or None for the default host
Yields:
A Playwright Page instance
A Selenium WebDriver instance
Raises:
ConfigError: If the host_id is invalid or not configured
@@ -48,17 +43,15 @@ class BrowserClient:
_logger = logging.getLogger(__name__)
_logger.info(f"[BrowserClient] open_page called for host_id: {host_id}")
context, page, should_close = await self.pool.allocate_context_and_page(host_id)
# Get driver from pool (synchronous call)
driver = self.pool.get_driver(host_id)
_logger.info(f"[BrowserClient] Got page from pool, should_close: {should_close}")
_logger.info(f"[BrowserClient] Got WebDriver from pool")
try:
yield page
yield driver
finally:
_logger.info(f"[BrowserClient] Cleaning up, should_close: {should_close}")
# Only close context for headless mode (not CDP)
if should_close:
with contextlib.suppress(Exception):
await context.close()
_logger.info(f"[BrowserClient] Request complete, driver remains in pool")
# WebDriver stays in pool, no cleanup needed per request
__all__ = ["BrowserClient"]

View File

@@ -7,78 +7,81 @@ when actions fail, enabling better debugging in headless/CI environments.
import base64
from datetime import datetime, timezone
from playwright.async_api import Page
from selenium.webdriver.remote.webdriver import WebDriver
from guide.app.models.domain.models import DebugInfo
async def capture_screenshot(page: Page) -> str | None:
async def capture_screenshot(driver: WebDriver) -> str | None:
"""Capture page screenshot as base64-encoded PNG.
Args:
page: The Playwright page instance
driver: The Selenium WebDriver instance
Returns:
Base64-encoded PNG image data, or None if capture fails
"""
try:
screenshot_bytes = await page.screenshot()
screenshot_bytes = driver.get_screenshot_as_png()
return base64.b64encode(screenshot_bytes).decode("utf-8")
except Exception:
# Return None if screenshot fails (e.g., page already closed)
# Return None if screenshot fails (e.g., driver session closed)
return None
async def capture_html(page: Page) -> str | None:
async def capture_html(driver: WebDriver) -> str | None:
"""Capture page HTML content.
Args:
page: The Playwright page instance
driver: The Selenium WebDriver instance
Returns:
HTML page content, or None if capture fails
"""
try:
return await page.content()
return driver.page_source
except Exception:
# Return None if HTML capture fails (e.g., page already closed)
# Return None if HTML capture fails (e.g., driver session closed)
return None
async def capture_console_logs(_page: Page) -> list[str]:
async def capture_console_logs(driver: WebDriver) -> list[str]:
"""Capture console messages logged during page lifecycle.
Note: This captures messages that were emitted during the page's
lifetime. To get comprehensive logs, attach listeners early in
the page lifecycle.
Note: Selenium captures browser console logs if logging capabilities
are enabled. Logs may not be available for all browser types or
configurations.
Args:
_page: The Playwright page instance
driver: The Selenium WebDriver instance
Returns:
List of console message strings (empty if none captured)
"""
# Playwright doesn't provide direct access to historical console logs,
# but we can provide a hook for attaching a console listener at page creation.
# For now, return empty list (see browser/client.py for listener attachment).
return []
try:
# Attempt to retrieve browser logs (requires 'goog:loggingPrefs' capability)
logs = driver.get_log("browser")
return [f"{log['level']}: {log['message']}" for log in logs]
except Exception:
# Logging may not be available for all browsers/configurations
return []
async def capture_all_diagnostics(page: Page) -> DebugInfo:
async def capture_all_diagnostics(driver: WebDriver) -> DebugInfo:
"""Capture all diagnostic information (screenshot, HTML, logs).
Attempts to capture screenshot, HTML, and console logs. If any
capture fails (e.g., page closed), that field is set to None or empty.
capture fails (e.g., driver session closed), that field is set to None or empty.
Args:
page: The Playwright page instance
driver: The Selenium WebDriver instance
Returns:
DebugInfo with all captured diagnostic data
"""
screenshot = await capture_screenshot(page)
html = await capture_html(page)
logs = await capture_console_logs(page)
screenshot = await capture_screenshot(driver)
html = await capture_html(driver)
logs = await capture_console_logs(driver)
return DebugInfo(
screenshot_base64=screenshot,

View File

@@ -1,6 +1,6 @@
"""High-level page interaction helpers for demo actions.
Provides a stateful wrapper around Playwright Page with:
Provides a stateful wrapper around Selenium WebDriver with:
- Integrated wait utilities for page conditions
- Diagnostics capture for debugging
- Accordion collapse and other UI patterns
@@ -9,7 +9,9 @@ Provides a stateful wrapper around Playwright Page with:
from typing import TypedDict
from playwright.async_api import Page, TimeoutError as PlaywrightTimeoutError
from selenium.common.exceptions import TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.remote.webdriver import WebDriver
from guide.app import errors
from guide.app.browser.wait import (
@@ -36,15 +38,15 @@ class AccordionCollapseResult(TypedDict):
class PageHelpers:
"""High-level page interaction wrapper for demo actions.
Wraps a Playwright Page instance with convenient methods for:
Wraps a Selenium WebDriver instance with convenient methods for:
- Waiting on page conditions (selector, network idle, stability)
- Capturing diagnostics (screenshot, HTML, logs)
- Common UI patterns (fill and advance, click and wait, search and select)
- Accordion collapse and similar UI operations
Usage:
async with browser_client.open_page(host_id) as page:
helpers = PageHelpers(page)
async with browser_client.open_page(host_id) as driver:
helpers = PageHelpers(driver)
await helpers.fill_and_advance(
selector=app_strings.intake.description_field,
value="My request",
@@ -52,13 +54,13 @@ class PageHelpers:
)
"""
def __init__(self, page: Page) -> None:
"""Initialize helpers with a Playwright page.
def __init__(self, driver: WebDriver) -> None:
"""Initialize helpers with a Selenium WebDriver.
Args:
page: The Playwright page instance to wrap
driver: The Selenium WebDriver instance to wrap
"""
self.page: Page = page
self.driver: WebDriver = driver
# --- Wait utilities (wrapped for convenience) ---
@@ -70,13 +72,13 @@ class PageHelpers:
"""Wait for selector to appear in the DOM.
Args:
selector: CSS or Playwright selector string
selector: CSS selector string
timeout_ms: Maximum time to wait in milliseconds (default: 5000)
Raises:
GuideError: If selector not found within timeout
"""
await wait_for_selector(self.page, selector, timeout_ms)
await wait_for_selector(self.driver, selector, timeout_ms)
async def wait_for_network_idle(self, timeout_ms: int = 5000) -> None:
"""Wait for network to become idle (no active requests).
@@ -87,7 +89,7 @@ class PageHelpers:
Raises:
GuideError: If network does not idle within timeout
"""
await wait_for_network_idle(self.page, timeout_ms)
await wait_for_network_idle(self.driver, timeout_ms)
async def wait_for_navigation(self, timeout_ms: int = 5000) -> None:
"""Wait for page navigation to complete.
@@ -98,7 +100,7 @@ class PageHelpers:
Raises:
GuideError: If navigation does not complete within timeout
"""
await wait_for_navigation(self.page, timeout_ms)
await wait_for_navigation(self.driver, timeout_ms)
async def wait_for_stable(
self,
@@ -114,7 +116,7 @@ class PageHelpers:
Raises:
GuideError: If page does not stabilize after retries
"""
await wait_for_stable_page(self.page, stability_check_ms, samples)
await wait_for_stable_page(self.driver, stability_check_ms, samples)
# --- Diagnostics ---
@@ -127,7 +129,7 @@ class PageHelpers:
# Lazy import to avoid circular dependencies with Pydantic models
from guide.app.browser.diagnostics import capture_all_diagnostics
return await capture_all_diagnostics(self.page)
return await capture_all_diagnostics(self.driver)
# --- High-level UI operations ---
@@ -149,8 +151,10 @@ class PageHelpers:
next_selector: Button selector to click after filling
wait_for_idle: Wait for network idle after click (default: True)
"""
await self.page.fill(selector, value)
await self.page.click(next_selector)
field = self.driver.find_element(By.CSS_SELECTOR, selector)
field.clear()
field.send_keys(value)
self.driver.find_element(By.CSS_SELECTOR, next_selector).click()
if wait_for_idle:
await self.wait_for_network_idle()
@@ -171,10 +175,12 @@ class PageHelpers:
result_selector: Result item selector
index: Which result to click (default: 0 for first)
"""
await self.page.fill(search_input, query)
field = self.driver.find_element(By.CSS_SELECTOR, search_input)
field.clear()
field.send_keys(query)
await self.wait_for_network_idle()
results = self.page.locator(result_selector)
await results.nth(index).click()
results = self.driver.find_elements(By.CSS_SELECTOR, result_selector)
results[index].click()
async def click_and_wait(
self,
@@ -189,7 +195,7 @@ class PageHelpers:
wait_for_idle: Wait for network idle after click (default: True)
wait_for_stable: Wait for page stability after click (default: False)
"""
await self.page.click(selector)
self.driver.find_element(By.CSS_SELECTOR, selector).click()
if wait_for_idle:
await self.wait_for_network_idle()
if wait_for_stable:
@@ -222,8 +228,8 @@ class PageHelpers:
ActionExecutionError: If no buttons found or all clicks failed
"""
# Find all buttons matching the selector
buttons = self.page.locator(selector)
count = await buttons.count()
buttons = self.driver.find_elements(By.CSS_SELECTOR, selector)
count = len(buttons)
if count == 0:
# No buttons found - return success with zero count
@@ -240,16 +246,16 @@ class PageHelpers:
# Material-UI uses KeyboardArrowUpOutlinedIcon when expanded
for i in range(count):
try:
button = buttons.nth(i)
button = buttons[i]
# Check if this specific button contains the up icon
up_icon = button.locator(
'svg[data-testid="KeyboardArrowUpOutlinedIcon"]'
up_icon = button.find_elements(
By.CSS_SELECTOR, 'svg[data-testid="KeyboardArrowUpOutlinedIcon"]'
)
# Fast check: if icon exists, button is expanded
if await up_icon.count() > 0:
await button.click()
if len(up_icon) > 0:
button.click()
collapsed_count += 1
except PlaywrightTimeoutError:
except TimeoutException:
# Timeout on click - track failure but continue
failed_indices.append(i)
except Exception:

View File

@@ -5,21 +5,19 @@ expensive overhead of launching/connecting to browsers on each request.
Architecture:
- BrowserPool: Manages the lifecycle of browser instances by host
- Per host: Single persistent browser connection
- Per action: Fresh BrowserContext for complete isolation
- No page/context pooling: Each action gets a clean slate
- Per host: Single persistent WebDriver connection
- Selenium-based: Uses debugger_address for CDP, standard launch for headless
"""
import contextlib
import logging
from playwright.async_api import (
Browser,
BrowserContext,
Page,
Playwright,
async_playwright,
)
from selenium import webdriver
from selenium.webdriver.chrome.options import Options as ChromeOptions
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.firefox.options import Options as FirefoxOptions
from selenium.webdriver.firefox.service import Service as FirefoxService
from selenium.webdriver.remote.webdriver import WebDriver
from guide.app.core.config import AppSettings, BrowserHostConfig, HostKind
from guide.app import errors
@@ -31,107 +29,50 @@ _logger = logging.getLogger(__name__)
class BrowserInstance:
"""Manages a single browser connection and its lifecycle.
Creates fresh contexts for each request to ensure complete isolation
between actions. No context pooling or reuse.
For CDP mode: Connects via debugger_address (no page refresh).
For headless mode: Launches browser with Selenium WebDriver.
"""
def __init__(
self, host_id: str, host_config: BrowserHostConfig, browser: Browser
self, host_id: str, host_config: BrowserHostConfig, driver: WebDriver
) -> None:
"""Initialize a browser instance for a host.
Args:
host_id: The host identifier
host_config: The host configuration
browser: The Playwright browser instance
driver: The Selenium WebDriver instance
"""
self.host_id: str = host_id
self.host_config: BrowserHostConfig = host_config
self.browser: Browser = browser
# Cache CDP context and page to avoid re-querying (which causes refresh)
self._cdp_context: BrowserContext | None = None
self._cdp_page: Page | None = None
self.driver: WebDriver = driver
async def allocate_context_and_page(self) -> tuple[BrowserContext, Page, bool]:
"""Allocate a fresh context and page for this request.
For CDP mode: returns cached context and page from initial connection.
For headless mode: creates a new context for complete isolation.
def get_driver(self) -> WebDriver:
"""Get the WebDriver instance for this browser.
Returns:
Tuple of (context, page, should_close):
- context: Browser context
- page: Page instance
- should_close: True if context should be closed after use, False for CDP
WebDriver instance
Raises:
BrowserConnectionError: If context/page creation fails
BrowserConnectionError: If driver is not available
"""
try:
if self.host_config.kind == HostKind.CDP:
_logger.debug(f"[CDP-{self.host_id}] allocate_context_and_page called")
# Return cached CDP context/page to avoid re-querying (which causes refresh)
if self._cdp_context is None or self._cdp_page is None:
_logger.info(f"[CDP-{self.host_id}] First access - querying browser.contexts")
# First time: get context and page, then cache them
contexts = self.browser.contexts
_logger.debug(f"[CDP-{self.host_id}] Got {len(contexts)} contexts")
if not contexts:
raise errors.BrowserConnectionError(
f"No contexts available in CDP browser for host {self.host_id}",
details={"host_id": self.host_id},
)
context = contexts[0]
_logger.debug(f"[CDP-{self.host_id}] Querying context.pages")
pages = context.pages
_logger.debug(f"[CDP-{self.host_id}] Got {len(pages)} pages: {[p.url for p in pages]}")
if not pages:
raise errors.BrowserConnectionError(
f"No pages available in CDP browser context for host {self.host_id}",
details={"host_id": self.host_id},
)
# Find the active (most recently used) non-devtools page
non_devtools_pages = [p for p in pages if not p.url.startswith("devtools://")]
if not non_devtools_pages:
raise errors.BrowserConnectionError(
f"No application pages found in CDP browser (only devtools pages)",
details={"host_id": self.host_id, "pages": [p.url for p in pages]},
)
# Cache the context and page for reuse
self._cdp_context = context
self._cdp_page = non_devtools_pages[-1]
_logger.info(f"[CDP-{self.host_id}] Cached page: {self._cdp_page.url}")
else:
_logger.debug(f"[CDP-{self.host_id}] Using cached page: {self._cdp_page.url}")
_logger.debug(f"[CDP-{self.host_id}] Returning cached context and page")
# Return cached references (doesn't trigger refresh)
return self._cdp_context, self._cdp_page, False # Don't close CDP contexts
# Headless mode: create new context for isolation
context = await self.browser.new_context()
page = await context.new_page()
return context, page, True # Close headless contexts
except errors.BrowserConnectionError:
raise
except Exception as exc:
if not self.driver:
raise errors.BrowserConnectionError(
f"Failed to allocate page for host {self.host_id}",
details={"host_id": self.host_id, "host_kind": self.host_config.kind},
) from exc
f"Driver not available for host {self.host_id}",
details={"host_id": self.host_id},
)
return self.driver
async def close(self) -> None:
def close(self) -> None:
"""Close the browser connection."""
with contextlib.suppress(Exception):
await self.browser.close()
self.driver.quit()
class BrowserPool:
"""Manages browser instances across multiple hosts.
Maintains one persistent browser connection per host. Browser connections are
reused, but contexts are created fresh for each request to ensure complete
isolation between actions.
Maintains one persistent WebDriver connection per host using Selenium.
"""
def __init__(self, settings: AppSettings) -> None:
@@ -142,56 +83,31 @@ class BrowserPool:
"""
self.settings: AppSettings = settings
self._instances: dict[str, BrowserInstance] = {}
self._playwright: Playwright | None = None
self._closed: bool = False
async def initialize(self) -> None:
def initialize(self) -> None:
"""Initialize the browser pool.
Starts the Playwright instance and eagerly connects to all CDP hosts
to avoid page refreshes during user interactions.
Uses lazy initialization - connections are created on first use.
"""
if self._playwright is not None:
return
self._playwright = await async_playwright().start()
# No eager connections - instances created lazily on first request
_logger.info("Browser pool initialized (lazy mode)")
# Eagerly connect to all CDP hosts to avoid refresh on first use
for host_id, host_config in self.settings.browser_hosts.items():
if host_config.kind == HostKind.CDP:
try:
instance = await self._create_instance(host_id, host_config)
self._instances[host_id] = instance
# Eagerly cache the page reference to avoid querying on first request
await instance.allocate_context_and_page()
_logger.info(f"Eagerly connected to CDP host '{host_id}' and cached page")
except Exception as exc:
_logger.warning(
f"Failed to eagerly connect to CDP host '{host_id}': {exc}"
)
_logger.info("Browser pool initialized")
async def close(self) -> None:
"""Close all browser connections and the Playwright instance."""
def close(self) -> None:
"""Close all browser connections."""
if self._closed:
return
self._closed = True
for instance in self._instances.values():
with contextlib.suppress(Exception):
await instance.close()
instance.close()
self._instances.clear()
if self._playwright:
with contextlib.suppress(Exception):
await self._playwright.stop()
self._playwright = None
_logger.info("Browser pool closed")
async def allocate_context_and_page(
self, host_id: str | None = None
) -> tuple[BrowserContext, Page, bool]:
"""Allocate a fresh context and page for the specified host.
def get_driver(self, host_id: str | None = None) -> WebDriver:
"""Get the WebDriver for the specified host.
Lazily creates browser connections on first request per host.
@@ -199,20 +115,12 @@ class BrowserPool:
host_id: The host identifier, or None for the default host
Returns:
Tuple of (context, page, should_close):
- context: Browser context
- page: Page instance
- should_close: True if context should be closed after use, False for CDP
WebDriver instance
Raises:
ConfigError: If the host_id is invalid or not configured
BrowserConnectionError: If the browser connection fails
"""
if self._playwright is None:
raise errors.ConfigError(
"Browser pool not initialized. Call initialize() first."
)
resolved_id = host_id or self.settings.default_browser_host_id
host_config = self.settings.browser_hosts.get(resolved_id)
if not host_config:
@@ -223,75 +131,75 @@ class BrowserPool:
# Get or create the browser instance for this host
if resolved_id not in self._instances:
instance = await self._create_instance(resolved_id, host_config)
instance = self._create_instance(resolved_id, host_config)
self._instances[resolved_id] = instance
return await self._instances[resolved_id].allocate_context_and_page()
return self._instances[resolved_id].get_driver()
async def _create_instance(
def _create_instance(
self, host_id: str, host_config: BrowserHostConfig
) -> BrowserInstance:
"""Create a new browser instance for the given host."""
assert self._playwright is not None
if host_config.kind == HostKind.CDP:
browser = await self._connect_cdp(host_config)
driver = self._connect_cdp(host_config)
else:
browser = await self._launch_headless(host_config)
driver = self._launch_headless(host_config)
instance = BrowserInstance(host_id, host_config, browser)
instance = BrowserInstance(host_id, host_config, driver)
_logger.info(
f"Created browser instance for host '{host_id}' ({host_config.kind})"
)
return instance
async def _connect_cdp(self, host_config: BrowserHostConfig) -> Browser:
"""Connect to a CDP host."""
assert self._playwright is not None
def _connect_cdp(self, host_config: BrowserHostConfig) -> WebDriver:
"""Connect to a CDP host using debugger_address.
This method connects to an existing browser via CDP without triggering
page refreshes, unlike Playwright's connect_over_cdp approach.
"""
if not host_config.host or host_config.port is None:
raise errors.ConfigError("CDP host requires 'host' and 'port' fields.")
cdp_url = f"http://{host_config.host}:{host_config.port}"
debugger_address = f"{host_config.host}:{host_config.port}"
try:
browser = await self._playwright.chromium.connect_over_cdp(cdp_url)
_logger.info(f"Connected to CDP endpoint: {cdp_url}")
return browser
chrome_options = ChromeOptions()
chrome_options.debugger_address = debugger_address
# Create WebDriver connected to existing browser
driver = webdriver.Chrome(options=chrome_options)
_logger.info(f"Connected to CDP endpoint via debugger_address: {debugger_address}")
return driver
except Exception as exc:
raise errors.BrowserConnectionError(
f"Cannot connect to CDP endpoint {cdp_url}",
f"Cannot connect to CDP endpoint {debugger_address}",
details={"host": host_config.host, "port": host_config.port},
) from exc
async def _launch_headless(self, host_config: BrowserHostConfig) -> Browser:
"""Launch a headless browser."""
assert self._playwright is not None
def _launch_headless(self, host_config: BrowserHostConfig) -> WebDriver:
"""Launch a headless browser using Selenium WebDriver."""
browser_name = (host_config.browser or "chromium").lower()
browser_type = self._resolve_browser_type(host_config.browser)
try:
browser = await browser_type.launch(headless=True)
_logger.info(
f"Launched headless browser: {host_config.browser or 'chromium'}"
)
return browser
if browser_name in ("chromium", "chrome"):
chrome_options = ChromeOptions()
chrome_options.add_argument("--headless=new")
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
driver = webdriver.Chrome(options=chrome_options)
elif browser_name == "firefox":
firefox_options = FirefoxOptions()
firefox_options.add_argument("--headless")
driver = webdriver.Firefox(options=firefox_options)
else:
raise errors.ConfigError(f"Unsupported browser type '{browser_name}'")
_logger.info(f"Launched headless browser: {browser_name}")
return driver
except Exception as exc:
raise errors.BrowserConnectionError(
f"Cannot launch headless browser: {host_config.browser}",
details={"browser_type": host_config.browser},
f"Cannot launch headless browser: {browser_name}",
details={"browser_type": browser_name},
) from exc
def _resolve_browser_type(self, browser: str | None):
"""Resolve browser type from configuration."""
assert self._playwright is not None
desired = (browser or "chromium").lower()
if desired == "chromium":
return self._playwright.chromium
if desired == "firefox":
return self._playwright.firefox
if desired == "webkit":
return self._playwright.webkit
raise errors.ConfigError(f"Unsupported browser type '{browser}'")
__all__ = ["BrowserPool", "BrowserInstance"]

View File

@@ -1,81 +1,99 @@
"""Playwright wait and stability helpers for reliable action execution.
"""Selenium wait and stability helpers for reliable action execution.
Provides utilities for waiting on page conditions, detecting network idle,
and verifying visual stability before proceeding with actions.
Provides utilities for waiting on page conditions and verifying visual
stability before proceeding with actions.
"""
import asyncio
import time
from playwright.async_api import Page, TimeoutError as PlaywrightTimeoutError
from selenium.common.exceptions import TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.remote.webdriver import WebDriver
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
from guide.app import errors
from guide.app.utils.retry import retry_async
async def wait_for_selector(
page: Page,
driver: WebDriver,
selector: str,
timeout_ms: int = 5000,
) -> None:
"""Wait for an element matching selector to be present in DOM.
Args:
page: The Playwright page instance
selector: CSS or Playwright selector string
driver: The Selenium WebDriver instance
selector: CSS selector string
timeout_ms: Maximum time to wait in milliseconds (default: 5000)
Raises:
GuideError: If selector not found within timeout
"""
try:
_ = await page.wait_for_selector(selector, timeout=timeout_ms)
except PlaywrightTimeoutError as exc:
wait = WebDriverWait(driver, timeout_ms / 1000)
wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, selector)))
except TimeoutException as exc:
msg = f"Selector '{selector}' not found within {timeout_ms}ms"
raise errors.GuideError(msg) from exc
async def wait_for_navigation(
page: Page,
driver: WebDriver,
timeout_ms: int = 5000,
) -> None:
"""Wait for page navigation to complete.
Args:
page: The Playwright page instance
driver: The Selenium WebDriver instance
timeout_ms: Maximum time to wait in milliseconds (default: 5000)
Raises:
GuideError: If navigation does not complete within timeout
"""
try:
await page.wait_for_load_state("networkidle", timeout=timeout_ms)
except PlaywrightTimeoutError as exc:
# Wait for document.readyState to be 'complete'
wait = WebDriverWait(driver, timeout_ms / 1000)
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
except TimeoutException as exc:
msg = f"Page navigation did not complete within {timeout_ms}ms"
raise errors.GuideError(msg) from exc
async def wait_for_network_idle(
page: Page,
driver: WebDriver,
timeout_ms: int = 5000,
) -> None:
"""Wait for network to become idle (no active requests).
Note: Selenium doesn't have native network idle support like Playwright.
This implementation waits for document.readyState and jQuery.active (if present).
Args:
page: The Playwright page instance
driver: The Selenium WebDriver instance
timeout_ms: Maximum time to wait in milliseconds (default: 5000)
Raises:
GuideError: If network does not idle within timeout
"""
try:
await page.wait_for_load_state("networkidle", timeout=timeout_ms)
except PlaywrightTimeoutError as exc:
wait = WebDriverWait(driver, timeout_ms / 1000)
# Wait for document ready state
wait.until(lambda d: d.execute_script("return document.readyState") == "complete")
# If jQuery is present, wait for ajax requests to complete
try:
wait.until(lambda d: d.execute_script("return typeof jQuery !== 'undefined' ? jQuery.active === 0 : true"))
except Exception:
# jQuery not present or script error, continue
pass
except TimeoutException as exc:
msg = f"Network did not idle within {timeout_ms}ms"
raise errors.GuideError(msg) from exc
async def is_page_stable(
page: Page,
driver: WebDriver,
stability_check_ms: int = 500,
samples: int = 3,
) -> bool:
@@ -85,7 +103,7 @@ async def is_page_stable(
indicating visual stability for reliable element interaction.
Args:
page: The Playwright page instance
driver: The Selenium WebDriver instance
stability_check_ms: Delay between samples in milliseconds (default: 500)
samples: Number of stable samples required (default: 3)
@@ -96,13 +114,13 @@ async def is_page_stable(
previous_content: str | None = None
for _ in range(samples):
current_content = await page.content()
current_content = driver.page_source
if previous_content and current_content != previous_content:
return False
previous_content = current_content
await asyncio.sleep(stability_check_ms / 1000)
time.sleep(stability_check_ms / 1000)
return True
except Exception:
@@ -112,7 +130,7 @@ async def is_page_stable(
@retry_async(retries=3, delay_seconds=0.2)
async def wait_for_stable_page(
page: Page,
driver: WebDriver,
stability_check_ms: int = 500,
samples: int = 3,
) -> None:
@@ -122,14 +140,14 @@ async def wait_for_stable_page(
Useful for SPAs or pages with animations/transitions.
Args:
page: The Playwright page instance
driver: The Selenium WebDriver instance
stability_check_ms: Delay between samples in milliseconds (default: 500)
samples: Number of stable samples required (default: 3)
Raises:
GuideError: If page does not stabilize after retries
"""
stable = await is_page_stable(page, stability_check_ms, samples)
stable = await is_page_stable(driver, stability_check_ms, samples)
if not stable:
msg = "Page did not stabilize after retries"
raise errors.GuideError(msg)

View File

@@ -40,9 +40,9 @@ def create_app() -> FastAPI:
@contextlib.asynccontextmanager
async def lifespan(_app: FastAPI):
"""Manage browser pool lifecycle."""
await browser_pool.initialize()
browser_pool.initialize()
yield
await browser_pool.close()
browser_pool.close()
app.router.lifespan_context = lifespan
app.include_router(api_router)

37
test_sb_minimal.py Normal file
View File

@@ -0,0 +1,37 @@
#!/usr/bin/env python3 -u
"""Minimal SeleniumBase CDP connection test."""
import sys
print("Starting minimal test...", flush=True)
try:
print("Importing seleniumbase...", flush=True)
from seleniumbase import sb_cdp
print("Import successful!", flush=True)
host = "192.168.50.185"
port = 9223
print(f"Attempting connection to {host}:{port}...", flush=True)
# Try connection
sb = sb_cdp.Chrome(host=host, port=port)
print("CONNECTION SUCCESSFUL!", flush=True)
# Try getting URL
url = sb.cdp.get_current_url()
print(f"Current URL: {url}", flush=True)
print("SUCCESS - No page refresh detected!", flush=True)
# Clean up
sb.driver.stop()
print("Connection closed.", flush=True)
except Exception as e:
print(f"ERROR: {type(e).__name__}: {e}", flush=True)
import traceback
traceback.print_exc()
sys.exit(1)
print("Test completed successfully!", flush=True)

View File

@@ -0,0 +1,81 @@
#!/usr/bin/env python3
"""Test Selenium debugger_address connection to verify no page refresh."""
import sys
import time
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
def main():
print("=== Selenium debugger_address Connection Test ===\n")
# CDP connection parameters
host = "192.168.50.185"
port = 9223
print(f"Connecting to Chrome via debugger_address at {host}:{port}...")
try:
# Configure Chrome options to connect to existing browser
chrome_options = Options()
chrome_options.debugger_address = f"{host}:{port}"
# Connect to existing browser (no WebDriver executable needed)
driver = webdriver.Chrome(options=chrome_options)
print("✓ Connected successfully!")
# Get initial state
initial_url = driver.current_url
print(f"\nInitial URL: {initial_url}")
# Get page title
title = driver.title
print(f"Page title: {title}")
# Wait a moment
print("\nWaiting 2 seconds to observe page state...")
time.sleep(2)
# Check URL again to see if page refreshed
current_url = driver.current_url
print(f"Current URL: {current_url}")
# Check if page refreshed
if initial_url == current_url:
print("\n✓✓✓ SUCCESS: Page did NOT refresh on connection! ✓✓✓")
else:
print("\n✗✗✗ FAILURE: Page URL changed (possible refresh) ✗✗✗")
print(f" Before: {initial_url}")
print(f" After: {current_url}")
# Try a simple interaction to further test
print("\nTrying to get HTML length as an interaction test...")
html = driver.page_source
print(f"HTML length: {len(html)} characters")
# Final URL check
final_url = driver.current_url
print(f"Final URL: {final_url}")
if initial_url == final_url:
print("\n✓✓✓ FINAL RESULT: No refresh detected! ✓✓✓")
else:
print(f"\n✗✗✗ FINAL RESULT: URL changed! ✗✗✗")
# Clean up
print("\nClosing connection...")
driver.quit()
print("Done!")
except Exception as e:
print(f"\n✗✗✗ ERROR: {type(e).__name__}: {e} ✗✗✗")
import traceback
traceback.print_exc()
return 1
return 0
if __name__ == "__main__":
exit(main())

76
test_seleniumbase_cdp.py Normal file
View File

@@ -0,0 +1,76 @@
#!/usr/bin/env python3
"""POC test to verify SeleniumBase CDP connection doesn't trigger page refresh."""
from seleniumbase import sb_cdp
import time
def main():
print("=== SeleniumBase CDP Connection POC ===\n")
# CDP connection parameters (matching your hosts.yaml config)
host = "192.168.50.185"
port = 9223
print(f"Connecting to Chrome via CDP at {host}:{port}...")
try:
# Connect to existing browser
sb = sb_cdp.Chrome(host=host, port=port)
print("✓ Connected successfully!")
# Get initial state
initial_url = sb.cdp.get_current_url()
print(f"\nInitial URL: {initial_url}")
# Get page title
title = sb.cdp.get_title()
print(f"Page title: {title}")
# Wait a moment
print("\nWaiting 2 seconds to observe page state...")
sb.sleep(2)
# Check URL again to see if page refreshed
current_url = sb.cdp.get_current_url()
print(f"Current URL: {current_url}")
# Check if page refreshed
if initial_url == current_url:
print("\n✓✓✓ SUCCESS: Page did NOT refresh on connection! ✓✓✓")
else:
print("\n✗✗✗ FAILURE: Page URL changed (possible refresh) ✗✗✗")
print(f" Before: {initial_url}")
print(f" After: {current_url}")
# Try a simple interaction to further test
print("\nTrying to get HTML length as an interaction test...")
html = sb.cdp.get_html()
print(f"HTML length: {len(html)} characters")
# Final URL check
final_url = sb.cdp.get_current_url()
print(f"Final URL: {final_url}")
if initial_url == final_url:
print("\n✓✓✓ FINAL RESULT: No refresh detected! ✓✓✓")
else:
print(f"\n✗✗✗ FINAL RESULT: URL changed! ✗✗✗")
# Clean up
print("\nClosing connection...")
sb.driver.stop()
print("Done!")
except Exception as e:
print(f"\n✗✗✗ ERROR: {type(e).__name__}: {e} ✗✗✗")
import traceback
traceback.print_exc()
return 1
return 0
if __name__ == "__main__":
exit(main())