forked from cleveragents/cleveragents-core
328 lines
11 KiB
Python
328 lines
11 KiB
Python
"""Step definitions for the Skill File Operation Tools feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import stat
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.skills.builtins.file_ops import (
|
|
MAX_READ_SIZE,
|
|
register_skill_file_tools,
|
|
)
|
|
from cleveragents.tool.registry import ToolRegistry
|
|
from cleveragents.tool.runner import ToolRunner
|
|
|
|
__all__: list[str] = []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _run_skill_tool(context: Any, tool_name: str, inputs: dict[str, Any]) -> None:
|
|
"""Execute a skill file tool through the runner."""
|
|
registry = ToolRegistry()
|
|
register_skill_file_tools(registry)
|
|
runner = ToolRunner(registry)
|
|
inputs["sandbox_root"] = context.skill_sandbox_dir
|
|
context.skill_tool_result = runner.execute(tool_name, inputs)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Givens
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a skill file ops sandbox directory")
|
|
def step_given_skill_sandbox(context: Any) -> None:
|
|
context.skill_sandbox_dir = tempfile.mkdtemp()
|
|
context._cleanup_handlers.append(
|
|
lambda: shutil.rmtree(context.skill_sandbox_dir, ignore_errors=True)
|
|
)
|
|
|
|
|
|
@given('a sandbox file "{name}" with content "{content}"')
|
|
def step_given_sandbox_file(context: Any, name: str, content: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content)
|
|
|
|
|
|
@given('a sandbox file "{name}" with lines "{csv_lines}"')
|
|
def step_given_sandbox_file_lines(context: Any, name: str, csv_lines: str) -> None:
|
|
lines = csv_lines.split(",")
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("\n".join(lines) + "\n")
|
|
|
|
|
|
@given('a binary file "{name}" in the sandbox')
|
|
def step_given_binary_file(context: Any, name: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
# PNG signature
|
|
path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
|
|
|
|
|
|
@given('an oversized file "{name}" in the sandbox')
|
|
def step_given_oversized_file(context: Any, name: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
# Write a file slightly over the limit
|
|
path.write_text("x" * (MAX_READ_SIZE + 1))
|
|
|
|
|
|
@given('a sandbox subdirectory "{name}"')
|
|
def step_given_sandbox_subdir(context: Any, name: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
@given('a sandbox file "{name}" with content "{content}" and mode {mode}')
|
|
def step_given_sandbox_file_with_mode(
|
|
context: Any, name: str, content: str, mode: str
|
|
) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content)
|
|
path.chmod(int(mode, 8))
|
|
|
|
|
|
@given("a skill file tools registry")
|
|
def step_given_skill_file_registry(context: Any) -> None:
|
|
context.skill_registry = ToolRegistry()
|
|
register_skill_file_tools(context.skill_registry)
|
|
|
|
|
|
@given('an ELF binary file "{name}" in the sandbox')
|
|
def step_given_elf_binary(context: Any, name: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(b"\x7fELF" + b"\x00" * 100)
|
|
|
|
|
|
@given('a control-heavy file "{name}" in the sandbox')
|
|
def step_given_control_heavy_file(context: Any, name: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
# >10% control characters
|
|
path.write_bytes(b"\x01\x02\x03\x04\x05\x06\x07" * 20)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Whens
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I execute the "skill/file-read" tool with path "{path}"')
|
|
def step_when_skill_read(context: Any, path: str) -> None:
|
|
_run_skill_tool(context, "skill/file-read", {"path": path})
|
|
|
|
|
|
@when('I execute the "skill/file-read" tool with an empty path')
|
|
def step_when_skill_read_empty(context: Any) -> None:
|
|
_run_skill_tool(context, "skill/file-read", {"path": ""})
|
|
|
|
|
|
@when(
|
|
'I execute the "skill/file-read" tool with path "{path}"'
|
|
" offset {offset:d} limit {limit:d}"
|
|
)
|
|
def step_when_skill_read_offset_limit(
|
|
context: Any, path: str, offset: int, limit: int
|
|
) -> None:
|
|
_run_skill_tool(
|
|
context,
|
|
"skill/file-read",
|
|
{"path": path, "offset": offset, "limit": limit},
|
|
)
|
|
|
|
|
|
@when(
|
|
'I execute the "skill/file-write" tool with path "{path}" and content "{content}"'
|
|
)
|
|
def step_when_skill_write(context: Any, path: str, content: str) -> None:
|
|
_run_skill_tool(context, "skill/file-write", {"path": path, "content": content})
|
|
|
|
|
|
@when(
|
|
'I write "skill/file-write" with path "{path}"'
|
|
' content "{content}" and newline_mode "{mode}"'
|
|
)
|
|
def step_when_skill_write_newline(
|
|
context: Any, path: str, content: str, mode: str
|
|
) -> None:
|
|
# Interpret escape sequences for test clarity
|
|
content = (
|
|
content.replace("\\r\\n", "\r\n").replace("\\n", "\n").replace("\\r", "\r")
|
|
)
|
|
_run_skill_tool(
|
|
context,
|
|
"skill/file-write",
|
|
{"path": path, "content": content, "newline_mode": mode},
|
|
)
|
|
|
|
|
|
@when('I execute the "skill/file-edit" tool replacing "{old}" with "{new}" in "{path}"')
|
|
def step_when_skill_edit(context: Any, old: str, new: str, path: str) -> None:
|
|
_run_skill_tool(
|
|
context,
|
|
"skill/file-edit",
|
|
{"path": path, "old_text": old, "new_text": new},
|
|
)
|
|
|
|
|
|
@when(
|
|
'I execute the "skill/file-edit" tool replacing all "{old}"'
|
|
' with "{new}" in "{path}"'
|
|
)
|
|
def step_when_skill_edit_all(context: Any, old: str, new: str, path: str) -> None:
|
|
_run_skill_tool(
|
|
context,
|
|
"skill/file-edit",
|
|
{
|
|
"path": path,
|
|
"old_text": old,
|
|
"new_text": new,
|
|
"replace_all": True,
|
|
},
|
|
)
|
|
|
|
|
|
@when('I execute the "skill/file-delete" tool with path "{path}"')
|
|
def step_when_skill_delete(context: Any, path: str) -> None:
|
|
_run_skill_tool(context, "skill/file-delete", {"path": path})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Thens
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the skill tool result should be successful")
|
|
def step_then_skill_result_success(context: Any) -> None:
|
|
result = context.skill_tool_result
|
|
assert result.success, f"Tool failed: {result.error}"
|
|
|
|
|
|
@then("the skill tool result should not be successful")
|
|
def step_then_skill_result_fail(context: Any) -> None:
|
|
result = context.skill_tool_result
|
|
assert not result.success, "Expected failure but tool succeeded"
|
|
|
|
|
|
@then('the skill output "{key}" should be "{value}"')
|
|
def step_then_skill_output_str(context: Any, key: str, value: str) -> None:
|
|
result = context.skill_tool_result
|
|
assert result.success, f"Tool failed: {result.error}"
|
|
assert str(result.output[key]) == value, (
|
|
f"Expected {key}={value}, got {result.output[key]}"
|
|
)
|
|
|
|
|
|
@then('the skill output "{key}" should be greater than {value:d}')
|
|
def step_then_skill_output_gt(context: Any, key: str, value: int) -> None:
|
|
assert context.skill_tool_result.output[key] > value
|
|
|
|
|
|
@then('the skill output "{key}" should be boolean true')
|
|
def step_then_skill_output_true(context: Any, key: str) -> None:
|
|
assert context.skill_tool_result.output[key] is True
|
|
|
|
|
|
@then('the skill output "{key}" should be boolean false')
|
|
def step_then_skill_output_false(context: Any, key: str) -> None:
|
|
assert context.skill_tool_result.output[key] is False
|
|
|
|
|
|
@then('the skill output "{key}" should equal {value:d}')
|
|
def step_then_skill_output_int(context: Any, key: str, value: int) -> None:
|
|
assert context.skill_tool_result.output[key] == value, (
|
|
f"Expected {key}={value}, got {context.skill_tool_result.output[key]}"
|
|
)
|
|
|
|
|
|
@then('the skill output "{key}" should contain "{text}"')
|
|
def step_then_skill_output_contains(context: Any, key: str, text: str) -> None:
|
|
assert text in context.skill_tool_result.output[key], (
|
|
f"Expected '{text}' in {key}: {context.skill_tool_result.output[key]}"
|
|
)
|
|
|
|
|
|
@then('the skill output "{key}" should not contain "{text}"')
|
|
def step_then_skill_output_not_contains(context: Any, key: str, text: str) -> None:
|
|
assert text not in context.skill_tool_result.output[key]
|
|
|
|
|
|
@then('the skill tool error should mention "{text}"')
|
|
def step_then_skill_error_mentions(context: Any, text: str) -> None:
|
|
error = context.skill_tool_result.error or ""
|
|
assert text.lower() in error.lower(), f"Expected '{text}' in error: {error}"
|
|
|
|
|
|
@then('the sandbox file "{name}" should exist')
|
|
def step_then_sandbox_file_exists(context: Any, name: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
assert path.exists(), f"File {path} does not exist"
|
|
|
|
|
|
@then('the sandbox file "{name}" should not exist')
|
|
def step_then_sandbox_file_not_exists(context: Any, name: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
assert not path.exists(), f"File {path} still exists"
|
|
|
|
|
|
@then('the sandbox file "{name}" should not contain carriage return')
|
|
def step_then_file_no_cr(context: Any, name: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
content = path.read_bytes()
|
|
assert b"\r" not in content, "File contains carriage returns"
|
|
|
|
|
|
@then('the sandbox file "{name}" should contain carriage return')
|
|
def step_then_file_has_cr(context: Any, name: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
content = path.read_bytes()
|
|
assert b"\r" in content, "File does not contain carriage returns"
|
|
|
|
|
|
@then('the sandbox file "{name}" should have mode {mode}')
|
|
def step_then_file_has_mode(context: Any, name: str, mode: str) -> None:
|
|
path = Path(context.skill_sandbox_dir) / name
|
|
actual = stat.S_IMODE(path.stat().st_mode)
|
|
expected = int(mode, 8)
|
|
assert actual == expected, f"Expected mode {oct(expected)}, got {oct(actual)}"
|
|
|
|
|
|
@then('the "{tool_name}" tool should have read_only capability')
|
|
def step_then_tool_read_only(context: Any, tool_name: str) -> None:
|
|
spec = context.skill_registry.get(tool_name)
|
|
assert spec is not None, f"Tool {tool_name} not found"
|
|
assert spec.capabilities.read_only, f"{tool_name} should be read_only"
|
|
|
|
|
|
@then('the "{tool_name}" tool should have writes capability')
|
|
def step_then_tool_writes(context: Any, tool_name: str) -> None:
|
|
spec = context.skill_registry.get(tool_name)
|
|
assert spec is not None, f"Tool {tool_name} not found"
|
|
assert spec.capabilities.writes, f"{tool_name} should have writes capability"
|
|
|
|
|
|
@then("the skill registry should contain {count:d} tools")
|
|
def step_then_skill_registry_count(context: Any, count: int) -> None:
|
|
tools = context.skill_registry.list_tools()
|
|
assert len(tools) == count, f"Expected {count} tools, got {len(tools)}"
|
|
|
|
|
|
@then('the skill registry should contain tool "{name}"')
|
|
def step_then_skill_registry_has(context: Any, name: str) -> None:
|
|
assert context.skill_registry.get(name) is not None, (
|
|
f"Tool '{name}' not found in registry"
|
|
)
|