mirror of
https://github.com/Skyvern-AI/skyvern.git
synced 2026-07-09 16:09:13 +00:00
test(SKY-11845): delete verified-redundant unit tests (zero coverage loss) (#7113)
Co-authored-by: Suchintan Singh <suchintan@skyvern.com>
This commit is contained in:
Suchintan
GitHub
parent
8ab0db16d1
commit
f2f3399e67
10 changed files with 28 additions and 706 deletions
|
|
@ -5,7 +5,6 @@ They do NOT require an LLM key or Playwright — they test infrastructure only.
|
|||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
|
@ -32,38 +31,6 @@ def test_no_cloud_in_sys_modules() -> None:
|
|||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bootstrap_creates_db() -> None:
|
||||
"""Embedded bootstrap creates SQLite tables, org, and token."""
|
||||
from skyvern import Skyvern
|
||||
|
||||
skyvern = Skyvern.local(use_in_memory_db=True)
|
||||
try:
|
||||
workflows = await skyvern.get_workflows()
|
||||
assert workflows is not None
|
||||
finally:
|
||||
await skyvern.aclose()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_artifact_tempdir_is_live() -> None:
|
||||
"""StorageFactory points to an existing temp directory after bootstrap."""
|
||||
from skyvern import Skyvern
|
||||
from skyvern.library.embedded_server_factory import EmbeddedClient
|
||||
|
||||
skyvern = Skyvern.local(use_in_memory_db=True)
|
||||
try:
|
||||
await skyvern.get_workflows()
|
||||
client = getattr(skyvern, "_embedded_client", None)
|
||||
assert isinstance(client, EmbeddedClient)
|
||||
artifact_dir = client.embedded_transport._artifact_dir
|
||||
assert artifact_dir is not None
|
||||
assert Path(artifact_dir).exists()
|
||||
assert "skyvern-artifacts-" in artifact_dir
|
||||
finally:
|
||||
await skyvern.aclose()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unique_llm_key_no_collision() -> None:
|
||||
"""Sequential clients (create -> close -> create again) use different LLM keys."""
|
||||
|
|
|
|||
|
|
@ -27,6 +27,12 @@ def _make_file_info(url: str, filename: str, checksum: str | None = None) -> Fil
|
|||
return FileInfo(url=url, filename=filename, checksum=checksum)
|
||||
|
||||
|
||||
def test_downloaded_file_signature_strips_query_params():
|
||||
file_info = _make_file_info("https://files/a.pdf?sig=x", "a.pdf", "abc")
|
||||
|
||||
assert to_downloaded_file_signature(file_info) == ("a.pdf", "abc", "https://files/a.pdf")
|
||||
|
||||
|
||||
def test_second_block_excludes_first_blocks_files():
|
||||
"""
|
||||
Simulate two sequential task blocks that each download a file.
|
||||
|
|
|
|||
|
|
@ -127,13 +127,6 @@ def test_planning_tool_still_trips_name_only_streak() -> None:
|
|||
assert "LOOP DETECTED" in msg
|
||||
|
||||
|
||||
def test_block_running_tool_is_blocked_by_pending_reconciliation() -> None:
|
||||
ctx = _ctx(pending_reconciliation_run_id="wr_123")
|
||||
msg = _tool_loop_error(ctx, "update_and_run_blocks")
|
||||
assert msg is not None
|
||||
assert "wr_123" in msg
|
||||
|
||||
|
||||
def test_block_running_tool_blocks_late_retry_after_failed_workflow() -> None:
|
||||
ctx = _ctx(
|
||||
last_failed_workflow_yaml="version: '1.0'",
|
||||
|
|
|
|||
|
|
@ -1,270 +1,13 @@
|
|||
"""Unit tests for iframe support: _locator_scope, frame_switch, frame_main, frame_list."""
|
||||
"""Unit tests for iframe support."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from skyvern.cli.core.browser_ops import do_frame_list, do_frame_main, do_frame_switch
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_fake_frame(*, name: str = "", url: str = "about:blank") -> MagicMock:
|
||||
frame = MagicMock()
|
||||
frame.name = name
|
||||
frame.url = url
|
||||
frame.locator = MagicMock(return_value=MagicMock())
|
||||
return frame
|
||||
|
||||
|
||||
def _make_fake_page(frames: list[MagicMock] | None = None) -> MagicMock:
|
||||
"""Build a mock Playwright Page with .locator(), .frames, .main_frame, .frame()."""
|
||||
page = MagicMock()
|
||||
all_frames = frames or [_make_fake_frame(name="main", url="https://example.com")]
|
||||
page.frames = all_frames
|
||||
page.main_frame = all_frames[0]
|
||||
page.locator = MagicMock(return_value=MagicMock())
|
||||
page.frame = MagicMock(return_value=None)
|
||||
return page
|
||||
|
||||
|
||||
class FakeSkyvernPage:
|
||||
"""Minimal stand-in for SkyvernPage to test _locator_scope without Playwright."""
|
||||
|
||||
def __init__(self, page: Any, working_frame: Any = None) -> None:
|
||||
self.page = page
|
||||
self._working_frame = working_frame
|
||||
|
||||
@property
|
||||
def _locator_scope(self) -> Any:
|
||||
frame = object.__getattribute__(self, "_working_frame")
|
||||
if frame is not None:
|
||||
return frame
|
||||
return object.__getattribute__(self, "page")
|
||||
|
||||
|
||||
class FakeSkyvernBrowserPage(FakeSkyvernPage):
|
||||
"""Minimal stand-in for SkyvernBrowserPage to test frame methods."""
|
||||
|
||||
async def frame_switch(
|
||||
self, *, selector: str | None = None, name: str | None = None, index: int | None = None
|
||||
) -> dict[str, Any]:
|
||||
params = sum(p is not None for p in (selector, name, index))
|
||||
if params != 1:
|
||||
raise ValueError("Exactly one of selector, name, or index is required.")
|
||||
|
||||
frame = None
|
||||
|
||||
if selector is not None:
|
||||
element = await self.page.query_selector(selector)
|
||||
if element is None:
|
||||
raise ValueError(f"Selector '{selector}' did not match any element.")
|
||||
frame = await element.content_frame()
|
||||
if frame is None:
|
||||
raise ValueError(f"Selector '{selector}' did not resolve to an iframe.")
|
||||
elif name is not None:
|
||||
frame = self.page.frame(name=name)
|
||||
if frame is None:
|
||||
raise ValueError(f"No frame found with name '{name}'.")
|
||||
elif index is not None:
|
||||
frames = self.page.frames
|
||||
if index < 0 or index >= len(frames):
|
||||
raise ValueError(f"Frame index {index} out of range (0-{len(frames) - 1}).")
|
||||
frame = frames[index]
|
||||
|
||||
self._working_frame = frame
|
||||
return {
|
||||
"name": frame.name if frame else None,
|
||||
"url": frame.url if frame else None,
|
||||
"selector": selector,
|
||||
"frame_name": name,
|
||||
"index": index,
|
||||
}
|
||||
|
||||
def frame_main(self) -> dict[str, str]:
|
||||
self._working_frame = None
|
||||
return {"status": "switched_to_main_frame"}
|
||||
|
||||
async def frame_list(self) -> list[dict[str, Any]]:
|
||||
frames = self.page.frames
|
||||
return [
|
||||
{"index": i, "name": f.name, "url": f.url, "is_main": f == self.page.main_frame}
|
||||
for i, f in enumerate(frames)
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _locator_scope tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestLocatorScope:
|
||||
def test_returns_page_when_no_frame(self) -> None:
|
||||
page = _make_fake_page()
|
||||
sp = FakeSkyvernPage(page)
|
||||
assert sp._locator_scope is page
|
||||
|
||||
def test_returns_frame_when_set(self) -> None:
|
||||
page = _make_fake_page()
|
||||
frame = _make_fake_frame(name="iframe1")
|
||||
sp = FakeSkyvernPage(page, working_frame=frame)
|
||||
assert sp._locator_scope is frame
|
||||
|
||||
def test_returns_page_after_clearing_frame(self) -> None:
|
||||
page = _make_fake_page()
|
||||
frame = _make_fake_frame()
|
||||
sp = FakeSkyvernPage(page, working_frame=frame)
|
||||
assert sp._locator_scope is frame
|
||||
sp._working_frame = None
|
||||
assert sp._locator_scope is page
|
||||
|
||||
def test_locator_call_delegates_to_frame(self) -> None:
|
||||
page = _make_fake_page()
|
||||
frame = _make_fake_frame()
|
||||
sp = FakeSkyvernPage(page, working_frame=frame)
|
||||
sp._locator_scope.locator("#btn")
|
||||
frame.locator.assert_called_once_with("#btn")
|
||||
page.locator.assert_not_called()
|
||||
|
||||
def test_locator_call_delegates_to_page_when_no_frame(self) -> None:
|
||||
page = _make_fake_page()
|
||||
sp = FakeSkyvernPage(page)
|
||||
sp._locator_scope.locator("#btn")
|
||||
page.locator.assert_called_once_with("#btn")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# frame_switch tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFrameSwitch:
|
||||
@pytest.mark.asyncio
|
||||
async def test_switch_by_selector(self) -> None:
|
||||
iframe = _make_fake_frame(name="payment", url="https://payment.example.com/v3")
|
||||
element_mock = MagicMock()
|
||||
element_mock.content_frame = AsyncMock(return_value=iframe)
|
||||
|
||||
page = _make_fake_page()
|
||||
page.query_selector = AsyncMock(return_value=element_mock)
|
||||
|
||||
sp = FakeSkyvernBrowserPage(page)
|
||||
result = await sp.frame_switch(selector="#payment-frame")
|
||||
|
||||
assert sp._working_frame is iframe
|
||||
assert result["name"] == "payment"
|
||||
assert result["url"] == "https://payment.example.com/v3"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_switch_by_name(self) -> None:
|
||||
iframe = _make_fake_frame(name="checkout", url="https://checkout.com")
|
||||
page = _make_fake_page()
|
||||
page.frame = MagicMock(return_value=iframe)
|
||||
|
||||
sp = FakeSkyvernBrowserPage(page)
|
||||
result = await sp.frame_switch(name="checkout")
|
||||
|
||||
assert sp._working_frame is iframe
|
||||
assert result["frame_name"] == "checkout"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_switch_by_index(self) -> None:
|
||||
main = _make_fake_frame(name="main", url="https://example.com")
|
||||
iframe = _make_fake_frame(name="embed", url="https://embed.com")
|
||||
page = _make_fake_page([main, iframe])
|
||||
|
||||
sp = FakeSkyvernBrowserPage(page)
|
||||
result = await sp.frame_switch(index=1)
|
||||
|
||||
assert sp._working_frame is iframe
|
||||
assert result["index"] == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_switch_no_params_raises(self) -> None:
|
||||
page = _make_fake_page()
|
||||
sp = FakeSkyvernBrowserPage(page)
|
||||
with pytest.raises(ValueError, match="Exactly one"):
|
||||
await sp.frame_switch()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_switch_multiple_params_raises(self) -> None:
|
||||
page = _make_fake_page()
|
||||
sp = FakeSkyvernBrowserPage(page)
|
||||
with pytest.raises(ValueError, match="Exactly one"):
|
||||
await sp.frame_switch(selector="#x", name="y")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_switch_selector_not_iframe_raises(self) -> None:
|
||||
element_mock = MagicMock()
|
||||
element_mock.content_frame = AsyncMock(return_value=None)
|
||||
|
||||
page = _make_fake_page()
|
||||
page.query_selector = AsyncMock(return_value=element_mock)
|
||||
|
||||
sp = FakeSkyvernBrowserPage(page)
|
||||
with pytest.raises(ValueError, match="did not resolve to an iframe"):
|
||||
await sp.frame_switch(selector="#not-iframe")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_switch_name_not_found_raises(self) -> None:
|
||||
page = _make_fake_page()
|
||||
page.frame = MagicMock(return_value=None)
|
||||
|
||||
sp = FakeSkyvernBrowserPage(page)
|
||||
with pytest.raises(ValueError, match="No frame found"):
|
||||
await sp.frame_switch(name="nonexistent")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_switch_index_out_of_range_raises(self) -> None:
|
||||
page = _make_fake_page([_make_fake_frame()])
|
||||
sp = FakeSkyvernBrowserPage(page)
|
||||
with pytest.raises(ValueError, match="out of range"):
|
||||
await sp.frame_switch(index=5)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# frame_main tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFrameMain:
|
||||
def test_clears_working_frame(self) -> None:
|
||||
page = _make_fake_page()
|
||||
frame = _make_fake_frame()
|
||||
sp = FakeSkyvernBrowserPage(page, working_frame=frame)
|
||||
assert sp._working_frame is frame
|
||||
sp.frame_main()
|
||||
assert sp._working_frame is None
|
||||
assert sp._locator_scope is page
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# frame_list tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFrameList:
|
||||
@pytest.mark.asyncio
|
||||
async def test_lists_all_frames(self) -> None:
|
||||
main = _make_fake_frame(name="", url="https://example.com")
|
||||
iframe1 = _make_fake_frame(name="ads", url="https://ads.com")
|
||||
iframe2 = _make_fake_frame(name="payment", url="https://payment.example.com")
|
||||
page = _make_fake_page([main, iframe1, iframe2])
|
||||
|
||||
sp = FakeSkyvernBrowserPage(page)
|
||||
frames = await sp.frame_list()
|
||||
|
||||
assert len(frames) == 3
|
||||
assert frames[0]["is_main"] is True
|
||||
assert frames[1]["name"] == "ads"
|
||||
assert frames[2]["name"] == "payment"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MCP tool tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -432,7 +175,7 @@ class TestCLIStateFrame:
|
|||
assert state.frame_name is None
|
||||
assert state.frame_index is None
|
||||
|
||||
def test_frame_fields_roundtrip(self, tmp_path: Any) -> None:
|
||||
def test_frame_fields_roundtrip(self, tmp_path) -> None:
|
||||
import skyvern.cli.commands._state as state_mod
|
||||
from skyvern.cli.commands._state import CLIState, load_state, save_state
|
||||
|
||||
|
|
|
|||
|
|
@ -79,68 +79,3 @@ def test_screenshot_attrs_reflect_post_filter_for_non_vision_config() -> None:
|
|||
_set_llm_context_attrs(span, screenshots=filtered, is_speculative_step=False)
|
||||
span.set_attribute.assert_any_call("screenshots_included", False)
|
||||
span.set_attribute.assert_any_call("screenshot_count", 0)
|
||||
|
||||
|
||||
def test_set_llm_context_attrs_invoked_in_all_three_handlers() -> None:
|
||||
# Invariant: router, non-router, and LLMCaller handlers must each record
|
||||
# context attrs; a missing call site silently drops attribution.
|
||||
from skyvern.forge.sdk.api.llm import api_handler_factory as mod
|
||||
|
||||
tree = ast.parse(inspect.getsource(mod))
|
||||
call_count = 0
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.Call):
|
||||
continue
|
||||
target = node.func
|
||||
fname = (
|
||||
target.id if isinstance(target, ast.Name) else (target.attr if isinstance(target, ast.Attribute) else None)
|
||||
)
|
||||
if fname == "_set_llm_context_attrs":
|
||||
call_count += 1
|
||||
assert call_count == 3, f"Expected exactly 3 call sites, found {call_count}"
|
||||
|
||||
|
||||
def _iter_function_defs(node: ast.AST) -> list[ast.FunctionDef | ast.AsyncFunctionDef]:
|
||||
funcs: list[ast.FunctionDef | ast.AsyncFunctionDef] = []
|
||||
for n in ast.walk(node):
|
||||
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
funcs.append(n)
|
||||
return funcs
|
||||
|
||||
|
||||
def _call_lineno(func: ast.FunctionDef | ast.AsyncFunctionDef, name: str) -> int | None:
|
||||
for n in ast.walk(func):
|
||||
if isinstance(n, ast.Call):
|
||||
target = n.func
|
||||
fname = (
|
||||
target.id
|
||||
if isinstance(target, ast.Name)
|
||||
else (target.attr if isinstance(target, ast.Attribute) else None)
|
||||
)
|
||||
if fname == name:
|
||||
return n.lineno
|
||||
return None
|
||||
|
||||
|
||||
def test_handlers_compute_screenshot_attrs_after_filter() -> None:
|
||||
# Invariant: in every handler that records screenshot attrs, the
|
||||
# _set_llm_context_attrs call must run AFTER _llm_screenshots_for_call so
|
||||
# the attrs reflect what was actually sent to the LLM, not the raw arg.
|
||||
from skyvern.forge.sdk.api.llm import api_handler_factory as mod
|
||||
|
||||
tree = ast.parse(inspect.getsource(mod))
|
||||
matches = 0
|
||||
for func in _iter_function_defs(tree):
|
||||
attrs_line = _call_lineno(func, "_set_llm_context_attrs")
|
||||
filter_line = _call_lineno(func, "_llm_screenshots_for_call")
|
||||
if attrs_line is None or filter_line is None:
|
||||
continue
|
||||
assert attrs_line > filter_line, (
|
||||
f"In `{func.name}` (defined at line {func.lineno}): "
|
||||
f"_set_llm_context_attrs (line {attrs_line}) must come AFTER "
|
||||
f"_llm_screenshots_for_call (line {filter_line})."
|
||||
)
|
||||
matches += 1
|
||||
# ast.walk yields the 3 inner handler closures plus their two enclosing
|
||||
# factory functions, so the ordering invariant must hold for >=3 matches.
|
||||
assert matches >= 3, f"Expected at least 3 handlers with both calls, found {matches}"
|
||||
|
|
|
|||
|
|
@ -1,111 +0,0 @@
|
|||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
from skyvern.forge.sdk.schemas.files import FileInfo
|
||||
from skyvern.forge.sdk.workflow.loop_download_filter import (
|
||||
filter_downloaded_files_for_current_iteration,
|
||||
to_downloaded_file_signature,
|
||||
)
|
||||
|
||||
|
||||
def _file(url: str, filename: str, checksum: str) -> FileInfo:
|
||||
return FileInfo(url=url, filename=filename, checksum=checksum)
|
||||
|
||||
|
||||
async def _capture_baseline(
|
||||
get_files_coro: asyncio.coroutines,
|
||||
timeout_seconds: float,
|
||||
) -> dict | None:
|
||||
"""Mirrors the exact baseline capture pattern from block.py / script_service.py."""
|
||||
sigs: list = []
|
||||
timed_out = False
|
||||
try:
|
||||
async with asyncio.timeout(timeout_seconds):
|
||||
sigs = [to_downloaded_file_signature(fi) for fi in await get_files_coro]
|
||||
except TimeoutError:
|
||||
timed_out = True
|
||||
|
||||
if timed_out:
|
||||
return None
|
||||
return {"downloaded_file_signatures_before_iteration": sigs}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_baseline_capture_returns_none_on_timeout() -> None:
|
||||
"""When get_downloaded_files hangs, the baseline must be None (not empty sigs)."""
|
||||
|
||||
async def _hang_forever() -> list[FileInfo]:
|
||||
await asyncio.Event().wait()
|
||||
return [] # never reached
|
||||
|
||||
result = await _capture_baseline(_hang_forever(), timeout_seconds=0.01)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_baseline_capture_returns_signatures_on_success() -> None:
|
||||
"""Normal completion should return the signature dict."""
|
||||
|
||||
async def _return_files() -> list[FileInfo]:
|
||||
return [
|
||||
_file("https://files/a.pdf?sig=x", "a.pdf", "abc"),
|
||||
_file("https://files/b.pdf?sig=y", "b.pdf", "def"),
|
||||
]
|
||||
|
||||
result = await _capture_baseline(_return_files(), timeout_seconds=5.0)
|
||||
assert result is not None
|
||||
sigs = result["downloaded_file_signatures_before_iteration"]
|
||||
assert len(sigs) == 2
|
||||
assert sigs[0] == ("a.pdf", "abc", "https://files/a.pdf")
|
||||
assert sigs[1] == ("b.pdf", "def", "https://files/b.pdf")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_baseline_capture_returns_empty_sigs_when_no_files() -> None:
|
||||
"""No files is distinct from a timeout — returns empty list, not None."""
|
||||
|
||||
async def _return_empty() -> list[FileInfo]:
|
||||
return []
|
||||
|
||||
result = await _capture_baseline(_return_empty(), timeout_seconds=5.0)
|
||||
assert result is not None
|
||||
assert result == {"downloaded_file_signatures_before_iteration": []}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_timeout_then_filter_returns_all_files_unfiltered() -> None:
|
||||
"""End-to-end: timeout → None state → filter returns all files."""
|
||||
|
||||
async def _hang_forever() -> list[FileInfo]:
|
||||
await asyncio.Event().wait()
|
||||
return []
|
||||
|
||||
loop_internal_state = await _capture_baseline(_hang_forever(), timeout_seconds=0.01)
|
||||
assert loop_internal_state is None
|
||||
|
||||
all_files = [
|
||||
_file("https://files/a.pdf", "a.pdf", "abc"),
|
||||
_file("https://files/b.pdf", "b.pdf", "def"),
|
||||
]
|
||||
result = filter_downloaded_files_for_current_iteration(all_files, loop_internal_state)
|
||||
assert result == all_files
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_success_then_filter_excludes_baseline_files() -> None:
|
||||
"""End-to-end: successful capture → filter excludes baseline files."""
|
||||
|
||||
async def _return_baseline() -> list[FileInfo]:
|
||||
return [_file("https://files/a.pdf?sig=old", "a.pdf", "abc")]
|
||||
|
||||
loop_internal_state = await _capture_baseline(_return_baseline(), timeout_seconds=5.0)
|
||||
assert loop_internal_state is not None
|
||||
|
||||
all_files = [
|
||||
_file("https://files/a.pdf?sig=old", "a.pdf", "abc"),
|
||||
_file("https://files/b.pdf?sig=new", "b.pdf", "def"),
|
||||
]
|
||||
result = filter_downloaded_files_for_current_iteration(all_files, loop_internal_state)
|
||||
assert len(result) == 1
|
||||
assert result[0].filename == "b.pdf"
|
||||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
from pathlib import Path
|
||||
|
||||
from skyvern.core.script_generations.script_validators import validate_missing_selectors
|
||||
from skyvern.services.script_reviewer import ScriptReviewer
|
||||
|
||||
_PROMPT_PATH = Path(__file__).resolve().parents[2] / "skyvern" / "forge" / "prompts" / "skyvern" / "script-reviewer.j2"
|
||||
|
|
@ -404,53 +405,7 @@ class TestValidateMissingSelectors:
|
|||
def setup_method(self) -> None:
|
||||
self.reviewer = ScriptReviewer()
|
||||
|
||||
def test_fallback_with_selector_is_fine(self) -> None:
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
await page.click(selector='button:has-text("Submit")', ai='fallback', prompt='submit')
|
||||
"""
|
||||
assert self.reviewer._validate_missing_selectors(code) is None
|
||||
|
||||
def test_fallback_without_selector_flagged(self) -> None:
|
||||
"""ai='fallback' with no selector= silently uses AI as primary path."""
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
await page.click(ai='fallback', prompt='Click Billing & Payments')
|
||||
"""
|
||||
error = self.reviewer._validate_missing_selectors(code)
|
||||
assert error is not None
|
||||
assert "page.click()" in error
|
||||
assert "Missing selector" in error
|
||||
|
||||
def test_fill_without_selector_flagged(self) -> None:
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
await page.fill(ai='fallback', prompt='Enter username', value='test')
|
||||
"""
|
||||
error = self.reviewer._validate_missing_selectors(code)
|
||||
assert error is not None
|
||||
assert "page.fill()" in error
|
||||
|
||||
def test_no_ai_arg_without_selector_flagged(self) -> None:
|
||||
"""Bare calls with no ai= and no selector= silently burn LLM tokens."""
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
await page.click(prompt='Click something')
|
||||
"""
|
||||
error = self.reviewer._validate_missing_selectors(code)
|
||||
assert error is not None
|
||||
assert "page.click()" in error
|
||||
assert "no ai= argument" in error
|
||||
|
||||
def test_proactive_without_selector_not_flagged(self) -> None:
|
||||
"""ai='proactive' without selector is intentional — AI always generates the value."""
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
await page.click(ai='proactive', prompt='Click something')
|
||||
"""
|
||||
assert self.reviewer._validate_missing_selectors(code) is None
|
||||
|
||||
def test_multiline_call_flagged(self) -> None:
|
||||
def test_delegates_to_shared_validator(self) -> None:
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
await page.click(
|
||||
|
|
@ -458,36 +413,9 @@ async def block_fn(page, context):
|
|||
prompt='Click Billing & Payments',
|
||||
)
|
||||
"""
|
||||
error = self.reviewer._validate_missing_selectors(code)
|
||||
assert error is not None
|
||||
assert "page.click()" in error
|
||||
|
||||
def test_multiline_with_selector_ok(self) -> None:
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
await page.click(
|
||||
selector='a:has-text("Billing")',
|
||||
ai='fallback',
|
||||
prompt='Click billing link',
|
||||
)
|
||||
"""
|
||||
assert self.reviewer._validate_missing_selectors(code) is None
|
||||
|
||||
def test_comments_ignored(self) -> None:
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
# await page.click(ai='fallback', prompt='old code')
|
||||
await page.click(selector='button', ai='fallback', prompt='submit')
|
||||
"""
|
||||
assert self.reviewer._validate_missing_selectors(code) is None
|
||||
|
||||
def test_non_interaction_methods_ignored(self) -> None:
|
||||
"""Methods like page.wait, page.complete are not interaction methods."""
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
await page.wait(ai='fallback', prompt='wait for page')
|
||||
"""
|
||||
assert self.reviewer._validate_missing_selectors(code) is None
|
||||
result = self.reviewer._validate_missing_selectors(code)
|
||||
assert result is not None
|
||||
assert result == validate_missing_selectors(code)
|
||||
|
||||
|
||||
class TestExtractCachedBlocks:
|
||||
|
|
|
|||
|
|
@ -107,6 +107,21 @@ async def block_fn(page, context):
|
|||
assert error is not None
|
||||
assert "page.fill()" in error
|
||||
|
||||
def test_comments_ignored(self) -> None:
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
# await page.click(ai='fallback', prompt='old code')
|
||||
await page.click(selector='button', ai='fallback', prompt='submit')
|
||||
"""
|
||||
assert validate_missing_selectors(code) is None
|
||||
|
||||
def test_non_interaction_methods_ignored(self) -> None:
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
await page.wait(ai='fallback', prompt='wait for page')
|
||||
"""
|
||||
assert validate_missing_selectors(code) is None
|
||||
|
||||
def test_multiple_methods_flagged(self) -> None:
|
||||
code = """
|
||||
async def block_fn(page, context):
|
||||
|
|
|
|||
|
|
@ -32,86 +32,6 @@ def _output_parameter(key: str) -> OutputParameter:
|
|||
)
|
||||
|
||||
|
||||
def test_taskv2_output_includes_downloaded_files_filtered_by_loop() -> None:
|
||||
"""When inside a loop, TaskV2Block output should include only THIS iteration's downloaded files."""
|
||||
# Simulate baseline: iteration started with a.pdf already downloaded
|
||||
loop_state = {
|
||||
"downloaded_file_signatures_before_iteration": [
|
||||
["a.pdf", "abc", "https://files/a.pdf"],
|
||||
],
|
||||
}
|
||||
|
||||
# Storage returns all files (a.pdf from before + b.pdf downloaded this iteration)
|
||||
all_files = [
|
||||
_file("https://files/a.pdf?sig=old", "a.pdf", "abc"),
|
||||
_file("https://files/b.pdf?sig=new", "b.pdf", "def"),
|
||||
]
|
||||
|
||||
# Apply the same filter that TaskV2Block now uses
|
||||
filtered = filter_downloaded_files_for_current_iteration(all_files, loop_state)
|
||||
|
||||
# Build output dict the same way TaskV2Block does
|
||||
task_v2_output = {
|
||||
"task_id": "oc_test",
|
||||
"status": "completed",
|
||||
"summary": None,
|
||||
"extracted_information": None,
|
||||
"failure_reason": None,
|
||||
"downloaded_files": [fi.model_dump() for fi in filtered],
|
||||
"downloaded_file_urls": [fi.url for fi in filtered],
|
||||
"task_screenshot_artifact_ids": [],
|
||||
"workflow_screenshot_artifact_ids": [],
|
||||
}
|
||||
|
||||
assert len(task_v2_output["downloaded_files"]) == 1
|
||||
assert task_v2_output["downloaded_files"][0]["filename"] == "b.pdf"
|
||||
assert task_v2_output["downloaded_file_urls"] == ["https://files/b.pdf?sig=new"]
|
||||
|
||||
|
||||
def test_taskv2_output_includes_all_files_outside_loop() -> None:
|
||||
"""Outside a loop (no loop_internal_state), all downloaded files should be included."""
|
||||
all_files = [
|
||||
_file("https://files/a.pdf", "a.pdf", "abc"),
|
||||
_file("https://files/b.pdf", "b.pdf", "def"),
|
||||
]
|
||||
|
||||
filtered = filter_downloaded_files_for_current_iteration(all_files, None)
|
||||
|
||||
task_v2_output = {
|
||||
"downloaded_files": [fi.model_dump() for fi in filtered],
|
||||
"downloaded_file_urls": [fi.url for fi in filtered],
|
||||
}
|
||||
|
||||
assert len(task_v2_output["downloaded_files"]) == 2
|
||||
assert task_v2_output["downloaded_file_urls"] == [
|
||||
"https://files/a.pdf",
|
||||
"https://files/b.pdf",
|
||||
]
|
||||
|
||||
|
||||
def test_taskv2_output_empty_when_no_new_downloads_in_iteration() -> None:
|
||||
"""If no new files were downloaded in this iteration, both lists should be empty."""
|
||||
loop_state = {
|
||||
"downloaded_file_signatures_before_iteration": [
|
||||
["a.pdf", "abc", "https://files/a.pdf"],
|
||||
],
|
||||
}
|
||||
|
||||
all_files = [
|
||||
_file("https://files/a.pdf?sig=old", "a.pdf", "abc"),
|
||||
]
|
||||
|
||||
filtered = filter_downloaded_files_for_current_iteration(all_files, loop_state)
|
||||
|
||||
task_v2_output = {
|
||||
"downloaded_files": [fi.model_dump() for fi in filtered],
|
||||
"downloaded_file_urls": [fi.url for fi in filtered],
|
||||
}
|
||||
|
||||
assert task_v2_output["downloaded_files"] == []
|
||||
assert task_v2_output["downloaded_file_urls"] == []
|
||||
|
||||
|
||||
def test_taskv2_context_loop_state_available_after_nested_task_execution() -> None:
|
||||
"""Verify loop_internal_state survives nested task execution and remains available for filtering."""
|
||||
loop_state = {
|
||||
|
|
|
|||
|
|
@ -1,74 +0,0 @@
|
|||
"""Track while-loop child blocks for script caching (mirrors for-loop SKY-7751 pattern)."""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from skyvern.forge.sdk.workflow.models.block import (
|
||||
BlockType,
|
||||
FileDownloadBlock,
|
||||
JinjaBranchCriteria,
|
||||
TaskBlock,
|
||||
WhileLoopBlock,
|
||||
)
|
||||
from skyvern.forge.sdk.workflow.models.parameter import OutputParameter
|
||||
from skyvern.forge.sdk.workflow.service import BLOCK_TYPES_THAT_SHOULD_BE_CACHED
|
||||
|
||||
|
||||
def _make_output_param(label: str) -> OutputParameter:
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
return OutputParameter(
|
||||
key=f"{label}_output",
|
||||
parameter_type="output",
|
||||
output_parameter_id=f"op_{label}",
|
||||
workflow_id="wf_test",
|
||||
created_at=now,
|
||||
modified_at=now,
|
||||
)
|
||||
|
||||
|
||||
def test_while_loop_is_cacheable() -> None:
|
||||
assert BlockType.WHILE_LOOP in BLOCK_TYPES_THAT_SHOULD_BE_CACHED
|
||||
|
||||
|
||||
def test_while_loop_child_labels_collected() -> None:
|
||||
inner = TaskBlock(label="inner_task", output_parameter=_make_output_param("inner_task"), url="https://x.test")
|
||||
wloop = WhileLoopBlock(
|
||||
label="w",
|
||||
output_parameter=_make_output_param("w"),
|
||||
loop_blocks=[inner],
|
||||
condition=JinjaBranchCriteria(expression="{{ true }}"),
|
||||
)
|
||||
script_blocks_by_label: dict[str, object] = {}
|
||||
blocks_to_update: set[str] = set()
|
||||
for loop_child in wloop.loop_blocks:
|
||||
if (
|
||||
loop_child.label
|
||||
and loop_child.label not in script_blocks_by_label
|
||||
and loop_child.block_type in BLOCK_TYPES_THAT_SHOULD_BE_CACHED
|
||||
):
|
||||
blocks_to_update.add(loop_child.label)
|
||||
assert "inner_task" in blocks_to_update
|
||||
|
||||
|
||||
def test_while_loop_file_download_child() -> None:
|
||||
dl = FileDownloadBlock(
|
||||
label="dl",
|
||||
output_parameter=_make_output_param("dl"),
|
||||
url="http://example.com",
|
||||
navigation_goal="get file",
|
||||
)
|
||||
wloop = WhileLoopBlock(
|
||||
label="w",
|
||||
output_parameter=_make_output_param("w"),
|
||||
loop_blocks=[dl],
|
||||
condition=JinjaBranchCriteria(expression="{{ false }}"),
|
||||
)
|
||||
script_blocks_by_label: dict[str, object] = {}
|
||||
blocks_to_update: set[str] = set()
|
||||
for loop_child in wloop.loop_blocks:
|
||||
if (
|
||||
loop_child.label
|
||||
and loop_child.label not in script_blocks_by_label
|
||||
and loop_child.block_type in BLOCK_TYPES_THAT_SHOULD_BE_CACHED
|
||||
):
|
||||
blocks_to_update.add(loop_child.label)
|
||||
assert "dl" in blocks_to_update
|
||||
Loading…
Add table
Add a link
Reference in a new issue