485 lines
18 KiB
Python
485 lines
18 KiB
Python
"""Step definitions for Skill context and registry tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.domain.models.core.skill import (
|
|
Skill,
|
|
SkillResolver,
|
|
)
|
|
from cleveragents.skills.context import SkillContext, SkillExecutionError
|
|
from cleveragents.skills.protocol import (
|
|
SkillDefinition,
|
|
SkillMetadata,
|
|
)
|
|
from cleveragents.skills.registry import SkillRegistry
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_skill_definition(name: str, **overrides: Any) -> SkillDefinition:
|
|
"""Create a SkillDefinition with sensible defaults."""
|
|
defaults: dict[str, Any] = {
|
|
"name": name,
|
|
"description": f"Test skill {name}",
|
|
}
|
|
defaults.update(overrides)
|
|
skill = Skill(**defaults)
|
|
resolver = SkillResolver()
|
|
resolved = resolver.resolve_tools(skill, {})
|
|
metadata = SkillMetadata.from_skill(skill, resolved=resolved)
|
|
return SkillDefinition(
|
|
skill=skill,
|
|
resolved_tools=resolved,
|
|
metadata=metadata,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SkillContext Creation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I create a skill_context with plan "{plan_id}" project "{project_id}" and sandbox "{sandbox}"'
|
|
)
|
|
def skill_context_create(
|
|
context: Context, plan_id: str, project_id: str, sandbox: str
|
|
) -> None:
|
|
"""Create a SkillContext with plan/project/sandbox info."""
|
|
context.skill_context = SkillContext(
|
|
plan_id=plan_id,
|
|
project_id=project_id,
|
|
sandbox_path=Path(sandbox),
|
|
)
|
|
context.skill_context_error = None
|
|
|
|
|
|
@when('I create a read_only skill_context with plan "{plan_id}" project "{project_id}"')
|
|
def skill_context_create_readonly(
|
|
context: Context, plan_id: str, project_id: str
|
|
) -> None:
|
|
"""Create a read-only SkillContext."""
|
|
context.skill_context = SkillContext(
|
|
plan_id=plan_id,
|
|
project_id=project_id,
|
|
sandbox_path=Path("/tmp/readonly-sandbox"),
|
|
read_only=True,
|
|
)
|
|
context.skill_context_error = None
|
|
|
|
|
|
@when('I create a skill_context with metadata key "{key}" value "{value}"')
|
|
def skill_context_create_with_metadata(context: Context, key: str, value: str) -> None:
|
|
"""Create a SkillContext with custom metadata."""
|
|
context.skill_context = SkillContext(
|
|
plan_id="plan-meta",
|
|
project_id="proj-meta",
|
|
sandbox_path=Path("/tmp/meta-sandbox"),
|
|
metadata={key: value},
|
|
)
|
|
|
|
|
|
@then("the skill_context should be created")
|
|
def skill_context_check_created(context: Context) -> None:
|
|
"""Verify context was created."""
|
|
assert context.skill_context is not None
|
|
|
|
|
|
@then('the skill_context plan_id should be "{expected}"')
|
|
def skill_context_check_plan_id(context: Context, expected: str) -> None:
|
|
"""Check context plan_id."""
|
|
assert context.skill_context.plan_id == expected
|
|
|
|
|
|
@then('the skill_context project_id should be "{expected}"')
|
|
def skill_context_check_project_id(context: Context, expected: str) -> None:
|
|
"""Check context project_id."""
|
|
assert context.skill_context.project_id == expected
|
|
|
|
|
|
@then('the skill_context sandbox_path should be "{expected}"')
|
|
def skill_context_check_sandbox(context: Context, expected: str) -> None:
|
|
"""Check context sandbox_path."""
|
|
assert str(context.skill_context.get_sandbox_path()) == expected
|
|
|
|
|
|
@then("the skill_context should not be read_only")
|
|
def skill_context_check_not_readonly(context: Context) -> None:
|
|
"""Check context is not read-only."""
|
|
assert context.skill_context.is_read_only() is False
|
|
|
|
|
|
@then("the skill_context should be read_only")
|
|
def skill_context_check_readonly(context: Context) -> None:
|
|
"""Check context is read-only."""
|
|
assert context.skill_context.is_read_only() is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plan Metadata
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the skill_context plan metadata should contain key "{key}"')
|
|
def skill_context_check_metadata_key(context: Context, key: str) -> None:
|
|
"""Check plan metadata contains a key."""
|
|
meta = context.skill_context.get_plan_metadata()
|
|
assert key in meta, f"Expected '{key}' in {list(meta.keys())}"
|
|
|
|
|
|
@then('the skill_context plan metadata key "{key}" should be "{expected}"')
|
|
def skill_context_check_metadata_value(
|
|
context: Context, key: str, expected: str
|
|
) -> None:
|
|
"""Check plan metadata value."""
|
|
meta = context.skill_context.get_plan_metadata()
|
|
actual = meta[key]
|
|
assert str(actual) == expected, f"Expected '{expected}', got '{actual}'"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Resource Resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a skill_context with resource "{name}" bound to "{value}"')
|
|
def skill_context_create_with_resource(context: Context, name: str, value: str) -> None:
|
|
"""Create a SkillContext with a resource binding."""
|
|
context.skill_context = SkillContext(
|
|
plan_id="plan-res",
|
|
project_id="proj-res",
|
|
sandbox_path=Path("/tmp/res-sandbox"),
|
|
resource_bindings={name: value},
|
|
)
|
|
|
|
|
|
@when('I resolve resource "{name}" from the skill_context')
|
|
def skill_context_resolve_resource(context: Context, name: str) -> None:
|
|
"""Resolve a resource from the context."""
|
|
context.resolved_resource = context.skill_context.resolve_resource(name)
|
|
|
|
|
|
@then('the resolved resource should be "{expected}"')
|
|
def skill_context_check_resolved_resource(context: Context, expected: str) -> None:
|
|
"""Check resolved resource value."""
|
|
assert context.resolved_resource == expected
|
|
|
|
|
|
@when("I create a skill_context with no resources")
|
|
def skill_context_create_no_resources(context: Context) -> None:
|
|
"""Create a SkillContext with empty resource bindings."""
|
|
context.skill_context = SkillContext(
|
|
plan_id="plan-empty",
|
|
project_id="proj-empty",
|
|
sandbox_path=Path("/tmp/empty-sandbox"),
|
|
)
|
|
|
|
|
|
@when('I try to resolve resource "{name}" from the skill_context')
|
|
def skill_context_try_resolve_resource(context: Context, name: str) -> None:
|
|
"""Try to resolve a missing resource."""
|
|
context.skill_context_error = None
|
|
try:
|
|
context.skill_context.resolve_resource(name)
|
|
except SkillExecutionError as e:
|
|
context.skill_context_error = e
|
|
|
|
|
|
@then("a skill_context resolution error should be raised")
|
|
def skill_context_check_resolution_error(context: Context) -> None:
|
|
"""Verify resolution error was raised."""
|
|
assert context.skill_context_error is not None
|
|
assert context.skill_context_error.error_type.value == "resolution_failure"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Change Tracking
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I register a tool invocation for "{tool_name}" with duration {duration:g}')
|
|
def skill_context_register_invocation(
|
|
context: Context, tool_name: str, duration: float
|
|
) -> None:
|
|
"""Register a tool invocation."""
|
|
context.skill_context.register_tool_invocation(
|
|
tool_name=tool_name,
|
|
input_data={"arg": "test"},
|
|
output_data={"result": "ok"},
|
|
duration_ms=duration,
|
|
)
|
|
|
|
|
|
@then("the skill_context change tracker should have {count:d} records")
|
|
def skill_context_check_tracker_count(context: Context, count: int) -> None:
|
|
"""Check change tracker record count."""
|
|
actual = len(context.skill_context.change_tracker)
|
|
assert actual == count, f"Expected {count} records, got {actual}"
|
|
|
|
|
|
@then('the last change tracker record tool_name should be "{expected}"')
|
|
def skill_context_check_last_tool(context: Context, expected: str) -> None:
|
|
"""Check last tracker record tool_name."""
|
|
record = context.skill_context.change_tracker[-1]
|
|
assert record["tool_name"] == expected
|
|
|
|
|
|
@then("the last change tracker record duration_ms should be {expected:g}")
|
|
def skill_context_check_last_duration(context: Context, expected: float) -> None:
|
|
"""Check last tracker record duration."""
|
|
record = context.skill_context.change_tracker[-1]
|
|
assert record["duration_ms"] == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Write Guard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I try to enforce write guard for tool "{tool_name}"')
|
|
def skill_context_try_write_guard(context: Context, tool_name: str) -> None:
|
|
"""Try to enforce write guard on a read-only context."""
|
|
context.skill_context_error = None
|
|
try:
|
|
context.skill_context.enforce_write_guard(tool_name)
|
|
except SkillExecutionError as e:
|
|
context.skill_context_error = e
|
|
|
|
|
|
@then("a skill_context permission denied error should be raised")
|
|
def skill_context_check_permission_error(context: Context) -> None:
|
|
"""Verify permission denied error was raised."""
|
|
assert context.skill_context_error is not None
|
|
assert context.skill_context_error.error_type.value == "permission_denied"
|
|
|
|
|
|
@then('the skill_context permission error tool_name should be "{expected}"')
|
|
def skill_context_check_permission_tool(context: Context, expected: str) -> None:
|
|
"""Check permission error tool_name."""
|
|
assert context.skill_context_error.tool_name == expected
|
|
|
|
|
|
@when('I enforce write guard for tool "{tool_name}" in writable context')
|
|
def skill_context_write_guard_writable(context: Context, tool_name: str) -> None:
|
|
"""Enforce write guard in a writable context (should not raise)."""
|
|
context.skill_context_error = None
|
|
try:
|
|
context.skill_context.enforce_write_guard(tool_name)
|
|
except SkillExecutionError as e:
|
|
context.skill_context_error = e
|
|
|
|
|
|
@then("no skill_context error should be raised")
|
|
def skill_context_check_no_error(context: Context) -> None:
|
|
"""Verify no error was raised."""
|
|
assert context.skill_context_error is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SkillRegistry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a skill_registry")
|
|
def skill_registry_create(context: Context) -> None:
|
|
"""Create a SkillRegistry."""
|
|
context.skill_registry = SkillRegistry()
|
|
context.skill_context_error = None
|
|
|
|
|
|
@when('I register a skill "{name}" in the registry')
|
|
def skill_registry_register(context: Context, name: str) -> None:
|
|
"""Register a skill in the registry."""
|
|
defn = _make_skill_definition(name)
|
|
context.skill_registry.register(defn)
|
|
|
|
|
|
@when('I register a skill "{name}" with tool refs in the registry')
|
|
def skill_registry_register_with_refs(context: Context, name: str) -> None:
|
|
"""Register a skill with tool refs in the registry."""
|
|
defn = _make_skill_definition(name, tool_refs=["local/tool-a", "local/tool-b"])
|
|
context.skill_registry.register(defn)
|
|
|
|
|
|
@when('I get skill "{name}" from the registry')
|
|
def skill_registry_get(context: Context, name: str) -> None:
|
|
"""Get a skill from the registry."""
|
|
context.registry_skill = context.skill_registry.get(name)
|
|
|
|
|
|
@when('I try to register a duplicate skill "{name}"')
|
|
def skill_registry_try_register_dup(context: Context, name: str) -> None:
|
|
"""Try to register a duplicate skill."""
|
|
context.skill_context_error = None
|
|
try:
|
|
defn = _make_skill_definition(name)
|
|
context.skill_registry.register(defn)
|
|
except SkillExecutionError as e:
|
|
context.skill_context_error = e
|
|
|
|
|
|
@when('I try to get skill "{name}" from the registry')
|
|
def skill_registry_try_get(context: Context, name: str) -> None:
|
|
"""Try to get a missing skill."""
|
|
context.skill_context_error = None
|
|
try:
|
|
context.skill_registry.get(name)
|
|
except SkillExecutionError as e:
|
|
context.skill_context_error = e
|
|
|
|
|
|
@when("I list all skills from the registry")
|
|
def skill_registry_list(context: Context) -> None:
|
|
"""List all skills from the registry."""
|
|
context.registry_skill_list = context.skill_registry.list_all()
|
|
|
|
|
|
@when('I unregister skill "{name}" from the registry')
|
|
def skill_registry_unregister(context: Context, name: str) -> None:
|
|
"""Unregister a skill from the registry."""
|
|
context.skill_registry.unregister(name)
|
|
|
|
|
|
@when('I try to unregister skill "{name}" from the registry')
|
|
def skill_registry_try_unregister(context: Context, name: str) -> None:
|
|
"""Try to unregister a missing skill."""
|
|
context.skill_context_error = None
|
|
try:
|
|
context.skill_registry.unregister(name)
|
|
except SkillExecutionError as e:
|
|
context.skill_context_error = e
|
|
|
|
|
|
@when('I resolve tools for "{name}" from the registry')
|
|
def skill_registry_resolve_tools(context: Context, name: str) -> None:
|
|
"""Resolve tools for a skill from the registry."""
|
|
context.resolved_tools = context.skill_registry.resolve_tools(name)
|
|
|
|
|
|
@then('the registry skill name should be "{expected}"')
|
|
def skill_registry_check_name(context: Context, expected: str) -> None:
|
|
"""Check registry skill name."""
|
|
assert context.registry_skill.skill.name == expected
|
|
|
|
|
|
@then("a skill_context validation error should be raised")
|
|
def skill_context_check_validation_error(context: Context) -> None:
|
|
"""Verify validation error was raised."""
|
|
assert context.skill_context_error is not None
|
|
assert context.skill_context_error.error_type.value == "validation_error"
|
|
|
|
|
|
@then("a skill_context not found error should be raised")
|
|
def skill_context_check_not_found_error(context: Context) -> None:
|
|
"""Verify not found error was raised."""
|
|
assert context.skill_context_error is not None
|
|
assert context.skill_context_error.error_type.value == "skill_not_found"
|
|
|
|
|
|
@then("the registry should have {count:d} skills")
|
|
def skill_registry_check_count(context: Context, count: int) -> None:
|
|
"""Check registry skill count."""
|
|
actual = len(context.registry_skill_list)
|
|
assert actual == count, f"Expected {count} skills, got {actual}"
|
|
|
|
|
|
@then('the registry skill list should contain "{name}"')
|
|
def skill_registry_check_list_contains(context: Context, name: str) -> None:
|
|
"""Check registry list contains a skill."""
|
|
names = [m.name for m in context.registry_skill_list]
|
|
assert name in names, f"Expected '{name}' in {names}"
|
|
|
|
|
|
@then("the resolved tools should have {count:d} entries")
|
|
def skill_registry_check_resolved_count(context: Context, count: int) -> None:
|
|
"""Check resolved tools count."""
|
|
actual = len(context.resolved_tools)
|
|
assert actual == count, f"Expected {count} entries, got {actual}"
|
|
|
|
|
|
@then('the resolved tools should contain "{name}"')
|
|
def skill_registry_check_resolved_contains(context: Context, name: str) -> None:
|
|
"""Check resolved tools contain a tool name."""
|
|
names = [e.name for e in context.resolved_tools]
|
|
assert name in names, f"Expected '{name}' in {names}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SkillRegistry Validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a skill_registry with a mock tool registry")
|
|
def skill_registry_create_with_mock(context: Context) -> None:
|
|
"""Create a SkillRegistry with a mock tool registry."""
|
|
mock_tool_reg = MagicMock()
|
|
mock_tool_reg.get_tool.return_value = None
|
|
context.skill_registry = SkillRegistry(tool_registry=mock_tool_reg)
|
|
context.skill_context_error = None
|
|
|
|
|
|
@when("I validate a skill with missing tool refs")
|
|
def skill_registry_validate_missing_refs(context: Context) -> None:
|
|
"""Validate a skill with tool refs that don't exist."""
|
|
defn = _make_skill_definition(
|
|
"local/validate-missing",
|
|
tool_refs=["local/nonexistent-tool"],
|
|
)
|
|
context.validation_errors = context.skill_registry.validate_skill(defn)
|
|
|
|
|
|
@when("I validate a skill with missing includes")
|
|
def skill_registry_validate_missing_includes(context: Context) -> None:
|
|
"""Validate a skill that includes a non-registered skill."""
|
|
from cleveragents.domain.models.core.skill import SkillInclude
|
|
|
|
skill = Skill(
|
|
name="local/validate-includes",
|
|
description="Skill with missing includes",
|
|
includes=[SkillInclude(name="local/not-registered")],
|
|
)
|
|
# Don't resolve tools through the resolver since the included
|
|
# skill doesn't exist — just build metadata from the skill alone
|
|
metadata = SkillMetadata(
|
|
name=skill.name,
|
|
description=skill.description,
|
|
)
|
|
defn = SkillDefinition(
|
|
skill=skill,
|
|
resolved_tools=[],
|
|
metadata=metadata,
|
|
)
|
|
context.validation_errors = context.skill_registry.validate_skill(defn)
|
|
|
|
|
|
@when("I validate a skill with no issues")
|
|
def skill_registry_validate_no_issues(context: Context) -> None:
|
|
"""Validate a skill with no issues."""
|
|
defn = _make_skill_definition("local/valid-skill")
|
|
context.validation_errors = context.skill_registry.validate_skill(defn)
|
|
|
|
|
|
@then('the validation errors should contain "{expected}"')
|
|
def skill_registry_check_validation_errors(context: Context, expected: str) -> None:
|
|
"""Check validation errors contain a message."""
|
|
assert any(expected in e for e in context.validation_errors), (
|
|
f"Expected error containing '{expected}' in {context.validation_errors}"
|
|
)
|
|
|
|
|
|
@then("the validation errors should be empty")
|
|
def skill_registry_check_validation_empty(context: Context) -> None:
|
|
"""Verify no validation errors."""
|
|
assert len(context.validation_errors) == 0, (
|
|
f"Expected empty errors, got {context.validation_errors}"
|
|
)
|