diff --git a/src/guide/app/actions/auth/login.py b/src/guide/app/actions/auth/login.py index 31ce13a..3f62238 100644 --- a/src/guide/app/actions/auth/login.py +++ b/src/guide/app/actions/auth/login.py @@ -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"}) diff --git a/src/guide/app/actions/base.py b/src/guide/app/actions/base.py index 3517842..9ad30c3 100644 --- a/src/guide/app/actions/base.py +++ b/src/guide/app/actions/base.py @@ -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 diff --git a/src/guide/app/actions/demo/__init__.py b/src/guide/app/actions/demo/__init__.py index dd8ba0a..acdb718 100644 --- a/src/guide/app/actions/demo/__init__.py +++ b/src/guide/app/actions/demo/__init__.py @@ -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 ) diff --git a/src/guide/app/actions/intake/sourcing_request.py b/src/guide/app/actions/intake/sourcing_request.py index e607822..a5a9651 100644 --- a/src/guide/app/actions/intake/sourcing_request.py +++ b/src/guide/app/actions/intake/sourcing_request.py @@ -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={ diff --git a/src/guide/app/actions/sourcing/add_suppliers.py b/src/guide/app/actions/sourcing/add_suppliers.py index 090d5ad..d1348b2 100644 --- a/src/guide/app/actions/sourcing/add_suppliers.py +++ b/src/guide/app/actions/sourcing/add_suppliers.py @@ -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)}) diff --git a/src/guide/app/api/routes/actions.py b/src/guide/app/api/routes/actions.py index 4f91eb5..ef0ba20 100644 --- a/src/guide/app/api/routes/actions.py +++ b/src/guide/app/api/routes/actions.py @@ -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, diff --git a/src/guide/app/auth/session.py b/src/guide/app/auth/session.py index 0aa387f..65662da 100644 --- a/src/guide/app/auth/session.py +++ b/src/guide/app/auth/session.py @@ -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 diff --git a/src/guide/app/browser/client.py b/src/guide/app/browser/client.py index e06cfbc..45ffe7c 100644 --- a/src/guide/app/browser/client.py +++ b/src/guide/app/browser/client.py @@ -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"] diff --git a/src/guide/app/browser/diagnostics.py b/src/guide/app/browser/diagnostics.py index 8ae350a..55762ac 100644 --- a/src/guide/app/browser/diagnostics.py +++ b/src/guide/app/browser/diagnostics.py @@ -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, diff --git a/src/guide/app/browser/helpers.py b/src/guide/app/browser/helpers.py index 52f2d1a..410a9f4 100644 --- a/src/guide/app/browser/helpers.py +++ b/src/guide/app/browser/helpers.py @@ -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: diff --git a/src/guide/app/browser/pool.py b/src/guide/app/browser/pool.py index e6a1295..a5eed35 100644 --- a/src/guide/app/browser/pool.py +++ b/src/guide/app/browser/pool.py @@ -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"] diff --git a/src/guide/app/browser/wait.py b/src/guide/app/browser/wait.py index ceb458c..984ac8b 100644 --- a/src/guide/app/browser/wait.py +++ b/src/guide/app/browser/wait.py @@ -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) diff --git a/src/guide/app/main.py b/src/guide/app/main.py index 1162d4e..50bf584 100644 --- a/src/guide/app/main.py +++ b/src/guide/app/main.py @@ -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) diff --git a/test_sb_minimal.py b/test_sb_minimal.py new file mode 100644 index 0000000..a97e12e --- /dev/null +++ b/test_sb_minimal.py @@ -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) diff --git a/test_selenium_debugger_address.py b/test_selenium_debugger_address.py new file mode 100644 index 0000000..5ea7e85 --- /dev/null +++ b/test_selenium_debugger_address.py @@ -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()) diff --git a/test_seleniumbase_cdp.py b/test_seleniumbase_cdp.py new file mode 100644 index 0000000..c78a7d7 --- /dev/null +++ b/test_seleniumbase_cdp.py @@ -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())