"""Step definitions for tool-level execution environment preference tests.""" from __future__ import annotations from enum import StrEnum from typing import Any from unittest.mock import create_autospec from behave import given, then, when from behave.runner import Context # --------------------------------------------------------------------------- # EnvironmentPreferenceMode enum steps # --------------------------------------------------------------------------- @given("I import EnvironmentPreferenceMode") def step_import_pref_mode(context: Context) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( EnvironmentPreferenceMode, ) context.pref_mode_cls = EnvironmentPreferenceMode @then('the NONE preference value should be "{value}"') def step_none_pref_value(context: Context, value: str) -> None: assert context.pref_mode_cls.NONE.value == value @then('the REQUIRED preference value should be "{value}"') def step_required_pref_value(context: Context, value: str) -> None: assert context.pref_mode_cls.REQUIRED.value == value @then('the PREFERRED preference value should be "{value}"') def step_preferred_pref_value(context: Context, value: str) -> None: assert context.pref_mode_cls.PREFERRED.value == value @then('the SPECIFIC preference value should be "{value}"') def step_specific_pref_value(context: Context, value: str) -> None: assert context.pref_mode_cls.SPECIFIC.value == value @then("EnvironmentPreferenceMode should be a StrEnum subclass") def step_pref_mode_is_strenum(context: Context) -> None: assert issubclass(context.pref_mode_cls, StrEnum) # --------------------------------------------------------------------------- # ExecutionEnvironmentPreference creation steps # --------------------------------------------------------------------------- @given("I create a default ExecutionEnvironmentPreference") def step_create_default_pref(context: Context) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( ExecutionEnvironmentPreference, ) context.pref = ExecutionEnvironmentPreference() @given('I create an ExecutionEnvironmentPreference with mode "{mode}"') def step_create_pref_with_mode(context: Context, mode: str) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( EnvironmentPreferenceMode, ExecutionEnvironmentPreference, ) context.pref = ExecutionEnvironmentPreference( mode=EnvironmentPreferenceMode(mode), ) @given('I create a specific ExecutionEnvironmentPreference targeting "{target}"') def step_create_specific_pref(context: Context, target: str) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( EnvironmentPreferenceMode, ExecutionEnvironmentPreference, ) context.pref = ExecutionEnvironmentPreference( mode=EnvironmentPreferenceMode.SPECIFIC, target_resource=target, ) @when( 'I try to create an ExecutionEnvironmentPreference with mode "{mode}"' ' and target "{target}"' ) def step_try_create_pref_with_target(context: Context, mode: str, target: str) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( EnvironmentPreferenceMode, ExecutionEnvironmentPreference, ) try: ExecutionEnvironmentPreference( mode=EnvironmentPreferenceMode(mode), target_resource=target, ) context.pref_error = None except ValueError as exc: context.pref_error = str(exc) @when( 'I try to create an ExecutionEnvironmentPreference with mode "specific"' " and no target" ) def step_try_create_specific_no_target(context: Context) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( EnvironmentPreferenceMode, ExecutionEnvironmentPreference, ) try: ExecutionEnvironmentPreference( mode=EnvironmentPreferenceMode.SPECIFIC, ) context.pref_error = None except ValueError as exc: context.pref_error = str(exc) @then('the preference creation should fail with "{message}"') def step_pref_creation_failed(context: Context, message: str) -> None: assert context.pref_error is not None, "Expected creation to fail" assert message in context.pref_error, ( f"Expected '{message}' in error, got: {context.pref_error}" ) @then('the preference mode should be "{mode}"') def step_pref_mode_is(context: Context, mode: str) -> None: assert context.pref.mode.value == mode @then("the preference target_resource should be None") def step_pref_target_none(context: Context) -> None: assert context.pref.target_resource is None @then('the preference target_resource should be "{target}"') def step_pref_target_is(context: Context, target: str) -> None: assert context.pref.target_resource == target # --------------------------------------------------------------------------- # Frozen model steps # --------------------------------------------------------------------------- @when("I try to mutate the preference mode") def step_try_mutate_pref(context: Context) -> None: from pydantic import ValidationError try: context.pref.mode = "required" # type: ignore[misc] context.mutation_failed = False except (ValidationError, TypeError): context.mutation_failed = True @then("the mutation should fail") def step_mutation_failed(context: Context) -> None: assert context.mutation_failed is True # --------------------------------------------------------------------------- # ToolSpec field steps # --------------------------------------------------------------------------- def _make_handler(inputs: dict[str, Any]) -> dict[str, Any]: return {"result": "ok"} @given("I create a ToolSpec with default execution_environment") def step_toolspec_default_env(context: Context) -> None: from cleveragents.tool.runtime import ToolSpec context.tool_spec = ToolSpec( name="test/tool", description="A test tool", handler=_make_handler, ) @given("I create a ToolSpec with required execution_environment") def step_toolspec_required_env(context: Context) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( EnvironmentPreferenceMode, ExecutionEnvironmentPreference, ) from cleveragents.tool.runtime import ToolSpec context.tool_spec = ToolSpec( name="test/tool", description="A test tool", handler=_make_handler, execution_environment=ExecutionEnvironmentPreference( mode=EnvironmentPreferenceMode.REQUIRED, ), ) @then('the ToolSpec execution_environment mode should be "{mode}"') def step_toolspec_env_mode(context: Context, mode: str) -> None: assert context.tool_spec.execution_environment.mode.value == mode # --------------------------------------------------------------------------- # Domain Tool field steps # --------------------------------------------------------------------------- @given("I create a domain Tool with default execution_environment") def step_domain_tool_default_env(context: Context) -> None: from cleveragents.domain.models.core.tool import Tool, ToolSource context.domain_tool = Tool( name="local/test-tool", description="A test tool", source=ToolSource.BUILTIN, ) @given("I create a domain Tool with required execution_environment") def step_domain_tool_required_env(context: Context) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( EnvironmentPreferenceMode, ExecutionEnvironmentPreference, ) from cleveragents.domain.models.core.tool import Tool, ToolSource context.domain_tool = Tool( name="local/test-tool", description="A test tool", source=ToolSource.BUILTIN, execution_environment=ExecutionEnvironmentPreference( mode=EnvironmentPreferenceMode.REQUIRED, ), ) @then('the domain Tool execution_environment mode should be "{mode}"') def step_domain_tool_env_mode(context: Context, mode: str) -> None: assert context.domain_tool.execution_environment.mode.value == mode # --------------------------------------------------------------------------- # Tool.from_config steps # --------------------------------------------------------------------------- @given('I have a tool config dict with execution_environment mode "{mode}"') def step_config_with_env_mode(context: Context, mode: str) -> None: context.tool_config = { "name": "local/test-tool", "description": "A test tool", "source": "builtin", "execution_environment": {"mode": mode}, } @given('I have a tool config dict with execution_environment targeting "{target}"') def step_config_with_env_mode_target(context: Context, target: str) -> None: context.tool_config = { "name": "local/test-tool", "description": "A test tool", "source": "builtin", "execution_environment": {"mode": "specific", "target_resource": target}, } @given("I have a tool config dict without execution_environment") def step_config_without_env(context: Context) -> None: context.tool_config = { "name": "local/test-tool", "description": "A test tool", "source": "builtin", } @when("I create a Tool from the config") def step_tool_from_config(context: Context) -> None: from cleveragents.domain.models.core.tool import Tool context.domain_tool = Tool.from_config(context.tool_config) @then('the tool execution_environment mode should be "{mode}"') def step_tool_env_mode(context: Context, mode: str) -> None: assert context.domain_tool.execution_environment.mode.value == mode @then('the tool execution_environment target_resource should be "{target}"') def step_tool_env_target(context: Context, target: str) -> None: assert context.domain_tool.execution_environment.target_resource == target # --------------------------------------------------------------------------- # as_cli_dict steps # --------------------------------------------------------------------------- @when("I call as_cli_dict on the tool") def step_call_cli_dict(context: Context) -> None: context.cli_dict = context.domain_tool.as_cli_dict() @then('the cli dict should not contain "{key}"') def step_cli_dict_no_key(context: Context, key: str) -> None: assert key not in context.cli_dict @then('the cli dict should contain execution_environment with mode "{mode}"') def step_cli_dict_env_mode(context: Context, mode: str) -> None: assert "execution_environment" in context.cli_dict assert context.cli_dict["execution_environment"]["mode"] == mode # --------------------------------------------------------------------------- # ToolRunner preference routing steps # --------------------------------------------------------------------------- @given("I have a ToolRunner with a tool with NONE preference") def step_runner_none_pref(context: Context) -> None: from cleveragents.tool.registry import ToolRegistry from cleveragents.tool.runner import ToolRunner from cleveragents.tool.runtime import ToolSpec registry = ToolRegistry() spec = ToolSpec( name="test/none-pref", description="Tool with no env preference", handler=_make_handler, ) registry.register(spec) context.runner = ToolRunner(registry=registry) context.runner.activate("test/none-pref") context.tool_name = "test/none-pref" @when("I execute the tool with no caller env") def step_execute_no_caller_env(context: Context) -> None: context.result = context.runner.execute(context.tool_name, {"input": "value"}) @then("the tool preference routing should succeed on host") def step_tool_success_host(context: Context) -> None: assert context.result.success is True @given("I have a ToolRunner with a tool with REQUIRED preference and no container") def step_runner_required_no_container(context: Context) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( EnvironmentPreferenceMode, ExecutionEnvironmentPreference, ) from cleveragents.tool.registry import ToolRegistry from cleveragents.tool.runner import ToolRunner from cleveragents.tool.runtime import ToolSpec registry = ToolRegistry() spec = ToolSpec( name="test/required-pref", description="Tool requiring container", handler=_make_handler, execution_environment=ExecutionEnvironmentPreference( mode=EnvironmentPreferenceMode.REQUIRED, ), ) registry.register(spec) context.runner = ToolRunner(registry=registry) context.runner.activate("test/required-pref") context.tool_name = "test/required-pref" @when("I execute the required-preference tool") def step_execute_required(context: Context) -> None: context.result = context.runner.execute(context.tool_name, {"input": "value"}) @then("the execution should fail with container required error") def step_fail_container_required(context: Context) -> None: assert context.result.success is False assert "requires container execution" in (context.result.error or "") @given("I have a ToolRunner with a tool with PREFERRED preference and no container") def step_runner_preferred_no_container(context: Context) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( EnvironmentPreferenceMode, ExecutionEnvironmentPreference, ) from cleveragents.tool.registry import ToolRegistry from cleveragents.tool.runner import ToolRunner from cleveragents.tool.runtime import ToolSpec registry = ToolRegistry() spec = ToolSpec( name="test/preferred-pref", description="Tool preferring container", handler=_make_handler, execution_environment=ExecutionEnvironmentPreference( mode=EnvironmentPreferenceMode.PREFERRED, ), ) registry.register(spec) context.runner = ToolRunner(registry=registry) context.runner.activate("test/preferred-pref") context.tool_name = "test/preferred-pref" @when("I execute the preferred-preference tool") def step_execute_preferred(context: Context) -> None: context.result = context.runner.execute(context.tool_name, {"input": "value"}) @given('I have a ToolRunner with a tool with SPECIFIC preference targeting "{target}"') def step_runner_specific_pref(context: Context, target: str) -> None: from cleveragents.domain.models.core.execution_environment_preference import ( EnvironmentPreferenceMode, ExecutionEnvironmentPreference, ) from cleveragents.tool.registry import ToolRegistry from cleveragents.tool.runner import ToolRunner from cleveragents.tool.runtime import ToolSpec registry = ToolRegistry() spec = ToolSpec( name="test/specific-pref", description="Tool targeting specific container", handler=_make_handler, execution_environment=ExecutionEnvironmentPreference( mode=EnvironmentPreferenceMode.SPECIFIC, target_resource=target, ), ) registry.register(spec) # Provide a mock container executor so container routing works from cleveragents.tool.container_executor import ContainerToolExecutor from cleveragents.tool.runtime import ToolResult mock_executor = create_autospec(ContainerToolExecutor, instance=True) mock_executor.execute_tool.return_value = ToolResult( success=True, output={"result": "container-ok"}, duration_ms=1.0, ) context.runner = ToolRunner(registry=registry, container_executor=mock_executor) context.runner.activate("test/specific-pref") context.tool_name = "test/specific-pref" @when("I execute the specific-preference tool with container available") def step_execute_specific_container(context: Context) -> None: context.result = context.runner.execute( context.tool_name, {"input": "value"}, linked_resource_types=["container-instance"], project_name="test-project", ) @then("the tool preference routing should succeed in container") def step_tool_container_mode(context: Context) -> None: assert context.result.success is True