85 lines
2.4 KiB
Python
85 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from workflow.models import (
|
|
BranchStep,
|
|
ConditionSpec,
|
|
PlannedGoalStep,
|
|
SkillInvocationStep,
|
|
WaitForConditionStep,
|
|
WorkflowDefinition,
|
|
)
|
|
|
|
|
|
def test_workflow_definition_accepts_heterogeneous_steps() -> None:
|
|
definition = WorkflowDefinition(
|
|
name="chat workflow",
|
|
entry_step_id="start",
|
|
steps=[
|
|
PlannedGoalStep("start", "open chat", next_step_id="skill"),
|
|
SkillInvocationStep("skill", "skill-1", {"message": "hello"}, "wait"),
|
|
WaitForConditionStep(
|
|
"wait",
|
|
ConditionSpec("scene_contains_text", {"text": "Done"}),
|
|
timeout_seconds=1,
|
|
poll_interval_seconds=0,
|
|
next_step_id="branch",
|
|
),
|
|
BranchStep(
|
|
"branch",
|
|
ConditionSpec("world_variable_equals", {"name": "ok", "value": True}),
|
|
on_true="start",
|
|
on_false="skill",
|
|
),
|
|
],
|
|
)
|
|
|
|
assert definition.step_by_id("skill").kind == "skill_invocation"
|
|
|
|
|
|
def test_workflow_definition_rejects_duplicate_step_id() -> None:
|
|
with pytest.raises(ValueError, match="duplicate"):
|
|
WorkflowDefinition(
|
|
name="bad",
|
|
entry_step_id="same",
|
|
steps=[
|
|
PlannedGoalStep("same", "one"),
|
|
PlannedGoalStep("same", "two"),
|
|
],
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"definition",
|
|
[
|
|
WorkflowDefinition,
|
|
],
|
|
)
|
|
def test_workflow_definition_rejects_dangling_references(definition) -> None:
|
|
with pytest.raises(ValueError, match="unknown|entry"):
|
|
definition(
|
|
name="bad next",
|
|
entry_step_id="start",
|
|
steps=[PlannedGoalStep("start", "one", next_step_id="missing")],
|
|
)
|
|
with pytest.raises(ValueError, match="unknown|entry"):
|
|
definition(
|
|
name="bad branch",
|
|
entry_step_id="branch",
|
|
steps=[
|
|
BranchStep(
|
|
"branch",
|
|
ConditionSpec("elapsed_seconds", {"seconds": 0}),
|
|
on_true="missing",
|
|
on_false="branch",
|
|
)
|
|
],
|
|
)
|
|
with pytest.raises(ValueError, match="entry"):
|
|
definition(
|
|
name="bad entry",
|
|
entry_step_id="missing",
|
|
steps=[PlannedGoalStep("start", "one")],
|
|
)
|