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).
475 lines
17 KiB
Python
475 lines
17 KiB
Python
"""Step definitions for the Skill Search and Directory Tools feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
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.search_ops import register_skill_search_tools
|
|
from cleveragents.tool.registry import ToolRegistry
|
|
from cleveragents.tool.runner import ToolRunner
|
|
|
|
__all__: list[str] = []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _run_search_tool(context: Any, tool_name: str, inputs: dict[str, Any]) -> None:
|
|
"""Execute a skill search tool through the runner."""
|
|
registry = ToolRegistry()
|
|
register_skill_search_tools(registry)
|
|
runner = ToolRunner(registry)
|
|
inputs["sandbox_root"] = context.search_sandbox_dir
|
|
context.search_tool_result = runner.execute(tool_name, inputs)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Givens
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a skill search sandbox directory")
|
|
def step_given_search_sandbox(context: Any) -> None:
|
|
context.search_sandbox_dir = tempfile.mkdtemp()
|
|
context._cleanup_handlers.append(
|
|
lambda: shutil.rmtree(context.search_sandbox_dir, ignore_errors=True)
|
|
)
|
|
|
|
|
|
@given('a search sandbox file "{name}" with content "{content}"')
|
|
def step_given_search_file(context: Any, name: str, content: str) -> None:
|
|
path = Path(context.search_sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content)
|
|
|
|
|
|
@given('a search sandbox file "{name}" with lines "{csv_lines}"')
|
|
def step_given_search_file_lines(context: Any, name: str, csv_lines: str) -> None:
|
|
lines = csv_lines.split(",")
|
|
path = Path(context.search_sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("\n".join(lines) + "\n")
|
|
|
|
|
|
@given('a search sandbox subdirectory "{name}"')
|
|
def step_given_search_subdir(context: Any, name: str) -> None:
|
|
path = Path(context.search_sandbox_dir) / name
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
@given('a search sandbox binary file "{name}"')
|
|
def step_given_search_binary(context: Any, name: str) -> None:
|
|
path = Path(context.search_sandbox_dir) / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
|
|
|
|
|
|
@given("a skill search tools registry")
|
|
def step_given_search_registry(context: Any) -> None:
|
|
context.search_registry = ToolRegistry()
|
|
register_skill_search_tools(context.search_registry)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Whens - ListDir
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I list-dir "{path}" in the search sandbox')
|
|
def step_when_list_dir(context: Any, path: str) -> None:
|
|
_run_search_tool(context, "skill/list-dir", {"path": path})
|
|
|
|
|
|
@when('I list-dir-typed "{path}" with type_filter "{type_filter}"')
|
|
def step_when_list_dir_type(context: Any, path: str, type_filter: str) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/list-dir",
|
|
{"path": path, "type_filter": type_filter},
|
|
)
|
|
|
|
|
|
@when('I list-dir-ignoring "{path}" ignoring "{pattern}"')
|
|
def step_when_list_dir_ignore(context: Any, path: str, pattern: str) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/list-dir",
|
|
{"path": path, "ignore_patterns": [pattern]},
|
|
)
|
|
|
|
|
|
@when('I list-dir-limited "{path}" with limit {limit:d}')
|
|
def step_when_list_dir_limit(context: Any, path: str, limit: int) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/list-dir",
|
|
{"path": path, "limit": limit},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Whens - Glob
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I glob-match "{path}" for "{pattern}"')
|
|
def step_when_glob_basic(context: Any, path: str, pattern: str) -> None:
|
|
_run_search_tool(context, "skill/glob", {"path": path, "pattern": pattern})
|
|
|
|
|
|
@when('I glob-exclude "{path}" for "{pattern}" excluding "{exclude}"')
|
|
def step_when_glob_excl(context: Any, path: str, pattern: str, exclude: str) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/glob",
|
|
{
|
|
"path": path,
|
|
"pattern": pattern,
|
|
"exclude_patterns": [exclude],
|
|
},
|
|
)
|
|
|
|
|
|
@when('I glob-include "{path}" for "{pattern}" including "{include}"')
|
|
def step_when_glob_incl(context: Any, path: str, pattern: str, include: str) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/glob",
|
|
{
|
|
"path": path,
|
|
"pattern": pattern,
|
|
"include_patterns": [include],
|
|
},
|
|
)
|
|
|
|
|
|
@when('I glob-limited "{path}" for "{pattern}" with limit {limit:d}')
|
|
def step_when_glob_lim(context: Any, path: str, pattern: str, limit: int) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/glob",
|
|
{"path": path, "pattern": pattern, "limit": limit},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Whens - Grep
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I grep-search "{path}" for "{pattern}"')
|
|
def step_when_grep_basic(context: Any, path: str, pattern: str) -> None:
|
|
_run_search_tool(context, "skill/grep", {"path": path, "pattern": pattern})
|
|
|
|
|
|
@when('I grep-context "{path}" for "{pattern}" with context {ctx:d}')
|
|
def step_when_grep_ctx(context: Any, path: str, pattern: str, ctx: int) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/grep",
|
|
{"path": path, "pattern": pattern, "context_lines": ctx},
|
|
)
|
|
|
|
|
|
@when('I grep-capped "{path}" for "{pattern}" with per_file_cap {cap:d}')
|
|
def step_when_grep_cap(context: Any, path: str, pattern: str, cap: int) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/grep",
|
|
{"path": path, "pattern": pattern, "per_file_cap": cap},
|
|
)
|
|
|
|
|
|
@when('I grep-maxed "{path}" for "{pattern}" with max_matches {max_m:d}')
|
|
def step_when_grep_max(context: Any, path: str, pattern: str, max_m: int) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/grep",
|
|
{"path": path, "pattern": pattern, "max_matches": max_m},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Thens
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the search tool result should be successful")
|
|
def step_then_search_success(context: Any) -> None:
|
|
result = context.search_tool_result
|
|
assert result.success, f"Tool failed: {result.error}"
|
|
|
|
|
|
@then("the search tool result should not be successful")
|
|
def step_then_search_fail(context: Any) -> None:
|
|
result = context.search_tool_result
|
|
assert not result.success, "Expected failure but tool succeeded"
|
|
|
|
|
|
@then('the search output "{key}" should be greater than or equal to {value:d}')
|
|
def step_then_search_output_gte(context: Any, key: str, value: int) -> None:
|
|
assert context.search_tool_result.output[key] >= value, (
|
|
f"Expected {key}>={value}, got {context.search_tool_result.output[key]}"
|
|
)
|
|
|
|
|
|
@then('the search output "{key}" should be boolean true')
|
|
def step_then_search_output_true(context: Any, key: str) -> None:
|
|
assert context.search_tool_result.output[key] is True
|
|
|
|
|
|
@then('the search output "{key}" should be boolean false')
|
|
def step_then_search_output_false(context: Any, key: str) -> None:
|
|
assert context.search_tool_result.output[key] is False
|
|
|
|
|
|
@then('the search output "{key}" should equal {value:d}')
|
|
def step_then_search_output_eq(context: Any, key: str, value: int) -> None:
|
|
assert context.search_tool_result.output[key] == value, (
|
|
f"Expected {key}={value}, got {context.search_tool_result.output[key]}"
|
|
)
|
|
|
|
|
|
@then("all search entries should not be directories")
|
|
def step_then_no_dirs(context: Any) -> None:
|
|
entries = context.search_tool_result.output["entries"]
|
|
for e in entries:
|
|
assert not e["is_dir"], f"{e['name']} is a directory"
|
|
|
|
|
|
@then("all search entries should be directories")
|
|
def step_then_all_dirs(context: Any) -> None:
|
|
entries = context.search_tool_result.output["entries"]
|
|
assert len(entries) > 0, "Expected at least one directory entry"
|
|
for e in entries:
|
|
assert e["is_dir"], f"{e['name']} is not a directory"
|
|
|
|
|
|
@then('no search entry should have name "{name}"')
|
|
def step_then_no_entry_name(context: Any, name: str) -> None:
|
|
entries = context.search_tool_result.output["entries"]
|
|
names = [e["name"] for e in entries]
|
|
assert name not in names, f"Found unwanted entry: {name}"
|
|
|
|
|
|
@then('the search tool error should mention "{text}"')
|
|
def step_then_search_error(context: Any, text: str) -> None:
|
|
error = context.search_tool_result.error or ""
|
|
assert text.lower() in error.lower(), f"Expected '{text}' in error: {error}"
|
|
|
|
|
|
@then("the search glob total should be {count:d}")
|
|
def step_then_glob_total(context: Any, count: int) -> None:
|
|
total = context.search_tool_result.output["total"]
|
|
assert total == count, f"Expected {count} matches, got {total}"
|
|
|
|
|
|
@then("the search glob total should be at least {count:d}")
|
|
def step_then_glob_total_min(context: Any, count: int) -> None:
|
|
total = context.search_tool_result.output["total"]
|
|
assert total >= count, f"Expected >={count} matches, got {total}"
|
|
|
|
|
|
@then('no search match should have name "{name}"')
|
|
def step_then_no_match_name(context: Any, name: str) -> None:
|
|
matches = context.search_tool_result.output["matches"]
|
|
names = [m["name"] for m in matches]
|
|
assert name not in names, f"Found unwanted match: {name}"
|
|
|
|
|
|
@then("the search glob results should be sorted by name")
|
|
def step_then_glob_sorted(context: Any) -> None:
|
|
matches = context.search_tool_result.output["matches"]
|
|
names = [m["name"] for m in matches]
|
|
assert names == sorted(names), f"Not sorted: {names}"
|
|
|
|
|
|
@then("the search grep total should be at least {count:d}")
|
|
def step_then_grep_total_min(context: Any, count: int) -> None:
|
|
total = context.search_tool_result.output["total"]
|
|
assert total >= count, f"Expected >={count} matches, got {total}"
|
|
|
|
|
|
@then("the search grep total should equal {count:d}")
|
|
def step_then_grep_total_eq(context: Any, count: int) -> None:
|
|
total = context.search_tool_result.output["total"]
|
|
assert total == count, f"Expected {count} matches, got {total}"
|
|
|
|
|
|
@then("the search grep first match should have context_before")
|
|
def step_then_grep_context_before(context: Any) -> None:
|
|
matches = context.search_tool_result.output["matches"]
|
|
assert len(matches) > 0, "No matches found"
|
|
assert "context_before" in matches[0], "Missing context_before"
|
|
|
|
|
|
@then("the search grep first match should have context_after")
|
|
def step_then_grep_context_after(context: Any) -> None:
|
|
matches = context.search_tool_result.output["matches"]
|
|
assert len(matches) > 0, "No matches found"
|
|
assert "context_after" in matches[0], "Missing context_after"
|
|
|
|
|
|
@then('the search grep matches should not reference "{name}"')
|
|
def step_then_grep_no_file(context: Any, name: str) -> None:
|
|
matches = context.search_tool_result.output["matches"]
|
|
for m in matches:
|
|
assert name not in m["file"], f"Found match in excluded file: {m['file']}"
|
|
|
|
|
|
@then('the search tool "{name}" should have read_only capability')
|
|
def step_then_search_tool_readonly(context: Any, name: str) -> None:
|
|
spec = context.search_registry.get(name)
|
|
assert spec is not None, f"Tool {name} not found"
|
|
assert spec.capabilities.read_only, f"{name} should be read_only"
|
|
|
|
|
|
@then("the search registry should contain {count:d} tools")
|
|
def step_then_search_registry_count(context: Any, count: int) -> None:
|
|
tools = context.search_registry.list_tools()
|
|
assert len(tools) == count, f"Expected {count} tools, got {len(tools)}"
|
|
|
|
|
|
@then('the search registry should contain tool "{name}"')
|
|
def step_then_search_registry_has(context: Any, name: str) -> None:
|
|
assert context.search_registry.get(name) is not None, (
|
|
f"Tool '{name}' not found in registry"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Coverage: OSError fallbacks, include patterns, directory skip
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I execute list-dir with a stat failure mock")
|
|
def step_list_dir_stat_failure(context: Any) -> None:
|
|
# Create a broken symlink so stat() raises OSError naturally
|
|
link = Path(context.search_sandbox_dir) / "statfail.txt"
|
|
if link.exists() or link.is_symlink():
|
|
link.unlink()
|
|
os.symlink("/nonexistent/path/target", str(link))
|
|
|
|
_run_search_tool(context, "skill/list-dir", {"path": "."})
|
|
|
|
|
|
@then('the list-dir result should contain "{name}" with size 0')
|
|
def step_then_listdir_zero_size(context: Any, name: str) -> None:
|
|
assert context.search_tool_result.success
|
|
entries = context.search_tool_result.output["entries"]
|
|
found = [e for e in entries if e["name"] == name]
|
|
assert len(found) == 1, f"Expected entry '{name}', found: {entries}"
|
|
assert found[0]["size"] == 0, f"Expected size 0, got {found[0]['size']}"
|
|
|
|
|
|
@when('I execute glob with pattern "{pattern}" and include "{include}"')
|
|
def step_glob_with_include(context: Any, pattern: str, include: str) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/glob",
|
|
{"path": ".", "pattern": pattern, "include_patterns": [include]},
|
|
)
|
|
|
|
|
|
@then('the glob results should contain "{name}"')
|
|
def step_then_glob_has(context: Any, name: str) -> None:
|
|
assert context.search_tool_result.success
|
|
entries = context.search_tool_result.output["matches"]
|
|
names = [e["name"] for e in entries]
|
|
assert name in names, f"Expected '{name}' in {names}"
|
|
|
|
|
|
@then('the glob results should not contain "{name}"')
|
|
def step_then_glob_not_has(context: Any, name: str) -> None:
|
|
assert context.search_tool_result.success
|
|
entries = context.search_tool_result.output["matches"]
|
|
names = [e["name"] for e in entries]
|
|
assert name not in names, f"Did not expect '{name}' in {names}"
|
|
|
|
|
|
@when('I execute glob with a stat failure mock and pattern "{pattern}"')
|
|
def step_glob_stat_failure(context: Any, pattern: str) -> None:
|
|
# Create a broken symlink so stat() raises OSError naturally
|
|
link = Path(context.search_sandbox_dir) / "globstat.txt"
|
|
if link.exists() or link.is_symlink():
|
|
link.unlink()
|
|
os.symlink("/nonexistent/path/target", str(link))
|
|
|
|
_run_search_tool(context, "skill/glob", {"path": ".", "pattern": pattern})
|
|
|
|
|
|
@then('the glob result should contain "{name}" with size 0')
|
|
def step_then_glob_zero_size(context: Any, name: str) -> None:
|
|
assert context.search_tool_result.success
|
|
entries = context.search_tool_result.output["matches"]
|
|
found = [e for e in entries if e["name"] == name]
|
|
assert len(found) == 1, f"Expected '{name}', found: {entries}"
|
|
assert found[0]["size"] == 0, f"Expected size 0, got {found[0]['size']}"
|
|
|
|
|
|
@when('I execute grep with pattern "{pattern}" and include "{include}"')
|
|
def step_grep_with_include(context: Any, pattern: str, include: str) -> None:
|
|
_run_search_tool(
|
|
context,
|
|
"skill/grep",
|
|
{"path": ".", "pattern": pattern, "include": include},
|
|
)
|
|
|
|
|
|
@then('the grep results should contain "{name}"')
|
|
def step_then_grep_has_file(context: Any, name: str) -> None:
|
|
assert context.search_tool_result.success
|
|
matches = context.search_tool_result.output["matches"]
|
|
files = [m["file"] for m in matches]
|
|
assert any(name in f for f in files), f"Expected '{name}' in {files}"
|
|
|
|
|
|
@then("the grep results should not contain a directory entry")
|
|
def step_then_grep_no_dir(context: Any) -> None:
|
|
assert context.search_tool_result.success
|
|
matches = context.search_tool_result.output["matches"]
|
|
for m in matches:
|
|
fpath = Path(context.search_sandbox_dir) / m["file"]
|
|
assert not fpath.is_dir(), f"Unexpected directory in results: {m['file']}"
|
|
|
|
|
|
@when('I execute grep with a read_text failure mock and pattern "{pattern}"')
|
|
def step_grep_read_failure(context: Any, pattern: str) -> None:
|
|
registry = ToolRegistry()
|
|
register_skill_search_tools(registry)
|
|
runner = ToolRunner(registry)
|
|
|
|
original_read_text = Path.read_text
|
|
|
|
def _fail_read(self: Path, *a: Any, **kw: Any) -> str:
|
|
if "unreadable" in str(self):
|
|
raise OSError("read failed")
|
|
return original_read_text(self, *a, **kw)
|
|
|
|
with patch.object(Path, "read_text", _fail_read):
|
|
context.search_tool_result = runner.execute(
|
|
"skill/grep",
|
|
{
|
|
"sandbox_root": context.search_sandbox_dir,
|
|
"path": ".",
|
|
"pattern": pattern,
|
|
"include": "*.txt",
|
|
},
|
|
)
|
|
|
|
|
|
@then("the grep result should be empty")
|
|
def step_then_grep_empty(context: Any) -> None:
|
|
assert context.search_tool_result.success
|
|
matches = context.search_tool_result.output["matches"]
|
|
assert len(matches) == 0, f"Expected empty matches, got {matches}"
|