brent.edwards
f0815dd126
CI / lint (pull_request) Has been cancelled
CI / typecheck (pull_request) Has been cancelled
CI / security (pull_request) Has been cancelled
CI / quality (pull_request) Has been cancelled
CI / unit_tests (pull_request) Has been cancelled
CI / integration_tests (pull_request) Has been cancelled
CI / coverage (pull_request) Has been cancelled
CI / build (pull_request) Has been cancelled
CI / docker (pull_request) Has been cancelled
980 lines
35 KiB
Python
980 lines
35 KiB
Python
"""Step definitions for Skill domain model and resolution tests."""
|
|
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.domain.models.core.skill import (
|
|
ResolvedToolEntry,
|
|
Skill,
|
|
SkillAgentSource,
|
|
SkillInclude,
|
|
SkillInlineTool,
|
|
SkillMcpSource,
|
|
SkillResolver,
|
|
SkillToolRef,
|
|
)
|
|
from cleveragents.domain.models.core.tool import (
|
|
ToolCapability,
|
|
ToolSource,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_skill(**overrides: Any) -> Skill:
|
|
"""Create a Skill with sensible defaults, allowing overrides."""
|
|
defaults: dict[str, Any] = {
|
|
"name": "local/test-skill",
|
|
"description": "A test skill",
|
|
}
|
|
defaults.update(overrides)
|
|
return Skill(**defaults)
|
|
|
|
|
|
def _skill_config(**overrides: Any) -> dict[str, Any]:
|
|
"""Return a minimal valid config dict for Skill.from_config."""
|
|
cfg: dict[str, Any] = {
|
|
"name": "local/my-skill",
|
|
"description": "A skill from config",
|
|
}
|
|
cfg.update(overrides)
|
|
return cfg
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Skill Creation Steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a skill_model with name "{name}" and description "{desc}"')
|
|
def skill_model_create_basic(context: Context, name: str, desc: str) -> None:
|
|
"""Create a skill with name and description."""
|
|
context.skill_model = Skill(name=name, description=desc)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@when('I create a skill_model with tool refs "{refs}"')
|
|
def skill_model_create_with_refs(context: Context, refs: str) -> None:
|
|
"""Create a skill with tool refs."""
|
|
ref_list = [r.strip() for r in refs.split(",") if r.strip()]
|
|
context.skill_model = _make_skill(tool_refs=ref_list)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@when("I create a skill_model with all fields")
|
|
def skill_model_create_all_fields(context: Context) -> None:
|
|
"""Create a skill with all fields populated."""
|
|
context.skill_model = Skill(
|
|
name="local/full-skill",
|
|
description="Full skill",
|
|
tool_refs=["local/edit-file", "local/read-file"],
|
|
includes=[SkillInclude(name="local/basic-tools")],
|
|
anonymous_tools=[
|
|
SkillInlineTool(
|
|
description="Inline helper",
|
|
source=ToolSource.CUSTOM,
|
|
code="print('hello')",
|
|
),
|
|
],
|
|
mcp_servers=[
|
|
SkillMcpSource(
|
|
server="npx @mcp/server-fs",
|
|
tools=["read_file", "write_file"],
|
|
),
|
|
],
|
|
agent_skills=[SkillAgentSource(path="./skills/code-review")],
|
|
overrides={"local/edit-file": {"timeout": 600}},
|
|
)
|
|
context.skill_model_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Skill Assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the skill_model should be created")
|
|
def skill_model_check_created(context: Context) -> None:
|
|
"""Verify the skill was created."""
|
|
assert context.skill_model is not None, "Skill should be created"
|
|
|
|
|
|
@then('the skill_model name should be "{expected}"')
|
|
def skill_model_check_name(context: Context, expected: str) -> None:
|
|
"""Check skill name."""
|
|
assert context.skill_model.name == expected, (
|
|
f"Expected name '{expected}', got '{context.skill_model.name}'"
|
|
)
|
|
|
|
|
|
@then('the skill_model description should be "{expected}"')
|
|
def skill_model_check_description(context: Context, expected: str) -> None:
|
|
"""Check skill description."""
|
|
assert context.skill_model.description == expected
|
|
|
|
|
|
@then('the skill_model namespace should be "{expected}"')
|
|
def skill_model_check_namespace(context: Context, expected: str) -> None:
|
|
"""Check skill namespace property."""
|
|
assert context.skill_model.namespace == expected
|
|
|
|
|
|
@then('the skill_model short_name should be "{expected}"')
|
|
def skill_model_check_short_name(context: Context, expected: str) -> None:
|
|
"""Check skill short_name property."""
|
|
assert context.skill_model.short_name == expected
|
|
|
|
|
|
@then("the skill_model should have {count:d} tool refs")
|
|
def skill_model_check_ref_count(context: Context, count: int) -> None:
|
|
"""Check skill tool_refs count."""
|
|
actual = len(context.skill_model.tool_refs)
|
|
assert actual == count, f"Expected {count} tool refs, got {actual}"
|
|
|
|
|
|
@then("the skill_model should have {count:d} include")
|
|
def skill_model_check_include_count(context: Context, count: int) -> None:
|
|
"""Check skill includes count."""
|
|
actual = len(context.skill_model.includes)
|
|
assert actual == count, f"Expected {count} includes, got {actual}"
|
|
|
|
|
|
@then("the skill_model should have {count:d} anonymous tool")
|
|
def skill_model_check_anon_count(context: Context, count: int) -> None:
|
|
"""Check skill anonymous_tools count."""
|
|
actual = len(context.skill_model.anonymous_tools)
|
|
assert actual == count, f"Expected {count} anonymous tools, got {actual}"
|
|
|
|
|
|
@then("the skill_model should have {count:d} mcp server")
|
|
def skill_model_check_mcp_count(context: Context, count: int) -> None:
|
|
"""Check skill mcp_servers count."""
|
|
actual = len(context.skill_model.mcp_servers)
|
|
assert actual == count, f"Expected {count} mcp servers, got {actual}"
|
|
|
|
|
|
@then("the skill_model should have {count:d} agent skill")
|
|
def skill_model_check_agent_count(context: Context, count: int) -> None:
|
|
"""Check skill agent_skills count."""
|
|
actual = len(context.skill_model.agent_skills)
|
|
assert actual == count, f"Expected {count} agent skills, got {actual}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Name Validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I try to create a skill_model with invalid name "{name}"')
|
|
def skill_model_try_invalid_name(context: Context, name: str) -> None:
|
|
"""Attempt to create a skill with an invalid name."""
|
|
context.skill_model_error = None
|
|
context.skill_model = None
|
|
try:
|
|
context.skill_model = _make_skill(name=name)
|
|
except ValidationError as e:
|
|
context.skill_model_error = e
|
|
|
|
|
|
@when("I try to create a skill_model with empty description")
|
|
def skill_model_try_empty_desc(context: Context) -> None:
|
|
"""Attempt to create a skill with empty description."""
|
|
context.skill_model_error = None
|
|
context.skill_model = None
|
|
try:
|
|
context.skill_model = Skill(name="local/bad", description="")
|
|
except ValidationError as e:
|
|
context.skill_model_error = e
|
|
|
|
|
|
@then("a skill_model validation error should be raised")
|
|
def skill_model_check_validation_error(context: Context) -> None:
|
|
"""Verify that a validation error was raised."""
|
|
assert context.skill_model_error is not None, (
|
|
"Expected a validation error to be raised"
|
|
)
|
|
|
|
|
|
@then('the skill_model error should mention "{text}"')
|
|
def skill_model_check_error_message(context: Context, text: str) -> None:
|
|
"""Check the error message contains expected text."""
|
|
error_str = str(context.skill_model_error)
|
|
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Include Resolution (flat)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a skill_resolver registry with "{name}" containing refs "{refs}"')
|
|
def skill_resolver_registry_with_refs(context: Context, name: str, refs: str) -> None:
|
|
"""Set up a registry with a skill containing tool refs."""
|
|
if not hasattr(context, "skill_registry"):
|
|
context.skill_registry = {}
|
|
ref_list = [r.strip() for r in refs.split(",") if r.strip()]
|
|
context.skill_registry[name] = Skill(
|
|
name=name,
|
|
description=f"Skill {name}",
|
|
tool_refs=ref_list,
|
|
)
|
|
|
|
|
|
@given(
|
|
'a skill_resolver root skill "{name}" including "{includes}" with own refs "{refs}"'
|
|
)
|
|
def skill_resolver_root_with_includes(
|
|
context: Context, name: str, includes: str, refs: str
|
|
) -> None:
|
|
"""Set up the root skill to resolve."""
|
|
if not hasattr(context, "skill_registry"):
|
|
context.skill_registry = {}
|
|
include_list = [
|
|
SkillInclude(name=i.strip()) for i in includes.split(",") if i.strip()
|
|
]
|
|
ref_list = [r.strip() for r in refs.split(",") if r.strip()]
|
|
context.skill_root = Skill(
|
|
name=name,
|
|
description=f"Root skill {name}",
|
|
tool_refs=ref_list,
|
|
includes=include_list,
|
|
)
|
|
context.skill_registry[name] = context.skill_root
|
|
|
|
|
|
@given('a skill_resolver root skill "{name}" including "{includes}" with no own refs')
|
|
def skill_resolver_root_no_refs(context: Context, name: str, includes: str) -> None:
|
|
"""Set up the root skill with includes and no own refs."""
|
|
if not hasattr(context, "skill_registry"):
|
|
context.skill_registry = {}
|
|
include_list = [
|
|
SkillInclude(name=i.strip()) for i in includes.split(",") if i.strip()
|
|
]
|
|
context.skill_root = Skill(
|
|
name=name,
|
|
description=f"Root skill {name}",
|
|
includes=include_list,
|
|
)
|
|
context.skill_registry[name] = context.skill_root
|
|
|
|
|
|
@when("I resolve the skill_resolver root skill")
|
|
def skill_resolver_resolve_root(context: Context) -> None:
|
|
"""Resolve the root skill."""
|
|
resolver = SkillResolver()
|
|
context.skill_resolved = resolver.resolve_tools(
|
|
context.skill_root, context.skill_registry
|
|
)
|
|
context.skill_resolver_error = None
|
|
|
|
|
|
@then("the skill_resolver result should have {count:d} tools")
|
|
def skill_resolver_check_tool_count(context: Context, count: int) -> None:
|
|
"""Check resolved tool count."""
|
|
actual = len(context.skill_resolved)
|
|
assert actual == count, (
|
|
f"Expected {count} tools, got {actual}: "
|
|
f"{[e.name for e in context.skill_resolved]}"
|
|
)
|
|
|
|
|
|
@then('the skill_resolver result should contain tool "{name}"')
|
|
def skill_resolver_check_contains_tool(context: Context, name: str) -> None:
|
|
"""Check resolved tools contain a specific tool."""
|
|
names = [e.name for e in context.skill_resolved]
|
|
assert name in names, f"Expected tool '{name}' in {names}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Include Resolution (nested)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
'a skill_resolver registry with "{name}" including "{includes}" and containing refs "{refs}"'
|
|
)
|
|
def skill_resolver_registry_with_includes_and_refs(
|
|
context: Context, name: str, includes: str, refs: str
|
|
) -> None:
|
|
"""Set up a registry skill with includes and refs."""
|
|
if not hasattr(context, "skill_registry"):
|
|
context.skill_registry = {}
|
|
include_list = [
|
|
SkillInclude(name=i.strip()) for i in includes.split(",") if i.strip()
|
|
]
|
|
ref_list = [r.strip() for r in refs.split(",") if r.strip()]
|
|
context.skill_registry[name] = Skill(
|
|
name=name,
|
|
description=f"Skill {name}",
|
|
tool_refs=ref_list,
|
|
includes=include_list,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cycle Detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a skill_resolver registry with "{name}" including "{includes}"')
|
|
def skill_resolver_registry_with_includes(
|
|
context: Context, name: str, includes: str
|
|
) -> None:
|
|
"""Set up a registry skill that includes another."""
|
|
if not hasattr(context, "skill_registry"):
|
|
context.skill_registry = {}
|
|
include_list = [
|
|
SkillInclude(name=i.strip()) for i in includes.split(",") if i.strip()
|
|
]
|
|
context.skill_registry[name] = Skill(
|
|
name=name,
|
|
description=f"Skill {name}",
|
|
includes=include_list,
|
|
)
|
|
|
|
|
|
@when('I try to resolve skill_resolver "{name}"')
|
|
def skill_resolver_try_resolve(context: Context, name: str) -> None:
|
|
"""Try to resolve a skill and capture any error."""
|
|
if not hasattr(context, "skill_registry"):
|
|
context.skill_registry = {}
|
|
# If root skill is set in registry, use it; otherwise find it
|
|
skill = context.skill_registry.get(name)
|
|
if skill is None:
|
|
# Must be set via root skill step
|
|
skill = getattr(context, "skill_root", None)
|
|
assert skill is not None, f"Skill '{name}' not found in registry or root"
|
|
resolver = SkillResolver()
|
|
context.skill_resolver_error = None
|
|
context.skill_resolved = None
|
|
try:
|
|
context.skill_resolved = resolver.resolve_tools(skill, context.skill_registry)
|
|
except ValueError as e:
|
|
context.skill_resolver_error = e
|
|
|
|
|
|
@then("a skill_resolver cycle error should be raised")
|
|
def skill_resolver_check_cycle_error(context: Context) -> None:
|
|
"""Verify that a cycle error was raised."""
|
|
assert context.skill_resolver_error is not None, (
|
|
"Expected a cycle error to be raised"
|
|
)
|
|
assert "ycle" in str(context.skill_resolver_error), (
|
|
f"Expected cycle error, got: {context.skill_resolver_error}"
|
|
)
|
|
|
|
|
|
@then('the skill_resolver error should mention "{text}"')
|
|
def skill_resolver_check_error_mention(context: Context, text: str) -> None:
|
|
"""Check the resolver error mentions text."""
|
|
error_str = str(context.skill_resolver_error)
|
|
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# De-duplication
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the skill_resolver tool "{name}" should come from "{source}"')
|
|
def skill_resolver_check_tool_source(context: Context, name: str, source: str) -> None:
|
|
"""Check which skill contributed a tool."""
|
|
for entry in context.skill_resolved:
|
|
if entry.name == name:
|
|
assert entry.source_skill == source, (
|
|
f"Expected tool '{name}' from '{source}', got '{entry.source_skill}'"
|
|
)
|
|
return
|
|
raise AssertionError(
|
|
f"Tool '{name}' not found in resolved: "
|
|
f"{[e.name for e in context.skill_resolved]}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Override Application
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a skill_model with overrides for "{tool_name}"')
|
|
def skill_model_create_with_overrides(context: Context, tool_name: str) -> None:
|
|
"""Create a skill with per-tool overrides."""
|
|
context.skill_model = Skill(
|
|
name="local/override-skill",
|
|
description="Skill with overrides",
|
|
tool_refs=[tool_name],
|
|
overrides={tool_name: {"timeout": 600}},
|
|
)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@when("I resolve the overridden skill_model")
|
|
def skill_model_resolve_overridden(context: Context) -> None:
|
|
"""Resolve a skill with overrides."""
|
|
resolver = SkillResolver()
|
|
context.skill_resolved = resolver.resolve_tools(context.skill_model, {})
|
|
|
|
|
|
@then('the skill_resolver tool "{name}" should have overrides')
|
|
def skill_resolver_check_tool_overrides(context: Context, name: str) -> None:
|
|
"""Check that a resolved tool has overrides."""
|
|
for entry in context.skill_resolved:
|
|
if entry.name == name:
|
|
assert entry.overrides, f"Expected overrides on tool '{name}', got empty"
|
|
return
|
|
raise AssertionError(f"Tool '{name}' not found in resolved")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Empty Skill
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I resolve the empty skill_model")
|
|
def skill_model_resolve_empty(context: Context) -> None:
|
|
"""Resolve an empty skill."""
|
|
resolver = SkillResolver()
|
|
context.skill_resolved = resolver.resolve_tools(context.skill_model, {})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# from_config Loading
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I load a skill_model from config with name "{name}"')
|
|
def skill_model_load_from_config(context: Context, name: str) -> None:
|
|
"""Load a skill from a config dict."""
|
|
config = _skill_config(name=name)
|
|
context.skill_model = Skill.from_config(config)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@when('I try to load a skill_model from config missing "{field}"')
|
|
def skill_model_try_load_missing(context: Context, field: str) -> None:
|
|
"""Try loading a skill config with missing field."""
|
|
config = _skill_config()
|
|
del config[field]
|
|
context.skill_model_config_error = None
|
|
try:
|
|
Skill.from_config(config)
|
|
except ValueError as e:
|
|
context.skill_model_config_error = e
|
|
|
|
|
|
@then('a skill_model config error should be raised with "{field}"')
|
|
def skill_model_check_config_error(context: Context, field: str) -> None:
|
|
"""Check that a config error mentioning field was raised."""
|
|
assert context.skill_model_config_error is not None, (
|
|
f"Expected ValueError for missing '{field}'"
|
|
)
|
|
assert field in str(context.skill_model_config_error), (
|
|
f"Expected error to mention '{field}', got: {context.skill_model_config_error}"
|
|
)
|
|
|
|
|
|
@when("I load a skill_model from full config")
|
|
def skill_model_load_full_config(context: Context) -> None:
|
|
"""Load a skill from a full config with all sections."""
|
|
config = {
|
|
"name": "local/full-config",
|
|
"description": "Full config skill",
|
|
"tools": ["local/edit-file", "local/read-file"],
|
|
"includes": [{"name": "local/basic-tools"}],
|
|
"anonymous_tools": [
|
|
{
|
|
"description": "Inline helper",
|
|
"source": "custom",
|
|
"code": "print('hi')",
|
|
},
|
|
],
|
|
"mcp_servers": [
|
|
{
|
|
"server": "npx @mcp/server-fs",
|
|
"tools": ["read_file"],
|
|
},
|
|
],
|
|
"agent_skills": [{"path": "./skills/review"}],
|
|
"overrides": {"local/edit-file": {"timeout": 600}},
|
|
}
|
|
context.skill_model = Skill.from_config(config)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@when("I load a skill_model from config with string includes")
|
|
def skill_model_load_config_string_includes(context: Context) -> None:
|
|
"""Load a skill from config where includes are plain strings."""
|
|
config = {
|
|
"name": "local/string-inc",
|
|
"description": "String includes",
|
|
"includes": ["local/basic-tools"],
|
|
}
|
|
context.skill_model = Skill.from_config(config)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@when("I load a skill_model from config with string agent skills")
|
|
def skill_model_load_config_string_agents(context: Context) -> None:
|
|
"""Load a skill from config where agent_skills are plain strings."""
|
|
config = {
|
|
"name": "local/string-agents",
|
|
"description": "String agents",
|
|
"agent_skills": ["./skills/review"],
|
|
}
|
|
context.skill_model = Skill.from_config(config)
|
|
context.skill_model_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# as_cli_dict
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a skill_model and call as_cli_dict")
|
|
def skill_model_create_and_cli_dict(context: Context) -> None:
|
|
"""Create a skill and get CLI dict."""
|
|
context.skill_model = _make_skill()
|
|
context.skill_cli_dict = context.skill_model.as_cli_dict()
|
|
|
|
|
|
@when("I create a skill_model with all fields and call as_cli_dict")
|
|
def skill_model_create_all_cli_dict(context: Context) -> None:
|
|
"""Create a skill with all fields and get CLI dict."""
|
|
context.skill_model = Skill(
|
|
name="local/full-cli",
|
|
description="Full CLI skill",
|
|
tool_refs=["local/edit-file"],
|
|
includes=[SkillInclude(name="local/base", overrides={"x": 1})],
|
|
anonymous_tools=[
|
|
SkillInlineTool(
|
|
description="Inline",
|
|
source=ToolSource.CUSTOM,
|
|
code="x = 1",
|
|
),
|
|
],
|
|
mcp_servers=[
|
|
SkillMcpSource(server="npx @mcp/fs", tools=["read"], env={"K": "V"}),
|
|
],
|
|
agent_skills=[SkillAgentSource(path="./skills/r")],
|
|
overrides={"local/edit-file": {"timeout": 600}},
|
|
)
|
|
context.skill_cli_dict = context.skill_model.as_cli_dict()
|
|
|
|
|
|
@when("I get the skill_model cli dict")
|
|
def skill_model_get_cli_dict(context: Context) -> None:
|
|
"""Get CLI dict from the current skill model."""
|
|
context.skill_cli_dict = context.skill_model.as_cli_dict()
|
|
|
|
|
|
@then('the skill_model cli dict should have key "{key}"')
|
|
def skill_model_check_cli_dict_key(context: Context, key: str) -> None:
|
|
"""Check CLI dict has expected key."""
|
|
assert key in context.skill_cli_dict, (
|
|
f"Expected key '{key}' in cli dict, keys: {list(context.skill_cli_dict)}"
|
|
)
|
|
|
|
|
|
@then('the skill_model cli dict should not have key "{key}"')
|
|
def skill_model_check_cli_dict_no_key(context: Context, key: str) -> None:
|
|
"""Check CLI dict does not have a key."""
|
|
assert key not in context.skill_cli_dict, f"Unexpected key '{key}' in cli dict"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Inline Tool Validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a skill_model inline tool with source "{source}" and code "{code}"')
|
|
def skill_model_create_inline_tool(context: Context, source: str, code: str) -> None:
|
|
"""Create a SkillInlineTool."""
|
|
context.skill_inline_tool = SkillInlineTool(
|
|
description="Inline test tool",
|
|
source=ToolSource(source),
|
|
code=code,
|
|
)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@then("the skill_model inline tool should be created")
|
|
def skill_model_check_inline_created(context: Context) -> None:
|
|
"""Verify inline tool was created."""
|
|
assert context.skill_inline_tool is not None
|
|
|
|
|
|
@then("the skill_model inline tool timeout should be {expected:d}")
|
|
def skill_model_check_inline_timeout(context: Context, expected: int) -> None:
|
|
"""Check inline tool timeout."""
|
|
assert context.skill_inline_tool.timeout == expected
|
|
|
|
|
|
@when("I try to create a skill_model inline tool with timeout {timeout:d}")
|
|
def skill_model_try_inline_bad_timeout(context: Context, timeout: int) -> None:
|
|
"""Attempt to create an inline tool with invalid timeout."""
|
|
context.skill_model_error = None
|
|
try:
|
|
SkillInlineTool(
|
|
description="Bad timeout",
|
|
source=ToolSource.CUSTOM,
|
|
code="x = 1",
|
|
timeout=timeout,
|
|
)
|
|
except ValidationError as e:
|
|
context.skill_model_error = e
|
|
|
|
|
|
@when("I try to create a skill_model inline tool with empty description")
|
|
def skill_model_try_inline_empty_desc(context: Context) -> None:
|
|
"""Attempt to create an inline tool with empty description."""
|
|
context.skill_model_error = None
|
|
try:
|
|
SkillInlineTool(
|
|
description="",
|
|
source=ToolSource.CUSTOM,
|
|
code="x = 1",
|
|
)
|
|
except ValidationError as e:
|
|
context.skill_model_error = e
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MCP Source Validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a skill_model mcp source with server "{server}" and tools "{tools}"')
|
|
def skill_model_create_mcp_source(context: Context, server: str, tools: str) -> None:
|
|
"""Create a SkillMcpSource with tools."""
|
|
tool_list = [t.strip() for t in tools.split(",") if t.strip()]
|
|
context.skill_mcp_source = SkillMcpSource(server=server, tools=tool_list)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@when('I create a skill_model mcp source with server "{server}" and no tools')
|
|
def skill_model_create_mcp_no_tools(context: Context, server: str) -> None:
|
|
"""Create a SkillMcpSource without specific tools (all tools)."""
|
|
context.skill_mcp_source = SkillMcpSource(server=server)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@when("I create a skill_model mcp source with env vars")
|
|
def skill_model_create_mcp_env(context: Context) -> None:
|
|
"""Create a SkillMcpSource with env vars."""
|
|
context.skill_mcp_source = SkillMcpSource(
|
|
server="npx @mcp/server",
|
|
env={"API_KEY": "test123"},
|
|
)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@when("I try to create a skill_model mcp source with empty server")
|
|
def skill_model_try_mcp_empty_server(context: Context) -> None:
|
|
"""Attempt to create MCP source with empty server."""
|
|
context.skill_model_error = None
|
|
try:
|
|
SkillMcpSource(server="")
|
|
except ValidationError as e:
|
|
context.skill_model_error = e
|
|
|
|
|
|
@then("the skill_model mcp source should be created")
|
|
def skill_model_check_mcp_created(context: Context) -> None:
|
|
"""Verify MCP source was created."""
|
|
assert context.skill_mcp_source is not None
|
|
|
|
|
|
@then("the skill_model mcp source should have {count:d} tools")
|
|
def skill_model_check_mcp_tool_count(context: Context, count: int) -> None:
|
|
"""Check MCP source tool count."""
|
|
assert context.skill_mcp_source.tools is not None
|
|
actual = len(context.skill_mcp_source.tools)
|
|
assert actual == count, f"Expected {count} tools, got {actual}"
|
|
|
|
|
|
@then("the skill_model mcp source tools should be none")
|
|
def skill_model_check_mcp_tools_none(context: Context) -> None:
|
|
"""Check MCP source tools is None."""
|
|
assert context.skill_mcp_source.tools is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Agent Skill Source Validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a skill_model agent source with path "{path}"')
|
|
def skill_model_create_agent_source(context: Context, path: str) -> None:
|
|
"""Create a SkillAgentSource."""
|
|
context.skill_agent_source = SkillAgentSource(path=path)
|
|
context.skill_model_error = None
|
|
|
|
|
|
@when("I try to create a skill_model agent source with empty path")
|
|
def skill_model_try_agent_empty_path(context: Context) -> None:
|
|
"""Attempt to create agent source with empty path."""
|
|
context.skill_model_error = None
|
|
try:
|
|
SkillAgentSource(path="")
|
|
except ValidationError as e:
|
|
context.skill_model_error = e
|
|
|
|
|
|
@then("the skill_model agent source should be created")
|
|
def skill_model_check_agent_created(context: Context) -> None:
|
|
"""Verify agent source was created."""
|
|
assert context.skill_agent_source is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CapabilitySummary Computation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I compute skill_model capability summary for a skill with inline tools")
|
|
def skill_model_compute_summary(context: Context) -> None:
|
|
"""Compute capability summary for a skill with various tools."""
|
|
skill = Skill(
|
|
name="local/summary-skill",
|
|
description="Summary test",
|
|
tool_refs=["local/some-ref"],
|
|
anonymous_tools=[
|
|
SkillInlineTool(
|
|
description="Read-only tool",
|
|
source=ToolSource.CUSTOM,
|
|
code="x = 1",
|
|
capability=ToolCapability(read_only=True),
|
|
),
|
|
SkillInlineTool(
|
|
description="Write tool",
|
|
source=ToolSource.CUSTOM,
|
|
code="x = 2",
|
|
capability=ToolCapability(
|
|
writes=True,
|
|
side_effects=["filesystem"],
|
|
),
|
|
),
|
|
],
|
|
mcp_servers=[SkillMcpSource(server="npx @mcp/fs")],
|
|
agent_skills=[SkillAgentSource(path="./skills/r")],
|
|
)
|
|
resolver = SkillResolver()
|
|
resolved = resolver.resolve_tools(skill, {})
|
|
context.skill_summary = resolver.compute_capability_summary(skill, resolved)
|
|
|
|
|
|
@when("I compute skill_model capability summary for an empty skill")
|
|
def skill_model_compute_empty_summary(context: Context) -> None:
|
|
"""Compute capability summary for an empty skill."""
|
|
skill = Skill(
|
|
name="local/empty-summary",
|
|
description="Empty summary test",
|
|
)
|
|
resolver = SkillResolver()
|
|
resolved = resolver.resolve_tools(skill, {})
|
|
context.skill_summary = resolver.compute_capability_summary(skill, resolved)
|
|
|
|
|
|
@then("the skill_model summary total tools should be {expected:d}")
|
|
def skill_model_check_summary_total(context: Context, expected: int) -> None:
|
|
"""Check summary total_tools."""
|
|
assert context.skill_summary.total_tools == expected
|
|
|
|
|
|
@then("the skill_model summary read_only_tools should be {expected:d}")
|
|
def skill_model_check_summary_readonly(context: Context, expected: int) -> None:
|
|
"""Check summary read_only_tools."""
|
|
assert context.skill_summary.read_only_tools == expected
|
|
|
|
|
|
@then("the skill_model summary write_tools should be {expected:d}")
|
|
def skill_model_check_summary_writes(context: Context, expected: int) -> None:
|
|
"""Check summary write_tools."""
|
|
assert context.skill_summary.write_tools == expected
|
|
|
|
|
|
@then("the skill_model summary has_side_effects should be true")
|
|
def skill_model_check_summary_side_effects(context: Context) -> None:
|
|
"""Check summary has_side_effects."""
|
|
assert context.skill_summary.has_side_effects is True
|
|
|
|
|
|
@then("the skill_model summary mcp_sources should be {expected:d}")
|
|
def skill_model_check_summary_mcp(context: Context, expected: int) -> None:
|
|
"""Check summary mcp_sources."""
|
|
assert context.skill_summary.mcp_sources == expected
|
|
|
|
|
|
@then("the skill_model summary agent_skill_sources should be {expected:d}")
|
|
def skill_model_check_summary_agents(context: Context, expected: int) -> None:
|
|
"""Check summary agent_skill_sources."""
|
|
assert context.skill_summary.agent_skill_sources == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SkillToolRef
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a skill_model tool ref with name "{name}"')
|
|
def skill_model_create_tool_ref(context: Context, name: str) -> None:
|
|
"""Create a SkillToolRef."""
|
|
context.skill_tool_ref = SkillToolRef(name=name)
|
|
|
|
|
|
@then("the skill_model tool ref should be created")
|
|
def skill_model_check_tool_ref(context: Context) -> None:
|
|
"""Verify tool ref was created."""
|
|
assert context.skill_tool_ref is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SkillInclude
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a skill_model include with name "{name}" and overrides')
|
|
def skill_model_create_include_with_overrides(context: Context, name: str) -> None:
|
|
"""Create a SkillInclude with overrides."""
|
|
context.skill_include = SkillInclude(name=name, overrides={"timeout": 600})
|
|
|
|
|
|
@when('I create a skill_model include with name "{name}" without overrides')
|
|
def skill_model_create_include_no_overrides(context: Context, name: str) -> None:
|
|
"""Create a SkillInclude without overrides."""
|
|
context.skill_include = SkillInclude(name=name)
|
|
|
|
|
|
@then("the skill_model include should be created")
|
|
def skill_model_check_include_created(context: Context) -> None:
|
|
"""Verify include was created."""
|
|
assert context.skill_include is not None
|
|
|
|
|
|
@then("the skill_model include should have overrides")
|
|
def skill_model_check_include_overrides(context: Context) -> None:
|
|
"""Check include has overrides."""
|
|
assert context.skill_include.overrides is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Missing included skill
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("a skill_resolver missing skill error should be raised")
|
|
def skill_resolver_check_missing_error(context: Context) -> None:
|
|
"""Verify a missing skill error was raised."""
|
|
assert context.skill_resolver_error is not None, "Expected a missing skill error"
|
|
assert "not found" in str(context.skill_resolver_error), (
|
|
f"Expected 'not found' in error, got: {context.skill_resolver_error}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MCP and agent skill resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create and resolve a skill_model with mcp and agent sources")
|
|
def skill_model_resolve_mcp_agent(context: Context) -> None:
|
|
"""Create and resolve a skill with MCP and agent sources."""
|
|
skill = Skill(
|
|
name="local/mcp-agent",
|
|
description="MCP and agent test",
|
|
tool_refs=["local/base-tool"],
|
|
mcp_servers=[
|
|
SkillMcpSource(
|
|
server="npx @mcp/fs",
|
|
tools=["read_file"],
|
|
),
|
|
],
|
|
agent_skills=[SkillAgentSource(path="./skills/review")],
|
|
)
|
|
resolver = SkillResolver()
|
|
context.skill_resolved = resolver.resolve_tools(skill, {})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ResolvedToolEntry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a skill_resolver resolved tool entry")
|
|
def skill_resolver_create_entry(context: Context) -> None:
|
|
"""Create a ResolvedToolEntry."""
|
|
context.skill_resolved_entry = ResolvedToolEntry(
|
|
name="local/test-tool",
|
|
source_skill="local/my-skill",
|
|
)
|
|
|
|
|
|
@then("the skill_resolver entry should be created")
|
|
def skill_resolver_check_entry(context: Context) -> None:
|
|
"""Verify entry was created."""
|
|
assert context.skill_resolved_entry is not None
|
|
assert context.skill_resolved_entry.name == "local/test-tool"
|
|
assert context.skill_resolved_entry.is_inline is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Robot Smoke Mirror Steps (Behave equivalents of robot/skill_resolution.robot)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the skill_resolver tool at index {index:d} should be "{expected}"')
|
|
def skill_resolver_check_tool_at_index(
|
|
context: Context, index: int, expected: str
|
|
) -> None:
|
|
"""Check the tool name at a specific index in the resolved list."""
|
|
resolved = context.skill_resolved
|
|
assert index < len(resolved), (
|
|
f"Index {index} out of range (resolved has {len(resolved)} tools)"
|
|
)
|
|
actual = resolved[index].name
|
|
assert actual == expected, (
|
|
f"Expected tool at index {index} to be '{expected}', got '{actual}'"
|
|
)
|
|
|
|
|
|
@when("I create and resolve a skill_model with inline tools")
|
|
def skill_model_resolve_inline_tools(context: Context) -> None:
|
|
"""Create and resolve a skill with inline tools."""
|
|
skill = Skill(
|
|
name="local/inline-test",
|
|
description="Inline tool test",
|
|
tool_refs=["local/ref-tool"],
|
|
anonymous_tools=[
|
|
SkillInlineTool(
|
|
description="First inline",
|
|
source=ToolSource.CUSTOM,
|
|
code="return 1",
|
|
),
|
|
SkillInlineTool(
|
|
description="Second inline",
|
|
source=ToolSource.CUSTOM,
|
|
code="return 2",
|
|
),
|
|
],
|
|
)
|
|
resolver = SkillResolver()
|
|
context.skill_resolved = resolver.resolve_tools(skill, {})
|
|
|
|
|
|
@then("the skill_resolver inline tool at index {index:d} should be marked inline")
|
|
def skill_resolver_check_inline_at_index(context: Context, index: int) -> None:
|
|
"""Check that the resolved entry at a specific index is marked inline."""
|
|
resolved = context.skill_resolved
|
|
assert index < len(resolved), (
|
|
f"Index {index} out of range (resolved has {len(resolved)} tools)"
|
|
)
|
|
entry = resolved[index]
|
|
assert entry.is_inline is True, (
|
|
f"Expected entry at index {index} ('{entry.name}') to be inline"
|
|
)
|