freemo
2980b14e7b
Add ~60 new behave scenarios across output rendering, tool router, file ops, and skill search features targeting uncovered lines and branches. Key additions: - ElementHandle validation guards (empty id, type, negative index, None args) - Handle close/context-manager edge cases - Table list-row and batch-row coverage - Color/box-draw/JSON/YAML materializer edge cases - Format selection paths (detect capabilities, explicit flag, empty format) - Session ColumnDef, double-close guard, force-close open handles - Tool router schema export and provider format scenarios - File ops edge cases and search stat-failure/glob-include tests All nox checks pass: lint, typecheck (0 errors), unit_tests (4730 scenarios), coverage_report (97.0% >= 97% threshold).
468 lines
16 KiB
Python
468 lines
16 KiB
Python
"""Step definitions for the Skill File Operation Tools feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.skills.builtins.file_ops import (
|
|
MAX_READ_SIZE,
|
|
_is_binary,
|
|
_is_fd_closed,
|
|
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"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Coverage: _is_binary OSError, create_dirs, atomic write cleanup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I check _is_binary on a nonexistent path")
|
|
def step_check_is_binary_nonexistent(context: Any) -> None:
|
|
context.is_binary_result = _is_binary(Path("/nonexistent/path/file.bin"))
|
|
|
|
|
|
@then("_is_binary should return false")
|
|
def step_then_is_binary_false(context: Any) -> None:
|
|
assert context.is_binary_result is False
|
|
|
|
|
|
@when('I execute write_file with create_dirs to "{path}" and content "{content}"')
|
|
def step_write_with_create_dirs(context: Any, path: str, content: str) -> None:
|
|
registry = ToolRegistry()
|
|
register_skill_file_tools(registry)
|
|
runner = ToolRunner(registry)
|
|
context.skill_tool_result = runner.execute(
|
|
"skill/file-write",
|
|
{
|
|
"sandbox_root": context.skill_sandbox_dir,
|
|
"path": path,
|
|
"content": content,
|
|
"create_dirs": True,
|
|
},
|
|
)
|
|
|
|
|
|
@then('the written file "{path}" should contain "{expected}"')
|
|
def step_then_written_file_contains(context: Any, path: str, expected: str) -> None:
|
|
full_path = Path(context.skill_sandbox_dir) / path
|
|
assert full_path.exists(), f"File {full_path} does not exist"
|
|
assert full_path.read_text() == expected
|
|
|
|
|
|
@when("I check _is_fd_closed with an open file descriptor")
|
|
def step_check_fd_open(context: Any) -> None:
|
|
fd = os.open(os.devnull, os.O_RDONLY)
|
|
context.fd_closed_result = _is_fd_closed(fd)
|
|
os.close(fd)
|
|
|
|
|
|
@then("_is_fd_closed should return false")
|
|
def step_then_fd_not_closed(context: Any) -> None:
|
|
assert context.fd_closed_result is False
|
|
|
|
|
|
@when("I check _is_fd_closed with a closed file descriptor")
|
|
def step_check_fd_closed(context: Any) -> None:
|
|
fd = os.open(os.devnull, os.O_RDONLY)
|
|
os.close(fd)
|
|
context.fd_closed_result = _is_fd_closed(fd)
|
|
|
|
|
|
@then("_is_fd_closed should return true")
|
|
def step_then_fd_closed(context: Any) -> None:
|
|
assert context.fd_closed_result is True
|
|
|
|
|
|
@when("I simulate atomic write failure during replace")
|
|
def step_simulate_write_replace_failure(context: Any) -> None:
|
|
registry = ToolRegistry()
|
|
register_skill_file_tools(registry)
|
|
runner = ToolRunner(registry)
|
|
|
|
def _fail_replace(src: str, dst: str) -> None:
|
|
raise OSError("Simulated replace failure")
|
|
|
|
context.atomic_write_error = None
|
|
with patch(
|
|
"cleveragents.skills.builtins.file_ops.os.replace", side_effect=_fail_replace
|
|
):
|
|
result = runner.execute(
|
|
"skill/file-write",
|
|
{
|
|
"sandbox_root": context.skill_sandbox_dir,
|
|
"path": "atomic_test.txt",
|
|
"content": "should fail",
|
|
},
|
|
)
|
|
context.skill_tool_result = result
|
|
|
|
|
|
@then("the temporary file should be cleaned up")
|
|
def step_then_tmp_cleaned(context: Any) -> None:
|
|
sandbox = Path(context.skill_sandbox_dir)
|
|
tmp_files = list(sandbox.glob("*.tmp"))
|
|
assert len(tmp_files) == 0, f"Temp files not cleaned up: {tmp_files}"
|
|
|
|
|
|
@then("the original exception should propagate")
|
|
def step_then_exception_propagated(context: Any) -> None:
|
|
assert not context.skill_tool_result.success
|
|
assert "Simulated replace failure" in (context.skill_tool_result.error or "")
|
|
|
|
|
|
@when("I simulate edit atomic write failure during chmod")
|
|
def step_simulate_edit_chmod_failure(context: Any) -> None:
|
|
registry = ToolRegistry()
|
|
register_skill_file_tools(registry)
|
|
runner = ToolRunner(registry)
|
|
|
|
def _fail_chmod(path: str, mode: int) -> None:
|
|
raise OSError("Simulated chmod failure")
|
|
|
|
with patch(
|
|
"cleveragents.skills.builtins.file_ops.os.chmod", side_effect=_fail_chmod
|
|
):
|
|
result = runner.execute(
|
|
"skill/file-edit",
|
|
{
|
|
"sandbox_root": context.skill_sandbox_dir,
|
|
"path": "editable.txt",
|
|
"old_text": "original",
|
|
"new_text": "modified",
|
|
},
|
|
)
|
|
context.skill_tool_result = result
|
|
|
|
|
|
@then("the temporary file should be cleaned up after edit")
|
|
def step_then_edit_tmp_cleaned(context: Any) -> None:
|
|
sandbox = Path(context.skill_sandbox_dir)
|
|
tmp_files = list(sandbox.glob("*.tmp"))
|
|
assert len(tmp_files) == 0, f"Temp files not cleaned up: {tmp_files}"
|
|
|
|
|
|
@then("the original edit exception should propagate")
|
|
def step_then_edit_exception_propagated(context: Any) -> None:
|
|
assert not context.skill_tool_result.success
|
|
assert "Simulated chmod failure" in (context.skill_tool_result.error or "")
|