3 Commits

30 changed files with 1278 additions and 295 deletions

View File

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

View File

@@ -1,7 +1,7 @@
from playwright.async_api import Page
from typing import ClassVar, override from typing import ClassVar, override
from selenium.webdriver.remote.webdriver import WebDriver
from guide.app.actions.base import DemoAction, register_action from guide.app.actions.base import DemoAction, register_action
from guide.app.auth import DummyMfaCodeProvider, ensure_persona from guide.app.auth import DummyMfaCodeProvider, ensure_persona
from guide.app import errors from guide.app import errors
@@ -19,17 +19,17 @@ class LoginAsPersonaAction(DemoAction):
_mfa_provider: DummyMfaCodeProvider _mfa_provider: DummyMfaCodeProvider
_login_url: str _login_url: str
def __init__(self, personas: PersonaStore, login_url: str) -> None: def __init__(self, persona_store: PersonaStore, login_url: str) -> None:
self._personas = personas self._personas = persona_store
self._mfa_provider = DummyMfaCodeProvider() self._mfa_provider = DummyMfaCodeProvider()
self._login_url = login_url self._login_url = login_url
@override @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: if context.persona_id is None:
raise errors.PersonaError("persona_id is required for login action") raise errors.PersonaError("persona_id is required for login action")
persona = self._personas.get(context.persona_id) persona = self._personas.get(context.persona_id)
await ensure_persona( ensure_persona(
page, persona, self._mfa_provider, login_url=self._login_url driver, persona, self._mfa_provider, login_url=self._login_url
) )
return ActionResult(details={"persona_id": persona.id, "status": "logged_in"}) 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 inspect import Parameter, signature
from typing import ClassVar, override, cast 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 import errors
from guide.app.models.domain import ActionContext, ActionMetadata, ActionResult from guide.app.models.domain import ActionContext, ActionMetadata, ActionResult
from guide.app.models.types import JSONValue
class DemoAction(ABC): class DemoAction(ABC):
@@ -22,7 +21,7 @@ class DemoAction(ABC):
category: ClassVar[str] category: ClassVar[str]
@abstractmethod @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.""" """Execute the action and return a result."""
... ...
@@ -85,11 +84,11 @@ class CompositeAction(DemoAction):
self.context: ActionContext | None = None self.context: ActionContext | None = None
@override @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. """Execute all child actions in sequence.
Args: Args:
page: The Playwright page instance driver: The Selenium WebDriver instance
context: The action context (shared across all steps) context: The action context (shared across all steps)
Returns: Returns:
@@ -97,12 +96,12 @@ class CompositeAction(DemoAction):
""" """
self.context = context self.context = context
results: dict[str, ActionResult] = {} results: dict[str, ActionResult] = {}
details: dict[str, JSONValue] = {} details: dict[str, object] = {}
for step_id in self.child_actions: for step_id in self.child_actions:
try: try:
action = self.registry.get(step_id) action = self.registry.get(step_id)
result = await action.run(page, context) result = await action.run(driver, context)
results[step_id] = result results[step_id] = result
details[step_id] = result.details 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"] __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 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.actions.base import DemoAction, register_action
from guide.app.models.domain import ActionContext, ActionResult from guide.app.models.domain import ActionContext, ActionResult
from guide.app.strings.registry import app_strings from guide.app.strings.registry import app_strings
@@ -14,9 +15,11 @@ class AddThreeSuppliersAction(DemoAction):
category: ClassVar[str] = "sourcing" category: ClassVar[str] = "sourcing"
@override @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 suppliers = app_strings.sourcing.default_trio
for supplier in suppliers: for supplier in suppliers:
await page.fill(app_strings.sourcing.supplier_search_input, supplier) search_input = driver.find_element(By.CSS_SELECTOR, app_strings.sourcing.supplier_search_input)
await page.click(app_strings.sourcing.add_supplier_button) 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)}) return ActionResult(details={"added_suppliers": list(suppliers)})

View File

@@ -94,12 +94,12 @@ async def execute_action(
mfa_provider = DummyMfaCodeProvider() mfa_provider = DummyMfaCodeProvider()
try: 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: if persona:
await ensure_persona( ensure_persona(
page, persona, mfa_provider, login_url=settings.raindrop_base_url 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: except errors.GuideError as exc:
return ActionEnvelope( return ActionEnvelope(
status=ActionStatus.ERROR, 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.auth.mfa import MfaCodeProvider
from guide.app.models.personas.models import DemoPersona from guide.app.models.personas.models import DemoPersona
from guide.app.strings.registry import app_strings 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.""" """Return the email/identifier of the currently signed-in user, if visible."""
element = page.locator(app_strings.auth.current_user_display) try:
if await element.count() == 0: element = driver.find_element(By.CSS_SELECTOR, app_strings.auth.current_user_display)
return None text = element.text
text = await element.first.text_content() if not text:
if text is None:
return None return None
prefix = app_strings.auth.current_user_display_prefix prefix = app_strings.auth.current_user_display_prefix
if prefix and text.startswith(prefix): if prefix and text.startswith(prefix):
return text.removeprefix(prefix).strip() return text.removeprefix(prefix).strip()
return text.strip() return text.strip()
except NoSuchElementException:
return None
async def login_with_mfa( def login_with_mfa(
page: Page, email: str, mfa_provider: MfaCodeProvider, login_url: str | None = None driver: WebDriver, email: str, mfa_provider: MfaCodeProvider, login_url: str | None = None
) -> None: ) -> None:
"""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: if login_url:
_response = await page.goto(login_url) driver.get(login_url)
del _response # Check again after navigation - user might already be logged in
await page.fill(app_strings.auth.email_input, email) email_input_exists = _element_exists(driver, app_strings.auth.email_input)
await page.click(app_strings.auth.send_code_button) 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) 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:
await page.click(app_strings.auth.logout_button) """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( def ensure_persona(
page: Page, driver: WebDriver,
persona: DemoPersona, persona: DemoPersona,
mfa_provider: MfaCodeProvider, mfa_provider: MfaCodeProvider,
login_url: str | None = None, login_url: str | None = 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(): if current and current.lower() == persona.email.lower():
return return
await logout(page) logout(driver)
await login_with_mfa(page, persona.email, mfa_provider, login_url=login_url) 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 import contextlib
from collections.abc import AsyncIterator 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 from guide.app.browser.pool import BrowserPool
class BrowserClient: 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 This client uses the BrowserPool to obtain WebDriver instances for each
request. Each context is isolated and closed after use to prevent state request. Selenium doesn't have separate contexts like Playwright, so we
pollution between actions. directly return the WebDriver.
Context lifecycle: For CDP mode: Returns persistent WebDriver connected to existing browser.
- Creation: Fresh context allocated from pool on request For headless mode: Returns WebDriver from pool.
- Usage: Exclusive use during action execution
- Cleanup: Context closed immediately after use
""" """
def __init__(self, pool: BrowserPool) -> None: def __init__(self, pool: BrowserPool) -> None:
@@ -28,31 +26,32 @@ class BrowserClient:
self.pool: BrowserPool = pool self.pool: BrowserPool = pool
@contextlib.asynccontextmanager @contextlib.asynccontextmanager
async def open_page(self, host_id: str | None = None) -> AsyncIterator[Page]: async def open_page(self, host_id: str | None = None) -> AsyncIterator[WebDriver]:
"""Get a fresh page from the pool with guaranteed isolation. """Get a WebDriver instance from the pool.
Allocates a new context and page for this request. The context is closed
after the with block completes, ensuring complete isolation from other
requests.
Args: Args:
host_id: The host identifier, or None for the default host host_id: The host identifier, or None for the default host
Yields: Yields:
A Playwright Page instance with a fresh, isolated context A Selenium WebDriver instance
Raises: Raises:
ConfigError: If the host_id is invalid or not configured ConfigError: If the host_id is invalid or not configured
BrowserConnectionError: If the browser connection fails 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: try:
yield page yield driver
finally: finally:
# Explicitly close the context to ensure complete cleanup _logger.info(f"[BrowserClient] Request complete, driver remains in pool")
# and prevent state leakage to subsequent requests # WebDriver stays in pool, no cleanup needed per request
with contextlib.suppress(Exception):
await context.close()
__all__ = ["BrowserClient"] __all__ = ["BrowserClient"]

View File

@@ -7,78 +7,81 @@ when actions fail, enabling better debugging in headless/CI environments.
import base64 import base64
from datetime import datetime, timezone 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 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. """Capture page screenshot as base64-encoded PNG.
Args: Args:
page: The Playwright page instance driver: The Selenium WebDriver instance
Returns: Returns:
Base64-encoded PNG image data, or None if capture fails Base64-encoded PNG image data, or None if capture fails
""" """
try: try:
screenshot_bytes = await page.screenshot() screenshot_bytes = driver.get_screenshot_as_png()
return base64.b64encode(screenshot_bytes).decode("utf-8") return base64.b64encode(screenshot_bytes).decode("utf-8")
except Exception: except Exception:
# Return None if screenshot fails (e.g., page already closed) # Return None if screenshot fails (e.g., driver session closed)
return None return None
async def capture_html(page: Page) -> str | None: async def capture_html(driver: WebDriver) -> str | None:
"""Capture page HTML content. """Capture page HTML content.
Args: Args:
page: The Playwright page instance driver: The Selenium WebDriver instance
Returns: Returns:
HTML page content, or None if capture fails HTML page content, or None if capture fails
""" """
try: try:
return await page.content() return driver.page_source
except Exception: 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 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. """Capture console messages logged during page lifecycle.
Note: This captures messages that were emitted during the page's Note: Selenium captures browser console logs if logging capabilities
lifetime. To get comprehensive logs, attach listeners early in are enabled. Logs may not be available for all browser types or
the page lifecycle. configurations.
Args: Args:
_page: The Playwright page instance driver: The Selenium WebDriver instance
Returns: Returns:
List of console message strings (empty if none captured) List of console message strings (empty if none captured)
""" """
# Playwright doesn't provide direct access to historical console logs, try:
# but we can provide a hook for attaching a console listener at page creation. # Attempt to retrieve browser logs (requires 'goog:loggingPrefs' capability)
# For now, return empty list (see browser/client.py for listener attachment). 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 [] return []
async def capture_all_diagnostics(page: Page) -> DebugInfo: async def capture_all_diagnostics(driver: WebDriver) -> DebugInfo:
"""Capture all diagnostic information (screenshot, HTML, logs). """Capture all diagnostic information (screenshot, HTML, logs).
Attempts to capture screenshot, HTML, and console logs. If any 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: Args:
page: The Playwright page instance driver: The Selenium WebDriver instance
Returns: Returns:
DebugInfo with all captured diagnostic data DebugInfo with all captured diagnostic data
""" """
screenshot = await capture_screenshot(page) screenshot = await capture_screenshot(driver)
html = await capture_html(page) html = await capture_html(driver)
logs = await capture_console_logs(page) logs = await capture_console_logs(driver)
return DebugInfo( return DebugInfo(
screenshot_base64=screenshot, 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: Architecture:
- BrowserPool: Manages the lifecycle of browser instances by host - BrowserPool: Manages the lifecycle of browser instances by host
- Per host: Single persistent browser connection - Per host: Single persistent WebDriver connection
- Per action: Fresh BrowserContext for complete isolation - Selenium-based: Uses debugger_address for CDP, standard launch for headless
- No page/context pooling: Each action gets a clean slate
""" """
import contextlib import contextlib
import logging import logging
from playwright.async_api import ( from selenium import webdriver
Browser, from selenium.webdriver.chrome.options import Options as ChromeOptions
BrowserContext, from selenium.webdriver.chrome.service import Service as ChromeService
Page, from selenium.webdriver.firefox.options import Options as FirefoxOptions
Playwright, from selenium.webdriver.firefox.service import Service as FirefoxService
async_playwright, from selenium.webdriver.remote.webdriver import WebDriver
)
from guide.app.core.config import AppSettings, BrowserHostConfig, HostKind from guide.app.core.config import AppSettings, BrowserHostConfig, HostKind
from guide.app import errors from guide.app import errors
@@ -31,57 +29,50 @@ _logger = logging.getLogger(__name__)
class BrowserInstance: class BrowserInstance:
"""Manages a single browser connection and its lifecycle. """Manages a single browser connection and its lifecycle.
Creates fresh contexts for each request to ensure complete isolation For CDP mode: Connects via debugger_address (no page refresh).
between actions. No context pooling or reuse. For headless mode: Launches browser with Selenium WebDriver.
""" """
def __init__( def __init__(
self, host_id: str, host_config: BrowserHostConfig, browser: Browser self, host_id: str, host_config: BrowserHostConfig, driver: WebDriver
) -> None: ) -> None:
"""Initialize a browser instance for a host. """Initialize a browser instance for a host.
Args: Args:
host_id: The host identifier host_id: The host identifier
host_config: The host configuration host_config: The host configuration
browser: The Playwright browser instance driver: The Selenium WebDriver instance
""" """
self.host_id: str = host_id self.host_id: str = host_id
self.host_config: BrowserHostConfig = host_config self.host_config: BrowserHostConfig = host_config
self.browser: Browser = browser self.driver: WebDriver = driver
async def allocate_context_and_page(self) -> tuple[BrowserContext, Page]: def get_driver(self) -> WebDriver:
"""Allocate a fresh context and page for this request. """Get the WebDriver instance for this browser.
Both CDP and headless modes create new contexts for complete isolation.
Returns: Returns:
Tuple of (context, page) - caller must close context when done WebDriver instance
Raises: Raises:
BrowserConnectionError: If context/page creation fails BrowserConnectionError: If driver is not available
""" """
try: if not self.driver:
context = await self.browser.new_context()
page = await context.new_page()
return context, page
except Exception as exc:
raise errors.BrowserConnectionError( raise errors.BrowserConnectionError(
f"Failed to allocate page for host {self.host_id}", f"Driver not available for host {self.host_id}",
details={"host_id": self.host_id, "host_kind": self.host_config.kind}, details={"host_id": self.host_id},
) from exc )
return self.driver
async def close(self) -> None: def close(self) -> None:
"""Close the browser connection.""" """Close the browser connection."""
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
await self.browser.close() self.driver.quit()
class BrowserPool: class BrowserPool:
"""Manages browser instances across multiple hosts. """Manages browser instances across multiple hosts.
Maintains one persistent browser connection per host. Browser connections are Maintains one persistent WebDriver connection per host using Selenium.
reused, but contexts are created fresh for each request to ensure complete
isolation between actions.
""" """
def __init__(self, settings: AppSettings) -> None: def __init__(self, settings: AppSettings) -> None:
@@ -92,41 +83,31 @@ class BrowserPool:
""" """
self.settings: AppSettings = settings self.settings: AppSettings = settings
self._instances: dict[str, BrowserInstance] = {} self._instances: dict[str, BrowserInstance] = {}
self._playwright: Playwright | None = None
self._closed: bool = False self._closed: bool = False
async def initialize(self) -> None: def initialize(self) -> None:
"""Initialize the browser pool. """Initialize the browser pool.
Starts the Playwright instance. Browser connections are created lazily Uses lazy initialization - connections are created on first use.
on first request to avoid startup delays.
""" """
if self._playwright is not None: # No eager connections - instances created lazily on first request
return _logger.info("Browser pool initialized (lazy mode)")
self._playwright = await async_playwright().start()
_logger.info("Browser pool initialized")
async def close(self) -> None: def close(self) -> None:
"""Close all browser connections and the Playwright instance.""" """Close all browser connections."""
if self._closed: if self._closed:
return return
self._closed = True self._closed = True
for instance in self._instances.values(): for instance in self._instances.values():
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
await instance.close() instance.close()
self._instances.clear() self._instances.clear()
if self._playwright:
with contextlib.suppress(Exception):
await self._playwright.stop()
self._playwright = None
_logger.info("Browser pool closed") _logger.info("Browser pool closed")
async def allocate_context_and_page( def get_driver(self, host_id: str | None = None) -> WebDriver:
self, host_id: str | None = None """Get the WebDriver for the specified host.
) -> tuple[BrowserContext, Page]:
"""Allocate a fresh context and page for the specified host.
Lazily creates browser connections on first request per 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 host_id: The host identifier, or None for the default host
Returns: Returns:
Tuple of (context, page) - caller must close context when done WebDriver instance
Raises: Raises:
ConfigError: If the host_id is invalid or not configured ConfigError: If the host_id is invalid or not configured
BrowserConnectionError: If the browser connection fails 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 resolved_id = host_id or self.settings.default_browser_host_id
host_config = self.settings.browser_hosts.get(resolved_id) host_config = self.settings.browser_hosts.get(resolved_id)
if not host_config: if not host_config:
@@ -155,75 +131,75 @@ class BrowserPool:
# Get or create the browser instance for this host # Get or create the browser instance for this host
if resolved_id not in self._instances: 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 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 self, host_id: str, host_config: BrowserHostConfig
) -> BrowserInstance: ) -> BrowserInstance:
"""Create a new browser instance for the given host.""" """Create a new browser instance for the given host."""
assert self._playwright is not None
if host_config.kind == HostKind.CDP: if host_config.kind == HostKind.CDP:
browser = await self._connect_cdp(host_config) driver = self._connect_cdp(host_config)
else: 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( _logger.info(
f"Created browser instance for host '{host_id}' ({host_config.kind})" f"Created browser instance for host '{host_id}' ({host_config.kind})"
) )
return instance return instance
async def _connect_cdp(self, host_config: BrowserHostConfig) -> Browser: def _connect_cdp(self, host_config: BrowserHostConfig) -> WebDriver:
"""Connect to a CDP host.""" """Connect to a CDP host using debugger_address.
assert self._playwright is not None
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: if not host_config.host or host_config.port is None:
raise errors.ConfigError("CDP host requires 'host' and 'port' fields.") 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: try:
browser = await self._playwright.chromium.connect_over_cdp(cdp_url) chrome_options = ChromeOptions()
_logger.info(f"Connected to CDP endpoint: {cdp_url}") chrome_options.debugger_address = debugger_address
return browser
# 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: except Exception as exc:
raise errors.BrowserConnectionError( 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}, details={"host": host_config.host, "port": host_config.port},
) from exc ) from exc
async def _launch_headless(self, host_config: BrowserHostConfig) -> Browser: def _launch_headless(self, host_config: BrowserHostConfig) -> WebDriver:
"""Launch a headless browser.""" """Launch a headless browser using Selenium WebDriver."""
assert self._playwright is not None browser_name = (host_config.browser or "chromium").lower()
browser_type = self._resolve_browser_type(host_config.browser)
try: try:
browser = await browser_type.launch(headless=True) if browser_name in ("chromium", "chrome"):
_logger.info( chrome_options = ChromeOptions()
f"Launched headless browser: {host_config.browser or 'chromium'}" chrome_options.add_argument("--headless=new")
) chrome_options.add_argument("--no-sandbox")
return browser 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: except Exception as exc:
raise errors.BrowserConnectionError( raise errors.BrowserConnectionError(
f"Cannot launch headless browser: {host_config.browser}", f"Cannot launch headless browser: {browser_name}",
details={"browser_type": host_config.browser}, details={"browser_type": browser_name},
) from exc ) 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"] __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, Provides utilities for waiting on page conditions and verifying visual
and verifying visual stability before proceeding with actions. 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 import errors
from guide.app.utils.retry import retry_async from guide.app.utils.retry import retry_async
async def wait_for_selector( async def wait_for_selector(
page: Page, driver: WebDriver,
selector: str, selector: str,
timeout_ms: int = 5000, timeout_ms: int = 5000,
) -> None: ) -> None:
"""Wait for an element matching selector to be present in DOM. """Wait for an element matching selector to be present in DOM.
Args: Args:
page: The Playwright page instance driver: The Selenium WebDriver instance
selector: CSS or Playwright selector string selector: CSS selector string
timeout_ms: Maximum time to wait in milliseconds (default: 5000) timeout_ms: Maximum time to wait in milliseconds (default: 5000)
Raises: Raises:
GuideError: If selector not found within timeout GuideError: If selector not found within timeout
""" """
try: try:
_ = await page.wait_for_selector(selector, timeout=timeout_ms) wait = WebDriverWait(driver, timeout_ms / 1000)
except PlaywrightTimeoutError as exc: 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" msg = f"Selector '{selector}' not found within {timeout_ms}ms"
raise errors.GuideError(msg) from exc raise errors.GuideError(msg) from exc
async def wait_for_navigation( async def wait_for_navigation(
page: Page, driver: WebDriver,
timeout_ms: int = 5000, timeout_ms: int = 5000,
) -> None: ) -> None:
"""Wait for page navigation to complete. """Wait for page navigation to complete.
Args: Args:
page: The Playwright page instance driver: The Selenium WebDriver instance
timeout_ms: Maximum time to wait in milliseconds (default: 5000) timeout_ms: Maximum time to wait in milliseconds (default: 5000)
Raises: Raises:
GuideError: If navigation does not complete within timeout GuideError: If navigation does not complete within timeout
""" """
try: try:
await page.wait_for_load_state("networkidle", timeout=timeout_ms) # Wait for document.readyState to be 'complete'
except PlaywrightTimeoutError as exc: 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" msg = f"Page navigation did not complete within {timeout_ms}ms"
raise errors.GuideError(msg) from exc raise errors.GuideError(msg) from exc
async def wait_for_network_idle( async def wait_for_network_idle(
page: Page, driver: WebDriver,
timeout_ms: int = 5000, timeout_ms: int = 5000,
) -> None: ) -> None:
"""Wait for network to become idle (no active requests). """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: Args:
page: The Playwright page instance driver: The Selenium WebDriver instance
timeout_ms: Maximum time to wait in milliseconds (default: 5000) timeout_ms: Maximum time to wait in milliseconds (default: 5000)
Raises: Raises:
GuideError: If network does not idle within timeout GuideError: If network does not idle within timeout
""" """
try: try:
await page.wait_for_load_state("networkidle", timeout=timeout_ms) wait = WebDriverWait(driver, timeout_ms / 1000)
except PlaywrightTimeoutError as exc: # 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" msg = f"Network did not idle within {timeout_ms}ms"
raise errors.GuideError(msg) from exc raise errors.GuideError(msg) from exc
async def is_page_stable( async def is_page_stable(
page: Page, driver: WebDriver,
stability_check_ms: int = 500, stability_check_ms: int = 500,
samples: int = 3, samples: int = 3,
) -> bool: ) -> bool:
@@ -85,7 +103,7 @@ async def is_page_stable(
indicating visual stability for reliable element interaction. indicating visual stability for reliable element interaction.
Args: Args:
page: The Playwright page instance driver: The Selenium WebDriver instance
stability_check_ms: Delay between samples in milliseconds (default: 500) stability_check_ms: Delay between samples in milliseconds (default: 500)
samples: Number of stable samples required (default: 3) samples: Number of stable samples required (default: 3)
@@ -96,13 +114,13 @@ async def is_page_stable(
previous_content: str | None = None previous_content: str | None = None
for _ in range(samples): for _ in range(samples):
current_content = await page.content() current_content = driver.page_source
if previous_content and current_content != previous_content: if previous_content and current_content != previous_content:
return False return False
previous_content = current_content previous_content = current_content
await asyncio.sleep(stability_check_ms / 1000) time.sleep(stability_check_ms / 1000)
return True return True
except Exception: except Exception:
@@ -112,7 +130,7 @@ async def is_page_stable(
@retry_async(retries=3, delay_seconds=0.2) @retry_async(retries=3, delay_seconds=0.2)
async def wait_for_stable_page( async def wait_for_stable_page(
page: Page, driver: WebDriver,
stability_check_ms: int = 500, stability_check_ms: int = 500,
samples: int = 3, samples: int = 3,
) -> None: ) -> None:
@@ -122,14 +140,14 @@ async def wait_for_stable_page(
Useful for SPAs or pages with animations/transitions. Useful for SPAs or pages with animations/transitions.
Args: Args:
page: The Playwright page instance driver: The Selenium WebDriver instance
stability_check_ms: Delay between samples in milliseconds (default: 500) stability_check_ms: Delay between samples in milliseconds (default: 500)
samples: Number of stable samples required (default: 3) samples: Number of stable samples required (default: 3)
Raises: Raises:
GuideError: If page does not stabilize after retries 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: if not stable:
msg = "Page did not stabilize after retries" msg = "Page did not stabilize after retries"
raise errors.GuideError(msg) raise errors.GuideError(msg)

View File

@@ -4,13 +4,22 @@ import os
from enum import Enum from enum import Enum
from pathlib import Path from pathlib import Path
from collections.abc import Mapping 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 import BaseModel, Field
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, SettingsConfigDict
from guide.app.models.personas.models import DemoPersona 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__) _logger = logging.getLogger(__name__)
CONFIG_DIR = Path(__file__).resolve().parents[4] / "config" CONFIG_DIR = Path(__file__).resolve().parents[4] / "config"
@@ -59,8 +68,8 @@ class AppSettings(BaseSettings):
case_sensitive=False, case_sensitive=False,
) )
raindrop_base_url: str = "https://app.raindrop.com" raindrop_base_url: str = "https://stg.raindrop.com"
raindrop_graphql_url: str = "https://app.raindrop.com/graphql" raindrop_graphql_url: str = "https://raindrop-staging.hasura.app/v1/graphql"
default_browser_host_id: str = "demo-cdp" default_browser_host_id: str = "demo-cdp"
browser_hosts: dict[str, BrowserHostConfig] = Field(default_factory=dict) browser_hosts: dict[str, BrowserHostConfig] = Field(default_factory=dict)
personas: dict[str, DemoPersona] = 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 # Load JSON overrides from environment
if browser_hosts_json := os.environ.get("RAINDROP_DEMO_BROWSER_HOSTS_JSON"): if browser_hosts_json := os.environ.get("RAINDROP_DEMO_BROWSER_HOSTS_JSON"):
try: try:
# Validate JSON is a list and process each record _settings_extraction_from_json(
decoded = cast(object, json.loads(browser_hosts_json)) browser_hosts_json,
if not isinstance(decoded, list): "RAINDROP_DEMO_BROWSER_HOSTS_JSON must be a JSON array",
raise ValueError( BrowserHostConfig,
"RAINDROP_DEMO_BROWSER_HOSTS_JSON must be a JSON array" hosts_dict,
) )
# 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
except (json.JSONDecodeError, ValueError) as exc: except (json.JSONDecodeError, ValueError) as exc:
_logger.warning(f"Failed to parse RAINDROP_DEMO_BROWSER_HOSTS_JSON: {exc}") _logger.warning(f"Failed to parse RAINDROP_DEMO_BROWSER_HOSTS_JSON: {exc}")
if personas_json := os.environ.get("RAINDROP_DEMO_PERSONAS_JSON"): if personas_json := os.environ.get("RAINDROP_DEMO_PERSONAS_JSON"):
try: try:
# Validate JSON is a list and process each record _settings_extraction_from_json(
decoded = cast(object, json.loads(personas_json)) personas_json,
if not isinstance(decoded, list): "RAINDROP_DEMO_PERSONAS_JSON must be a JSON array",
raise ValueError("RAINDROP_DEMO_PERSONAS_JSON must be a JSON array") DemoPersona,
# Iterate only over validated list personas_dict,
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
except (json.JSONDecodeError, ValueError) as exc: except (json.JSONDecodeError, ValueError) as exc:
_logger.warning(f"Failed to parse RAINDROP_DEMO_PERSONAS_JSON: {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" f"Loaded {len(settings.browser_hosts)} browser hosts, {len(settings.personas)} personas"
) )
return settings 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): class GuideError(Exception):
code: str = "UNKNOWN_ERROR" code: str = "UNKNOWN_ERROR"
message: str 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) super().__init__(message)
self.message = message self.message = message
self.details = details or {} self.details = details or {}

View File

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

View File

@@ -4,7 +4,6 @@ from typing import Literal
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from guide.app.core.config import HostKind from guide.app.core.config import HostKind
from guide.app.models.types import JSONValue
from guide.app import utils from guide.app import utils
@@ -29,42 +28,52 @@ class DebugInfo(BaseModel):
class ActionRequest(BaseModel): class ActionRequest(BaseModel):
"""Request to execute a demo action."""
persona_id: str | None = None persona_id: str | None = None
browser_host_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): class ActionContext(BaseModel):
"""Action execution context passed to action handlers."""
action_id: str action_id: str
persona_id: str | None = None persona_id: str | None = None
browser_host_id: str 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) 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, default_factory=dict,
description="Shared state for composite actions (multi-step flows)", description="Shared state for composite actions (multi-step flows)",
) )
class ActionResult(BaseModel): class ActionResult(BaseModel):
"""Result of a single action execution."""
status: Literal["ok", "error"] = "ok" 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 error: str | None = None
class ActionMetadata(BaseModel): class ActionMetadata(BaseModel):
"""Metadata about an action for discovery."""
id: str id: str
description: str description: str
category: str category: str
class ActionResponse(BaseModel): class ActionResponse(BaseModel):
"""Response envelope for action execution."""
status: Literal["ok", "error"] status: Literal["ok", "error"]
action_id: str action_id: str
browser_host_id: str browser_host_id: str
persona_id: str | None = None persona_id: str | None = None
correlation_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 error: str | None = None
@@ -74,13 +83,15 @@ class ActionStatus(str, Enum):
class ActionEnvelope(BaseModel): class ActionEnvelope(BaseModel):
"""Complete envelope with diagnostics for API responses."""
status: ActionStatus status: ActionStatus
action_id: str action_id: str
correlation_id: str correlation_id: str
result: dict[str, JSONValue] | None = None result: dict[str, object] | None = None
error_code: str | None = None error_code: str | None = None
message: str | None = None message: str | None = None
details: dict[str, JSONValue] | None = None details: dict[str, object] | None = None
debug_info: DebugInfo | None = None debug_info: DebugInfo | None = None
"""Diagnostic data captured on action failure (screenshots, HTML, logs)""" """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 import errors
from guide.app.core.config import AppSettings from guide.app.core.config import AppSettings
from guide.app.models.personas.models import DemoPersona from guide.app.models.personas.models import DemoPersona
from guide.app.models.types import JSONValue
class GraphQLClient: class GraphQLClient:
@@ -17,10 +16,10 @@ class GraphQLClient:
self, self,
*, *,
query: str, query: str,
variables: Mapping[str, JSONValue] | None, variables: Mapping[str, object] | None,
persona: DemoPersona | None, persona: DemoPersona | None,
operation_name: str | None = None, operation_name: str | None = None,
) -> dict[str, JSONValue]: ) -> dict[str, object]:
url = self._settings.raindrop_graphql_url url = self._settings.raindrop_graphql_url
headers = self._build_headers(persona) headers = self._build_headers(persona)
async with httpx.AsyncClient(timeout=10.0) as client: async with httpx.AsyncClient(timeout=10.0) as client:
@@ -47,12 +46,12 @@ class GraphQLClient:
data = cast(dict[str, object], resp.json()) data = cast(dict[str, object], resp.json())
if errors_list := data.get("errors"): 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( raise errors.GraphQLOperationError(
"GraphQL operation failed", details=details "GraphQL operation failed", details=details
) )
payload = data.get("data", {}) 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]: def _build_headers(self, persona: DemoPersona | None) -> dict[str, str]:
headers: dict[str, str] = {"Content-Type": "application/json"} 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." "Requesting 500 tons of replacement conveyor belts for Q4 maintenance window."
) )
ALT_REQUEST: ClassVar[str] = "Intake for rapid supplier onboarding and risk review." 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.intake import IntakeLabels
from guide.app.strings.labels.sourcing import SourcingLabels from guide.app.strings.labels.sourcing import SourcingLabels
from guide.app.strings.selectors.auth import AuthSelectors 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.intake import IntakeSelectors
from guide.app.strings.selectors.navigation import NavigationSelectors from guide.app.strings.selectors.navigation import NavigationSelectors
from guide.app.strings.selectors.sourcing import SourcingSelectors from guide.app.strings.selectors.sourcing import SourcingSelectors
@@ -40,16 +41,54 @@ class IntakeStrings:
Provides direct access to all intake-related UI constants. Provides direct access to all intake-related UI constants.
""" """
# Selectors # Selectors - General
description_field: ClassVar[str] = IntakeSelectors.DESCRIPTION_FIELD description_field: ClassVar[str] = IntakeSelectors.DESCRIPTION_FIELD
next_button: ClassVar[str] = IntakeSelectors.NEXT_BUTTON 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 # Labels
description_placeholder: ClassVar[str] = IntakeLabels.DESCRIPTION_PLACEHOLDER description_placeholder: ClassVar[str] = IntakeLabels.DESCRIPTION_PLACEHOLDER
# Demo text # Demo text
conveyor_belt_request: ClassVar[str] = IntakeTexts.CONVEYOR_BELT_REQUEST conveyor_belt_request: ClassVar[str] = IntakeTexts.CONVEYOR_BELT_REQUEST
alt_request: ClassVar[str] = IntakeTexts.ALT_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: class SourcingStrings:
@@ -105,11 +144,22 @@ class AuthStrings:
current_user_display_prefix: ClassVar[str] = AuthLabels.CURRENT_USER_DISPLAY_PREFIX 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: class AppStrings:
"""Root registry for all application strings. """Root registry for all application strings.
Provides hierarchical, type-safe access to selectors, labels, and demo texts. 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 GraphQL queries are maintained separately in raindrop/generated/queries.py
and loaded from .graphql files for better maintainability. and loaded from .graphql files for better maintainability.
@@ -119,6 +169,7 @@ class AppStrings:
sourcing: ClassVar[type[SourcingStrings]] = SourcingStrings sourcing: ClassVar[type[SourcingStrings]] = SourcingStrings
navigation: ClassVar[type[NavigationStrings]] = NavigationStrings navigation: ClassVar[type[NavigationStrings]] = NavigationStrings
auth: ClassVar[type[AuthStrings]] = AuthStrings auth: ClassVar[type[AuthStrings]] = AuthStrings
common: ClassVar[type[CommonStrings]] = CommonStrings
# Module-level instance for convenience # Module-level instance for convenience
@@ -131,4 +182,5 @@ __all__ = [
"SourcingStrings", "SourcingStrings",
"NavigationStrings", "NavigationStrings",
"AuthStrings", "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: class IntakeSelectors:
# General intake selectors
DESCRIPTION_FIELD: ClassVar[str] = '[data-test="intake-description"]' DESCRIPTION_FIELD: ClassVar[str] = '[data-test="intake-description"]'
NEXT_BUTTON: ClassVar[str] = '[data-test="intake-next"]' 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 import pytest
from unittest.mock import AsyncMock, MagicMock from unittest.mock import AsyncMock, MagicMock
from typing import TYPE_CHECKING
@pytest.fixture @pytest.fixture

View File

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