cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

test: boost combined branch coverage to 97% with additional behave scenarios

Browse Source 
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).
This commit is contained in:
Jeffrey Phillips Freeman
2026-02-21 04:52:19 +00:00
parent 2881b0cf7a
commit 2980b14e7b
8 changed files with 1766 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/output_rendering.feature
+459
View File
@@ -370,3 +370,462 @@ Feature: Output Rendering Framework
Given a non-TTY terminal environment
When I select a materializer for format "rich" with explicit flag
Then the selected materializer is RichMaterializer
# ================================================================
# Coverage boost: handle validation guards
# ================================================================
Scenario: PanelHandle rejects empty key in set_entry
Given an OutputSession with format "plain"
When I create a panel handle with title "V"
Then setting entry with empty key raises ValueError
Scenario: PanelHandle rejects None entries in set_entries
Given an OutputSession with format "plain"
When I create a panel handle with title "V"
Then batch setting None entries raises ValueError
Scenario: PanelHandle rejects empty key in remove_entry
Given an OutputSession with format "plain"
When I create a panel handle with title "V"
Then removing entry with empty key raises ValueError
Scenario: TableHandle rejects None row in add_row
Given an OutputSession with format "plain"
When I create a table handle with columns "A,B"
Then adding a None row raises ValueError
Scenario: TableHandle rejects None rows in add_rows
Given an OutputSession with format "plain"
When I create a table handle with columns "A,B"
Then adding None rows raises ValueError
Scenario: TableHandle rejects None summary
Given an OutputSession with format "plain"
When I create a table handle with columns "A,B"
Then setting None summary raises ValueError
Scenario: TableHandle rejects empty sort key
Given an OutputSession with format "plain"
When I create a table handle with columns "A,B"
Then setting empty sort key raises ValueError
Scenario: StatusHandle rejects empty message
Given an OutputSession with format "plain"
When I create a status handle with message "ok"
Then setting empty status message raises ValueError
Scenario: StatusHandle rejects empty level
Given an OutputSession with format "plain"
When I create a status handle with message "ok"
Then setting empty status level raises ValueError
Scenario: ProgressHandle rejects empty label
Given an OutputSession with format "plain"
When I create a progress handle labelled "Loading"
Then setting empty progress label raises ValueError
Scenario: ProgressHandle rejects negative delta
Given an OutputSession with format "plain"
When I create a progress handle labelled "Loading" with total 10
Then incrementing by negative delta raises ValueError
Scenario: ProgressHandle rejects empty step label
Given an OutputSession with format "plain"
When I create a progress handle labelled "Loading" with steps "A,B"
Then setting step status with empty label raises ValueError
# ================================================================
# Coverage boost: handle data paths
# ================================================================
Scenario: PanelHandle set_entries updates existing entry
Given an OutputSession with format "plain"
When I create a panel handle with title "UP"
And I set entry "K" to "old"
And I batch set entries with keys "K" and values "new"
Then the panel entry "K" has value "new"
Scenario: TableHandle add_rows with list-type rows
Given an OutputSession with format "plain"
When I create a table handle with columns "A,B"
And I batch add list rows with values "x,y" and "m,n"
Then the table has 2 rows
Scenario: ProgressHandle increment adds to current
Given an OutputSession with format "plain"
When I create a progress handle labelled "Down" with total 100
And I set progress to 10 of 100
And I increment progress by 5
Then the progress current is 15
Scenario: ProgressHandle set_step_status creates steps list
Given an OutputSession with format "plain"
When I create a progress handle labelled "Steps"
And I set step "Alpha" status to "done"
Then step "Alpha" has status "done"
# ================================================================
# Coverage boost: color materializer rendering
# ================================================================
Scenario: Color materializer renders table with ANSI codes
Given an OutputSession with format "color" using ColorMaterializer
When I create a table handle with columns "Name,Status"
And I add a dict row with Name "foo" and Status "ok"
And I close the table handle
And the session is closed
Then the color output contains ANSI codes
And the color output contains "foo"
Scenario: Color materializer renders empty table
Given an OutputSession with format "color" using ColorMaterializer
When I create a table with no columns
And I close the table handle
And the session is closed
Then the color output contains "empty table"
Scenario: Color materializer renders status with detail
Given an OutputSession with format "color" using ColorMaterializer
When I create a status handle with message "Building"
And I set the status detail to "step 3 of 10"
And I close the status handle
And the session is closed
Then the color output contains ANSI codes
And the color output contains "Building"
Scenario: Color materializer renders progress indicator
Given an OutputSession with format "color" using ColorMaterializer
When I create a progress handle labelled "Downloading" with total 50
And I set progress to 25 of 50
And I close the progress handle
And the session is closed
Then the color output contains ANSI codes
And the color output contains "Downloading"
Scenario: Color materializer renders panel with style hints
Given an OutputSession with format "color" using ColorMaterializer
When I create a panel handle with title "Results"
And I set entry with key "passed" value "yes" and style hint "success"
And I set entry with key "warnings" value "3" and style hint "warning"
And I set entry with key "errors" value "1" and style hint "error"
And I close the panel handle
And the session is closed
Then the color output contains ANSI codes
And the color output contains "passed"
# ================================================================
# Coverage boost: plain materializer edge cases
# ================================================================
Scenario: Plain materializer renders empty table
Given an OutputSession with format "plain" using PlainMaterializer
When I create a table with no columns
And I close the table handle
And the session is closed
Then the plain output contains "empty table"
Scenario: Plain materializer renders table with title and summary
Given an OutputSession with format "plain" using PlainMaterializer
When I create a table handle with columns "Name,Count"
And I set table title to "Report"
And I add a dict row with Name "A" and Status "10"
And I set summary with total "10"
And I close the table handle
And the session is closed
Then the plain output contains "Report"
Scenario: Plain materializer renders status with detail
Given an OutputSession with format "plain" using PlainMaterializer
When I create a status handle with message "Processing"
And I set the status detail to "file 3 of 10"
And I close the status handle
And the session is closed
Then the plain output contains "Processing"
And the plain output contains "file 3 of 10"
Scenario: Plain materializer renders indeterminate progress
Given an OutputSession with format "plain" using PlainMaterializer
When I create a progress handle labelled "Scanning"
And I set progress to indeterminate mode
And I close the progress handle
And the session is closed
Then the plain output contains "Scanning"
Scenario: Plain materializer renders current-only progress
Given an OutputSession with format "plain" using PlainMaterializer
When I create a progress handle labelled "Items"
And I set progress current to 42
And I close the progress handle
And the session is closed
Then the plain output contains "Items"
And the plain output contains "42"
Scenario: Plain materializer renders progress with steps
Given an OutputSession with format "plain" using PlainMaterializer
When I create a progress handle labelled "Deploy" with steps "Build,Test,Ship"
And I set step "Build" status to "done"
And I set step "Test" status to "active"
And I close the progress handle
And the session is closed
Then the plain output contains "Build"
And the plain output contains "Test"
# ================================================================
# Coverage boost: serialization helpers (JSON/YAML)
# ================================================================
Scenario: JSON materializer serializes status element
Given an OutputSession with format "json" using JsonMaterializer
When I create a status handle with message "Done"
And I set the status detail to "All passed"
And I close the status handle
And the session is closed
Then the json output is valid JSON
And the json output contains element type "status"
And the json output contains "Done"
Scenario: JSON materializer serializes progress with steps
Given an OutputSession with format "json" using JsonMaterializer
When I create a progress handle labelled "Pipeline" with steps "A,B"
And I set step "A" status to "done"
And I close the progress handle
And the session is closed
Then the json output is valid JSON
And the json output contains element type "progress"
And the json output contains "Pipeline"
Scenario: JSON materializer serializes panel with style hint and icon
Given an OutputSession with format "json" using JsonMaterializer
When I create a panel handle with title "Info"
And I set entry with key "status" value "ok" and style hint "success"
And I set entry with icon key "icon_entry" value "check" and icon "check_mark"
And I close the panel handle
And the session is closed
Then the json output is valid JSON
And the json output contains "style_hint"
And the json output contains "icon"
Scenario: JSON error envelope includes details
Given a JsonMaterializer
When I generate an error envelope with code "ERR" message "fail" and details "traceback"
Then the json error output contains "traceback"
# ================================================================
# Coverage boost: box-drawing materializer edge cases
# ================================================================
Scenario: Box-drawing materializer renders empty table
Given an OutputSession with format "table" using TableMaterializer
When I create a table with no columns
And I close the table handle
And the session is closed
Then the strategy output contains "empty table"
Scenario: Box-drawing materializer renders table with title
Given an OutputSession with format "table" using TableMaterializer
When I create a table handle with columns "X,Y"
And I set table title to "Summary"
And I add a dict row with Name "X" and Status "1"
And I close the table handle
And the session is closed
Then the strategy output contains "Summary"
# ================================================================
# Coverage boost: format selection paths
# ================================================================
Scenario: Select yaml materializer
When I select a materializer for format "yaml"
Then the selected materializer is YamlMaterializer
Scenario: Select materializer rejects empty format
When I attempt to select a materializer with empty format
Then a ValueError is raised
Scenario: Rich format falls back to color when no cursor support
Given a terminal with ANSI but no cursor support
When I select a materializer for format "rich"
Then the selected materializer is ColorMaterializer
Scenario: Color format uses color materializer with ANSI support
Given a terminal with ANSI support
When I select a materializer for format "color"
Then the selected materializer is ColorMaterializer
Scenario: Table format uses table materializer on TTY
Given a TTY terminal with ANSI support
When I select a materializer for format "table"
Then the selected materializer is TableMaterializer
Scenario: Table format falls back to plain on non-TTY
Given a non-TTY terminal environment
When I select a materializer for format "table"
Then the selected materializer is PlainMaterializer
# ================================================================
# Coverage boost: session edge cases
# ================================================================
Scenario: Session rejects creating panel with empty title
Given an OutputSession with format "plain"
Then creating a panel with empty title raises ValueError
Scenario: Session rejects creating status with empty message
Given an OutputSession with format "plain"
Then creating a status with empty message raises ValueError
Scenario: Session rejects creating progress with empty label
Given an OutputSession with format "plain"
Then creating a progress with empty label raises ValueError
Scenario: Closed session rejects new handles
Given an OutputSession with format "plain"
When the session is closed
Then creating a panel on closed session raises RuntimeError
Scenario: Session close force-closes open handles
Given an OutputSession with format "plain" using PlainMaterializer
When I create a panel handle with title "Unclosed"
And I set entry "A" to "B"
And the session is closed
Then the plain output contains "Unclosed"
# ================================================================
# Coverage boost: ElementHandle validation guards
# ================================================================
Scenario: ElementHandle rejects empty handle_id
Given an OutputSession with format "plain"
Then constructing a handle with empty handle_id raises ValueError
Scenario: ElementHandle rejects empty element_type
Given an OutputSession with format "plain"
Then constructing a handle with empty element_type raises ValueError
Scenario: ElementHandle rejects negative declaration_index
Given an OutputSession with format "plain"
Then constructing a handle with negative index raises ValueError
Scenario: ElementHandle rejects None session
Given an OutputSession with format "plain"
Then constructing a handle with None session raises ValueError
Scenario: ElementHandle rejects None element
Given an OutputSession with format "plain"
Then constructing a handle with None element raises ValueError
# ================================================================
# Coverage boost: handle close and __exit__ edge cases
# ================================================================
Scenario: Closing an already-closed handle is a no-op
Given an OutputSession with format "plain" using PlainMaterializer
When I create a panel handle with title "CloseMe"
And I close the panel handle
And I close the panel handle again
Then the handle state is "closed"
Scenario: Handle used as context manager auto-closes
Given an OutputSession with format "plain" using PlainMaterializer
When I use a panel handle as context manager with title "AutoClose"
Then the handle state is "closed"
And the plain output contains "AutoClose"
# ================================================================
# Coverage boost: table rows as list (not dict)
# ================================================================
Scenario: TableHandle accepts list rows
Given an OutputSession with format "plain" using PlainMaterializer
When I create a table handle with columns "A,B"
And I add a list row with values "x,y"
And I close the table handle
And the session is closed
Then the plain output contains "x"
Scenario: TableHandle batch add accepts list rows
Given an OutputSession with format "plain" using PlainMaterializer
When I create a table handle with columns "A,B"
And I batch add list rows with values "1,2" and "3,4"
And I close the table handle
And the session is closed
Then the plain output contains "1"
# ================================================================
# Coverage boost: color materializer table with title
# ================================================================
Scenario: Color materializer renders table title
Given an OutputSession with format "color" using ColorMaterializer
When I create a table handle with columns "Col1,Col2"
And I set table title to "Colored Title"
And I add a dict row with Name "Col1" and Status "val"
And I close the table handle
And the session is closed
Then the color output contains "Colored Title"
# ================================================================
# Coverage boost: box-draw table dispatch for non-table
# ================================================================
Scenario: Table materializer renders panel as plain text
Given an OutputSession with format "table" using TableMaterializer
When I create a panel handle with title "PanelInTable"
And I set entry "info" to "details"
And I close the panel handle
And the session is closed
Then the strategy output contains "PanelInTable"
# ================================================================
# Coverage boost: session creates table with ColumnDef objects
# ================================================================
Scenario: Session table accepts ColumnDef objects
Given an OutputSession with format "plain" using PlainMaterializer
When I create a table with ColumnDef objects "Alpha,Beta"
And I add a dict row with Name "Alpha" and Status "val1"
And I close the table handle
And the session is closed
Then the plain output contains "Alpha"
# ================================================================
# Coverage boost: detect_capabilities function
# ================================================================
Scenario: Detect terminal capabilities returns TerminalCapabilities
When I detect terminal capabilities
Then the capabilities object has is_tty field
And the capabilities object has supports_ansi field
# ================================================================
# Coverage boost: explicit format flag
# ================================================================
Scenario: Explicit format flag creates strategy without fallback
When I select a materializer for format "rich" with explicit flag
Then the selected materializer is RichMaterializer
# ================================================================
# Coverage boost: serializer on_session_end with buffered content
# ================================================================
Scenario: Plain materializer flushes remaining buffers on session end
Given an OutputSession with format "plain" using PlainMaterializer
When I create a panel handle with title "First"
And I create a second panel handle with title "Second"
And I set entry "a" to "1" on handle 2
And I close handle 2
And I set entry "b" to "2"
And I close the panel handle
And the session is closed
Then the plain output contains "First"
And the plain output contains "Second"
# ================================================================
# Coverage boost: serializer get_output with None session
# ================================================================
Scenario: JSON serializer returns empty when session is None
Given a JsonMaterializer with no session
Then the serializer get_output returns empty string
features/skill_file_ops.feature
+34
View File
@@ -241,3 +241,37 @@ Feature: Skill File Operation Tools
When I execute the "skill/file-read" tool with path "binary.dat"
Then the skill tool result should not be successful
And the skill tool error should mention "Binary file detected"
# ---- Coverage: _is_binary OSError, create_dirs, atomic write cleanup ----
Scenario: _is_binary returns false for unreadable path
Given a skill file ops sandbox directory
When I check _is_binary on a nonexistent path
Then _is_binary should return false
Scenario: WriteFile creates parent directories with create_dirs
Given a skill file ops sandbox directory
When I execute write_file with create_dirs to "deep/nested/dir/file.txt" and content "hello dirs"
Then the skill tool result should be successful
And the written file "deep/nested/dir/file.txt" should contain "hello dirs"
Scenario: _is_fd_closed returns false for open descriptor
When I check _is_fd_closed with an open file descriptor
Then _is_fd_closed should return false
Scenario: _is_fd_closed returns true for closed descriptor
When I check _is_fd_closed with a closed file descriptor
Then _is_fd_closed should return true
Scenario: WriteFile atomic write cleanup on os.replace failure
Given a skill file ops sandbox directory
When I simulate atomic write failure during replace
Then the temporary file should be cleaned up
And the original exception should propagate
Scenario: EditFile atomic write cleanup on os.chmod failure
Given a skill file ops sandbox directory
And a sandbox file "editable.txt" with content "original content"
When I simulate edit atomic write failure during chmod
Then the temporary file should be cleaned up after edit
And the original edit exception should propagate
features/skill_search.feature
+36
View File
@@ -211,3 +211,39 @@ Feature: Skill Search and Directory Tools
And the search registry should contain tool "skill/list-dir"
And the search registry should contain tool "skill/glob"
And the search registry should contain tool "skill/grep"
# ---- Coverage: OSError fallbacks, include patterns, directory skip ----
Scenario: ListDir handles stat failure gracefully
Given a skill search sandbox directory
And a search sandbox file "statfail.txt" with content "test"
When I execute list-dir with a stat failure mock
Then the list-dir result should contain "statfail.txt" with size 0
Scenario: Glob filters by include pattern
Given a skill search sandbox directory
And a search sandbox file "keep.txt" with content "a"
And a search sandbox file "skip.py" with content "b"
When I execute glob with pattern "*" and include "*.txt"
Then the glob results should contain "keep.txt"
And the glob results should not contain "skip.py"
Scenario: Glob handles stat failure gracefully
Given a skill search sandbox directory
And a search sandbox file "globstat.txt" with content "test"
When I execute glob with a stat failure mock and pattern "*.txt"
Then the glob result should contain "globstat.txt" with size 0
Scenario: Grep skips directories in rglob
Given a skill search sandbox directory
And a search sandbox subdirectory "subdir"
And a search sandbox file "subdir/match.txt" with content "findme here"
When I execute grep with pattern "findme" and include "*"
Then the grep results should contain "match.txt"
And the grep results should not contain a directory entry
Scenario: Grep skips unreadable files gracefully
Given a skill search sandbox directory
And a search sandbox file "unreadable.txt" with content "content"
When I execute grep with a read_text failure mock and pattern "content"
Then the grep result should be empty
features/steps/output_rendering_steps.py
+445
View File
@@ -15,8 +15,11 @@ from cleveragents.cli.formatting import (
format_output_session,
)
from cleveragents.cli.output.handles import (
ColumnDef,
ElementClosedError,
ElementHandle,
Panel,
PanelEntry,
ProgressIndicator,
StatusMessage,
Table,
@@ -31,6 +34,7 @@ from cleveragents.cli.output.materializers import (
)
from cleveragents.cli.output.selection import (
TerminalCapabilities,
detect_terminal_capabilities,
select_materializer,
)
from cleveragents.cli.output.session import OutputSession
@@ -695,6 +699,15 @@ def step_select_invalid_format(context: Context, fmt: str) -> None:
context.selection_error = exc
@when("I attempt to select a materializer with empty format")
def step_select_empty_format(context: Context) -> None:
context.selection_error = None
try:
select_materializer("")
except ValueError as exc:
context.selection_error = exc
@then("a ValueError is raised")
def step_value_error_raised(context: Context) -> None:
assert context.selection_error is not None
@@ -762,3 +775,435 @@ def step_snapshot_count(context: Context, count: int) -> None:
assert len(snap.elements) == count, (
f"Expected {count} elements, got {len(snap.elements)}"
)
# ===========================================================================
# Coverage boost: handle validation guards
# ===========================================================================
@then("setting entry with empty key raises ValueError")
def step_set_entry_empty_key(context: Context) -> None:
try:
context.panel_handle.set_entry("", "value")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("batch setting None entries raises ValueError")
def step_batch_set_none(context: Context) -> None:
try:
context.panel_handle.set_entries(None) # type: ignore[arg-type]
raise AssertionError("Expected ValueError")
except (ValueError, TypeError):
pass
@then("removing entry with empty key raises ValueError")
def step_remove_empty_key(context: Context) -> None:
try:
context.panel_handle.remove_entry("")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("adding a None row raises ValueError")
def step_add_none_row(context: Context) -> None:
try:
context.table_handle.add_row(None) # type: ignore[arg-type]
raise AssertionError("Expected ValueError")
except (ValueError, TypeError):
pass
@then("adding None rows raises ValueError")
def step_add_none_rows(context: Context) -> None:
try:
context.table_handle.add_rows(None) # type: ignore[arg-type]
raise AssertionError("Expected ValueError")
except (ValueError, TypeError):
pass
@then("setting None summary raises ValueError")
def step_set_none_summary(context: Context) -> None:
try:
context.table_handle.set_summary(None) # type: ignore[arg-type]
raise AssertionError("Expected ValueError")
except (ValueError, TypeError):
pass
@then("setting empty sort key raises ValueError")
def step_set_empty_sort_key(context: Context) -> None:
try:
context.table_handle.set_sort_key("")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("setting empty status message raises ValueError")
def step_set_empty_status_msg(context: Context) -> None:
try:
context.status_handle.set_message("")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("setting empty status level raises ValueError")
def step_set_empty_status_level(context: Context) -> None:
try:
context.status_handle.set_level("")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("setting empty progress label raises ValueError")
def step_set_empty_progress_label(context: Context) -> None:
try:
context.progress_handle.set_label("")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("incrementing by negative delta raises ValueError")
def step_incr_negative(context: Context) -> None:
try:
context.progress_handle.increment(-1)
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("setting step status with empty label raises ValueError")
def step_set_step_empty_label(context: Context) -> None:
try:
context.progress_handle.set_step_status("", "complete")
raise AssertionError("Expected ValueError")
except ValueError:
pass
# ===========================================================================
# Coverage boost: handle data paths
# ===========================================================================
@when('I batch add list rows with values "{r1}" and "{r2}"')
def step_batch_list_rows(context: Context, r1: str, r2: str) -> None:
rows = [r1.split(","), r2.split(",")]
context.table_handle.add_rows(rows)
@when("I set progress current to {val:d}")
def step_set_progress_current(context: Context, val: int) -> None:
context.progress_handle.set_progress(val)
# ===========================================================================
# Coverage boost: materializer step helpers
# ===========================================================================
@when("I create a table with no columns")
def step_create_empty_table(context: Context) -> None:
context.table_handle = context.session.table(columns=[])
@when('I set table title to "{title}"')
def step_set_table_title(context: Context, title: str) -> None:
tbl: Table = context.table_handle.element # type: ignore[assignment]
tbl.title = title
@when('I set entry with key "{key}" value "{val}" and style hint "{hint}"')
def step_set_entry_style_hint(context: Context, key: str, val: str, hint: str) -> None:
panel: Panel = context.panel_handle.element # type: ignore[assignment]
panel.entries.append(PanelEntry(key=key, value=val, style_hint=hint))
@when('I set entry with icon key "{key}" value "{val}" and icon "{icon}"')
def step_set_entry_icon(context: Context, key: str, val: str, icon: str) -> None:
panel: Panel = context.panel_handle.element # type: ignore[assignment]
panel.entries.append(PanelEntry(key=key, value=val, icon=icon))
@when(
'I generate an error envelope with code "{code}" message "{msg}" and details "{det}"'
)
def step_json_error_with_details(
context: Context, code: str, msg: str, det: str
) -> None:
context.json_error_output = context.json_mat.get_error_output(
code, msg, details={"info": det}
)
@then('the strategy output contains "{text}"')
def step_strategy_output_contains(context: Context, text: str) -> None:
output = context.strategy.get_output()
assert text in output, f"Expected {text!r} in:\n{output}"
# ===========================================================================
# Coverage boost: format selection
# ===========================================================================
@given("a terminal with ANSI but no cursor support")
def step_ansi_no_cursor(context: Context) -> None:
context.terminal_caps = TerminalCapabilities(
is_tty=True, supports_ansi=True, supports_cursor=False, term="xterm"
)
@given("a terminal with ANSI support")
def step_ansi_support(context: Context) -> None:
context.terminal_caps = TerminalCapabilities(
is_tty=True, supports_ansi=True, supports_cursor=False, term="xterm"
)
@given("a TTY terminal with ANSI support")
def step_tty_ansi(context: Context) -> None:
context.terminal_caps = TerminalCapabilities(
is_tty=True, supports_ansi=True, supports_cursor=True, term="xterm-256color"
)
@then("the selected materializer is ColorMaterializer")
def step_mat_is_color(context: Context) -> None:
assert isinstance(context.selected_mat, ColorMaterializer)
@then("the selected materializer is YamlMaterializer")
def step_mat_is_yaml(context: Context) -> None:
assert isinstance(context.selected_mat, YamlMaterializer)
@then("the selected materializer is TableMaterializer")
def step_mat_is_table(context: Context) -> None:
assert isinstance(context.selected_mat, TableMaterializer)
# ===========================================================================
# Coverage boost: session edge cases
# ===========================================================================
@then("creating a panel with empty title raises ValueError")
def step_panel_empty_title(context: Context) -> None:
try:
context.session.panel("")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("creating a status with empty message raises ValueError")
def step_status_empty_msg(context: Context) -> None:
try:
context.session.status("")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("creating a progress with empty label raises ValueError")
def step_progress_empty_label(context: Context) -> None:
try:
context.session.progress("")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("creating a panel on closed session raises RuntimeError")
def step_closed_session_panel(context: Context) -> None:
try:
context.session.panel("Should fail")
raise AssertionError("Expected RuntimeError")
except RuntimeError:
pass
# ===========================================================================
# ElementHandle validation guards
# ===========================================================================
@then("constructing a handle with empty handle_id raises ValueError")
def step_handle_empty_id(context: Context) -> None:
try:
ElementHandle(
handle_id="",
element_type="panel",
declaration_index=0,
session=context.session,
element=Panel(title="T"),
)
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("constructing a handle with empty element_type raises ValueError")
def step_handle_empty_type(context: Context) -> None:
try:
ElementHandle(
handle_id="h1",
element_type="",
declaration_index=0,
session=context.session,
element=Panel(title="T"),
)
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("constructing a handle with negative index raises ValueError")
def step_handle_negative_index(context: Context) -> None:
try:
ElementHandle(
handle_id="h1",
element_type="panel",
declaration_index=-1,
session=context.session,
element=Panel(title="T"),
)
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("constructing a handle with None session raises ValueError")
def step_handle_none_session(context: Context) -> None:
try:
ElementHandle(
handle_id="h1",
element_type="panel",
declaration_index=0,
session=None, # type: ignore[arg-type]
element=Panel(title="T"),
)
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("constructing a handle with None element raises ValueError")
def step_handle_none_element(context: Context) -> None:
try:
ElementHandle(
handle_id="h1",
element_type="panel",
declaration_index=0,
session=context.session,
element=None, # type: ignore[arg-type]
)
raise AssertionError("Expected ValueError")
except ValueError:
pass
# ===========================================================================
# Handle close / __exit__ edge cases
# ===========================================================================
@when("I close the panel handle again")
def step_close_panel_again(context: Context) -> None:
context.panel_handle.close()
@then('the handle state is "{state}"')
def step_handle_state(context: Context, state: str) -> None:
handle = getattr(context, "panel_handle", None)
assert handle is not None
assert handle._state == state, f"Expected {state}, got {handle._state}"
@when('I use a panel handle as context manager with title "{title}"')
def step_use_context_manager(context: Context, title: str) -> None:
with context.session.panel(title) as handle:
handle.set_entry("key", "value")
context.panel_handle = handle
# ===========================================================================
# Table list rows
# ===========================================================================
# ===========================================================================
# Table with ColumnDef objects
# ===========================================================================
@when('I create a table with ColumnDef objects "{cols}"')
def step_table_columndef(context: Context, cols: str) -> None:
col_defs = [ColumnDef(name=c.strip()) for c in cols.split(",")]
context.table_handle = context.session.table(columns=col_defs)
# ===========================================================================
# Detect capabilities
# ===========================================================================
@when("I detect terminal capabilities")
def step_detect_caps(context: Context) -> None:
context.detected_caps = detect_terminal_capabilities()
@then("the capabilities object has is_tty field")
def step_caps_has_tty(context: Context) -> None:
assert hasattr(context.detected_caps, "is_tty")
@then("the capabilities object has supports_ansi field")
def step_caps_has_ansi(context: Context) -> None:
assert hasattr(context.detected_caps, "supports_ansi")
# ===========================================================================
# Second panel handle for out-of-order flush
# ===========================================================================
@when('I create a second panel handle with title "{title}"')
def step_create_second_panel(context: Context, title: str) -> None:
context.panel_handle_2 = context.session.panel(title)
@when('I set entry "{key}" to "{val}" on handle 2')
def step_set_entry_handle2(context: Context, key: str, val: str) -> None:
context.panel_handle_2.set_entry(key, val)
@when("I close handle 2")
def step_close_handle2(context: Context) -> None:
context.panel_handle_2.close()
# ===========================================================================
# JSON serializer with no session
# ===========================================================================
@given("a JsonMaterializer with no session")
def step_json_no_session(context: Context) -> None:
context.strategy = JsonMaterializer()
@then("the serializer get_output returns empty string")
def step_serializer_empty(context: Context) -> None:
output = context.strategy.get_output()
assert output == "", f"Expected empty string, got: {output!r}"
features/steps/skill_file_ops_steps.py
+140
View File
@@ -2,16 +2,20 @@
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
@@ -325,3 +329,139 @@ 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 "")
features/steps/skill_search_steps.py
+130
View File
@@ -2,10 +2,12 @@
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
@@ -342,3 +344,131 @@ 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}"
features/steps/tool_router_steps.py
+398 -2
View File
@@ -23,7 +23,7 @@ from cleveragents.tool.router import (
normalize_tool_schema_for_provider,
)
from cleveragents.tool.runner import ToolRunner
from cleveragents.tool.runtime import ToolSpec
from cleveragents.tool.runtime import ToolResult, ToolSpec
# -- Helpers ----------------------------------------------------------------
@@ -282,8 +282,9 @@ def step_registry_with_failing_val(context: Context, name: str) -> None:
@given('a tool call router for plan "{plan_id}"')
def step_create_router(context: Context, plan_id: str) -> None:
registry = getattr(context, "tool_registry", None) or context.registry
context.router = ToolCallRouter(
registry=context.registry,
registry=registry,
runner=context.runner,
plan_id=plan_id,
)
@@ -620,3 +621,398 @@ def step_classify_empty_error(context: Context) -> None:
@given("a tool call with invalid JSON arguments")
def step_payload_invalid_json(context: Context) -> None:
context.payload = {"name": "test/echo", "arguments": "not valid json"}
# -- Coverage: provider branches and validation ---------------------------------
@given('a tool call payload with name "{name}" and dict arguments')
def step_payload_dict_args(context: Context, name: str) -> None:
context.payload = {"name": name, "arguments": {"msg": "hello"}}
@given("an OpenAI payload with integer arguments")
def step_payload_int_args(context: Context) -> None:
context.payload = {"name": "test/echo", "arguments": 42}
@when("I try to normalize the invalid openai arguments")
def step_normalize_invalid_openai(context: Context) -> None:
try:
normalize_tool_call(context.payload)
context.raised_error = None
except ValueError as exc:
context.raised_error = exc
@given("an Anthropic payload with string input")
def step_payload_anthropic_str(context: Context) -> None:
context.payload = {
"name": "test/echo",
"id": "call-1",
"type": "tool_use",
"input": "bad",
}
@when("I try to normalize the invalid anthropic input")
def step_normalize_invalid_anthropic(context: Context) -> None:
try:
normalize_tool_call(context.payload)
context.raised_error = None
except ValueError as exc:
context.raised_error = exc
@given("a LangChain payload with dict args")
def step_payload_langchain_dict(context: Context) -> None:
context.payload = {"name": "test/echo", "args": {"key": "val"}, "id": "lc-1"}
@given("a LangChain payload with list args")
def step_payload_langchain_list(context: Context) -> None:
context.payload = {"name": "test/echo", "args": [1, 2, 3], "id": "lc-2"}
@when("I try to normalize the invalid langchain args")
def step_normalize_invalid_langchain(context: Context) -> None:
try:
normalize_tool_call(context.payload)
context.raised_error = None
except ValueError as exc:
context.raised_error = exc
@given("an unknown format payload with parameters key")
def step_payload_unknown_params(context: Context) -> None:
context.payload = {"name": "test/echo", "parameters": {"query": "hello"}}
@given("an unknown format payload with JSON string in args key")
def step_payload_unknown_json_str(context: Context) -> None:
# Use "arguments" with a non-parseable string first then "input" as JSON string
# to trigger the unknown branch with a JSON string in a fallback key.
# "arguments" with int makes it not str/dict -> unknown; "input" as JSON string.
context.payload = {"name": "test/echo", "input": '{"data": "value"}'}
@given("an unknown format payload with bad JSON and valid fallback")
def step_payload_unknown_bad_json(context: Context) -> None:
# arguments=42 makes detect_provider_format return UNKNOWN.
# In normalize unknown branch: "arguments" (int) skipped,
# "input" (bad JSON str) triggers JSONDecodeError→continue,
# "parameters" (dict) is used successfully.
context.payload = {
"name": "test/echo",
"arguments": 42,
"input": "not json",
"parameters": {"ok": True},
}
@when('I normalize the tool call for "{fmt}" format')
def step_normalize_for_format(context: Context, fmt: str) -> None:
context.normalized_request = normalize_tool_call(context.payload)
@then('the normalized request should have tool name "{name}"')
def step_then_request_tool_name(context: Context, name: str) -> None:
assert context.normalized_request.tool_name == name, (
f"Expected '{name}', got '{context.normalized_request.tool_name}'"
)
@then('the normalized arguments should be a dict with key "{key}"')
def step_then_args_dict_key(context: Context, key: str) -> None:
assert key in context.normalized_request.arguments
@then('the normalized arguments should have key "{key}"')
def step_then_args_key(context: Context, key: str) -> None:
assert key in context.normalized_request.arguments
@given('a tool spec named "{name}"')
def step_given_tool_spec(context: Context, name: str) -> None:
context.tool_spec = ToolSpec(
name=name,
description="test tool",
input_schema={"type": "object", "properties": {"msg": {"type": "string"}}},
output_schema={},
capabilities=ToolCapability(read_only=False, writes=True),
handler=lambda inputs: inputs,
)
@then('the schema should have key "{key}"')
def step_then_schema_key(context: Context, key: str) -> None:
assert key in context.normalized_schema, (
f"Missing key '{key}' in {context.normalized_schema}"
)
@then('the router plan_id should be "{expected}"')
def step_then_router_plan_id(context: Context, expected: str) -> None:
assert context.router.plan_id == expected
# -- Tools that raise or fail for coverage tests ---------
@given("a tool registry with a tool that raises RuntimeError")
def step_registry_runtime_error(context: Context) -> None:
registry = ToolRegistry()
def _noop(inputs: dict[str, Any]) -> dict[str, Any]:
return {}
registry.register(
ToolSpec(
name="test/boom",
description="exploding tool",
input_schema={},
output_schema={},
capabilities=ToolCapability(read_only=True),
handler=_noop,
)
)
context.tool_registry = registry
runner = ToolRunner(registry)
# Override execute to raise RuntimeError, bypassing ToolRunner's catch
def _raising_execute(name: str, arguments: dict[str, Any]) -> ToolResult:
raise RuntimeError("boom")
runner.execute = _raising_execute # type: ignore[assignment]
context.runner = runner
@when("I route an OpenAI payload for the error tool")
def step_route_error_tool(context: Context) -> None:
context.route_result = context.router.route(
{"name": "test/boom", "arguments": "{}"}
)
@then("the route result should have success false")
def step_then_result_fail(context: Context) -> None:
assert context.route_result.result.success is False
@then('the route result error should contain "{text}"')
def step_then_result_error_contains(context: Context, text: str) -> None:
assert text in (context.route_result.result.error or "")
@given("a tool registry with a tool that returns failed result")
def step_registry_fail_result(context: Context) -> None:
registry = ToolRegistry()
def _fail(inputs: dict[str, Any]) -> dict[str, Any]:
return {"error": "tool not found: missing"}
spec = ToolSpec(
name="test/fail",
description="failing tool",
input_schema={},
output_schema={},
capabilities=ToolCapability(read_only=True),
handler=_fail,
)
registry.register(spec)
context.tool_registry = registry
runner = ToolRunner(registry)
# Override execute to return a ToolResult with success=False
def _execute_fail(name: str, arguments: dict[str, Any]) -> ToolResult:
return ToolResult(
success=False,
output={},
error="tool not found: missing",
duration_ms=1.0,
)
runner.execute = _execute_fail # type: ignore[assignment]
context.runner = runner
@when("I route an OpenAI payload for the fail tool")
def step_route_fail_tool(context: Context) -> None:
context.route_result = context.router.route(
{"name": "test/fail", "arguments": "{}"}
)
@then("the route result should have error_category set")
def step_then_error_category_set(context: Context) -> None:
assert context.route_result.error_category is not None
@given("a tool registry with a validation tool")
def step_registry_validation_tool(context: Context) -> None:
registry = ToolRegistry()
def _validate(inputs: dict[str, Any]) -> dict[str, Any]:
return {"passed": True}
registry.register(
ToolSpec(
name="validate/check",
description="validate something",
input_schema={},
output_schema={"validation_mode": "strict"},
capabilities=ToolCapability(read_only=True, writes=False),
handler=_validate,
)
)
context.tool_registry = registry
context.runner = ToolRunner(registry)
@when("I route an OpenAI payload for the validation tool")
def step_route_validation_tool(context: Context) -> None:
context.route_result = context.router.route(
{"name": "validate/check", "arguments": "{}"}
)
@then("the route result should have is_validation true")
def step_then_is_validation(context: Context) -> None:
assert context.route_result.is_validation is True
@then("the route result should have validation_passed true")
def step_then_validation_passed(context: Context) -> None:
assert context.route_result.validation_passed is True
# -- Streaming coverage ----
@when("I stream-route an OpenAI payload for the error tool")
def step_stream_route_error(context: Context) -> None:
results = list(
context.router.route_streaming({"name": "test/boom", "arguments": "{}"})
)
context.streaming_results = results
@then("the streaming updates should include ERROR status")
def step_then_streaming_error_status(context: Context) -> None:
statuses = [
r.status
for r in context.streaming_results
if isinstance(r, StreamingToolUpdate)
]
assert StreamingStatus.ERROR in statuses
@then("the streaming final result should have success false")
def step_then_streaming_fail(context: Context) -> None:
final = [
r for r in context.streaming_results if isinstance(r, NormalizedToolCallResult)
]
assert len(final) == 1
assert final[0].result.success is False
@when("I try to stream-route a non-dict payload")
def step_try_stream_non_dict(context: Context) -> None:
try:
list(context.router.route_streaming("not a dict")) # type: ignore[arg-type]
context.raised_error = None
except ValueError as exc:
context.raised_error = exc
@when("I stream-route an OpenAI payload for the validation tool")
def step_stream_route_validation(context: Context) -> None:
results = list(
context.router.route_streaming({"name": "validate/check", "arguments": "{}"})
)
context.streaming_results = results
@then("the streaming final result should have is_validation true")
def step_then_streaming_validation(context: Context) -> None:
final = [
r for r in context.streaming_results if isinstance(r, NormalizedToolCallResult)
]
assert len(final) == 1
assert final[0].is_validation is True
# -- Schema export coverage ----
@when('I export tool schemas for "{provider}"')
def step_export_schemas(context: Context, provider: str) -> None:
fmt = ProviderFormat(provider)
context.exported_schemas = context.router.export_schemas(fmt)
@then('the exported schemas should include a tool with type "{tool_type}"')
def step_then_schema_tool_type(context: Context, tool_type: str) -> None:
types = [s.get("tool_type") for s in context.exported_schemas]
assert tool_type in types, f"Expected type '{tool_type}' in {types}"
@given('a tool registry with a validation tool with mode "{mode}"')
def step_registry_validation_with_mode(context: Context, mode: str) -> None:
registry = ToolRegistry()
def _validate(inputs: dict[str, Any]) -> dict[str, Any]:
return {"passed": True}
registry.register(
ToolSpec(
name="validate/strict",
description="strict validator",
input_schema={},
output_schema={"validation_mode": mode},
capabilities=ToolCapability(read_only=True, writes=False),
handler=_validate,
)
)
context.tool_registry = registry
context.runner = ToolRunner(registry)
@then('the exported validation schema should have mode "{mode}"')
def step_then_schema_mode(context: Context, mode: str) -> None:
validation_schemas = [
s for s in context.exported_schemas if s.get("tool_type") == "validation"
]
assert len(validation_schemas) >= 1
assert validation_schemas[0].get("validation_mode") == mode
@given("a tool registry with a validation tool with numeric mode")
def step_registry_validation_numeric_mode(context: Context) -> None:
registry = ToolRegistry()
def _validate(inputs: dict[str, Any]) -> dict[str, Any]:
return {"passed": True}
registry.register(
ToolSpec(
name="validate/numeric",
description="numeric mode validator",
input_schema={},
output_schema={"validation_mode": 123},
capabilities=ToolCapability(read_only=True, writes=False),
handler=_validate,
)
)
context.tool_registry = registry
context.runner = ToolRunner(registry)
@then("the exported validation schema should not have a mode")
def step_then_schema_no_mode(context: Context) -> None:
validation_schemas = [
s for s in context.exported_schemas if s.get("tool_type") == "validation"
]
assert len(validation_schemas) >= 1
assert "validation_mode" not in validation_schemas[0]
features/tool_router.feature
+124
View File
@@ -322,3 +322,127 @@ Feature: Tool Call Router
Scenario: Detect format for non-dict returns unknown
When I detect format for a non-dict value
Then the detected format should be "unknown"
# ---- Coverage: normalize with dict args, Anthropic/LangChain branches ----
Scenario: Normalize OpenAI payload with dict arguments
Given a tool call payload with name "test/echo" and dict arguments
When I normalize the tool call for "openai" format
Then the normalized request should have tool name "test/echo"
And the normalized arguments should be a dict with key "msg"
Scenario: Normalize LangChain payload with dict args
Given a LangChain payload with dict args
When I normalize the tool call for "langchain" format
Then the normalized request should have tool name "test/echo"
Scenario: Normalize LangChain payload with non-dict args raises ValueError
Given a LangChain payload with list args
When I try to normalize the invalid langchain args
Then a router ValueError should be raised containing "must be a dict"
Scenario: Normalize unknown payload with parameters key
Given an unknown format payload with parameters key
When I normalize the tool call for "unknown" format
Then the normalized arguments should have key "query"
Scenario: Normalize unknown payload with JSON string args
Given an unknown format payload with JSON string in args key
When I normalize the tool call for "unknown" format
Then the normalized arguments should have key "data"
Scenario: Normalize unknown payload with bad JSON skips to next key
Given an unknown format payload with bad JSON and valid fallback
When I normalize the tool call for "unknown" format
Then the normalized arguments should have key "ok"
# ---- Coverage: schema normalization LangChain and unknown ----
Scenario: Schema normalization for LangChain uses args_schema
Given a tool spec named "test/echo"
When I normalize the schema for "langchain" provider
Then the schema should have key "args_schema"
Scenario: Schema normalization for unknown uses input_schema
Given a tool spec named "test/echo"
When I normalize the schema for "unknown" provider
Then the schema should have key "input_schema"
# ---- Coverage: plan_id property ----
Scenario: Router plan_id property returns correct value
Given a tool registry with an echo tool "test/echo"
And a tool call router for plan "plan-prop-test"
Then the router plan_id should be "plan-prop-test"
# ---- Coverage: route generic exception, validation surfacing, error classification ----
Scenario: Route catches generic RuntimeError during execution
Given a tool registry with a tool that raises RuntimeError
And a tool call router for plan "plan-runtime-err"
When I route an OpenAI payload for the error tool
Then the route result should have success false
And the route result error should contain "boom"
Scenario: Route returns error classification for failed ToolResult
Given a tool registry with a tool that returns failed result
And a tool call router for plan "plan-fail-result"
When I route an OpenAI payload for the fail tool
Then the route result should have success false
And the route result should have error_category set
Scenario: Route validates validation tool with passed result
Given a tool registry with a validation tool
And a tool call router for plan "plan-valid-check"
When I route an OpenAI payload for the validation tool
Then the route result should have is_validation true
And the route result should have validation_passed true
# ---- Coverage: streaming exception and validation ----
Scenario: Streaming route catches exception and yields ERROR
Given a tool registry with a tool that raises RuntimeError
And a tool call router for plan "plan-stream-err"
When I stream-route an OpenAI payload for the error tool
Then the streaming updates should include ERROR status
And the streaming final result should have success false
Scenario: Streaming route non-dict payload raises ValueError
Given a tool registry with an echo tool "test/echo"
And a tool call router for plan "plan-stream-ndict"
When I try to stream-route a non-dict payload
Then a router ValueError should be raised containing "payload must be a dict"
Scenario: Streaming route validates validation tool result
Given a tool registry with a validation tool
And a tool call router for plan "plan-stream-valid"
When I stream-route an OpenAI payload for the validation tool
Then the streaming final result should have is_validation true
# ---- Coverage: get_tool_schemas validation annotation ----
Scenario: Export schemas annotates validation tools
Given a tool registry with a validation tool
And a tool call router for plan "plan-schema-export"
When I export tool schemas for "openai"
Then the exported schemas should include a tool with type "validation"
Scenario: Export schemas annotates regular tools
Given a tool registry with an echo tool "test/echo"
And a tool call router for plan "plan-schema-regular"
When I export tool schemas for "openai"
Then the exported schemas should include a tool with type "tool"
# ---- Coverage: _get_validation_mode ----
Scenario: Validation mode extracted from output_schema
Given a tool registry with a validation tool with mode "strict"
And a tool call router for plan "plan-vmode"
When I export tool schemas for "openai"
Then the exported validation schema should have mode "strict"
Scenario: Validation mode returns None for non-string mode
Given a tool registry with a validation tool with numeric mode
And a tool call router for plan "plan-vmode-num"
When I export tool schemas for "openai"
Then the exported validation schema should not have a mode