From fe5f1839524fc414aac3ebcb1d306d85083ea9bb Mon Sep 17 00:00:00 2001 From: Marc Kelechava Date: Mon, 6 Jul 2026 17:10:12 -0700 Subject: [PATCH] SKY-11527 Collapse custom select fanout (#7143) --- skyvern/webeye/actions/handler.py | 581 +++++++++++++++++ tests/unit/test_no_available_option.py | 854 +++++++++++++++++++++++++ 2 files changed, 1435 insertions(+) diff --git a/skyvern/webeye/actions/handler.py b/skyvern/webeye/actions/handler.py index 1babae2ab..27e452df7 100644 --- a/skyvern/webeye/actions/handler.py +++ b/skyvern/webeye/actions/handler.py @@ -159,6 +159,7 @@ UPLOAD_PENDING_FOLLOWUP_MESSAGE = "Upload is not complete yet. Continue the uplo FIX_TEL_INPUT_DIGIT_DROP_FLAG = "FIX_TEL_INPUT_DIGIT_DROP" COLLAPSE_SELECT_FANOUT_FLAG = "COLLAPSE_SELECT_FANOUT" +COLLAPSE_CUSTOM_SELECT_FANOUT_FLAG = "COLLAPSE_CUSTOM_SELECT_FANOUT" COLLAPSE_AUTOCOMPLETE_FANOUT_FLAG = "COLLAPSE_AUTOCOMPLETE_FANOUT" DOWNLOAD_EVENT_ACTIVE_DIR_GRACE_SECONDS = 60 @@ -1118,6 +1119,30 @@ async def _is_collapse_select_fanout_enabled(task: Task) -> bool: return False +async def _is_collapse_custom_select_fanout_enabled(task: Task) -> bool: + organization_id = task.organization_id + if not organization_id: + return False + experimentation_provider = getattr(app, "EXPERIMENTATION_PROVIDER", None) + if not experimentation_provider: + return False + try: + return bool( + await experimentation_provider.is_feature_enabled_cached( + COLLAPSE_CUSTOM_SELECT_FANOUT_FLAG, + organization_id, + properties={"organization_id": organization_id}, + ) + ) + except Exception: + LOG.warning( + "Failed to evaluate collapse-custom-select-fanout flag; defaulting to disabled", + organization_id=organization_id, + exc_info=True, + ) + return False + + async def _is_collapse_autocomplete_fanout_enabled(task: Task) -> bool: organization_id = task.organization_id if not organization_id: @@ -3706,6 +3731,8 @@ async def handle_select_option_action( assert result is not None assert result.action_result is not None results.append(result.action_result) + if result.action_result.skip_remaining_actions: + return results if isinstance(result.action_result, ActionSuccess) or result.value is None: return results suggested_value = result.value @@ -5749,6 +5776,525 @@ def _collect_option_texts(elements: list[dict]) -> list[str]: return out +def _custom_select_choice_input_ids(node: dict) -> set[str]: + input_ids: set[str] = set() + queue: deque[dict] = deque(node.get("children") or []) + while queue: + child = queue.popleft() + if not isinstance(child, dict): + continue + tag = str(child.get("tagName") or "").lower() + attrs = child.get("attributes") or {} + input_type = str(attrs.get("type") or "").lower() + element_id = str(child.get("id") or "") + if tag == "input" and input_type in ("checkbox", "radio") and element_id: + input_ids.add(element_id) + for grandchild in child.get("children") or []: + queue.append(grandchild) + return input_ids + + +def _custom_select_choice_value(node: dict) -> str | None: + attrs = node.get("attributes") or {} + value = " ".join(str(attrs.get("value") or "").split()) + if value: + return value + queue: deque[dict] = deque(node.get("children") or []) + while queue: + child = queue.popleft() + if not isinstance(child, dict): + continue + child_attrs = child.get("attributes") or {} + child_value = " ".join(str(child_attrs.get("value") or "").split()) + if child_value: + return child_value + for grandchild in child.get("children") or []: + queue.append(grandchild) + return None + + +_CUSTOM_SELECT_CONTAINER_ROLES = frozenset({"combobox", "listbox", "menu", "radiogroup", "tree"}) +_CUSTOM_SELECT_CHOICE_ROLES = frozenset({"menuitem", "menuitemcheckbox", "menuitemradio", "option", "treeitem"}) + + +def _is_custom_select_choice_surface(role: str) -> bool: + return role in {"listbox", "menu", "radiogroup", "tree"} + + +def _custom_select_candidates_from_elements(elements: list[dict]) -> list[dict[str, str | None]]: + queue: deque[tuple[dict, bool]] = deque((element, False) for element in elements) + candidates: list[dict[str, str | None]] = [] + seen: set[tuple[str | None, str | None, str | None]] = set() + covered_choice_input_ids: set[str] = set() + + while queue: + node, in_choice_surface = queue.popleft() + if not isinstance(node, dict): + continue + + attrs = node.get("attributes") or {} + role = str(attrs.get("role") or "").lower() + tag = str(node.get("tagName") or "").lower() + input_type = str(attrs.get("type") or "").lower() + element_id = str(node.get("id") or "") or None + value = _custom_select_choice_value(node) + label = _select_shadow_label_from_node(node) or value + choice_input_ids = _custom_select_choice_input_ids(node) + is_choice_input = tag == "input" and input_type in ("checkbox", "radio") + is_option_node = role in _CUSTOM_SELECT_CHOICE_ROLES or tag == "option" or (tag == "li" and in_choice_surface) + is_label_choice = tag == "label" and bool(choice_input_ids) + has_choice_state = "aria-selected" in attrs or "aria-checked" in attrs + is_clickable_choice = ( + bool(node.get("interactable")) + and label + and ( + role not in _CUSTOM_SELECT_CONTAINER_ROLES + and tag not in {"input", "select", "textarea"} + and not (tag == "a" and bool(attrs.get("href"))) + and (in_choice_surface or has_choice_state) + ) + ) + + if is_choice_input and element_id in covered_choice_input_ids: + pass + elif element_id and label and (is_option_node or is_choice_input or is_label_choice or is_clickable_choice): + candidate = _select_shadow_candidate(label, element_id=element_id, value=value) + if candidate is not None: + key = (candidate.get("element_id"), candidate.get("label"), candidate.get("value")) + if key not in seen: + seen.add(key) + candidates.append(candidate) + if is_label_choice: + covered_choice_input_ids.update(choice_input_ids) + + child_in_choice_surface = in_choice_surface or _is_custom_select_choice_surface(role) + for child in node.get("children") or []: + queue.append((child, child_in_choice_surface)) + + return candidates + + +def _split_selected_label_values(value: str) -> set[str]: + normalized = _normalize_select_shadow_text(value) + if not normalized: + return set() + parts = {part.strip() for part in normalized.split(",") if part.strip()} | {normalized} + # Widgets that reflect a committed pick by relabelling a trigger/wrapper (e.g. aria-label + # flips from "Search sources" to "Selected WAKE") should still match the bare option label. + for prefix in _SELECTED_LABEL_PREFIXES: + if normalized.startswith(prefix): + parts.add(normalized[len(prefix) :].strip()) + return {part for part in parts if part} + + +_SELECTED_LABEL_PREFIXES = ("selected ", "selected:") + +_CUSTOM_SELECT_MATCHED_STATE_JS = r""" +(el) => { + const normalize = (value) => (value ?? "").replace(/\s+/g, " ").trim().toLowerCase(); + const label = [ + el.textContent, + el.getAttribute("aria-label"), + el.getAttribute("title"), + el.getAttribute("value"), + el.value + ].map(normalize).find(Boolean) || ""; + const role = normalize(el.getAttribute("role")); + const nestedChoice = el.querySelector?.("input[type='checkbox'], input[type='radio']"); + const ariaSelected = el.getAttribute("aria-selected") === "true"; + const ariaChecked = el.getAttribute("aria-checked") === "true"; + const selectedAttr = el.hasAttribute("selected") || el.selected === true; + const checked = Boolean( + (el.matches?.("input[type='checkbox'], input[type='radio']") && el.checked) + || nestedChoice?.checked + ); + return { label, role, ariaSelected, ariaChecked, selectedAttr, checked }; +} +""" + +_CUSTOM_SELECT_COMMITTED_STATE_JS = r""" +([anchor, args]) => { + const expectedLabel = args.expectedLabel; + const anchorIsComboboxInput = args.anchorIsComboboxInput; + const allowAriaSelectedOptionTokens = args.allowAriaSelectedOptionTokens !== false; + const normalize = (value) => (value ?? "").replace(/\s+/g, " ").trim().toLowerCase(); + const splitValues = (value) => { + const normalized = normalize(value); + if (!normalized) return []; + const parts = [normalized, ...normalized.split(",").map((part) => part.trim()).filter(Boolean)]; + for (const prefix of ["selected ", "selected:"]) { + if (normalized.startsWith(prefix)) parts.push(normalized.slice(prefix.length).trim()); + } + return parts.filter(Boolean); + }; + const matchesExpected = (value) => splitValues(value).includes(expectedLabel); + const triggerSelector = [ + "[role='combobox']", + "[aria-haspopup='listbox']", + "[aria-haspopup='menu']", + "[aria-haspopup='true']", + "button[aria-expanded]", + "input[role='combobox']", + "select" + ].join(","); + const scopeSelectors = [ + "[data-uxi-widget-type]", + "[data-automation-id*='formField']", + "[role='group']", + "fieldset", + ".field" + ]; + // Nearest matching ancestor wins; never scope to the whole form or a bare + // parent container — sibling fields showing the target label must not + // pre-confirm this one. With no recognized field wrapper, fall back to the + // anchor itself (a miss routes to the LLM path, never a cross-field match). + const scopeCandidates = scopeSelectors + .map((selector) => anchor.closest?.(selector)) + .filter(Boolean); + const scopeRoot = ( + scopeCandidates.reduce((closest, el) => (!closest || closest.contains(el) ? el : closest), null) + || anchor + ); + const tokenSelectors = [ + ...(allowAriaSelectedOptionTokens ? ["[role='option'][aria-selected='true']"] : []), + "[data-automation-id='selectedItem']", + ".pill", + ".chip", + "[class*='token']" + ].join(","); + for (const token of scopeRoot.querySelectorAll(tokenSelectors)) { + if (matchesExpected(token.textContent) || matchesExpected(token.getAttribute("aria-label"))) return true; + } + for (const hidden of scopeRoot.querySelectorAll("input[type='hidden']")) { + if (matchesExpected(hidden.value)) return true; + } + const activeId = anchor.getAttribute?.("aria-activedescendant"); + if (allowAriaSelectedOptionTokens && activeId) { + const active = scopeRoot.querySelector(`#${CSS.escape(activeId)}`); + if (active && active.getAttribute("aria-selected") === "true") { + if (matchesExpected(active.textContent) || matchesExpected(active.getAttribute("aria-label"))) { + return true; + } + } + } + const reflectedValues = (el) => [ + el.textContent, + el.getAttribute("aria-label"), + el.getAttribute("aria-valuetext"), + el.getAttribute("title"), + ]; + const seen = new Set(); + const triggerCandidates = [ + anchor, + anchor.closest?.(triggerSelector), + ...(scopeRoot.matches?.(triggerSelector) ? [scopeRoot] : []), + ...scopeRoot.querySelectorAll(triggerSelector) + ]; + for (const el of triggerCandidates) { + if (!el || seen.has(el) || !scopeRoot.contains(el)) continue; + seen.add(el); + if (reflectedValues(el).some(matchesExpected)) return true; + } + // A combobox may still hold the user-typed filter text; raw value equality alone is not + // a committed signal. Only trust it when the dropdown has closed (aria-expanded=false). + if (anchorIsComboboxInput) { + const valueMatches = matchesExpected(anchor.value) || matchesExpected(anchor.getAttribute("value")); + if (valueMatches) { + const expanded = anchor.getAttribute("aria-expanded") + || anchor.closest?.("[aria-expanded]")?.getAttribute("aria-expanded"); + if (expanded === "false") return true; + } + return false; + } + for (const el of seen) { + if (reflectedValues(el).some((value) => normalize(value))) return false; + } + return false; +} +""" + + +async def _evaluate_element_scoped( + element: SkyvernElement, + expression: str, + arg: Any | None = None, +) -> Any: + handler = await element.get_element_handler() + payload = handler if arg is None else [handler, arg] + return await SkyvernFrame.evaluate(frame=element.get_frame(), expression=expression, arg=payload) + + +async def _read_custom_select_matched_state(element: SkyvernElement) -> dict | None: + try: + if await element.get_locator().count() != 1: + return None + state = await _evaluate_element_scoped(element, _CUSTOM_SELECT_MATCHED_STATE_JS) + except Exception: + LOG.info( + "Failed to read custom-select matched element state", + exc_info=True, + ) + return None + return state if isinstance(state, dict) else None + + +def _custom_select_matched_state_confirms(state: dict | None, expected_label: str) -> bool: + if not isinstance(state, dict): + return False + label_matches = expected_label in _split_selected_label_values(str(state.get("label") or "")) + return label_matches and any( + bool(state.get(field)) for field in ("ariaSelected", "ariaChecked", "selectedAttr", "checked") + ) + + +def _custom_select_matched_state_confirms_pre_click(state: dict | None, expected_label: str) -> bool: + if not isinstance(state, dict): + return False + label_matches = expected_label in _split_selected_label_values(str(state.get("label") or "")) + if not label_matches: + return False + if any(bool(state.get(field)) for field in ("ariaChecked", "selectedAttr", "checked")): + return True + if str(state.get("role") or "").lower() == "option": + return False + return bool(state.get("ariaSelected")) + + +async def _custom_select_scope_confirms_committed( + *, + readback_scope_element: SkyvernElement | None, + anchor_is_combobox_input: bool, + matched_element_id: str, + matched_label: str | None, + expected_label: str, + allow_aria_selected_option_tokens: bool, +) -> bool: + if readback_scope_element is None: + return False + + try: + committed = await _evaluate_element_scoped( + readback_scope_element, + _CUSTOM_SELECT_COMMITTED_STATE_JS, + { + "expectedLabel": expected_label, + "anchorIsComboboxInput": anchor_is_combobox_input, + "allowAriaSelectedOptionTokens": allow_aria_selected_option_tokens, + }, + ) + except Exception: + LOG.info( + "Failed to read custom-select committed label", + matched_element_id=matched_element_id, + matched_label=matched_label, + exc_info=True, + ) + return False + + return committed is True + + +async def _verify_custom_select_option( + *, + matched_element: SkyvernElement, + readback_scope_element: SkyvernElement | None, + anchor_is_combobox_input: bool, + matched_element_id: str, + matched_label: str | None, +) -> bool: + expected_label = _normalize_select_shadow_text(matched_label) + if not expected_label: + return False + + if _custom_select_matched_state_confirms(await _read_custom_select_matched_state(matched_element), expected_label): + return True + + return await _custom_select_scope_confirms_committed( + readback_scope_element=readback_scope_element, + anchor_is_combobox_input=anchor_is_combobox_input, + matched_element_id=matched_element_id, + matched_label=matched_label, + expected_label=expected_label, + allow_aria_selected_option_tokens=True, + ) + + +async def _resolve_custom_select_readback_scope_element( + *, + get_readback_scope_element: Callable[[], Awaitable[SkyvernElement | None]] | None, + target_value: str, + matched_element_id: str, + matched_label: str | None, +) -> SkyvernElement | None: + if get_readback_scope_element is None: + return None + + try: + return await get_readback_scope_element() + except Exception: + LOG.info( + "Failed to resolve custom-select read-back scope element; continuing with matched-element read-back", + target_value=target_value, + matched_element_id=matched_element_id, + matched_label=matched_label, + exc_info=True, + ) + return None + + +def _readback_scope_element_provider( + element: SkyvernElement, +) -> Callable[[], Awaitable[SkyvernElement | None]]: + async def _provide() -> SkyvernElement | None: + return element + + return _provide + + +async def _anchor_is_combobox_input(element: SkyvernElement | None) -> bool: + if element is None: + return False + try: + return str(element.get_tag_name() or "").lower() == "input" + except Exception: + return False + + +async def _select_deterministic_custom_option( + *, + target_value: str | None, + get_option_candidates: Callable[[], list[dict[str, str | None]]], + field_context: Any, + page: Page, + get_skyvern_element: Callable[[str], Awaitable[SkyvernElement]], + get_readback_scope_element: Callable[[], Awaitable[SkyvernElement | None]] | None = None, + task: Task, +) -> tuple[ActionResult, str | None] | None: + if not target_value: + return None + if isinstance(field_context, dict) and field_context.get("is_date_related") is True: + return None + if not await _is_collapse_custom_select_fanout_enabled(task): + return None + + option_candidates = get_option_candidates() + if not option_candidates: + return None + + option_labels = [candidate.get("label") or "" for candidate in option_candidates] + option_values = [candidate.get("value") for candidate in option_candidates] + resolution = await app.AGENT_FUNCTION.resolve_field_option( + target_value=target_value, + option_labels=option_labels, + option_values=option_values, + field_context=field_context, + url=task.url, + organization_id=task.organization_id, + ) + if resolution.fallback_to_llm or resolution.matched_index is None: + return None + if resolution.matched_index >= len(option_candidates): + return None + + matched_candidate = option_candidates[resolution.matched_index] + element_id = matched_candidate.get("element_id") + matched_label = resolution.matched_label + if not element_id: + return None + + readback_scope_element: SkyvernElement | None = None + anchor_is_combobox_input = False + try: + selected_element = await get_skyvern_element(element_id) + if await selected_element.get_attr("role") == "listbox": + return None + + readback_scope_element = await _resolve_custom_select_readback_scope_element( + get_readback_scope_element=get_readback_scope_element, + target_value=target_value, + matched_element_id=element_id, + matched_label=matched_label, + ) + anchor_is_combobox_input = await _anchor_is_combobox_input(readback_scope_element) + + expected_label = _normalize_select_shadow_text(matched_label) + if expected_label: + matched_state = await _read_custom_select_matched_state(selected_element) + if _custom_select_matched_state_confirms_pre_click(matched_state, expected_label): + return ActionSuccess(), matched_label + if await _custom_select_scope_confirms_committed( + readback_scope_element=readback_scope_element, + anchor_is_combobox_input=anchor_is_combobox_input, + matched_element_id=element_id, + matched_label=matched_label, + expected_label=expected_label, + allow_aria_selected_option_tokens=False, + ): + return ActionSuccess(), matched_label + + await selected_element.scroll_into_view() + await selected_element.click(page=page) + verified = await _verify_custom_select_option( + matched_element=selected_element, + readback_scope_element=readback_scope_element, + anchor_is_combobox_input=anchor_is_combobox_input, + matched_element_id=element_id, + matched_label=matched_label, + ) + if verified: + return ActionSuccess(), matched_label + except Exception: + LOG.info( + "Deterministic custom-select failed; falling back to LLM path", + target_value=target_value, + matched_element_id=element_id, + matched_label=matched_label, + exc_info=True, + ) + return None + + if anchor_is_combobox_input: + # Text-input comboboxes can be safely reset, so an unconfirmed read-back routes to the LLM + # mini-agent (which clears/reopens the field) instead of hard-failing the whole action. + await _reset_custom_select_combobox_input(readback_scope_element, page) + LOG.info( + "Deterministic custom-select read-back inconclusive on combobox input; routing to LLM fallback", + target_value=target_value, + matched_element_id=element_id, + matched_label=matched_label, + ) + return None + + LOG.info( + "Deterministic custom-select read-back failed after click; returning failure to avoid replaying over mutated widget", + target_value=target_value, + matched_element_id=element_id, + matched_label=matched_label, + ) + action_failure = ActionFailure( + NoElementMatchedForTargetOption( + target=target_value, + reason="Deterministic custom-select click could not be verified by matched element read-back", + ) + ) + action_failure.skip_remaining_actions = True + return action_failure, matched_label + + +async def _reset_custom_select_combobox_input(element: SkyvernElement | None, page: Page) -> None: + if element is None: + return + try: + locator = element.get_locator() + await locator.fill("") + await element.click(page=page) + except Exception: + LOG.info( + "Failed to reset custom-select combobox input before LLM fallback", + exc_info=True, + ) + + def _no_match_exception_for_dropdown( *, reasoning: str | None, @@ -5863,6 +6409,22 @@ async def select_from_emerging_elements( html_length=len(incremental_html), ) + async def get_readback_scope_element() -> SkyvernElement | None: + return await dom_after_open.get_skyvern_element_by_id(current_element_id) + + deterministic_result = await _select_deterministic_custom_option( + target_value=options.target_value, + get_option_candidates=lambda: _custom_select_candidates_from_elements(shadow_candidate_elements), + field_context=options.model_dump(), + page=page, + get_skyvern_element=dom_after_open.get_skyvern_element_by_id, + get_readback_scope_element=get_readback_scope_element, + task=task, + ) + if deterministic_result is not None: + action_result, _matched_label = deterministic_result + return action_result + prompt = prompt_engine.load_prompt( "custom-select", is_date_related=options.is_date_related, @@ -6015,6 +6577,25 @@ async def select_from_dropdown( incremental_scraped.set_element_tree_trimmed(trimmed_element_tree) html = incremental_scraped.build_element_tree(html_need_skyvern_attrs=True) + deterministic_result = await _select_deterministic_custom_option( + target_value=target_value, + get_option_candidates=lambda: _custom_select_candidates_from_elements(trimmed_element_tree), + field_context=context.model_dump(), + page=page, + get_skyvern_element=lambda element_id: SkyvernElement.create_from_incremental(incremental_scraped, element_id), + get_readback_scope_element=_readback_scope_element_provider(skyvern_element), + task=task, + ) + if deterministic_result is not None: + action_result, matched_label = deterministic_result + single_select_result.reasoning = "Deterministic exact/stem custom-select match" + single_select_result.value = matched_label or target_value + single_select_result.action_type = ActionType.CLICK + single_select_result.action_result = action_result + if isinstance(action_result, ActionSuccess): + single_select_result.dropdown_menu = None + return single_select_result + skyvern_context = ensure_context() prompt = prompt_engine.load_prompt( "custom-select", diff --git a/tests/unit/test_no_available_option.py b/tests/unit/test_no_available_option.py index d391a2606..b3433bf9d 100644 --- a/tests/unit/test_no_available_option.py +++ b/tests/unit/test_no_available_option.py @@ -2,16 +2,118 @@ from __future__ import annotations +from datetime import UTC, datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + import pytest from skyvern.exceptions import ( NoAvailableOptionFoundForCustomSelection, NoIncrementalElementFoundForCustomSelection, ) +from skyvern.forge.agent_functions import AgentFunction +from skyvern.webeye.actions import handler +from skyvern.webeye.actions.actions import InputOrSelectContext, SelectOption, SelectOptionAction from skyvern.webeye.actions.handler import ( _collect_option_texts, + _custom_select_candidates_from_elements, _no_match_exception_for_dropdown, + _select_deterministic_custom_option, + _verify_custom_select_option, ) +from tests.unit.helpers import make_organization, make_task + + +class _FakeCustomElement: + def __init__(self, *, role: str | None = None, tag_name: str = "div") -> None: + self.get_attr = AsyncMock(return_value=role) + self.get_tag_name = MagicMock(return_value=tag_name) + self.get_frame = MagicMock(return_value=MagicMock()) + self.get_element_handler = AsyncMock(return_value=MagicMock()) + self.scroll_into_view = AsyncMock() + self.click = AsyncMock() + self._locator = MagicMock() + + def get_locator(self) -> MagicMock: + return self._locator + + +class _FakeAnchorElement: + def __init__(self, *, tag_name: str = "button") -> None: + self._locator = MagicMock() + self._locator.evaluate = AsyncMock(return_value=False) + self._locator.fill = AsyncMock() + self.get_attr = AsyncMock(return_value=None) + self.get_tag_name = MagicMock(return_value=tag_name) + self.get_id = MagicMock(return_value="field-control") + self.get_frame = MagicMock(return_value=MagicMock()) + self.get_element_handler = AsyncMock(return_value=MagicMock()) + self.is_custom_option = AsyncMock(return_value=False) + self.is_selectable = AsyncMock(return_value=True) + self.is_disabled = AsyncMock(return_value=False) + self.is_checkbox = AsyncMock(return_value=False) + self.is_radio = AsyncMock(return_value=False) + self.is_btn_input = AsyncMock(return_value=False) + self.is_visible = AsyncMock(return_value=False) + self.scroll_into_view = AsyncMock() + self.click = AsyncMock() + self.coordinate_click = AsyncMock() + self.press_key = AsyncMock() + self.blur = AsyncMock() + + def get_locator(self) -> MagicMock: + return self._locator + + +def _stub_evaluate( + *, + matched_state: dict | list[dict | None] | None = None, + committed: bool | None = None, +) -> AsyncMock: + """Stub handler.SkyvernFrame.evaluate, dispatching by which PR JS body is being run. + + matched_state may be a single value (repeated) or a list consumed one entry per call, so a + test can distinguish the pre-click idempotence read from the post-click verification read. + """ + matched_states = list(matched_state) if isinstance(matched_state, list) else None + + async def _evaluate(*, frame: object, expression: str, arg: object = None) -> object: + if "return { label," in expression: + if matched_states is not None: + return matched_states.pop(0) if matched_states else None + return matched_state + if "anchorIsComboboxInput" in expression: + return committed + return None + + return AsyncMock(side_effect=_evaluate) + + +class _FakeIncrementalScrapePage: + def __init__(self, element_trees: list[list[dict]]) -> None: + self._element_trees = list(element_trees) + self.start_listen_dom_increment = AsyncMock() + self.stop_listen_dom_increment = AsyncMock() + self.set_element_tree_trimmed = MagicMock() + self.build_element_tree = MagicMock(return_value="
") + + async def get_incremental_element_tree(self, *_args: object, **_kwargs: object) -> list[dict]: + return self._element_trees.pop(0) + + +def _task() -> object: + now = datetime.now(UTC) + organization = make_organization(now) + return make_task(now, organization) + + +def _select_action(label: str = "Choice") -> SelectOptionAction: + return SelectOptionAction( + element_id="field-control", + option=SelectOption(label=label), + input_or_select_context=InputOrSelectContext(field="Field", is_required=True), + ) class TestCollectOptionTexts: @@ -151,6 +253,758 @@ class TestCollectOptionTexts: assert _collect_option_texts(tree) == ["Alpha", "Bravo"] +class TestCustomSelectCandidates: + def test_extracts_role_option_dedupes_checkbox_and_skips_listbox_container(self) -> None: + tree = [ + { + "id": "source-panel", + "tagName": "div", + "attributes": {"role": "listbox"}, + "text": "Job Board Referral", + "interactable": True, + "children": [ + { + "id": "source-job-board", + "tagName": "div", + "attributes": {"role": "option", "value": "job-board"}, + "text": "Job Board", + "interactable": True, + }, + { + "id": "source-help-link", + "tagName": "a", + "attributes": {"href": "/help"}, + "text": "Job Board", + "interactable": True, + }, + ], + }, + { + "id": "county-panel", + "tagName": "div", + "children": [ + { + "id": "county-a-label", + "tagName": "label", + "text": "County A", + "interactable": True, + "children": [ + { + "id": "county-a-input", + "tagName": "input", + "attributes": {"type": "checkbox", "value": "County A"}, + "interactable": True, + } + ], + } + ], + }, + { + "id": "nav-job-board", + "tagName": "a", + "attributes": {"href": "/jobs"}, + "text": "Job Board", + "interactable": True, + }, + { + "id": "nav-list", + "tagName": "ul", + "children": [ + { + "id": "nav-list-job-board", + "tagName": "li", + "text": "Job Board", + "interactable": True, + } + ], + }, + ] + + assert _custom_select_candidates_from_elements(tree) == [ + {"label": "Job Board", "element_id": "source-job-board", "value": "job-board"}, + {"label": "County A", "element_id": "county-a-label", "value": "County A"}, + ] + + def test_extracts_menuitemradio_with_aria_checked_state(self) -> None: + tree = [ + { + "id": "choice-menu", + "tagName": "div", + "attributes": {"role": "menu"}, + "children": [ + { + "id": "choice-radio", + "tagName": "div", + "attributes": {"role": "menuitemradio", "aria-checked": "false"}, + "text": "Choice", + "interactable": True, + } + ], + } + ] + + assert _custom_select_candidates_from_elements(tree) == [ + {"label": "Choice", "element_id": "choice-radio", "value": None} + ] + + +class TestDeterministicCustomSelect: + @pytest.mark.asyncio + async def test_ambiguous_match_returns_none_for_llm_fallback(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + get_skyvern_element = AsyncMock() + get_readback_scope_element = AsyncMock() + + result = await _select_deterministic_custom_option( + target_value="United States", + get_option_candidates=lambda: [ + {"label": "United States", "element_id": "us-1", "value": "US"}, + {"label": "United States", "element_id": "us-2", "value": "USA"}, + ], + field_context={}, + page=MagicMock(), + get_skyvern_element=get_skyvern_element, + get_readback_scope_element=get_readback_scope_element, + task=_task(), # type: ignore[arg-type] + ) + + assert result is None + get_skyvern_element.assert_not_awaited() + get_readback_scope_element.assert_not_awaited() + + @pytest.mark.asyncio + async def test_feature_flag_off_does_not_resolve_readback_scope(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=False)), + ) + get_skyvern_element = AsyncMock() + get_readback_scope_element = AsyncMock() + + result = await _select_deterministic_custom_option( + target_value="Job Board", + get_option_candidates=lambda: [{"label": "Job Board", "element_id": "source-job-board", "value": None}], + field_context={}, + page=MagicMock(), + get_skyvern_element=get_skyvern_element, + get_readback_scope_element=get_readback_scope_element, + task=_task(), # type: ignore[arg-type] + ) + + assert result is None + get_skyvern_element.assert_not_awaited() + get_readback_scope_element.assert_not_awaited() + + @pytest.mark.asyncio + async def test_date_related_context_skips_deterministic_path(self) -> None: + get_skyvern_element = AsyncMock() + + result = await _select_deterministic_custom_option( + target_value="15", + get_option_candidates=lambda: [{"label": "15", "element_id": "day-15", "value": None}], + field_context={"is_date_related": True}, + page=MagicMock(), + get_skyvern_element=get_skyvern_element, + task=_task(), # type: ignore[arg-type] + ) + + assert result is None + get_skyvern_element.assert_not_awaited() + + @pytest.mark.asyncio + async def test_listbox_container_candidate_returns_none_without_clicking( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + selected_element = _FakeCustomElement(role="listbox") + + result = await _select_deterministic_custom_option( + target_value="Job Board", + get_option_candidates=lambda: [{"label": "Job Board", "element_id": "source-panel", "value": None}], + field_context={}, + page=MagicMock(), + get_skyvern_element=AsyncMock(return_value=selected_element), + task=_task(), # type: ignore[arg-type] + ) + + assert result is None + selected_element.scroll_into_view.assert_not_awaited() + selected_element.click.assert_not_awaited() + + @pytest.mark.asyncio + async def test_aria_checked_menuitemradio_returns_success_without_clicking( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + monkeypatch.setattr( + handler.SkyvernFrame, + "evaluate", + _stub_evaluate( + matched_state={ + "label": "Choice", + "role": "menuitemradio", + "ariaSelected": False, + "ariaChecked": True, + "selectedAttr": False, + "checked": False, + } + ), + ) + selected_element = _FakeCustomElement(role="menuitemradio") + selected_element.get_locator().count = AsyncMock(return_value=1) + + result = await _select_deterministic_custom_option( + target_value="Choice", + get_option_candidates=lambda: [{"label": "Choice", "element_id": "choice-radio", "value": None}], + field_context={}, + page=MagicMock(), + get_skyvern_element=AsyncMock(return_value=selected_element), + task=_task(), # type: ignore[arg-type] + ) + + assert result is not None + action_result, matched_label = result + assert action_result.success + assert matched_label == "Choice" + selected_element.click.assert_not_awaited() + selected_element.scroll_into_view.assert_not_awaited() + + @pytest.mark.asyncio + async def test_already_selected_target_returns_success_without_clicking( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + monkeypatch.setattr( + handler.SkyvernFrame, + "evaluate", + _stub_evaluate( + matched_state={"label": "WAKE", "ariaChecked": False, "selectedAttr": False, "checked": True} + ), + ) + selected_element = _FakeCustomElement() + selected_element.get_locator().count = AsyncMock(return_value=1) + + result = await _select_deterministic_custom_option( + target_value="WAKE", + get_option_candidates=lambda: [{"label": "WAKE", "element_id": "county-wake", "value": "WAKE"}], + field_context={}, + page=MagicMock(), + get_skyvern_element=AsyncMock(return_value=selected_element), + task=_task(), # type: ignore[arg-type] + ) + + assert result is not None + action_result, matched_label = result + assert action_result.success + assert matched_label == "WAKE" + selected_element.click.assert_not_awaited() + selected_element.scroll_into_view.assert_not_awaited() + + @pytest.mark.asyncio + async def test_highlighted_role_option_with_bare_aria_selected_proceeds_to_click( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + matched_states = [ + { + "label": "Job Board", + "role": "option", + "ariaSelected": True, + "ariaChecked": False, + "selectedAttr": False, + "checked": False, + }, + { + "label": "Job Board", + "role": "option", + "ariaSelected": True, + "ariaChecked": False, + "selectedAttr": False, + "checked": False, + }, + ] + scope_args: list[dict[str, object]] = [] + + async def evaluate(*, frame: object, expression: str, arg: object = None) -> object: + if "return { label," in expression: + return matched_states.pop(0) + if "anchorIsComboboxInput" in expression: + assert isinstance(arg, list) + assert isinstance(arg[1], dict) + scope_args.append(arg[1]) + return False + return None + + monkeypatch.setattr(handler.SkyvernFrame, "evaluate", AsyncMock(side_effect=evaluate)) + selected_element = _FakeCustomElement(role="option") + selected_element.get_locator().count = AsyncMock(return_value=1) + readback_scope_element = _FakeAnchorElement() + + result = await _select_deterministic_custom_option( + target_value="Job Board", + get_option_candidates=lambda: [{"label": "Job Board", "element_id": "source-job-board", "value": None}], + field_context={}, + page=MagicMock(), + get_skyvern_element=AsyncMock(return_value=selected_element), + get_readback_scope_element=AsyncMock(return_value=readback_scope_element), + task=_task(), # type: ignore[arg-type] + ) + + assert result is not None + action_result, matched_label = result + assert action_result.success + assert matched_label == "Job Board" + selected_element.click.assert_awaited_once() + assert scope_args[0]["allowAriaSelectedOptionTokens"] is False + + @pytest.mark.asyncio + async def test_unchecked_target_proceeds_to_click(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + monkeypatch.setattr( + handler.SkyvernFrame, + "evaluate", + _stub_evaluate( + matched_state=[ + {"label": "WAKE", "ariaChecked": False, "selectedAttr": False, "checked": False}, + {"label": "WAKE", "ariaChecked": False, "selectedAttr": False, "checked": True}, + ] + ), + ) + selected_element = _FakeCustomElement() + selected_element.get_locator().count = AsyncMock(return_value=1) + + result = await _select_deterministic_custom_option( + target_value="WAKE", + get_option_candidates=lambda: [{"label": "WAKE", "element_id": "county-wake", "value": "WAKE"}], + field_context={}, + page=MagicMock(), + get_skyvern_element=AsyncMock(return_value=selected_element), + task=_task(), # type: ignore[arg-type] + ) + + assert result is not None + action_result, matched_label = result + assert action_result.success + assert matched_label == "WAKE" + selected_element.click.assert_awaited_once() + + @pytest.mark.asyncio + async def test_readback_scope_resolution_failure_still_verifies_matched_element( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + monkeypatch.setattr( + handler.SkyvernFrame, + "evaluate", + _stub_evaluate( + matched_state=[ + {"label": "Job Board", "ariaSelected": False, "selectedAttr": False, "checked": False}, + {"label": "Job Board", "ariaSelected": True, "selectedAttr": False, "checked": False}, + ] + ), + ) + selected_element = _FakeCustomElement() + selected_element.get_locator().count = AsyncMock(return_value=1) + + result = await _select_deterministic_custom_option( + target_value="Job Board", + get_option_candidates=lambda: [{"label": "Job Board", "element_id": "source-job-board", "value": None}], + field_context={}, + page=MagicMock(), + get_skyvern_element=AsyncMock(return_value=selected_element), + get_readback_scope_element=AsyncMock(side_effect=RuntimeError("anchor disappeared")), + task=_task(), # type: ignore[arg-type] + ) + + assert result is not None + action_result, matched_label = result + assert action_result.success + assert matched_label == "Job Board" + selected_element.click.assert_awaited_once() + + @pytest.mark.asyncio + async def test_clicked_but_unverified_checkbox_returns_failure_without_llm_fallback( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + monkeypatch.setattr( + handler.SkyvernFrame, + "evaluate", + _stub_evaluate( + matched_state={"label": "Job Board", "ariaSelected": False, "selectedAttr": False, "checked": False}, + committed=False, + ), + ) + selected_element = _FakeCustomElement() + selected_element.get_locator().count = AsyncMock(return_value=1) + # A checkbox-panel anchor (button) is non-resettable, so an unverified click hard-fails. + readback_scope_element = _FakeAnchorElement(tag_name="button") + + result = await _select_deterministic_custom_option( + target_value="Job Board", + get_option_candidates=lambda: [{"label": "Job Board", "element_id": "source-job-board", "value": None}], + field_context={}, + page=MagicMock(), + get_skyvern_element=AsyncMock(return_value=selected_element), + get_readback_scope_element=AsyncMock(return_value=readback_scope_element), + task=_task(), # type: ignore[arg-type] + ) + + assert result is not None + action_result, matched_label = result + assert not action_result.success + assert action_result.skip_remaining_actions is True + assert matched_label == "Job Board" + assert action_result.exception_message is not None + assert "could not be verified" in action_result.exception_message + selected_element.click.assert_awaited_once() + + @pytest.mark.asyncio + async def test_inconclusive_combobox_routes_to_llm_fallback_after_reset( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + monkeypatch.setattr( + handler.SkyvernFrame, + "evaluate", + _stub_evaluate( + matched_state={"label": "WAKE", "ariaSelected": False, "selectedAttr": False, "checked": False}, + committed=False, + ), + ) + selected_element = _FakeCustomElement() + selected_element.get_locator().count = AsyncMock(return_value=1) + # A text-input combobox anchor is resettable, so an inconclusive read-back must NOT hard-fail. + readback_scope_element = _FakeAnchorElement(tag_name="input") + + result = await _select_deterministic_custom_option( + target_value="WAKE", + get_option_candidates=lambda: [{"label": "WAKE", "element_id": "source-wake", "value": "WAKE"}], + field_context={}, + page=MagicMock(), + get_skyvern_element=AsyncMock(return_value=selected_element), + get_readback_scope_element=AsyncMock(return_value=readback_scope_element), + task=_task(), # type: ignore[arg-type] + ) + + assert result is None + selected_element.click.assert_awaited_once() + readback_scope_element.get_locator().fill.assert_awaited_once_with("") + readback_scope_element.click.assert_awaited_once() + + @pytest.mark.asyncio + async def test_readback_accepts_aria_checked_matched_menuitemradio(self, monkeypatch: pytest.MonkeyPatch) -> None: + matched_element = _FakeCustomElement() + matched_element.get_locator().count = AsyncMock(return_value=1) + monkeypatch.setattr( + handler.SkyvernFrame, + "evaluate", + _stub_evaluate( + matched_state={ + "label": "Choice", + "ariaSelected": False, + "ariaChecked": True, + "selectedAttr": False, + "checked": False, + }, + committed=False, + ), + ) + readback_scope_element = _FakeAnchorElement() + + assert ( + await _verify_custom_select_option( + matched_element=matched_element, # type: ignore[arg-type] + readback_scope_element=readback_scope_element, # type: ignore[arg-type] + anchor_is_combobox_input=False, + matched_element_id="choice-radio", + matched_label="Choice", + ) + is True + ) + + @pytest.mark.asyncio + async def test_readback_accepts_scoped_trigger_reflection_without_synthetic_token(self) -> None: + matched_element = _FakeCustomElement() + matched_element.get_locator().count = AsyncMock(return_value=0) + captured: dict[str, object] = {} + + async def evaluate(*, frame: object, expression: str, arg: object = None) -> object: + captured["expression"] = expression + captured["arg"] = arg + if "return { label," in expression: + return None + return True + + readback_scope_element = _FakeAnchorElement() + + with pytest.MonkeyPatch.context() as mp: + mp.setattr(handler.SkyvernFrame, "evaluate", AsyncMock(side_effect=evaluate)) + outcome = await _verify_custom_select_option( + matched_element=matched_element, # type: ignore[arg-type] + readback_scope_element=readback_scope_element, # type: ignore[arg-type] + anchor_is_combobox_input=False, + matched_element_id="choice-radio", + matched_label="Choice", + ) + + assert outcome is True + script = str(captured["expression"]) + assert "document.querySelectorAll" not in script + assert "data-selected-option-id" not in script + assert "data-selected-element-id" not in script + assert "data-skyvern-selected-id" not in script + assert "unique_id" not in script + assert "scopeRoot.querySelectorAll(triggerSelector)" in script + assert "aria-valuetext" in script + assert "aria-activedescendant" in script + + @pytest.mark.asyncio + async def test_combobox_filter_text_value_is_not_a_committed_signal(self) -> None: + # A combobox whose .value still holds the typed filter text must NOT be accepted as + # committed; the JS returns False so the caller routes to the safe LLM fallback. + matched_element = _FakeCustomElement() + matched_element.get_locator().count = AsyncMock(return_value=0) + readback_scope_element = _FakeAnchorElement(tag_name="input") + + with pytest.MonkeyPatch.context() as mp: + mp.setattr( + handler.SkyvernFrame, + "evaluate", + _stub_evaluate(matched_state=None, committed=False), + ) + outcome = await _verify_custom_select_option( + matched_element=matched_element, # type: ignore[arg-type] + readback_scope_element=readback_scope_element, # type: ignore[arg-type] + anchor_is_combobox_input=True, + matched_element_id="source-wake", + matched_label="WAKE", + ) + + assert outcome is False + + @pytest.mark.asyncio + async def test_handle_select_option_action_does_not_value_fallback_after_deterministic_readback_failure( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + monkeypatch.setattr( + handler.SkyvernFrame, + "evaluate", + _stub_evaluate( + matched_state={"label": "Choice", "ariaSelected": False, "selectedAttr": False, "checked": False}, + committed=False, + ), + ) + + anchor_element = _FakeAnchorElement() + selected_element = _FakeCustomElement() + selected_element.get_locator().count = AsyncMock(return_value=1) + fake_frame = MagicMock() + fake_frame.safe_wait_for_animation_end = AsyncMock() + fake_incremental = _FakeIncrementalScrapePage( + [ + [{"id": "opened-option"}], + [ + { + "id": "choice-option", + "tagName": "div", + "attributes": {"role": "option"}, + "text": "Choice", + "interactable": True, + } + ], + ] + ) + + class FakeDomUtil: + def __init__(self, *_args: object, **_kwargs: object) -> None: + pass + + async def get_skyvern_element_by_id(self, _element_id: str) -> _FakeAnchorElement: + return anchor_element + + select_from_dropdown_by_value = AsyncMock(return_value=handler.ActionSuccess()) + monkeypatch.setattr(handler, "DomUtil", FakeDomUtil) + monkeypatch.setattr(handler.SkyvernFrame, "create_instance", AsyncMock(return_value=fake_frame)) + monkeypatch.setattr(handler, "IncrementalScrapePage", MagicMock(return_value=fake_incremental)) + monkeypatch.setattr( + handler, + "_get_input_or_select_context", + AsyncMock(return_value=InputOrSelectContext(field="Field", is_required=True)), + ) + monkeypatch.setattr(handler, "locate_dropdown_menu", AsyncMock(return_value=None)) + monkeypatch.setattr(handler.SkyvernElement, "create_from_incremental", AsyncMock(return_value=selected_element)) + monkeypatch.setattr(handler, "select_from_dropdown_by_value", select_from_dropdown_by_value) + + results = await handler.handle_select_option_action( + action=_select_action(), + page=MagicMock(), + scraped_page=SimpleNamespace( + id_to_element_dict={"field-control": {"id": "field-control"}}, + id_to_css_dict={}, + ), + task=_task(), # type: ignore[arg-type] + step=MagicMock(), + ) + + assert len(results) == 1 + assert not results[0].success + assert results[0].skip_remaining_actions is True + selected_element.click.assert_awaited_once() + select_from_dropdown_by_value.assert_not_awaited() + + @pytest.mark.asyncio + async def test_select_from_dropdown_deterministic_success_skips_custom_select_prompt_and_llm( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + handler.app, + "EXPERIMENTATION_PROVIDER", + SimpleNamespace(is_feature_enabled_cached=AsyncMock(return_value=True)), + ) + monkeypatch.setattr(handler.app, "AGENT_FUNCTION", AgentFunction()) + monkeypatch.setattr( + handler.SkyvernFrame, + "evaluate", + _stub_evaluate( + matched_state=[ + {"label": "Choice", "ariaSelected": False, "selectedAttr": False, "checked": False}, + {"label": "Choice", "ariaSelected": True, "selectedAttr": False, "checked": False}, + ] + ), + ) + load_prompt = MagicMock(return_value="prompt") + custom_select_llm = AsyncMock(return_value={}) + monkeypatch.setattr(handler.prompt_engine, "load_prompt", load_prompt) + monkeypatch.setattr(handler.app, "CUSTOM_SELECT_AGENT_LLM_API_HANDLER", custom_select_llm) + monkeypatch.setattr(handler, "locate_dropdown_menu", AsyncMock(return_value=None)) + + anchor_element = _FakeAnchorElement() + selected_element = _FakeCustomElement() + selected_element.get_locator().count = AsyncMock(return_value=1) + fake_incremental = _FakeIncrementalScrapePage( + [ + [ + { + "id": "choice-option", + "tagName": "div", + "attributes": {"role": "option"}, + "text": "Choice", + "interactable": True, + } + ] + ] + ) + monkeypatch.setattr(handler.SkyvernElement, "create_from_incremental", AsyncMock(return_value=selected_element)) + + result = await handler.select_from_dropdown( + context=InputOrSelectContext(field="Field", is_required=True), + page=MagicMock(), + skyvern_element=anchor_element, # type: ignore[arg-type] + skyvern_frame=MagicMock(), + incremental_scraped=fake_incremental, # type: ignore[arg-type] + check_filter_funcs=[], + step=MagicMock(), + task=_task(), # type: ignore[arg-type] + force_select=True, + target_value="Choice", + ) + + assert isinstance(result.action_result, handler.ActionSuccess) + assert result.value == "Choice" + selected_element.click.assert_awaited_once() + load_prompt.assert_not_called() + custom_select_llm.assert_not_awaited() + + @pytest.mark.asyncio + async def test_readback_scope_scanned_via_skyvernframe_only(self) -> None: + matched_element = _FakeCustomElement() + matched_element.get_locator().count = AsyncMock(return_value=0) + captured: dict[str, object] = {} + + async def evaluate(*, frame: object, expression: str, arg: object = None) -> object: + if "anchorIsComboboxInput" in expression: + captured["expression"] = expression + if "return { label," in expression: + return None + return False + + readback_scope_element = _FakeAnchorElement() + + with pytest.MonkeyPatch.context() as mp: + mp.setattr(handler.SkyvernFrame, "evaluate", AsyncMock(side_effect=evaluate)) + outcome = await _verify_custom_select_option( + matched_element=matched_element, # type: ignore[arg-type] + readback_scope_element=readback_scope_element, # type: ignore[arg-type] + anchor_is_combobox_input=False, + matched_element_id="matched-option", + matched_label="Job Board", + ) + + assert outcome is False + script = str(captured["expression"]) + assert script.count('"output"') == 0 + assert script.count("document.querySelectorAll") == 0 + assert script.count("scopeRoot.querySelectorAll") >= 1 + assert script.count("triggerSelector") >= 1 + assert script.count("aria-valuetext") >= 1 + assert script.count("data-selected-option-id") == 0 + assert script.count("data-selected-element-id") == 0 + assert script.count("data-skyvern-selected-id") == 0 + + class TestNoAvailableOptionFoundForCustomSelection: def test_message_includes_code_target_count_excerpt_and_reason(self) -> None: exc = NoAvailableOptionFoundForCustomSelection(