HAL9000
2e3e116b42
Remove duplicate step function definition in tool_builtins_steps.py that was shadowing the decorated @when handler, causing the Behave step to be orphaned at runtime. Add CHANGELOG entry documenting the encoding fix for issue #7559. ISSUES CLOSED: #7559 # Conflicts: # CHANGELOG.md
479 lines
17 KiB
Python
479 lines
17 KiB
Python
"""Step definitions for the Built-in File Tools feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.tool.builtins import (
|
|
register_file_tools,
|
|
register_file_tools_with_changeset,
|
|
)
|
|
from cleveragents.tool.builtins.changeset import (
|
|
ChangeSetCapture,
|
|
ChangeSetEntry,
|
|
_detect_operation,
|
|
)
|
|
from cleveragents.tool.registry import ToolRegistry
|
|
from cleveragents.tool.runner import ToolRunner
|
|
|
|
__all__: list[str] = []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _run_tool(context: Any, tool_name: str, inputs: dict[str, Any]) -> None:
|
|
"""Execute a built-in file tool through the runner."""
|
|
registry = ToolRegistry()
|
|
if hasattr(context, "changeset_capture") and context.changeset_capture is not None:
|
|
register_file_tools_with_changeset(registry, context.changeset_capture)
|
|
else:
|
|
register_file_tools(registry)
|
|
runner = ToolRunner(registry)
|
|
inputs["sandbox_root"] = context.sandbox_dir
|
|
context.tool_result = runner.execute(tool_name, inputs)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Givens
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a temporary sandbox directory")
|
|
def step_given_sandbox(context: Any) -> None:
|
|
context.sandbox_dir = tempfile.mkdtemp()
|
|
context._cleanup_handlers.append(
|
|
lambda: shutil.rmtree(context.sandbox_dir, ignore_errors=True)
|
|
)
|
|
|
|
|
|
@given('a file "{name}" with content "{content}"')
|
|
def step_given_file_with_content(context: Any, name: str, content: str) -> None:
|
|
path = Path(context.sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content)
|
|
|
|
|
|
@given('an encoded file "{name}" with encoding "{encoding}" and content "{content}"')
|
|
def step_given_file_with_content_encoding(
|
|
context: Any, name: str, encoding: str, content: str
|
|
) -> None:
|
|
path = Path(context.sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding=encoding)
|
|
|
|
|
|
@given('a file "{name}" with lines "{csv_lines}"')
|
|
def step_given_file_with_lines(context: Any, name: str, csv_lines: str) -> None:
|
|
lines = csv_lines.split(",")
|
|
path = Path(context.sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("\n".join(lines) + "\n")
|
|
|
|
|
|
@given('a changeset capture with plan_id "{plan_id}"')
|
|
def step_given_changeset_capture(context: Any, plan_id: str) -> None:
|
|
context.changeset_capture = ChangeSetCapture(plan_id=plan_id)
|
|
|
|
|
|
@given('a subdirectory "{name}" in the sandbox')
|
|
def step_given_subdirectory(context: Any, name: str) -> None:
|
|
path = Path(context.sandbox_dir) / name
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
@given("a sibling directory with a name that is a prefix of the sandbox name")
|
|
def step_given_sibling_prefix_dir(context: Any) -> None:
|
|
"""Create a sibling directory whose name is a string prefix of the sandbox dir.
|
|
|
|
For example, if the sandbox is /tmp/abc123, this creates /tmp/abc123-escape.
|
|
This exercises the path traversal bypass where str.startswith() would
|
|
incorrectly allow /tmp/abc123-escape to pass a /tmp/abc123 sandbox check.
|
|
The step stores the relative escape path on context for use in When steps.
|
|
"""
|
|
sandbox_path = Path(context.sandbox_dir)
|
|
sibling_name = sandbox_path.name + "-escape"
|
|
sibling_path = sandbox_path.parent / sibling_name
|
|
sibling_path.mkdir(parents=True, exist_ok=True)
|
|
context._cleanup_handlers.append(
|
|
lambda: shutil.rmtree(str(sibling_path), ignore_errors=True)
|
|
)
|
|
context.sibling_escape_dir = str(sibling_path)
|
|
# Relative path from sandbox root that resolves into the sibling directory
|
|
context.sibling_escape_rel_path = f"../{sibling_name}/secret.txt"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Whens
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I execute the "builtin/file-read" tool with path "{path}"')
|
|
def step_when_file_read(context: Any, path: str) -> None:
|
|
_run_tool(context, "builtin/file-read", {"path": path})
|
|
|
|
|
|
@when("I attempt to read a file in the sibling escape directory")
|
|
def step_when_read_sibling_escape(context: Any) -> None:
|
|
"""Attempt to read a file using the dynamic sibling-escape relative path.
|
|
|
|
Uses the path stored by the 'sibling directory with a name that is a prefix'
|
|
Given step to exercise the prefix-collision path traversal bypass.
|
|
"""
|
|
_run_tool(context, "builtin/file-read", {"path": context.sibling_escape_rel_path})
|
|
|
|
|
|
@when(
|
|
'I execute the "builtin/file-read" tool with path "{path}" offset {offset:d} limit {limit:d}'
|
|
)
|
|
def step_when_file_read_offset_limit(
|
|
context: Any, path: str, offset: int, limit: int
|
|
) -> None:
|
|
_run_tool(
|
|
context, "builtin/file-read", {"path": path, "offset": offset, "limit": limit}
|
|
)
|
|
|
|
|
|
@when(
|
|
'I execute the "builtin/file-write" tool with path "{path}" and content "{content}"'
|
|
)
|
|
def step_when_file_write(context: Any, path: str, content: str) -> None:
|
|
_run_tool(context, "builtin/file-write", {"path": path, "content": content})
|
|
|
|
|
|
@when('I execute the "builtin/file-edit" tool replacing "{old}" with "{new}"')
|
|
def step_when_file_edit(context: Any, old: str, new: str) -> None:
|
|
_run_tool(
|
|
context,
|
|
"builtin/file-edit",
|
|
{"path": "edit.txt", "old_text": old, "new_text": new},
|
|
)
|
|
|
|
|
|
@when(
|
|
'I edit file "edit-enc.txt" replacing "{old}" with "{new}" specifying encoding "{encoding}"'
|
|
)
|
|
def step_when_file_edit_with_encoding(
|
|
context: Any, old: str, new: str, encoding: str
|
|
) -> None:
|
|
_run_tool(
|
|
context,
|
|
"builtin/file-edit",
|
|
{
|
|
"path": "edit-enc.txt",
|
|
"old_text": old,
|
|
"new_text": new,
|
|
"encoding": encoding,
|
|
},
|
|
)
|
|
|
|
|
|
@when('I execute the "builtin/file-edit" tool replacing all "{old}" with "{new}"')
|
|
def step_when_file_edit_all(context: Any, old: str, new: str) -> None:
|
|
_run_tool(
|
|
context,
|
|
"builtin/file-edit",
|
|
{"path": "edit.txt", "old_text": old, "new_text": new, "replace_all": True},
|
|
)
|
|
|
|
|
|
@when('I execute the "builtin/file-delete" tool with path "{path}"')
|
|
def step_when_file_delete(context: Any, path: str) -> None:
|
|
_run_tool(context, "builtin/file-delete", {"path": path})
|
|
|
|
|
|
@when('I list directory "{path}" with the file-list tool')
|
|
def step_when_file_list(context: Any, path: str) -> None:
|
|
_run_tool(context, "builtin/file-list", {"path": path})
|
|
|
|
|
|
@when('I list directory "{path}" filtering by pattern "{pattern}"')
|
|
def step_when_file_list_pattern(context: Any, path: str, pattern: str) -> None:
|
|
_run_tool(context, "builtin/file-list", {"path": path, "pattern": pattern})
|
|
|
|
|
|
@when('I list directory "{path}" recursively with the file-list tool')
|
|
def step_when_file_list_recursive(context: Any, path: str) -> None:
|
|
_run_tool(context, "builtin/file-list", {"path": path, "recursive": True})
|
|
|
|
|
|
@when('I search directory "{path}" for pattern "{pattern}"')
|
|
def step_when_file_search(context: Any, path: str, pattern: str) -> None:
|
|
_run_tool(context, "builtin/file-search", {"path": path, "pattern": pattern})
|
|
|
|
|
|
@when(
|
|
'I search directory "{path}" for pattern "{pattern}" with max_results {max_results:d}'
|
|
)
|
|
def step_when_file_search_max(
|
|
context: Any, path: str, pattern: str, max_results: int
|
|
) -> None:
|
|
_run_tool(
|
|
context,
|
|
"builtin/file-search",
|
|
{"path": path, "pattern": pattern, "max_results": max_results},
|
|
)
|
|
|
|
|
|
@when("I write a file through changeset-wrapped tools")
|
|
def step_when_write_changeset(context: Any) -> None:
|
|
_run_tool(
|
|
context, "builtin/file-write", {"path": "cs-new.txt", "content": "changeset"}
|
|
)
|
|
|
|
|
|
@when("I edit a file through changeset-wrapped tools")
|
|
def step_when_edit_changeset(context: Any) -> None:
|
|
_run_tool(
|
|
context,
|
|
"builtin/file-edit",
|
|
{"path": "cs-edit.txt", "old_text": "hello", "new_text": "goodbye"},
|
|
)
|
|
|
|
|
|
@when("I delete a file through changeset-wrapped tools")
|
|
def step_when_delete_changeset(context: Any) -> None:
|
|
_run_tool(context, "builtin/file-delete", {"path": "cs-delete.txt"})
|
|
|
|
|
|
@when("I perform multiple changeset-tracked operations")
|
|
def step_when_multiple_changeset_ops(context: Any) -> None:
|
|
_run_tool(context, "builtin/file-write", {"path": "multi1.txt", "content": "a"})
|
|
_run_tool(context, "builtin/file-write", {"path": "multi2.txt", "content": "b"})
|
|
# Edit multi1.txt
|
|
_run_tool(
|
|
context,
|
|
"builtin/file-edit",
|
|
{"path": "multi1.txt", "old_text": "a", "new_text": "c"},
|
|
)
|
|
|
|
|
|
@when("I clear the changeset")
|
|
def step_when_clear_changeset(context: Any) -> None:
|
|
context.changeset_capture.clear()
|
|
|
|
|
|
@when("I register all file tools")
|
|
def step_when_register_all(context: Any) -> None:
|
|
register_file_tools(context.registry)
|
|
|
|
|
|
@when("I register file tools with changeset")
|
|
def step_when_register_with_changeset(context: Any) -> None:
|
|
register_file_tools_with_changeset(context.registry, context.changeset_capture)
|
|
|
|
|
|
@when('I edit "{filename}" through changeset-wrapped tools')
|
|
def step_when_edit_specific_changeset(context: Any, filename: str) -> None:
|
|
_run_tool(
|
|
context,
|
|
"builtin/file-edit",
|
|
{"path": filename, "old_text": "original", "new_text": "modified"},
|
|
)
|
|
|
|
|
|
@when("I write a new file through raw changeset capture")
|
|
def step_when_raw_capture_write_new(context: Any) -> None:
|
|
"""Test _detect_operation fallback when before is None and after exists."""
|
|
# Simulate: before=None, after=hash, output has no 'created' key
|
|
op = _detect_operation("builtin/file-write", None, "abc123hash", {})
|
|
context.changeset_capture._entries.append(
|
|
ChangeSetEntry(
|
|
operation=op,
|
|
path="new-fallback.txt",
|
|
before_hash=None,
|
|
after_hash="abc123hash",
|
|
metadata={"tool": "builtin/file-write"},
|
|
)
|
|
)
|
|
|
|
|
|
@when("I overwrite a file through raw changeset capture")
|
|
def step_when_raw_capture_overwrite(context: Any) -> None:
|
|
"""Test _detect_operation fallback when before and after both exist."""
|
|
# Simulate: before=hash, after=different_hash, output has no 'created' key
|
|
op = _detect_operation("builtin/file-write", "hash1", "hash2", {})
|
|
context.changeset_capture._entries.append(
|
|
ChangeSetEntry(
|
|
operation=op,
|
|
path="existing-cs.txt",
|
|
before_hash="hash1",
|
|
after_hash="hash2",
|
|
metadata={"tool": "builtin/file-write"},
|
|
)
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Thens
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the output "{key}" should be "{value}"')
|
|
def step_then_output_str(context: Any, key: str, value: str) -> None:
|
|
assert context.tool_result.success, f"Tool failed: {context.tool_result.error}"
|
|
assert str(context.tool_result.output[key]) == value, (
|
|
f"Expected {key}={value}, got {context.tool_result.output[key]}"
|
|
)
|
|
|
|
|
|
@then('the output "{key}" should be greater than {value:d}')
|
|
def step_then_output_gt(context: Any, key: str, value: int) -> None:
|
|
assert context.tool_result.output[key] > value
|
|
|
|
|
|
@then('the output "{key}" should be boolean true')
|
|
def step_then_output_true(context: Any, key: str) -> None:
|
|
assert context.tool_result.output[key] is True
|
|
|
|
|
|
@then('the output "{key}" should be boolean false')
|
|
def step_then_output_false(context: Any, key: str) -> None:
|
|
assert context.tool_result.output[key] is False
|
|
|
|
|
|
@then('the output "{key}" should equal {value:d}')
|
|
def step_then_output_int(context: Any, key: str, value: int) -> None:
|
|
assert context.tool_result.output[key] == value, (
|
|
f"Expected {key}={value}, got {context.tool_result.output[key]}"
|
|
)
|
|
|
|
|
|
@then('the output "{key}" should contain "{text}"')
|
|
def step_then_output_contains(context: Any, key: str, text: str) -> None:
|
|
assert text in context.tool_result.output[key], (
|
|
f"Expected '{text}' in {key}: {context.tool_result.output[key]}"
|
|
)
|
|
|
|
|
|
@then('the output "{key}" should not contain "{text}"')
|
|
def step_then_output_not_contains(context: Any, key: str, text: str) -> None:
|
|
assert text not in context.tool_result.output[key]
|
|
|
|
|
|
@then('the file "{name}" should exist in the sandbox')
|
|
def step_then_file_exists(context: Any, name: str) -> None:
|
|
path = Path(context.sandbox_dir) / name
|
|
assert path.exists(), f"File {path} does not exist"
|
|
|
|
|
|
@then('the file "{name}" should not exist in the sandbox')
|
|
def step_then_file_not_exists(context: Any, name: str) -> None:
|
|
path = Path(context.sandbox_dir) / name
|
|
assert not path.exists(), f"File {path} still exists"
|
|
|
|
|
|
@then("the output files list should contain at least {count:d} entries")
|
|
def step_then_files_count(context: Any, count: int) -> None:
|
|
assert context.tool_result.success
|
|
files = context.tool_result.output["files"]
|
|
assert len(files) >= count, f"Expected >= {count} files, got {len(files)}"
|
|
|
|
|
|
@then('all listed files should match pattern "{pattern}"')
|
|
def step_then_files_match_pattern(context: Any, pattern: str) -> None:
|
|
assert context.tool_result.success
|
|
files = context.tool_result.output["files"]
|
|
for f in files:
|
|
assert fnmatch.fnmatch(f["name"], pattern), (
|
|
f"File {f['name']} does not match pattern {pattern}"
|
|
)
|
|
|
|
|
|
@then("the search matches should contain at least {count:d} result")
|
|
def step_then_search_min(context: Any, count: int) -> None:
|
|
assert context.tool_result.success
|
|
matches = context.tool_result.output["matches"]
|
|
assert len(matches) >= count, f"Expected >= {count} matches, got {len(matches)}"
|
|
|
|
|
|
@then("the search matches should contain exactly {count:d} results")
|
|
def step_then_search_exact(context: Any, count: int) -> None:
|
|
assert context.tool_result.success
|
|
matches = context.tool_result.output["matches"]
|
|
assert len(matches) == count, f"Expected {count} matches, got {len(matches)}"
|
|
|
|
|
|
@then("the changeset should have {count:d} entry")
|
|
def step_then_changeset_count_singular(context: Any, count: int) -> None:
|
|
cs = context.changeset_capture.get_changeset()
|
|
assert len(cs.entries) == count, f"Expected {count} entries, got {len(cs.entries)}"
|
|
|
|
|
|
@then("the changeset should have {count:d} entries")
|
|
def step_then_changeset_count_plural(context: Any, count: int) -> None:
|
|
cs = context.changeset_capture.get_changeset()
|
|
assert len(cs.entries) == count, f"Expected {count} entries, got {len(cs.entries)}"
|
|
|
|
|
|
@then('the changeset entry operation should be "{operation}"')
|
|
def step_then_changeset_operation(context: Any, operation: str) -> None:
|
|
cs = context.changeset_capture.get_changeset()
|
|
assert len(cs.entries) > 0, "No changeset entries"
|
|
assert cs.entries[-1].operation == operation, (
|
|
f"Expected operation '{operation}', got '{cs.entries[-1].operation}'"
|
|
)
|
|
|
|
|
|
@then("the changeset summary should mention the total count")
|
|
def step_then_summary_count(context: Any) -> None:
|
|
cs = context.changeset_capture.get_changeset()
|
|
summary = cs.summary()
|
|
count = len(cs.entries)
|
|
assert str(count) in summary, f"Expected count {count} in summary: {summary}"
|
|
|
|
|
|
@then("the changeset summary should mention operation types")
|
|
def step_then_summary_ops(context: Any) -> None:
|
|
cs = context.changeset_capture.get_changeset()
|
|
summary = cs.summary()
|
|
assert "create" in summary or "modify" in summary or "delete" in summary, (
|
|
f"No operation types in summary: {summary}"
|
|
)
|
|
|
|
|
|
@then('the changeset summary should be "{expected}"')
|
|
def step_then_summary_exact(context: Any, expected: str) -> None:
|
|
cs = context.changeset_capture.get_changeset()
|
|
assert cs.summary() == expected, f"Expected '{expected}', got '{cs.summary()}'"
|
|
|
|
|
|
@then("the registry should contain {count:d} tools")
|
|
def step_then_registry_count(context: Any, count: int) -> None:
|
|
tools = context.registry.list_tools()
|
|
assert len(tools) == count, f"Expected {count} tools, got {len(tools)}"
|
|
|
|
|
|
@then('the registry should contain tool "{name}"')
|
|
def step_then_registry_has_tool(context: Any, name: str) -> None:
|
|
assert context.registry.get(name) is not None, (
|
|
f"Tool '{name}' not found in registry"
|
|
)
|
|
|
|
|
|
@then("the changeset entry should have a before_hash")
|
|
def step_then_has_before_hash(context: Any) -> None:
|
|
cs = context.changeset_capture.get_changeset()
|
|
assert cs.entries[-1].before_hash is not None
|
|
|
|
|
|
@then("the changeset entry should have an after_hash")
|
|
def step_then_has_after_hash(context: Any) -> None:
|
|
cs = context.changeset_capture.get_changeset()
|
|
assert cs.entries[-1].after_hash is not None
|
|
|
|
|
|
@then("the before_hash and after_hash should differ")
|
|
def step_then_hashes_differ(context: Any) -> None:
|
|
cs = context.changeset_capture.get_changeset()
|
|
entry = cs.entries[-1]
|
|
assert entry.before_hash != entry.after_hash, "Hashes should differ after edit"
|