Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 741059009f | |||
| 1aac3a9a3e | |||
| ff3b9c7edb |
@@ -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:
|
||||
|
||||
@@ -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"})
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
115
src/guide/app/actions/demo/__init__.py
Normal file
115
src/guide/app/actions/demo/__init__.py
Normal 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"]
|
||||
@@ -1,3 +1,3 @@
|
||||
from guide.app.actions.intake.basic import FillIntakeBasicAction
|
||||
from guide.app.actions.intake.sourcing_request import FillIntakeBasicAction
|
||||
|
||||
__all__ = ["FillIntakeBasicAction"]
|
||||
|
||||
@@ -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"})
|
||||
190
src/guide/app/actions/intake/sourcing_request.py
Normal file
190
src/guide/app/actions/intake/sourcing_request.py
Normal 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,
|
||||
},
|
||||
}
|
||||
)
|
||||
@@ -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)})
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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,
|
||||
|
||||
285
src/guide/app/browser/helpers.py
Normal file
285
src/guide/app/browser/helpers.py
Normal 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"]
|
||||
@@ -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"]
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)"""
|
||||
|
||||
|
||||
@@ -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"}
|
||||
|
||||
@@ -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"
|
||||
@@ -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",
|
||||
]
|
||||
|
||||
30
src/guide/app/strings/selectors/common.py
Normal file
30
src/guide/app/strings/selectors/common.py
Normal 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"]
|
||||
@@ -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
37
test_sb_minimal.py
Normal 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)
|
||||
81
test_selenium_debugger_address.py
Normal file
81
test_selenium_debugger_address.py
Normal 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
76
test_seleniumbase_cdp.py
Normal 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())
|
||||
@@ -5,7 +5,6 @@ from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user