cleveragents-core/features/steps/skill_schema_steps.py

"""Step definitions for skill YAML schema validation.

Tests for features/skill_schema.feature — validates the SkillConfigSchema
Pydantic model, YAML loading, key normalization, environment variable
interpolation, and error messages.
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

from behave import given, then, when  # type: ignore[import-untyped]
from behave.runner import Context  # type: ignore[import-untyped]
from pydantic import ValidationError

from cleveragents.skills.schema import SkillConfigSchema

# ────────────────────────────────────────────────────────────
# Helpers
# ────────────────────────────────────────────────────────────

_NAME_ONLY_YAML = """\
name: local/empty-skill
"""

_TOOLS_YAML = """\
name: local/file-reader
description: "Basic file reading operations"
tools:
  - name: builtin/read_file
  - name: builtin/list_directory
  - name: builtin/search_files
"""

_FULL_YAML = """\
name: local/full-skill
description: "A fully populated skill"
tools:
  - name: builtin/read_file
  - name: builtin/write_file
    writes: true
inline_tools:
  - name: helper_tool
    description: "Inline helper"
    source: custom
    code: |
      def run(input_data):
          return {"ok": True}
includes:
  - name: local/file-reader
mcp_servers:
  - name: github
    transport: stdio
    command: npx
    args: ["-y", "@modelcontextprotocol/server-github"]
agent_skill_folders:
  - path: ./skills/deploy
"""

_TOOL_OVERRIDES_YAML = """\
name: local/tool-overrides
tools:
  - name: builtin/shell_execute
    description: "Execute commands in sandbox"
    writes: true
    checkpointable: false
"""

_FULL_INLINE_TOOL_YAML = """\
name: local/inline-full
inline_tools:
  - name: run_migrations
    description: "Run database migrations"
    source: custom
    code: |
      def run(input_data):
          return {"ok": True}
    input_schema:
      type: object
      properties:
        direction:
          type: string
    writes: true
    checkpointable: true
    side_effects: ["schema_mutation"]
"""

_STDIO_MCP_YAML = """\
name: local/mcp-stdio
mcp_servers:
  - name: github
    transport: stdio
    command: npx
    args: ["-y", "@modelcontextprotocol/server-github"]
"""

_SSE_MCP_YAML = """\
name: local/mcp-sse
mcp_servers:
  - name: remote-server
    transport: sse
    url: "https://mcp.example.com/sse"
    headers:
      Authorization: "Bearer token"
"""

_MCP_FILTER_YAML = """\
name: local/mcp-filter
mcp_servers:
  - name: filtered
    transport: stdio
    command: npx
    tool_filter:
      include:
        - tool_a
        - tool_b
      exclude:
        - tool_c
"""

_AGENT_FOLDERS_YAML = """\
name: local/agent-folders
agent_skill_folders:
  - path: ./skills/deploy
  - path: ./skills/code-review-bundle
    name: code-review
"""

_CAMEL_CASE_YAML = """\
name: local/camel-skill
inlineTools:
  - name: my_tool
    source: custom
    code: |
      def run(input_data):
          return {}
mcpServers:
  - name: srv
    transport: stdio
    command: echo
"""


# ────────────────────────────────────────────────────────────
# Given steps
# ────────────────────────────────────────────────────────────


@given("a skill YAML string with only the name field")
def step_given_name_only_yaml(context: Context) -> None:
    """Provide a minimal skill YAML with only name."""
    context.skill_yaml_string = _NAME_ONLY_YAML


@given("a skill YAML string with tool references")
def step_given_tools_yaml(context: Context) -> None:
    """Provide a skill YAML with tool references."""
    context.skill_yaml_string = _TOOLS_YAML


@given("a skill YAML string with all sections populated")
def step_given_full_yaml(context: Context) -> None:
    """Provide a fully populated skill YAML."""
    context.skill_yaml_string = _FULL_YAML


@given('the skill YAML file "{rel_path}"')
def step_given_skill_yaml_file(context: Context, rel_path: str) -> None:
    """Set the path to a skill YAML file relative to project root."""
    project_root = Path(__file__).resolve().parents[2]
    context.skill_yaml_file = str(project_root / rel_path)


@given("a skill YAML string with tool overrides")
def step_given_tool_overrides_yaml(context: Context) -> None:
    """Provide a skill YAML with tool reference overrides."""
    context.skill_yaml_string = _TOOL_OVERRIDES_YAML


@given("a skill YAML string with a full inline tool")
def step_given_full_inline_tool_yaml(context: Context) -> None:
    """Provide a skill YAML with a fully populated inline tool."""
    context.skill_yaml_string = _FULL_INLINE_TOOL_YAML


@given("a skill YAML string with a stdio MCP server")
def step_given_stdio_mcp_yaml(context: Context) -> None:
    """Provide a skill YAML with a stdio MCP server."""
    context.skill_yaml_string = _STDIO_MCP_YAML


@given("a skill YAML string with an SSE MCP server")
def step_given_sse_mcp_yaml(context: Context) -> None:
    """Provide a skill YAML with an SSE MCP server."""
    context.skill_yaml_string = _SSE_MCP_YAML


@given("a skill YAML string with MCP tool filter")
def step_given_mcp_filter_yaml(context: Context) -> None:
    """Provide a skill YAML with MCP tool filter."""
    context.skill_yaml_string = _MCP_FILTER_YAML


@given("a skill YAML string with agent skill folders")
def step_given_agent_folders_yaml(context: Context) -> None:
    """Provide a skill YAML with agent skill folders."""
    context.skill_yaml_string = _AGENT_FOLDERS_YAML


@given("a skill YAML string missing the name field")
def step_given_missing_name_yaml(context: Context) -> None:
    """Provide a skill YAML without the name field."""
    context.skill_yaml_string = "description: Missing the name\n"


@given('a skill YAML string with name "{name}"')
def step_given_yaml_with_name(context: Context, name: str) -> None:
    """Provide a skill YAML with a custom name value."""
    context.skill_yaml_string = f"name: {name}\n"


@given('a skill YAML string with an invalid tool ref name "{name}"')
def step_given_invalid_tool_ref(context: Context, name: str) -> None:
    """Provide a skill YAML with an invalid tool reference name."""
    context.skill_yaml_string = f"""\
name: local/bad-ref
tools:
  - name: {name}
"""


@given("a skill YAML string with an inline tool missing source")
def step_given_inline_missing_source(context: Context) -> None:
    """Provide a skill YAML with an inline tool missing the source field."""
    context.skill_yaml_string = """\
name: local/bad-inline
inline_tools:
  - name: my_tool
    code: |
      def run(input_data):
          return {}
"""


@given("a skill YAML string with an inline tool missing code")
def step_given_inline_missing_code(context: Context) -> None:
    """Provide a skill YAML with an inline tool missing the code field."""
    context.skill_yaml_string = """\
name: local/bad-inline
inline_tools:
  - name: my_tool
    source: custom
"""


@given('a skill YAML string with an inline tool source "{source}"')
def step_given_inline_invalid_source(context: Context, source: str) -> None:
    """Provide a skill YAML with an invalid inline tool source."""
    context.skill_yaml_string = f"""\
name: local/bad-source
inline_tools:
  - name: my_tool
    source: {source}
    code: |
      def run(input_data):
          return {{}}
"""


@given('a skill YAML string with an invalid include name "{name}"')
def step_given_invalid_include(context: Context, name: str) -> None:
    """Provide a skill YAML with an invalid include name."""
    context.skill_yaml_string = f"""\
name: local/bad-include
includes:
  - name: {name}
"""


@given('a skill YAML string with MCP transport "{transport}"')
def step_given_invalid_mcp_transport(context: Context, transport: str) -> None:
    """Provide a skill YAML with an invalid MCP transport."""
    context.skill_yaml_string = f"""\
name: local/bad-mcp
mcp_servers:
  - name: srv
    transport: {transport}
"""


@given("a skill YAML string with MCP server missing transport")
def step_given_mcp_missing_transport(context: Context) -> None:
    """Provide a skill YAML with an MCP server missing transport."""
    context.skill_yaml_string = """\
name: local/bad-mcp
mcp_servers:
  - name: srv
"""


@given("a skill YAML string with agent folder missing path")
def step_given_agent_folder_missing_path(context: Context) -> None:
    """Provide a skill YAML with an agent folder missing path."""
    context.skill_yaml_string = """\
name: local/bad-folder
agent_skill_folders:
  - name: missing-path-folder
"""


@given("a skill YAML string with an unknown top-level field")
def step_given_unknown_field(context: Context) -> None:
    """Provide a skill YAML with an extra unknown field."""
    context.skill_yaml_string = """\
name: local/extra-field
unknown_field: should fail
"""


@given("a None skill YAML string")
def step_given_none_yaml(context: Context) -> None:
    """Set YAML string to None."""
    context.skill_yaml_string = None  # type: ignore[assignment]


@given("an empty skill YAML string")
def step_given_empty_yaml(context: Context) -> None:
    """Set YAML string to empty."""
    context.skill_yaml_string = ""


@given("a skill YAML string that is a list")
def step_given_yaml_list(context: Context) -> None:
    """Set YAML string to a list instead of a mapping."""
    context.skill_yaml_string = "- item1\n- item2\n"


@given("a None skill YAML file path")
def step_given_none_file_path(context: Context) -> None:
    """Set file path to None."""
    context.skill_yaml_file = None  # type: ignore[assignment]


@given('the skill YAML directory path "{dir_path}"')
def step_given_yaml_directory_path(context: Context, dir_path: str) -> None:
    """Set file path to a directory."""
    project_root = Path(__file__).resolve().parents[2]
    context.skill_yaml_file = str(project_root / dir_path)


@given("a skill YAML string with camelCase keys")
def step_given_camel_case_yaml(context: Context) -> None:
    """Provide a skill YAML with camelCase keys."""
    context.skill_yaml_string = _CAMEL_CASE_YAML


@given('a skill YAML string using env var "${{{var}}}" for mcp command')
def step_given_yaml_with_env_var(context: Context, var: str) -> None:
    """Provide a skill YAML with an environment variable reference in MCP."""
    context.skill_yaml_string = f"""\
name: local/env-skill
mcp_servers:
  - name: srv
    transport: stdio
    command: ${{{var}}}
"""


# ────────────────────────────────────────────────────────────
# When steps
# ────────────────────────────────────────────────────────────


@when("I validate the skill schema")
def step_when_validate_skill_schema(context: Context) -> None:
    """Validate the skill YAML string via SkillConfigSchema."""
    try:
        context.skill_config = SkillConfigSchema.from_yaml(context.skill_yaml_string)
        context.skill_schema_error = None
    except (ValidationError, ValueError) as exc:
        context.skill_config = None
        context.skill_schema_error = str(exc)


@when("I validate the skill schema from file")
def step_when_validate_from_file(context: Context) -> None:
    """Validate a skill YAML file via SkillConfigSchema."""
    try:
        context.skill_config = SkillConfigSchema.from_yaml_file(context.skill_yaml_file)
        context.skill_schema_error = None
    except (ValidationError, ValueError, FileNotFoundError) as exc:
        context.skill_config = None
        context.skill_schema_error = str(exc)


@when("I validate the skill schema expecting failure")
def step_when_validate_expecting_failure(context: Context) -> None:
    """Validate the skill YAML, expecting it to fail."""
    try:
        context.skill_config = SkillConfigSchema.from_yaml(context.skill_yaml_string)
        context.skill_schema_error = None
    except (ValidationError, ValueError) as exc:
        context.skill_config = None
        context.skill_schema_error = str(exc)


@when("I validate the skill schema from file expecting failure")
def step_when_validate_file_expecting_failure(context: Context) -> None:
    """Validate a skill YAML file, expecting it to fail."""
    try:
        context.skill_config = SkillConfigSchema.from_yaml_file(context.skill_yaml_file)
        context.skill_schema_error = None
    except (ValidationError, ValueError, FileNotFoundError, TypeError) as exc:
        context.skill_config = None
        context.skill_schema_error = str(exc)


# ────────────────────────────────────────────────────────────
# Then steps
# ────────────────────────────────────────────────────────────


@then("the skill schema validation should succeed")
def step_then_validation_succeeds(context: Context) -> None:
    """Assert that schema validation succeeded."""
    assert context.skill_config is not None, (
        f"Expected validation to succeed but got error: {context.skill_schema_error}"
    )


@then("the skill schema validation should fail")
def step_then_validation_fails(context: Context) -> None:
    """Assert that schema validation failed."""
    assert context.skill_config is None, "Expected validation to fail but it succeeded"
    assert context.skill_schema_error is not None


@then('the skill schema error should mention "{text}"')
def step_then_error_mentions(context: Context, text: str) -> None:
    """Assert the error message contains expected text."""
    assert context.skill_schema_error is not None, "No error was captured"
    assert text.lower() in context.skill_schema_error.lower(), (
        f"Expected error to mention '{text}', got: {context.skill_schema_error}"
    )


@then('the skill config name should be "{expected}"')
def step_then_config_name(context: Context, expected: str) -> None:
    """Assert the parsed config name."""
    assert context.skill_config is not None
    assert context.skill_config.name == expected


@then('the skill config description should be "{expected}"')
def step_then_config_description(context: Context, expected: str) -> None:
    """Assert the parsed config description."""
    assert context.skill_config is not None
    assert context.skill_config.description == expected


@then("the skill config should have {count:d} tools")
def step_then_tool_count(context: Context, count: int) -> None:
    """Assert the number of tool references."""
    assert context.skill_config is not None
    assert len(context.skill_config.tools) == count


@then("the skill config should have {count:d} inline tools")
def step_then_inline_tool_count(context: Context, count: int) -> None:
    """Assert the number of inline tools."""
    assert context.skill_config is not None
    assert len(context.skill_config.inline_tools) == count


@then("the skill config should have {count:d} includes")
def step_then_include_count(context: Context, count: int) -> None:
    """Assert the number of includes."""
    assert context.skill_config is not None
    assert len(context.skill_config.includes) == count


@then("the skill config should have {count:d} mcp servers")
def step_then_mcp_count(context: Context, count: int) -> None:
    """Assert the number of MCP servers."""
    assert context.skill_config is not None
    assert len(context.skill_config.mcp_servers) == count


@then("the skill config should have {count:d} agent skill folders")
def step_then_agent_folder_count(context: Context, count: int) -> None:
    """Assert the number of agent skill folders."""
    assert context.skill_config is not None
    assert len(context.skill_config.agent_skill_folders) == count


@then("the skill config tools list should be empty")
def step_then_tools_empty(context: Context) -> None:
    """Assert tools list is empty."""
    assert context.skill_config is not None
    assert context.skill_config.tools == []


@then("the skill config inline_tools list should be empty")
def step_then_inline_tools_empty(context: Context) -> None:
    """Assert inline_tools list is empty."""
    assert context.skill_config is not None
    assert context.skill_config.inline_tools == []


@then("the skill config includes list should be empty")
def step_then_includes_empty(context: Context) -> None:
    """Assert includes list is empty."""
    assert context.skill_config is not None
    assert context.skill_config.includes == []


@then("the skill config mcp_servers list should be empty")
def step_then_mcp_servers_empty(context: Context) -> None:
    """Assert mcp_servers list is empty."""
    assert context.skill_config is not None
    assert context.skill_config.mcp_servers == []


@then("the skill config agent_skill_folders list should be empty")
def step_then_agent_folders_empty(context: Context) -> None:
    """Assert agent_skill_folders list is empty."""
    assert context.skill_config is not None
    assert context.skill_config.agent_skill_folders == []


@then('skill tool {idx:d} name should be "{expected}"')
def step_then_tool_name(context: Context, idx: int, expected: str) -> None:
    """Assert a tool reference name by index."""
    assert context.skill_config is not None
    assert context.skill_config.tools[idx].name == expected


@then('skill tool {idx:d} description should be "{expected}"')
def step_then_tool_description(context: Context, idx: int, expected: str) -> None:
    """Assert a tool reference description by index."""
    assert context.skill_config is not None
    assert context.skill_config.tools[idx].description == expected


@then("skill tool {idx:d} writes should be true")
def step_then_tool_writes_true(context: Context, idx: int) -> None:
    """Assert a tool reference writes flag is True."""
    assert context.skill_config is not None
    assert context.skill_config.tools[idx].writes is True


@then("skill tool {idx:d} checkpointable should be false")
def step_then_tool_checkpointable_false(context: Context, idx: int) -> None:
    """Assert a tool reference checkpointable flag is False."""
    assert context.skill_config is not None
    assert context.skill_config.tools[idx].checkpointable is False


@then('skill inline tool {idx:d} name should be "{expected}"')
def step_then_inline_name(context: Context, idx: int, expected: str) -> None:
    """Assert an inline tool name by index."""
    assert context.skill_config is not None
    assert context.skill_config.inline_tools[idx].name == expected


@then('skill inline tool {idx:d} source should be "{expected}"')
def step_then_inline_source(context: Context, idx: int, expected: str) -> None:
    """Assert an inline tool source by index."""
    assert context.skill_config is not None
    assert context.skill_config.inline_tools[idx].source == expected


@then("skill inline tool {idx:d} writes should be true")
def step_then_inline_writes_true(context: Context, idx: int) -> None:
    """Assert an inline tool writes flag is True."""
    assert context.skill_config is not None
    assert context.skill_config.inline_tools[idx].writes is True


@then("skill inline tool {idx:d} checkpointable should be true")
def step_then_inline_checkpointable_true(context: Context, idx: int) -> None:
    """Assert an inline tool checkpointable flag is True."""
    assert context.skill_config is not None
    assert context.skill_config.inline_tools[idx].checkpointable is True


@then('skill mcp server {idx:d} name should be "{expected}"')
def step_then_mcp_name(context: Context, idx: int, expected: str) -> None:
    """Assert an MCP server name by index."""
    assert context.skill_config is not None
    assert context.skill_config.mcp_servers[idx].name == expected


@then('skill mcp server {idx:d} transport should be "{expected}"')
def step_then_mcp_transport(context: Context, idx: int, expected: str) -> None:
    """Assert an MCP server transport by index."""
    assert context.skill_config is not None
    assert context.skill_config.mcp_servers[idx].transport == expected


@then('skill mcp server {idx:d} command should be "{expected}"')
def step_then_mcp_command(context: Context, idx: int, expected: str) -> None:
    """Assert an MCP server command by index."""
    assert context.skill_config is not None
    assert context.skill_config.mcp_servers[idx].command == expected


@then('skill mcp server {idx:d} url should be "{expected}"')
def step_then_mcp_url(context: Context, idx: int, expected: str) -> None:
    """Assert an MCP server url by index."""
    assert context.skill_config is not None
    assert context.skill_config.mcp_servers[idx].url == expected


@then("skill mcp server {idx:d} tool filter include should have {count:d} items")
def step_then_mcp_filter_include_count(context: Context, idx: int, count: int) -> None:
    """Assert the number of tool filter include entries."""
    assert context.skill_config is not None
    mcp = context.skill_config.mcp_servers[idx]
    assert mcp.tool_filter is not None
    assert len(mcp.tool_filter.include) == count


@then("skill mcp server {idx:d} tool filter exclude should have {count:d} items")
def step_then_mcp_filter_exclude_count(context: Context, idx: int, count: int) -> None:
    """Assert the number of tool filter exclude entries."""
    assert context.skill_config is not None
    mcp = context.skill_config.mcp_servers[idx]
    assert mcp.tool_filter is not None
    assert len(mcp.tool_filter.exclude) == count


@then('skill agent folder {idx:d} path should be "{expected}"')
def step_then_agent_folder_path(context: Context, idx: int, expected: str) -> None:
    """Assert an agent skill folder path by index."""
    assert context.skill_config is not None
    assert context.skill_config.agent_skill_folders[idx].path == expected


@then('skill agent folder {idx:d} name should be "{expected}"')
def step_then_agent_folder_name(context: Context, idx: int, expected: str) -> None:
    """Assert an agent skill folder name by index."""
    assert context.skill_config is not None
    assert context.skill_config.agent_skill_folders[idx].name == expected


@then('the skill config model_dump should contain key "{key}"')
def step_then_model_dump_contains_key(context: Context, key: str) -> None:
    """Assert the model_dump dict contains a key."""
    assert context.skill_config is not None
    data: dict[str, Any] = context.skill_config.model_dump()
    assert key in data, f"Key '{key}' not found in model_dump: {list(data.keys())}"