84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from core.models import Bounds, Scene, SceneElement, utc_now
|
|
from workflow.conditions import UnknownConditionKindError, evaluate_condition
|
|
from workflow.models import ConditionSpec, WorkflowStepResult
|
|
|
|
|
|
def _scene(text: str) -> Scene:
|
|
return Scene(
|
|
width=10,
|
|
height=20,
|
|
elements=[
|
|
SceneElement(
|
|
id="label",
|
|
type="text",
|
|
text=text,
|
|
bounds=Bounds(1, 2, 3, 4),
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def test_scene_contains_text_evaluator_true_and_false() -> None:
|
|
spec = ConditionSpec("scene_contains_text", {"text": "hello"})
|
|
|
|
assert evaluate_condition(spec, scene=_scene("Hello world"), world_state=None)
|
|
assert not evaluate_condition(spec, scene=_scene("Goodbye"), world_state=None)
|
|
|
|
|
|
def test_world_variable_equals_evaluator_true_false_and_absent() -> None:
|
|
spec = ConditionSpec("world_variable_equals", {"name": "ready", "value": True})
|
|
|
|
assert evaluate_condition(
|
|
spec,
|
|
scene=None,
|
|
world_state=SimpleNamespace(variables={"ready": True}),
|
|
)
|
|
assert not evaluate_condition(
|
|
spec,
|
|
scene=None,
|
|
world_state=SimpleNamespace(variables={"ready": False}),
|
|
)
|
|
assert not evaluate_condition(spec, scene=None, world_state=None)
|
|
|
|
|
|
def test_elapsed_seconds_evaluator_true_and_false() -> None:
|
|
spec = ConditionSpec("elapsed_seconds", {"seconds": 1})
|
|
|
|
assert evaluate_condition(
|
|
spec,
|
|
scene=None,
|
|
world_state=None,
|
|
started_at=utc_now() - timedelta(seconds=2),
|
|
)
|
|
assert not evaluate_condition(
|
|
spec,
|
|
scene=None,
|
|
world_state=None,
|
|
started_at=utc_now(),
|
|
)
|
|
|
|
|
|
def test_step_result_success_evaluator_true_and_false() -> None:
|
|
spec = ConditionSpec("step_result_success", {"step_id": "first"})
|
|
results = [WorkflowStepResult("first", "planned_goal", True)]
|
|
|
|
assert evaluate_condition(spec, scene=None, world_state=None, step_results=results)
|
|
assert not evaluate_condition(
|
|
ConditionSpec("step_result_success", {"step_id": "missing"}),
|
|
scene=None,
|
|
world_state=None,
|
|
step_results=results,
|
|
)
|
|
|
|
|
|
def test_unknown_condition_kind_raises_distinguishable_error() -> None:
|
|
with pytest.raises(UnknownConditionKindError):
|
|
evaluate_condition(ConditionSpec("unknown"), scene=None, world_state=None)
|