80 lines
2.2 KiB
Python
80 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from skills_learning.models import FlowStep, FlowTemplateSkill, SkillMetadata
|
|
from workflow.skill_exec import (
|
|
SkillExecutionError,
|
|
resolve_skill_steps,
|
|
run_flow_template_skill,
|
|
)
|
|
|
|
|
|
def _skill() -> FlowTemplateSkill:
|
|
return FlowTemplateSkill(
|
|
metadata=SkillMetadata(
|
|
name="send message",
|
|
description="Send a message",
|
|
kind="flow_template",
|
|
),
|
|
steps=[
|
|
FlowStep("input_text", {"text": "{message}"}),
|
|
FlowStep("tap", {"x": 1, "y": 2}),
|
|
],
|
|
parameters={"message": {"type": "string"}},
|
|
)
|
|
|
|
|
|
def test_resolve_skill_steps_substitutes_valid_args() -> None:
|
|
resolved = resolve_skill_steps(_skill(), {"message": "hello"})
|
|
|
|
assert resolved == [
|
|
{"tool_name": "input_text", "args": {"text": "hello"}},
|
|
{"tool_name": "tap", "args": {"x": 1, "y": 2}},
|
|
]
|
|
|
|
|
|
def test_missing_required_parameter_fails_before_tool_call() -> None:
|
|
calls: list[str] = []
|
|
|
|
with pytest.raises(SkillExecutionError, match="missing required"):
|
|
run_flow_template_skill(
|
|
_skill(),
|
|
{},
|
|
tools={"input_text": lambda **kwargs: calls.append("input_text")},
|
|
)
|
|
|
|
assert calls == []
|
|
|
|
|
|
def test_non_flow_template_skill_is_rejected() -> None:
|
|
skill = FlowTemplateSkill(
|
|
metadata=SkillMetadata(
|
|
name="knowledge",
|
|
description="Not executable",
|
|
kind="knowledge",
|
|
)
|
|
)
|
|
|
|
with pytest.raises(SkillExecutionError, match="flow_template"):
|
|
resolve_skill_steps(skill, {})
|
|
|
|
|
|
def test_run_flow_template_skill_dispatches_resolved_tools() -> None:
|
|
calls: list[tuple[str, dict[str, object]]] = []
|
|
|
|
results = run_flow_template_skill(
|
|
_skill(),
|
|
{"message": "hello"},
|
|
tools={
|
|
"input_text": lambda **kwargs: calls.append(("input_text", kwargs)) or {"ok": True},
|
|
"tap": lambda **kwargs: calls.append(("tap", kwargs)) or {"ok": True},
|
|
},
|
|
)
|
|
|
|
assert [result.success for result in results] == [True, True]
|
|
assert calls == [
|
|
("input_text", {"text": "hello"}),
|
|
("tap", {"x": 1, "y": 2}),
|
|
]
|