forked from HAL9000/cleveragents-core
CoreRasurae
1e987a2009
Co-authored-by: Luis Mendes <luis.mendes@cleverthis.com> Co-committed-by: Luis Mendes <luis.mendes@cleverthis.com>
413 lines
15 KiB
Python
413 lines
15 KiB
Python
"""Step definitions for tool lifecycle coverage boost feature tests.
|
|
|
|
Targets uncovered lines and branches in src/cleveragents/tool/lifecycle.py:
|
|
- Line 253: ToolLifecycleCache.remove returns None when tool not in cache
|
|
- Line 394: ToolRuntime.activate re-raises ToolCancelledError
|
|
- Line 473: ToolRuntime.execute raises ToolNotActivatedError for None instance
|
|
- Lines 496-502: ToolRuntime.execute handles ToolCancelledError with trace
|
|
- Line 550: ToolRuntime.deactivate early return when instance not in cache
|
|
- Branch 585:exit: deactivate_plan with empty instances dict
|
|
- Branch 235:exit: get_plan_tools iterating over empty cache
|
|
- Branch 248:253: remove when tool not found / plan not found
|
|
- Branch 250:252: remove pops entry successfully
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.domain.models.core.tool import (
|
|
Tool,
|
|
ToolCapability,
|
|
ToolSource,
|
|
ToolType,
|
|
)
|
|
from cleveragents.infrastructure.events.reactive import ReactiveEventBus
|
|
from cleveragents.tool.context import (
|
|
CancellationToken,
|
|
ToolCancelledError,
|
|
ToolExecutionContext,
|
|
)
|
|
from cleveragents.tool.lifecycle import (
|
|
ToolDescriptor,
|
|
ToolExecutionError,
|
|
ToolLifecycleCache,
|
|
ToolNotActivatedError,
|
|
ToolResult,
|
|
ToolRuntime,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mock tool instances for coverage-boost scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _CBMockToolInstance:
|
|
"""Basic mock ToolInstance for coverage-boost tests."""
|
|
|
|
def __init__(self, name: str = "mock/tool", read_only: bool = True) -> None:
|
|
self.name = name
|
|
self.read_only = read_only
|
|
self.activated = False
|
|
self.deactivated = False
|
|
|
|
def discover(self) -> ToolDescriptor:
|
|
return ToolDescriptor(
|
|
name=self.name,
|
|
description=f"Coverage-boost mock {self.name}",
|
|
capability=ToolCapability(read_only=self.read_only),
|
|
)
|
|
|
|
def activate(self, ctx: ToolExecutionContext) -> None:
|
|
self.activated = True
|
|
|
|
def execute(self, params: dict[str, Any], ctx: ToolExecutionContext) -> ToolResult:
|
|
return ToolResult(success=True, data={"result": "ok"})
|
|
|
|
def deactivate(self, ctx: ToolExecutionContext) -> None:
|
|
self.deactivated = True
|
|
|
|
|
|
class _CancellingActivateInstance(_CBMockToolInstance):
|
|
"""Mock that raises ToolCancelledError on activate()."""
|
|
|
|
def activate(self, ctx: ToolExecutionContext) -> None:
|
|
raise ToolCancelledError("Cancelled during activation")
|
|
|
|
|
|
class _CancellingExecuteInstance(_CBMockToolInstance):
|
|
"""Mock that raises ToolCancelledError on execute()."""
|
|
|
|
def activate(self, ctx: ToolExecutionContext) -> None:
|
|
self.activated = True
|
|
|
|
def execute(self, params: dict[str, Any], ctx: ToolExecutionContext) -> ToolResult:
|
|
raise ToolCancelledError("Cancelled during execution")
|
|
|
|
|
|
class _NullCacheInstance(_CBMockToolInstance):
|
|
"""Mock whose activate succeeds but we manipulate the cache afterwards."""
|
|
|
|
def activate(self, ctx: ToolExecutionContext) -> None:
|
|
self.activated = True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper to create Tool domain models
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _cb_make_tool(
|
|
name: str,
|
|
*,
|
|
read_only: bool = True,
|
|
writes: bool = False,
|
|
) -> Tool:
|
|
"""Create a minimal Tool domain model for coverage-boost tests."""
|
|
return Tool(
|
|
name=name,
|
|
description=f"Coverage-boost test tool {name}",
|
|
source=ToolSource.BUILTIN,
|
|
tool_type=ToolType.TOOL,
|
|
capability=ToolCapability(
|
|
read_only=read_only,
|
|
writes=writes,
|
|
),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ToolLifecycleCache steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I create a coverage-boost lifecycle cache")
|
|
def step_create_cb_cache(context: Context) -> None:
|
|
context.cb_cache = ToolLifecycleCache()
|
|
context.cb_remove_result = None
|
|
|
|
|
|
@given('I have a coverage-boost mock tool instance named "{name}"')
|
|
def step_create_cb_mock(context: Context, name: str) -> None:
|
|
context.cb_mock_instance = _CBMockToolInstance(name=name)
|
|
|
|
|
|
@given('I have a second coverage-boost mock tool instance named "{name}"')
|
|
def step_create_cb_mock_second(context: Context, name: str) -> None:
|
|
context.cb_mock_instance_2 = _CBMockToolInstance(name=name)
|
|
|
|
|
|
@when(
|
|
'I put the coverage-boost instance in cache for plan "{plan_id}" and tool "{tool_name}"'
|
|
)
|
|
def step_cb_cache_put(context: Context, plan_id: str, tool_name: str) -> None:
|
|
context.cb_cache.put(plan_id, tool_name, context.cb_mock_instance)
|
|
|
|
|
|
@when(
|
|
'I put the second coverage-boost instance in cache for plan "{plan_id}" and tool "{tool_name}"'
|
|
)
|
|
def step_cb_cache_put_second(context: Context, plan_id: str, tool_name: str) -> None:
|
|
context.cb_cache.put(plan_id, tool_name, context.cb_mock_instance_2)
|
|
|
|
|
|
@when('I remove tool "{tool_name}" from coverage-boost cache plan "{plan_id}"')
|
|
def step_cb_cache_remove(context: Context, tool_name: str, plan_id: str) -> None:
|
|
context.cb_remove_result = context.cb_cache.remove(plan_id, tool_name)
|
|
|
|
|
|
@then("the coverage-boost cache should have {count:d} plans")
|
|
def step_cb_cache_plan_count(context: Context, count: int) -> None:
|
|
assert context.cb_cache.plan_count == count, (
|
|
f"Expected {count} plans, got {context.cb_cache.plan_count}"
|
|
)
|
|
|
|
|
|
@then('the coverage-boost cache internal dict should not contain plan "{plan_id}"')
|
|
def step_cb_cache_no_plan_key(context: Context, plan_id: str) -> None:
|
|
# Access internal _cache dict to verify the plan key was fully deleted
|
|
assert plan_id not in context.cb_cache._cache, (
|
|
f"Plan '{plan_id}' should have been removed from internal _cache dict"
|
|
)
|
|
|
|
|
|
@then("the coverage-boost remove result should be None")
|
|
def step_cb_remove_result_none(context: Context) -> None:
|
|
assert context.cb_remove_result is None, (
|
|
f"Expected None, got {context.cb_remove_result}"
|
|
)
|
|
|
|
|
|
@then('the coverage-boost plan tools for plan "{plan_id}" should be an empty list')
|
|
def step_cb_plan_tools_empty(context: Context, plan_id: str) -> None:
|
|
tools = context.cb_cache.get_plan_tools(plan_id)
|
|
assert tools == [], f"Expected empty list, got {tools}"
|
|
|
|
|
|
@then(
|
|
'the coverage-boost cache get for plan "{plan_id}" and tool "{tool_name}" should not be None'
|
|
)
|
|
def step_cb_cache_get_not_none(context: Context, plan_id: str, tool_name: str) -> None:
|
|
result = context.cb_cache.get(plan_id, tool_name)
|
|
assert result is not None, f"Expected non-None for plan={plan_id} tool={tool_name}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ToolRuntime steps — coverage-boost context and runtime setup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I create a coverage-boost tool runtime")
|
|
def step_create_cb_runtime(context: Context) -> None:
|
|
context.cb_runtime = ToolRuntime(event_bus=ReactiveEventBus())
|
|
context.cb_mocks: dict[str, _CBMockToolInstance] = {}
|
|
context.cb_error = None
|
|
|
|
|
|
@given('I create a coverage-boost execution context with plan_id "{plan_id}"')
|
|
def step_create_cb_ctx(context: Context, plan_id: str) -> None:
|
|
token = CancellationToken()
|
|
context.cb_ctx = ToolExecutionContext(
|
|
plan_id=plan_id,
|
|
cancellation_token=token,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ToolRuntime.activate — ToolCancelledError re-raise (line 394)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('I register a coverage-boost cancelling-activate tool "{name}"')
|
|
def step_register_cb_cancel_activate(context: Context, name: str) -> None:
|
|
tool = _cb_make_tool(name, read_only=True)
|
|
mock = _CancellingActivateInstance(name=name)
|
|
context.cb_runtime.register_tool(tool, mock)
|
|
context.cb_mocks[name] = mock
|
|
|
|
|
|
@when('I try to activate the cancelling tool "{name}" in coverage-boost runtime')
|
|
def step_try_cb_activate_cancel(context: Context, name: str) -> None:
|
|
try:
|
|
context.cb_runtime.activate(name, context.cb_ctx)
|
|
context.cb_error = None
|
|
except ToolCancelledError as exc:
|
|
context.cb_error = exc
|
|
except Exception as exc:
|
|
context.cb_error = exc
|
|
|
|
|
|
@then("a ToolCancelledError should have been raised in coverage-boost activation")
|
|
def step_check_cb_activation_cancelled(context: Context) -> None:
|
|
assert isinstance(context.cb_error, ToolCancelledError), (
|
|
f"Expected ToolCancelledError, got {type(context.cb_error).__name__}: {context.cb_error}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ToolRuntime.execute — None cached instance (line 473)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('I register a coverage-boost tool "{name}" with a mock that nullifies cache')
|
|
def step_register_cb_null_cache(context: Context, name: str) -> None:
|
|
tool = _cb_make_tool(name, read_only=True)
|
|
mock = _NullCacheInstance(name=name)
|
|
context.cb_runtime.register_tool(tool, mock)
|
|
context.cb_mocks[name] = mock
|
|
|
|
|
|
@when('I try to execute the null-cache tool "{name}" in coverage-boost runtime')
|
|
def step_try_cb_execute_null_cache(context: Context, name: str) -> None:
|
|
# We need activate to succeed (so it passes the auto-activate gate)
|
|
# but cache.get to return None right after.
|
|
# Strategy: patch the cache's get method to return None after activate
|
|
# has put the instance into the cache.
|
|
original_get = context.cb_runtime._cache.get
|
|
|
|
def _patched_get(plan_id: str, tool_name: str) -> Any:
|
|
# First call is in activate() to check if already cached — return None
|
|
# so activate proceeds. After activate puts it, the second call is in
|
|
# execute() at line 471 — we remove it from cache first so it returns None.
|
|
result = original_get(plan_id, tool_name)
|
|
if result is not None:
|
|
# Remove it so the next get (line 471) returns None
|
|
context.cb_runtime._cache.remove(plan_id, tool_name)
|
|
return None
|
|
return result
|
|
|
|
try:
|
|
with patch.object(context.cb_runtime._cache, "get", side_effect=_patched_get):
|
|
context.cb_runtime.execute(name, {}, context.cb_ctx)
|
|
context.cb_error = None
|
|
except ToolNotActivatedError as exc:
|
|
context.cb_error = exc
|
|
except Exception as exc:
|
|
context.cb_error = exc
|
|
|
|
|
|
@then("a ToolNotActivatedError should have been raised in coverage-boost execution")
|
|
def step_check_cb_not_activated(context: Context) -> None:
|
|
# ToolNotActivatedError is raised at line 473 but caught by the generic
|
|
# except Exception handler at line 504 and wrapped in ToolExecutionError.
|
|
is_direct = isinstance(context.cb_error, ToolNotActivatedError)
|
|
is_wrapped = (
|
|
isinstance(context.cb_error, ToolExecutionError)
|
|
and "not activated" in str(context.cb_error).lower()
|
|
)
|
|
assert is_direct or is_wrapped, (
|
|
f"Expected ToolNotActivatedError (or ToolExecutionError wrapping it), "
|
|
f"got {type(context.cb_error).__name__}: {context.cb_error}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ToolRuntime.execute — ToolCancelledError during execution (lines 496-502)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('I register a coverage-boost cancelling-execute tool "{name}"')
|
|
def step_register_cb_cancel_execute(context: Context, name: str) -> None:
|
|
tool = _cb_make_tool(name, read_only=True)
|
|
mock = _CancellingExecuteInstance(name=name)
|
|
context.cb_runtime.register_tool(tool, mock)
|
|
context.cb_mocks[name] = mock
|
|
|
|
|
|
@when('I try to execute the cancelling-execute tool "{name}" in coverage-boost runtime')
|
|
def step_try_cb_execute_cancel(context: Context, name: str) -> None:
|
|
try:
|
|
context.cb_runtime.execute(name, {}, context.cb_ctx)
|
|
context.cb_error = None
|
|
except ToolCancelledError as exc:
|
|
context.cb_error = exc
|
|
except Exception as exc:
|
|
context.cb_error = exc
|
|
|
|
|
|
@then("a ToolCancelledError should have been raised in coverage-boost execution")
|
|
def step_check_cb_execution_cancelled(context: Context) -> None:
|
|
assert isinstance(context.cb_error, ToolCancelledError), (
|
|
f"Expected ToolCancelledError, got {type(context.cb_error).__name__}: {context.cb_error}"
|
|
)
|
|
|
|
|
|
@then("the coverage-boost context should have {count:d} traces")
|
|
def step_check_cb_ctx_traces(context: Context, count: int) -> None:
|
|
actual = len(context.cb_ctx.traces)
|
|
assert actual == count, f"Expected {count} traces, got {actual}"
|
|
|
|
|
|
@then("the coverage-boost last trace should show success {expected}")
|
|
def step_check_cb_last_trace_success(context: Context, expected: str) -> None:
|
|
traces = context.cb_ctx.traces
|
|
assert len(traces) > 0, "No traces recorded"
|
|
last = traces[-1]
|
|
expected_bool = expected == "True"
|
|
assert last.success == expected_bool, (
|
|
f"Expected trace success={expected_bool}, got {last.success}"
|
|
)
|
|
|
|
|
|
@then('the coverage-boost last trace error should be "{expected}"')
|
|
def step_check_cb_last_trace_error(context: Context, expected: str) -> None:
|
|
traces = context.cb_ctx.traces
|
|
assert len(traces) > 0, "No traces recorded"
|
|
last = traces[-1]
|
|
assert last.error == expected, (
|
|
f"Expected trace error='{expected}', got '{last.error}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ToolRuntime.deactivate — tool not in cache (line 550)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('I register a coverage-boost read-only tool "{name}" with a mock')
|
|
def step_register_cb_readonly(context: Context, name: str) -> None:
|
|
tool = _cb_make_tool(name, read_only=True)
|
|
mock = _CBMockToolInstance(name=name, read_only=True)
|
|
context.cb_runtime.register_tool(tool, mock)
|
|
context.cb_mocks[name] = mock
|
|
|
|
|
|
@when('I deactivate tool "{name}" in the coverage-boost runtime')
|
|
def step_cb_deactivate_tool(context: Context, name: str) -> None:
|
|
try:
|
|
context.cb_runtime.deactivate(name, context.cb_ctx)
|
|
context.cb_error = None
|
|
except Exception as exc:
|
|
context.cb_error = exc
|
|
|
|
|
|
@then("no coverage-boost exception should have been raised")
|
|
def step_check_cb_no_exception(context: Context) -> None:
|
|
assert context.cb_error is None, (
|
|
f"Expected no exception, got {type(context.cb_error).__name__}: {context.cb_error}"
|
|
)
|
|
|
|
|
|
@then('the coverage-boost mock "{name}" should not have been deactivated')
|
|
def step_check_cb_mock_not_deactivated(context: Context, name: str) -> None:
|
|
mock = context.cb_mocks[name]
|
|
assert not mock.deactivated, f"Mock '{name}' should NOT have been deactivated"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ToolRuntime.deactivate_plan — empty plan (branch 585:exit)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I deactivate plan in the coverage-boost runtime")
|
|
def step_cb_deactivate_plan(context: Context) -> None:
|
|
try:
|
|
context.cb_runtime.deactivate_plan(context.cb_ctx)
|
|
context.cb_error = None
|
|
except Exception as exc:
|
|
context.cb_error = exc
|