3 Commits

30 changed files with 1278 additions and 295 deletions

View File

@@ -1,7 +1,7 @@
personas:
buyer:
role: buyer
email: buyer.demo@example.com
email: travis@raindrop.com
login_method: mfa_email
browser_host_id: demo-cdp
supplier:

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
@@ -19,17 +19,17 @@ class LoginAsPersonaAction(DemoAction):
_mfa_provider: DummyMfaCodeProvider
_login_url: str
def __init__(self, personas: PersonaStore, login_url: str) -> None:
self._personas = personas
def __init__(self, persona_store: PersonaStore, login_url: str) -> None:
self._personas = persona_store
self._mfa_provider = DummyMfaCodeProvider()
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,11 +3,10 @@ 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
from guide.app.models.types import JSONValue
class DemoAction(ABC):
@@ -22,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."""
...
@@ -85,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:
@@ -97,12 +96,12 @@ class CompositeAction(DemoAction):
"""
self.context = context
results: dict[str, ActionResult] = {}
details: dict[str, JSONValue] = {}
details: dict[str, object] = {}
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

@@ -0,0 +1,115 @@
"""Demo actions for POC features and testing PageHelpers patterns.
This module demonstrates how to use PageHelpers class for high-level
browser interactions with minimal imports.
"""
from typing import ClassVar, override
from selenium.webdriver.remote.webdriver import WebDriver
from guide.app.actions.base import DemoAction, register_action
from guide.app.browser.helpers import PageHelpers, AccordionCollapseResult
from guide.app.models.domain import ActionContext, ActionResult
from guide.app.strings.registry import app_strings
@register_action
class CollapseAccordionsDemoAction(DemoAction):
"""Collapse all expanded accordion buttons on the current page.
This action demonstrates the PageHelpers pattern for browser interactions.
It finds all accordion buttons matching the selector and collapses those
that are currently expanded.
Supports optional custom selector and timeout via action parameters.
"""
id: ClassVar[str] = "demo.collapse-accordions"
description: ClassVar[str] = "Collapse all expanded accordion buttons on the page."
category: ClassVar[str] = "demo"
@override
async def run(self, driver: WebDriver, context: ActionContext) -> ActionResult:
"""Collapse accordions on the current page.
Parameters:
selector (str, optional): CSS selector for accordion buttons
(default: page header accordion button with data-cy attribute)
timeout_ms (int, optional): Timeout for finding elements in ms
(default: 5000)
Returns:
ActionResult with details including:
- collapsed_count: Number of successfully collapsed buttons
- total_found: Total number of matching buttons found
- failed_indices: Comma-separated string of button indices that failed
- message: Human-readable summary
Note:
Expanded state detection is automatic via SVG icon data-testid.
The method only clicks buttons with KeyboardArrowUpOutlinedIcon.
"""
# Get selector from params or use default
selector: str = _coerce_to_str(
context.params.get("selector"),
app_strings.common.page_header_accordion,
)
# Get timeout from params or use default
timeout_ms: int = _coerce_to_int(context.params.get("timeout_ms"), 5000)
# Use PageHelpers for the interaction (single import!)
helpers = PageHelpers(driver)
result: AccordionCollapseResult = await helpers.collapse_accordions(
selector, timeout_ms
)
# Extract result values
collapsed_count = result["collapsed_count"]
total_found = result["total_found"]
failed_indices_list = result["failed_indices"]
# Format failed indices as comma-separated string
failed_indices_str: str = ",".join(str(idx) for idx in failed_indices_list)
# Format result message
if total_found == 0:
message = f"No accordion buttons found with selector: {selector}"
elif collapsed_count == 0:
message = f"Found {total_found} accordions but failed to collapse any"
else:
message = f"Collapsed {collapsed_count} of {total_found} accordion(s)"
return ActionResult(
details={
"message": message,
"selector": selector,
"collapsed_count": collapsed_count,
"total_found": total_found,
"failed_indices": failed_indices_str,
}
)
def _coerce_to_str(value: object, default: str) -> str:
"""Coerce a value to str, or return default if None."""
if value is None:
return default
return str(value)
def _coerce_to_int(value: object, default: int) -> int:
"""Coerce a value to int, or return default if None."""
if value is None:
return default
if isinstance(value, int):
return value
if isinstance(value, str):
return int(value)
if isinstance(value, float):
return int(value)
return default
__all__ = ["CollapseAccordionsDemoAction"]

View File

@@ -1,3 +1,3 @@
from guide.app.actions.intake.basic import FillIntakeBasicAction
from guide.app.actions.intake.sourcing_request import FillIntakeBasicAction
__all__ = ["FillIntakeBasicAction"]

View File

@@ -1,23 +0,0 @@
from playwright.async_api import Page
from typing import ClassVar, override
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
@register_action
class FillIntakeBasicAction(DemoAction):
id: ClassVar[str] = "fill-intake-basic"
description: ClassVar[str] = (
"Fill the intake description and advance to the next step."
)
category: ClassVar[str] = "intake"
@override
async def run(self, page: Page, 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)
return ActionResult(details={"message": "Intake filled"})

View File

@@ -0,0 +1,190 @@
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"
description: ClassVar[str] = (
"Fill the intake description and advance to the next step."
)
category: ClassVar[str] = "intake"
@override
async def run(self, driver: WebDriver, context: ActionContext) -> ActionResult:
description_val = app_strings.intake.conveyor_belt_request
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"})
@register_action
class FillSourcingRequestAction(DemoAction):
id: ClassVar[str] = "fill-sourcing-request"
description: ClassVar[str] = (
"Fill the complete sourcing request intake form with demo data."
)
category: ClassVar[str] = "intake"
@override
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.
"""
_logger.info("[FORM-FILL] Starting form fill action")
# Fill commodities (multi-select autocomplete)
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:
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)
driver.find_element(By.CSS_SELECTOR, app_strings.intake.planned_field).click()
# Wait for the dropdown menu to appear
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
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:
driver.find_element(By.CSS_SELECTOR, app_strings.intake.regions_field).click()
# Wait for the dropdown menu to appear
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)
driver.find_element(By.CSS_SELECTOR, app_strings.intake.page_header_title).click()
time.sleep(0.5)
# Fill OpEx/CapEx (single-select dropdown)
driver.find_element(By.CSS_SELECTOR, app_strings.intake.opex_capex_field).click()
# Wait for the dropdown menu to appear
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
time.sleep(0.5)
# Fill description (required textarea)
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)
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)
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)
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_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
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={
"message": "Sourcing request form filled",
"fields_filled": {
"commodities": list(app_strings.intake.commodity_request),
"planned": app_strings.intake.planned_request,
"regions": list(app_strings.intake.regions_request),
"opex_capex": app_strings.intake.opex_capex_request,
"description": app_strings.intake.description_request,
"desired_supplier_name": app_strings.intake.desired_supplier_name_request,
"desired_supplier_contact": app_strings.intake.desired_supplier_contact_request,
"reseller": app_strings.intake.reseller_request,
"entity": app_strings.intake.entity_request,
},
}
)

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,49 +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:
if login_url:
_response = await page.goto(login_url)
del _response
await page.fill(app_strings.auth.email_input, email)
await page.click(app_strings.auth.send_code_button)
"""Log in with MFA. Only proceeds if email input exists after navigation."""
# 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 not email_input_exists:
if login_url:
driver.get(login_url)
# Check again after navigation - user might already be logged in
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
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:
await page.click(app_strings.auth.logout_button)
def logout(driver: WebDriver) -> None:
"""Log out if the logout button exists."""
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

@@ -0,0 +1,11 @@
"""Browser module exports.
Note: PageHelpers is NOT exported from this module to avoid import-time
schema generation issues with Pydantic. Import directly:
from guide.app.browser.helpers import PageHelpers
"""
from guide.app.browser.client import BrowserClient
from guide.app.browser.pool import BrowserPool
__all__ = ["BrowserClient", "BrowserPool"]

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,31 +26,32 @@ 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.
Allocates a new context and page for this request. The context is closed
after the with block completes, ensuring complete isolation from other
requests.
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 with a fresh, isolated context
A Selenium WebDriver instance
Raises:
ConfigError: If the host_id is invalid or not configured
BrowserConnectionError: If the browser connection fails
"""
context, page = await self.pool.allocate_context_and_page(host_id)
import logging
_logger = logging.getLogger(__name__)
_logger.info(f"[BrowserClient] open_page called for host_id: {host_id}")
# Get driver from pool (synchronous call)
driver = self.pool.get_driver(host_id)
_logger.info(f"[BrowserClient] Got WebDriver from pool")
try:
yield page
yield driver
finally:
# Explicitly close the context to ensure complete cleanup
# and prevent state leakage to subsequent requests
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

@@ -0,0 +1,285 @@
"""High-level page interaction helpers for demo actions.
Provides a stateful wrapper around Selenium WebDriver with:
- Integrated wait utilities for page conditions
- Diagnostics capture for debugging
- Accordion collapse and other UI patterns
- Fluent API for common interaction sequences
"""
from typing import TypedDict
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 (
wait_for_network_idle,
wait_for_navigation,
wait_for_selector,
wait_for_stable_page,
)
class AccordionCollapseResult(TypedDict):
"""Result from collapse_accordions operation."""
collapsed_count: int
"""Number of successfully collapsed buttons."""
total_found: int
"""Total number of buttons found."""
failed_indices: list[int]
"""List of button indices that failed to click."""
class PageHelpers:
"""High-level page interaction wrapper for demo actions.
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 driver:
helpers = PageHelpers(driver)
await helpers.fill_and_advance(
selector=app_strings.intake.description_field,
value="My request",
next_selector=app_strings.intake.next_button,
)
"""
def __init__(self, driver: WebDriver) -> None:
"""Initialize helpers with a Selenium WebDriver.
Args:
driver: The Selenium WebDriver instance to wrap
"""
self.driver: WebDriver = driver
# --- Wait utilities (wrapped for convenience) ---
async def wait_for_selector(
self,
selector: str,
timeout_ms: int = 5000,
) -> None:
"""Wait for selector to appear in the DOM.
Args:
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.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).
Args:
timeout_ms: Maximum time to wait in milliseconds (default: 5000)
Raises:
GuideError: If network does not idle within timeout
"""
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.
Args:
timeout_ms: Maximum time to wait in milliseconds (default: 5000)
Raises:
GuideError: If navigation does not complete within timeout
"""
await wait_for_navigation(self.driver, timeout_ms)
async def wait_for_stable(
self,
stability_check_ms: int = 500,
samples: int = 3,
) -> None:
"""Wait for page to become visually stable (DOM not changing).
Args:
stability_check_ms: Delay between stability checks in ms (default: 500)
samples: Number of stable samples required (default: 3)
Raises:
GuideError: If page does not stabilize after retries
"""
await wait_for_stable_page(self.driver, stability_check_ms, samples)
# --- Diagnostics ---
async def capture_diagnostics(self):
"""Capture all diagnostic information (screenshot, HTML, logs).
Returns:
DebugInfo with screenshot, HTML content, and console logs
"""
# Lazy import to avoid circular dependencies with Pydantic models
from guide.app.browser.diagnostics import capture_all_diagnostics
return await capture_all_diagnostics(self.driver)
# --- High-level UI operations ---
async def fill_and_advance(
self,
selector: str,
value: str,
next_selector: str,
wait_for_idle: bool = True,
) -> None:
"""Fill a field and click next button (common pattern).
Fills an input field with a value, then clicks a next/continue button.
Optionally waits for network idle after the click.
Args:
selector: Field selector to fill
value: Value to enter in the field
next_selector: Button selector to click after filling
wait_for_idle: Wait for network idle after click (default: True)
"""
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()
async def search_and_select(
self,
search_input: str,
query: str,
result_selector: str,
index: int = 0,
) -> None:
"""Type in search box and select result (common pattern).
Fills a search input, waits for network idle, then clicks a result item.
Args:
search_input: Search input field selector
query: Search query text to type
result_selector: Result item selector
index: Which result to click (default: 0 for first)
"""
field = self.driver.find_element(By.CSS_SELECTOR, search_input)
field.clear()
field.send_keys(query)
await self.wait_for_network_idle()
results = self.driver.find_elements(By.CSS_SELECTOR, result_selector)
results[index].click()
async def click_and_wait(
self,
selector: str,
wait_for_idle: bool = True,
wait_for_stable: bool = False,
) -> None:
"""Click element and optionally wait for page state.
Args:
selector: Element to click
wait_for_idle: Wait for network idle after click (default: True)
wait_for_stable: Wait for page stability after click (default: False)
"""
self.driver.find_element(By.CSS_SELECTOR, selector).click()
if wait_for_idle:
await self.wait_for_network_idle()
if wait_for_stable:
await self.wait_for_stable()
# --- Accordion operations ---
async def collapse_accordions(
self,
selector: str,
_timeout_ms: int = 5000,
) -> AccordionCollapseResult:
"""Collapse all expanded accordion buttons matching selector.
Detects expanded state by checking for SVG with
data-testid="KeyboardArrowUpOutlinedIcon" (Material-UI chevron up icon).
Clicks expanded buttons to collapse them.
Args:
selector: CSS selector for accordion buttons
_timeout_ms: Reserved for future timeout implementation (not currently used)
Returns:
Dict with keys:
- collapsed_count: Number of successfully collapsed buttons
- total_found: Total number of buttons found
- failed_indices: List of indices that failed to click
Raises:
ActionExecutionError: If no buttons found or all clicks failed
"""
# Find all buttons matching the selector
buttons = self.driver.find_elements(By.CSS_SELECTOR, selector)
count = len(buttons)
if count == 0:
# No buttons found - return success with zero count
return {
"collapsed_count": 0,
"total_found": 0,
"failed_indices": [],
}
collapsed_count = 0
failed_indices: list[int] = []
# Click each button that contains the "up" chevron icon (expanded state)
# Material-UI uses KeyboardArrowUpOutlinedIcon when expanded
for i in range(count):
try:
button = buttons[i]
# Check if this specific button contains the up icon
up_icon = button.find_elements(
By.CSS_SELECTOR, 'svg[data-testid="KeyboardArrowUpOutlinedIcon"]'
)
# Fast check: if icon exists, button is expanded
if len(up_icon) > 0:
button.click()
collapsed_count += 1
except TimeoutException:
# Timeout on click - track failure but continue
failed_indices.append(i)
except Exception:
# Other errors (element gone, stale reference, etc.)
failed_indices.append(i)
# If total failure (found buttons but couldn't collapse any),
# raise error with details
if count > 0 and collapsed_count == 0:
failed_indices_str = ",".join(str(i) for i in failed_indices)
raise errors.ActionExecutionError(
f"Failed to collapse any accordions (found {count}, all failed)",
details={
"selector": selector,
"found_count": count,
"failed_indices": failed_indices_str,
},
)
return {
"collapsed_count": collapsed_count,
"total_found": count,
"failed_indices": failed_indices,
}
__all__ = ["PageHelpers", "AccordionCollapseResult"]

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,57 +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
self.driver: WebDriver = driver
async def allocate_context_and_page(self) -> tuple[BrowserContext, Page]:
"""Allocate a fresh context and page for this request.
Both CDP and headless modes create new contexts for complete isolation.
def get_driver(self) -> WebDriver:
"""Get the WebDriver instance for this browser.
Returns:
Tuple of (context, page) - caller must close context when done
WebDriver instance
Raises:
BrowserConnectionError: If context/page creation fails
BrowserConnectionError: If driver is not available
"""
try:
context = await self.browser.new_context()
page = await context.new_page()
return context, page
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:
@@ -92,41 +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. Browser connections are created lazily
on first request to avoid startup delays.
Uses lazy initialization - connections are created on first use.
"""
if self._playwright is not None:
return
self._playwright = await async_playwright().start()
_logger.info("Browser pool initialized")
# No eager connections - instances created lazily on first request
_logger.info("Browser pool initialized (lazy mode)")
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]:
"""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.
@@ -134,17 +115,12 @@ class BrowserPool:
host_id: The host identifier, or None for the default host
Returns:
Tuple of (context, page) - caller must close context when done
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:
@@ -155,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

@@ -4,13 +4,22 @@ import os
from enum import Enum
from pathlib import Path
from collections.abc import Mapping
from typing import ClassVar, TypeAlias, cast
from typing import ClassVar, Protocol, TypeAlias, TypeVar, cast
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from guide.app.models.personas.models import DemoPersona
class _ModelWithId(Protocol):
"""Protocol for models that have an id attribute."""
id: str
_T = TypeVar("_T", bound=BaseModel)
_logger = logging.getLogger(__name__)
CONFIG_DIR = Path(__file__).resolve().parents[4] / "config"
@@ -59,8 +68,8 @@ class AppSettings(BaseSettings):
case_sensitive=False,
)
raindrop_base_url: str = "https://app.raindrop.com"
raindrop_graphql_url: str = "https://app.raindrop.com/graphql"
raindrop_base_url: str = "https://stg.raindrop.com"
raindrop_graphql_url: str = "https://raindrop-staging.hasura.app/v1/graphql"
default_browser_host_id: str = "demo-cdp"
browser_hosts: dict[str, BrowserHostConfig] = Field(default_factory=dict)
personas: dict[str, DemoPersona] = Field(default_factory=dict)
@@ -208,33 +217,23 @@ def load_settings() -> AppSettings:
# Load JSON overrides from environment
if browser_hosts_json := os.environ.get("RAINDROP_DEMO_BROWSER_HOSTS_JSON"):
try:
# Validate JSON is a list and process each record
decoded = cast(object, json.loads(browser_hosts_json))
if not isinstance(decoded, list):
raise ValueError(
"RAINDROP_DEMO_BROWSER_HOSTS_JSON must be a JSON array"
)
# Iterate only over validated list
decoded_list = cast(list[object], decoded)
for item in decoded_list:
if isinstance(item, Mapping):
host = BrowserHostConfig.model_validate(item)
hosts_dict[host.id] = host
_settings_extraction_from_json(
browser_hosts_json,
"RAINDROP_DEMO_BROWSER_HOSTS_JSON must be a JSON array",
BrowserHostConfig,
hosts_dict,
)
except (json.JSONDecodeError, ValueError) as exc:
_logger.warning(f"Failed to parse RAINDROP_DEMO_BROWSER_HOSTS_JSON: {exc}")
if personas_json := os.environ.get("RAINDROP_DEMO_PERSONAS_JSON"):
try:
# Validate JSON is a list and process each record
decoded = cast(object, json.loads(personas_json))
if not isinstance(decoded, list):
raise ValueError("RAINDROP_DEMO_PERSONAS_JSON must be a JSON array")
# Iterate only over validated list
decoded_list = cast(list[object], decoded)
for item in decoded_list:
if isinstance(item, Mapping):
persona = DemoPersona.model_validate(item)
personas_dict[persona.id] = persona
_settings_extraction_from_json(
personas_json,
"RAINDROP_DEMO_PERSONAS_JSON must be a JSON array",
DemoPersona,
personas_dict,
)
except (json.JSONDecodeError, ValueError) as exc:
_logger.warning(f"Failed to parse RAINDROP_DEMO_PERSONAS_JSON: {exc}")
@@ -251,3 +250,33 @@ def load_settings() -> AppSettings:
f"Loaded {len(settings.browser_hosts)} browser hosts, {len(settings.personas)} personas"
)
return settings
# TODO Rename this here and in `load_settings`
def _settings_extraction_from_json(
json_str: str,
error_message: str,
model_class: type[_T],
target_dict: dict[str, _T],
) -> None:
"""Extract and validate records from JSON string into target dictionary.
Args:
json_str: JSON string containing an array of records
error_message: Error message to raise if JSON is not an array
model_class: Pydantic model class to validate each record
target_dict: Dictionary to update with validated records
"""
# Validate JSON is a list and process each record
decoded = cast(object, json.loads(json_str))
if not isinstance(decoded, list):
raise ValueError(error_message)
# Iterate only over validated list
decoded_list = cast(list[object], decoded)
for item in decoded_list:
if isinstance(item, Mapping):
record = model_class.model_validate(item)
# Both BrowserHostConfig and DemoPersona have id: str per contract
# Cast through object to satisfy type checker (structural typing enforced at runtime)
record_with_id = cast(_ModelWithId, cast(object, record))
target_dict[record_with_id.id] = record

View File

@@ -1,12 +1,9 @@
from guide.app.models.types import JSONValue
class GuideError(Exception):
code: str = "UNKNOWN_ERROR"
message: str
details: dict[str, JSONValue]
details: dict[str, object]
def __init__(self, message: str, *, details: dict[str, JSONValue] | None = None):
def __init__(self, message: str, *, details: dict[str, object] | None = None):
super().__init__(message)
self.message = message
self.details = details or {}

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)

View File

@@ -4,7 +4,6 @@ from typing import Literal
from pydantic import BaseModel, Field
from guide.app.core.config import HostKind
from guide.app.models.types import JSONValue
from guide.app import utils
@@ -29,42 +28,52 @@ class DebugInfo(BaseModel):
class ActionRequest(BaseModel):
"""Request to execute a demo action."""
persona_id: str | None = None
browser_host_id: str | None = None
params: dict[str, JSONValue] = Field(default_factory=dict)
params: dict[str, object] = Field(default_factory=dict)
class ActionContext(BaseModel):
"""Action execution context passed to action handlers."""
action_id: str
persona_id: str | None = None
browser_host_id: str
params: dict[str, JSONValue] = Field(default_factory=dict)
params: dict[str, object] = Field(default_factory=dict)
correlation_id: str = Field(default_factory=utils.ids.new_correlation_id)
shared_state: dict[str, JSONValue] = Field(
shared_state: dict[str, object] = Field(
default_factory=dict,
description="Shared state for composite actions (multi-step flows)",
)
class ActionResult(BaseModel):
"""Result of a single action execution."""
status: Literal["ok", "error"] = "ok"
details: dict[str, JSONValue] = Field(default_factory=dict)
details: dict[str, object] = Field(default_factory=dict)
error: str | None = None
class ActionMetadata(BaseModel):
"""Metadata about an action for discovery."""
id: str
description: str
category: str
class ActionResponse(BaseModel):
"""Response envelope for action execution."""
status: Literal["ok", "error"]
action_id: str
browser_host_id: str
persona_id: str | None = None
correlation_id: str | None = None
details: dict[str, JSONValue] | None = None
details: dict[str, object] | None = None
error: str | None = None
@@ -74,13 +83,15 @@ class ActionStatus(str, Enum):
class ActionEnvelope(BaseModel):
"""Complete envelope with diagnostics for API responses."""
status: ActionStatus
action_id: str
correlation_id: str
result: dict[str, JSONValue] | None = None
result: dict[str, object] | None = None
error_code: str | None = None
message: str | None = None
details: dict[str, JSONValue] | None = None
details: dict[str, object] | None = None
debug_info: DebugInfo | None = None
"""Diagnostic data captured on action failure (screenshots, HTML, logs)"""

View File

@@ -6,7 +6,6 @@ from typing import cast
from guide.app import errors
from guide.app.core.config import AppSettings
from guide.app.models.personas.models import DemoPersona
from guide.app.models.types import JSONValue
class GraphQLClient:
@@ -17,10 +16,10 @@ class GraphQLClient:
self,
*,
query: str,
variables: Mapping[str, JSONValue] | None,
variables: Mapping[str, object] | None,
persona: DemoPersona | None,
operation_name: str | None = None,
) -> dict[str, JSONValue]:
) -> dict[str, object]:
url = self._settings.raindrop_graphql_url
headers = self._build_headers(persona)
async with httpx.AsyncClient(timeout=10.0) as client:
@@ -47,12 +46,12 @@ class GraphQLClient:
data = cast(dict[str, object], resp.json())
if errors_list := data.get("errors"):
details: dict[str, JSONValue] = {"errors": cast(JSONValue, errors_list)}
details: dict[str, object] = {"errors": errors_list}
raise errors.GraphQLOperationError(
"GraphQL operation failed", details=details
)
payload = data.get("data", {})
return cast(dict[str, JSONValue], payload) if isinstance(payload, dict) else {}
return cast(dict[str, object], payload) if isinstance(payload, dict) else {}
def _build_headers(self, persona: DemoPersona | None) -> dict[str, str]:
headers: dict[str, str] = {"Content-Type": "application/json"}

View File

@@ -6,3 +6,12 @@ class IntakeTexts:
"Requesting 500 tons of replacement conveyor belts for Q4 maintenance window."
)
ALT_REQUEST: ClassVar[str] = "Intake for rapid supplier onboarding and risk review."
COMMODITY_REQUEST: ClassVar[tuple[str, ...]] = ("IT Managed Services", "Infrastructure", "Services")
PLANNED_REQUEST: ClassVar[str] = "Planned"
REGIONS_REQUEST: ClassVar[tuple[str, ...]] = ("North America", "APAC")
OPEX_CAPEX_REQUEST: ClassVar[str] = "Both"
DESCRIPTION_REQUEST: ClassVar[str] = "Requesting IT Support and Management Telecom support for Q4 expansion into Asia."
DESIRED_SUPPLIER_NAME_REQUEST: ClassVar[str] = "Solid"
DESIRED_SUPPLIER_CONTACT_REQUEST: ClassVar[str] = "Sivrat E."
RESELLER_REQUEST: ClassVar[str] = "Erf, Weend, & Fyre"
ENTITY_REQUEST: ClassVar[str] = "GP AG"

View File

@@ -29,6 +29,7 @@ from guide.app.strings.labels.auth import AuthLabels
from guide.app.strings.labels.intake import IntakeLabels
from guide.app.strings.labels.sourcing import SourcingLabels
from guide.app.strings.selectors.auth import AuthSelectors
from guide.app.strings.selectors.common import CommonSelectors
from guide.app.strings.selectors.intake import IntakeSelectors
from guide.app.strings.selectors.navigation import NavigationSelectors
from guide.app.strings.selectors.sourcing import SourcingSelectors
@@ -40,16 +41,54 @@ class IntakeStrings:
Provides direct access to all intake-related UI constants.
"""
# Selectors
# Selectors - General
description_field: ClassVar[str] = IntakeSelectors.DESCRIPTION_FIELD
next_button: ClassVar[str] = IntakeSelectors.NEXT_BUTTON
# Selectors - Sourcing Request Form
form: ClassVar[str] = IntakeSelectors.FORM
requester_field: ClassVar[str] = IntakeSelectors.REQUESTER_FIELD
assigned_owner_field: ClassVar[str] = IntakeSelectors.ASSIGNED_OWNER_FIELD
legal_contact_field: ClassVar[str] = IntakeSelectors.LEGAL_CONTACT_FIELD
commodity_field: ClassVar[str] = IntakeSelectors.COMMODITY_FIELD
planned_field: ClassVar[str] = IntakeSelectors.PLANNED_FIELD
regions_field: ClassVar[str] = IntakeSelectors.REGIONS_FIELD
opex_capex_field: ClassVar[str] = IntakeSelectors.OPEX_CAPEX_FIELD
description_text_field: ClassVar[str] = IntakeSelectors.DESCRIPTION_TEXT_FIELD
description_textarea: ClassVar[str] = IntakeSelectors.DESCRIPTION_TEXTAREA
reseller_field: ClassVar[str] = IntakeSelectors.RESELLER_FIELD
reseller_textarea: ClassVar[str] = IntakeSelectors.RESELLER_TEXTAREA
target_date_field: ClassVar[str] = IntakeSelectors.TARGET_DATE_FIELD
entity_field: ClassVar[str] = IntakeSelectors.ENTITY_FIELD
desired_supplier_name_field: ClassVar[str] = IntakeSelectors.DESIRED_SUPPLIER_NAME_FIELD
desired_supplier_name_textarea: ClassVar[str] = IntakeSelectors.DESIRED_SUPPLIER_NAME_TEXTAREA
desired_supplier_contact_field: ClassVar[str] = IntakeSelectors.DESIRED_SUPPLIER_CONTACT_FIELD
desired_supplier_contact_textarea: ClassVar[str] = IntakeSelectors.DESIRED_SUPPLIER_CONTACT_TEXTAREA
select_document_field: ClassVar[str] = IntakeSelectors.SELECT_DOCUMENT_FIELD
drop_zone_container: ClassVar[str] = IntakeSelectors.DROP_ZONE_CONTAINER
drop_zone_input: ClassVar[str] = IntakeSelectors.DROP_ZONE_INPUT
back_button: ClassVar[str] = IntakeSelectors.BACK_BUTTON
submit_button: ClassVar[str] = IntakeSelectors.SUBMIT_BUTTON
close_button: ClassVar[str] = IntakeSelectors.CLOSE_BUTTON
page_header: ClassVar[str] = IntakeSelectors.PAGE_HEADER
page_header_card: ClassVar[str] = IntakeSelectors.PAGE_HEADER_CARD
page_header_title: ClassVar[str] = IntakeSelectors.PAGE_HEADER_TITLE
# Labels
description_placeholder: ClassVar[str] = IntakeLabels.DESCRIPTION_PLACEHOLDER
# Demo text
conveyor_belt_request: ClassVar[str] = IntakeTexts.CONVEYOR_BELT_REQUEST
alt_request: ClassVar[str] = IntakeTexts.ALT_REQUEST
commodity_request: ClassVar[tuple[str, ...]] = IntakeTexts.COMMODITY_REQUEST
planned_request: ClassVar[str] = IntakeTexts.PLANNED_REQUEST
regions_request: ClassVar[tuple[str, ...]] = IntakeTexts.REGIONS_REQUEST
opex_capex_request: ClassVar[str] = IntakeTexts.OPEX_CAPEX_REQUEST
description_request: ClassVar[str] = IntakeTexts.DESCRIPTION_REQUEST
desired_supplier_name_request: ClassVar[str] = IntakeTexts.DESIRED_SUPPLIER_NAME_REQUEST
desired_supplier_contact_request: ClassVar[str] = IntakeTexts.DESIRED_SUPPLIER_CONTACT_REQUEST
reseller_request: ClassVar[str] = IntakeTexts.RESELLER_REQUEST
entity_request: ClassVar[str] = IntakeTexts.ENTITY_REQUEST
class SourcingStrings:
@@ -105,11 +144,22 @@ class AuthStrings:
current_user_display_prefix: ClassVar[str] = AuthLabels.CURRENT_USER_DISPLAY_PREFIX
class CommonStrings:
"""Common UI strings: selectors for cross-domain elements.
Provides direct access to selectors for generic UI patterns used
across multiple domains (accordions, modals, etc.).
"""
# Selectors
page_header_accordion: ClassVar[str] = CommonSelectors.PAGE_HEADER_ACCORDION
class AppStrings:
"""Root registry for all application strings.
Provides hierarchical, type-safe access to selectors, labels, and demo texts.
Each namespace (intake, sourcing, navigation, auth) exposes nested classes.
Each namespace (intake, sourcing, navigation, auth, common) exposes nested classes.
GraphQL queries are maintained separately in raindrop/generated/queries.py
and loaded from .graphql files for better maintainability.
@@ -119,6 +169,7 @@ class AppStrings:
sourcing: ClassVar[type[SourcingStrings]] = SourcingStrings
navigation: ClassVar[type[NavigationStrings]] = NavigationStrings
auth: ClassVar[type[AuthStrings]] = AuthStrings
common: ClassVar[type[CommonStrings]] = CommonStrings
# Module-level instance for convenience
@@ -131,4 +182,5 @@ __all__ = [
"SourcingStrings",
"NavigationStrings",
"AuthStrings",
"CommonStrings",
]

View File

@@ -0,0 +1,30 @@
"""Common UI element selectors used across multiple domains.
Provides selectors for generic UI patterns that appear across the application,
such as accordion buttons, modals, and other reusable components.
"""
from typing import ClassVar
class CommonSelectors:
"""Common selectors for cross-domain UI elements.
These selectors are domain-agnostic and used by multiple workflows.
Note: Accordion state detection (expanded vs collapsed) is handled via
SVG icon data-testid attributes in PageHelpers.collapse_accordions:
- Expanded: SVG with data-testid="KeyboardArrowUpOutlinedIcon"
- Collapsed: SVG with data-testid="KeyboardArrowDownOutlinedIcon"
"""
PAGE_HEADER_ACCORDION: ClassVar[str] = 'button[data-cy="page-header-chervon-button"]'
"""Page header accordion chevron button (Material-UI IconButton).
State is indicated by contained SVG icon:
- Expanded: KeyboardArrowUpOutlinedIcon
- Collapsed: KeyboardArrowDownOutlinedIcon
"""
__all__ = ["CommonSelectors"]

View File

@@ -4,5 +4,51 @@ from typing import ClassVar
class IntakeSelectors:
# General intake selectors
DESCRIPTION_FIELD: ClassVar[str] = '[data-test="intake-description"]'
NEXT_BUTTON: ClassVar[str] = '[data-test="intake-next"]'
# Sourcing Request Form
FORM: ClassVar[str] = '[data-testid="what-do-you-want-form-form"]'
# User/Assignment Fields
REQUESTER_FIELD: ClassVar[str] = '[data-cy="board-item-field-user-requester"]'
ASSIGNED_OWNER_FIELD: ClassVar[str] = '[data-cy="board-item-field-user-owner"]'
LEGAL_CONTACT_FIELD: ClassVar[str] = '[data-cy="board-item-field-user-legal_contact"]'
# Category/Classification Fields
COMMODITY_FIELD: ClassVar[str] = '[data-cy="board-item-field-commodities-commodity"]'
PLANNED_FIELD: ClassVar[str] = '[data-cy="board-item-field-menu-planned"]'
REGIONS_FIELD: ClassVar[str] = '[data-cy="board-item-field-regions-f32"]'
OPEX_CAPEX_FIELD: ClassVar[str] = '[data-cy="board-item-field-menu-opex_capex"]'
# Text/Description Fields
DESCRIPTION_TEXT_FIELD: ClassVar[str] = '[data-cy="board-item-field-text-description"]'
DESCRIPTION_TEXTAREA: ClassVar[str] = 'textarea[name="description"]'
RESELLER_FIELD: ClassVar[str] = '[data-cy="board-item-field-text-reseller"]'
RESELLER_TEXTAREA: ClassVar[str] = 'textarea[name="reseller"]'
# Date Fields
TARGET_DATE_FIELD: ClassVar[str] = 'input[name="target_date"]'
# Entity/Supplier Fields
ENTITY_FIELD: ClassVar[str] = '[data-cy="board-item-field-entity-f31"]'
DESIRED_SUPPLIER_NAME_FIELD: ClassVar[str] = '[data-cy="board-item-field-text-f45"]'
DESIRED_SUPPLIER_NAME_TEXTAREA: ClassVar[str] = 'textarea[name="f45"]'
DESIRED_SUPPLIER_CONTACT_FIELD: ClassVar[str] = '[data-cy="board-item-field-text-f46"]'
DESIRED_SUPPLIER_CONTACT_TEXTAREA: ClassVar[str] = 'textarea[name="f46"]'
# Document/Attachment Fields
SELECT_DOCUMENT_FIELD: ClassVar[str] = '[data-cy="rddropzone-add-document"]'
DROP_ZONE_CONTAINER: ClassVar[str] = '#rd-drop-zone-input-div'
DROP_ZONE_INPUT: ClassVar[str] = '#rd-drop-zone-input'
# Action Buttons
BACK_BUTTON: ClassVar[str] = 'button:has-text("Back")'
SUBMIT_BUTTON: ClassVar[str] = 'button[type="submit"]:has-text("Submit")'
CLOSE_BUTTON: ClassVar[str] = '[data-cy="page-header-close-button"]'
# Page Header Elements
PAGE_HEADER: ClassVar[str] = '#page-header'
PAGE_HEADER_CARD: ClassVar[str] = '[data-cy^="page-header-card-"]'
PAGE_HEADER_TITLE: ClassVar[str] = '[data-cy="page-header-name-title"]'

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())

View File

@@ -5,7 +5,6 @@ from __future__ import annotations
import pytest
from unittest.mock import AsyncMock, MagicMock
from typing import TYPE_CHECKING
@pytest.fixture

View File

@@ -171,8 +171,8 @@ class TestAppSettingsDefaults:
from guide.app.core.config import AppSettings
settings = AppSettings()
assert settings.raindrop_base_url == "https://app.raindrop.com"
assert settings.raindrop_graphql_url == "https://app.raindrop.com/graphql"
assert settings.raindrop_base_url == "https://stg.raindrop.com"
assert settings.raindrop_graphql_url == "https://raindrop-staging.hasura.app/v1/graphql"
assert settings.default_browser_host_id == "demo-cdp"
assert isinstance(settings.browser_hosts, dict)
assert isinstance(settings.personas, dict)