299 lines
12 KiB
Python
299 lines
12 KiB
Python
"""Tests for CLI commands infrastructure: _state.py and _output.py."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
import typer
|
|
|
|
from skyvern.cli.commands._state import CLIState, clear_state, load_state, save_state
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _state.py
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _patch_state_dir(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
monkeypatch.setattr("skyvern.cli.commands._state.STATE_DIR", tmp_path)
|
|
monkeypatch.setattr("skyvern.cli.commands._state.STATE_FILE", tmp_path / "state.json")
|
|
|
|
|
|
class TestCLIState:
|
|
def test_save_load_roundtrip(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_patch_state_dir(monkeypatch, tmp_path)
|
|
save_state(CLIState(session_id="pbs_123", cdp_url=None, mode="cloud"))
|
|
loaded = load_state()
|
|
assert loaded is not None
|
|
assert loaded.session_id == "pbs_123"
|
|
assert loaded.cdp_url is None
|
|
assert loaded.mode == "cloud"
|
|
|
|
def test_save_load_roundtrip_cdp(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_patch_state_dir(monkeypatch, tmp_path)
|
|
save_state(CLIState(session_id=None, cdp_url="ws://localhost:9222/devtools/browser/abc", mode="cdp"))
|
|
loaded = load_state()
|
|
assert loaded is not None
|
|
assert loaded.session_id is None
|
|
assert loaded.cdp_url == "ws://localhost:9222/devtools/browser/abc"
|
|
assert loaded.mode == "cdp"
|
|
|
|
def test_load_returns_none_when_missing(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr("skyvern.cli.commands._state.STATE_FILE", tmp_path / "nonexistent.json")
|
|
assert load_state() is None
|
|
|
|
def test_24h_ttl_expires(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_patch_state_dir(monkeypatch, tmp_path)
|
|
state_file = tmp_path / "state.json"
|
|
state_file.write_text(
|
|
json.dumps(
|
|
{
|
|
"session_id": "pbs_old",
|
|
"mode": "cloud",
|
|
"created_at": "2020-01-01T00:00:00+00:00",
|
|
}
|
|
)
|
|
)
|
|
assert load_state() is None
|
|
|
|
def test_clear_state(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_patch_state_dir(monkeypatch, tmp_path)
|
|
save_state(CLIState(session_id="pbs_123"))
|
|
clear_state()
|
|
assert not (tmp_path / "state.json").exists()
|
|
|
|
def test_load_ignores_corrupt_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
state_file = tmp_path / "state.json"
|
|
monkeypatch.setattr("skyvern.cli.commands._state.STATE_FILE", state_file)
|
|
state_file.write_text("not-json")
|
|
assert load_state() is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _output.py
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOutput:
|
|
def test_json_envelope(self, capsys: pytest.CaptureFixture) -> None:
|
|
from skyvern.cli.commands._output import output
|
|
|
|
output({"key": "value"}, action="test", json_mode=True)
|
|
parsed = json.loads(capsys.readouterr().out)
|
|
assert parsed["ok"] is True
|
|
assert parsed["action"] == "test"
|
|
assert parsed["data"]["key"] == "value"
|
|
|
|
def test_json_error(self, capsys: pytest.CaptureFixture) -> None:
|
|
from skyvern.cli.commands._output import output_error
|
|
|
|
with pytest.raises(SystemExit, match="1"):
|
|
output_error("bad thing", hint="fix it", json_mode=True)
|
|
parsed = json.loads(capsys.readouterr().out)
|
|
assert parsed["ok"] is False
|
|
assert parsed["error"]["message"] == "bad thing"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Connection resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestResolveConnection:
|
|
def test_explicit_session_wins(self) -> None:
|
|
from skyvern.cli.commands.browser import _resolve_connection
|
|
|
|
result = _resolve_connection("pbs_explicit", None)
|
|
assert result.mode == "cloud"
|
|
assert result.session_id == "pbs_explicit"
|
|
assert result.cdp_url is None
|
|
|
|
def test_explicit_cdp_wins(self) -> None:
|
|
from skyvern.cli.commands.browser import _resolve_connection
|
|
|
|
result = _resolve_connection(None, "ws://localhost:9222/devtools/browser/abc")
|
|
assert result.mode == "cdp"
|
|
assert result.session_id is None
|
|
assert result.cdp_url == "ws://localhost:9222/devtools/browser/abc"
|
|
|
|
def test_rejects_both_connection_flags(self) -> None:
|
|
from skyvern.cli.commands.browser import _resolve_connection
|
|
|
|
with pytest.raises(typer.BadParameter, match="Pass only one of --session or --cdp"):
|
|
_resolve_connection("pbs_explicit", "ws://localhost:9222/devtools/browser/abc")
|
|
|
|
def test_state_fallback(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from skyvern.cli.commands.browser import _resolve_connection
|
|
|
|
_patch_state_dir(monkeypatch, tmp_path)
|
|
save_state(CLIState(session_id="pbs_from_state", mode="cloud"))
|
|
result = _resolve_connection(None, None)
|
|
assert result.mode == "cloud"
|
|
assert result.session_id == "pbs_from_state"
|
|
|
|
def test_state_fallback_cdp(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from skyvern.cli.commands.browser import _resolve_connection
|
|
|
|
_patch_state_dir(monkeypatch, tmp_path)
|
|
save_state(CLIState(session_id=None, cdp_url="ws://localhost:9222/devtools/browser/abc", mode="cdp"))
|
|
result = _resolve_connection(None, None)
|
|
assert result.mode == "cdp"
|
|
assert result.cdp_url == "ws://localhost:9222/devtools/browser/abc"
|
|
|
|
def test_no_session_raises(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from skyvern.cli.commands.browser import _resolve_connection
|
|
|
|
monkeypatch.setattr("skyvern.cli.commands._state.STATE_FILE", tmp_path / "nonexistent.json")
|
|
with pytest.raises(typer.BadParameter, match="No active browser connection"):
|
|
_resolve_connection(None, None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Browser command helpers and command behavior
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBrowserCommandGuards:
|
|
def test_resolve_ai_target_requires_selector_or_intent(self) -> None:
|
|
from skyvern.cli.commands.browser import _resolve_ai_target
|
|
from skyvern.cli.core.guards import GuardError
|
|
|
|
with pytest.raises(GuardError, match="Must provide intent, selector, or both"):
|
|
_resolve_ai_target(None, None, operation="click")
|
|
|
|
def test_validate_wait_state_rejects_invalid(self) -> None:
|
|
from skyvern.cli.commands.browser import _validate_wait_state
|
|
from skyvern.cli.core.guards import GuardError
|
|
|
|
with pytest.raises(GuardError, match="Invalid state"):
|
|
_validate_wait_state("bad-state")
|
|
|
|
|
|
class TestBrowserCommands:
|
|
def test_session_get_outputs_session_details(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture
|
|
) -> None:
|
|
from skyvern.cli.commands import browser as browser_cmd
|
|
|
|
session_obj = SimpleNamespace(
|
|
browser_session_id="pbs_123",
|
|
status="active",
|
|
started_at=datetime(2026, 2, 17, 12, 0, tzinfo=timezone.utc),
|
|
completed_at=None,
|
|
timeout=60,
|
|
runnable_id=None,
|
|
)
|
|
skyvern = SimpleNamespace(get_browser_session=AsyncMock(return_value=session_obj))
|
|
monkeypatch.setattr(browser_cmd, "get_skyvern", lambda: skyvern)
|
|
monkeypatch.setattr(browser_cmd, "load_state", lambda: CLIState(session_id="pbs_123", mode="cloud"))
|
|
|
|
browser_cmd.session_get(session="pbs_123", json_output=True)
|
|
|
|
parsed = json.loads(capsys.readouterr().out)
|
|
assert parsed["ok"] is True
|
|
assert parsed["action"] == "session_get"
|
|
assert parsed["data"]["session_id"] == "pbs_123"
|
|
assert parsed["data"]["is_current"] is True
|
|
|
|
def test_evaluate_blocks_password_js_before_connection(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture
|
|
) -> None:
|
|
from skyvern.cli.commands import browser as browser_cmd
|
|
|
|
monkeypatch.setattr(
|
|
browser_cmd,
|
|
"_resolve_connection",
|
|
lambda _session, _cdp: (_ for _ in ()).throw(AssertionError("should not resolve connection")),
|
|
)
|
|
|
|
with pytest.raises(SystemExit, match="1"):
|
|
browser_cmd.evaluate(
|
|
expression='document.querySelector("input[type=password]").value = ""', json_output=True
|
|
)
|
|
|
|
parsed = json.loads(capsys.readouterr().out)
|
|
assert parsed["ok"] is False
|
|
assert "Cannot set password field values" in parsed["error"]["message"]
|
|
|
|
def test_click_requires_target_before_connection(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture
|
|
) -> None:
|
|
from skyvern.cli.commands import browser as browser_cmd
|
|
|
|
monkeypatch.setattr(
|
|
browser_cmd,
|
|
"_resolve_connection",
|
|
lambda _session, _cdp: (_ for _ in ()).throw(AssertionError("should not resolve connection")),
|
|
)
|
|
|
|
with pytest.raises(SystemExit, match="1"):
|
|
browser_cmd.click(
|
|
intent=None,
|
|
selector=None,
|
|
session=None,
|
|
cdp=None,
|
|
timeout=30000,
|
|
button=None,
|
|
click_count=None,
|
|
json_output=True,
|
|
)
|
|
|
|
parsed = json.loads(capsys.readouterr().out)
|
|
assert parsed["ok"] is False
|
|
assert "Must provide intent, selector, or both" in parsed["error"]["message"]
|
|
|
|
def test_click_with_intent_uses_proactive_ai_mode(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture
|
|
) -> None:
|
|
from skyvern.cli.commands import browser as browser_cmd
|
|
|
|
page = MagicMock()
|
|
page.click = AsyncMock(return_value="xpath=//button[@id='submit']")
|
|
browser = SimpleNamespace(get_working_page=AsyncMock(return_value=page))
|
|
|
|
monkeypatch.setattr(
|
|
browser_cmd,
|
|
"_resolve_connection",
|
|
lambda _session, _cdp: browser_cmd.ConnectionTarget(mode="cloud", session_id="pbs_123"),
|
|
)
|
|
monkeypatch.setattr(browser_cmd, "_connect_browser", AsyncMock(return_value=browser))
|
|
|
|
browser_cmd.click(
|
|
intent="the Submit button",
|
|
selector=None,
|
|
session="pbs_123",
|
|
cdp=None,
|
|
timeout=30000,
|
|
button=None,
|
|
click_count=None,
|
|
json_output=True,
|
|
)
|
|
|
|
parsed = json.loads(capsys.readouterr().out)
|
|
assert parsed["ok"] is True
|
|
assert parsed["action"] == "click"
|
|
assert parsed["data"]["ai_mode"] == "proactive"
|
|
assert parsed["data"]["resolved_selector"] == "xpath=//button[@id='submit']"
|
|
|
|
def test_wait_rejects_invalid_state_before_connection(
|
|
self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture
|
|
) -> None:
|
|
from skyvern.cli.commands import browser as browser_cmd
|
|
|
|
monkeypatch.setattr(
|
|
browser_cmd,
|
|
"_resolve_connection",
|
|
lambda _session, _cdp: (_ for _ in ()).throw(AssertionError("should not resolve connection")),
|
|
)
|
|
|
|
with pytest.raises(SystemExit, match="1"):
|
|
browser_cmd.wait(state="bad-state", time_ms=1000, json_output=True)
|
|
|
|
parsed = json.loads(capsys.readouterr().out)
|
|
assert parsed["ok"] is False
|
|
assert "Invalid state" in parsed["error"]["message"]
|