55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from runtime.executor import Executor, ExecutorConfig
|
|
from runtime.planner import PlannedStep
|
|
|
|
|
|
def test_executor_retries_until_transient_tool_succeeds() -> None:
|
|
calls = {"count": 0}
|
|
|
|
def flaky_tool() -> dict[str, bool]:
|
|
calls["count"] += 1
|
|
if calls["count"] < 3:
|
|
raise RuntimeError("transient")
|
|
return {"ok": True}
|
|
|
|
executor = Executor(
|
|
tools={"flaky": flaky_tool},
|
|
config=ExecutorConfig(max_retries=3, backoff_seconds=0),
|
|
)
|
|
|
|
result = executor.execute(PlannedStep(action="flaky", description="retry"))
|
|
|
|
assert result.success is True
|
|
assert result.attempts == 3
|
|
assert result.result == {"ok": True}
|
|
|
|
|
|
def test_executor_keeps_action_metadata_out_of_device_tool_arguments() -> None:
|
|
calls: list[tuple[int, int]] = []
|
|
|
|
def tap(*, x: int, y: int) -> dict[str, bool]:
|
|
calls.append((x, y))
|
|
return {"ok": True}
|
|
|
|
executor = Executor(
|
|
tools={"tap": tap},
|
|
config=ExecutorConfig(max_retries=1, backoff_seconds=0),
|
|
)
|
|
step = PlannedStep(
|
|
action="tap",
|
|
description="Open settings.",
|
|
args={"x": 12, "y": 34},
|
|
purpose="Open settings.",
|
|
expected_outcome="The settings page is visible.",
|
|
)
|
|
|
|
result = executor.execute(step)
|
|
|
|
assert result.success is True
|
|
assert calls == [(12, 34)]
|
|
assert result.to_dict()["step"]["purpose"] == "Open settings."
|
|
assert (
|
|
result.to_dict()["step"]["expected_outcome"] == "The settings page is visible."
|
|
)
|