HAL9000
d51f3a05ae
- Renamed 'it should contain' steps to 'the container types should contain' for specificity - Updated execution_environment.feature to use the new step names - This fixes the AmbiguousStep error that was preventing unit tests from running
538 lines
18 KiB
Python
538 lines
18 KiB
Python
"""Step definitions for execution environment routing feature tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from enum import StrEnum
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Enum steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I import ExecutionEnvironment")
|
|
def step_import_enum(context: Context) -> None:
|
|
from cleveragents.domain.models.core.plan import ExecutionEnvironment
|
|
|
|
context.exec_env_cls = ExecutionEnvironment
|
|
|
|
|
|
@then('the HOST value should be "{value}"')
|
|
def step_host_value(context: Context, value: str) -> None:
|
|
assert context.exec_env_cls.HOST.value == value
|
|
|
|
|
|
@then('the CONTAINER value should be "{value}"')
|
|
def step_container_value(context: Context, value: str) -> None:
|
|
assert context.exec_env_cls.CONTAINER.value == value
|
|
|
|
|
|
@then("ExecutionEnvironment should be a StrEnum subclass")
|
|
def step_is_strenum(context: Context) -> None:
|
|
assert issubclass(context.exec_env_cls, StrEnum)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plan model steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I create a plan with default fields")
|
|
def step_plan_default(context: Context) -> None:
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
)
|
|
|
|
context.plan = Plan(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(namespace="local", name="test-plan"),
|
|
description="Test plan",
|
|
action_name="local/test-action",
|
|
definition_of_done="All tests pass",
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
created_by="test",
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
|
|
|
|
@given('I create a plan with execution_environment "{env}"')
|
|
def step_plan_with_env(context: Context, env: str) -> None:
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
)
|
|
|
|
context.plan = Plan(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(namespace="local", name="test-plan"),
|
|
description="Test plan",
|
|
action_name="local/test-action",
|
|
execution_environment=env,
|
|
definition_of_done="All tests pass",
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
created_by="test",
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
|
|
|
|
@then("the plan execution_environment should be None")
|
|
def step_plan_env_none(context: Context) -> None:
|
|
assert context.plan.execution_environment is None
|
|
|
|
|
|
@then('the plan execution_environment should be "{env}"')
|
|
def step_plan_env_value(context: Context, env: str) -> None:
|
|
assert context.plan.execution_environment == env
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ContextConfig model steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I create a ContextConfig with default fields")
|
|
def step_context_config_default(context: Context) -> None:
|
|
from cleveragents.domain.models.core.project import ContextConfig
|
|
|
|
context.context_config = ContextConfig()
|
|
|
|
|
|
@given('I create a ContextConfig with execution_environment "{env}"')
|
|
def step_context_config_with_env(context: Context, env: str) -> None:
|
|
from cleveragents.domain.models.core.project import ContextConfig
|
|
|
|
context.context_config = ContextConfig(execution_environment=env)
|
|
|
|
|
|
@then("the context config execution_environment should be None")
|
|
def step_context_config_env_none(context: Context) -> None:
|
|
assert context.context_config.execution_environment is None
|
|
|
|
|
|
@then('the context config execution_environment should be "{env}"')
|
|
def step_context_config_env_value(context: Context, env: str) -> None:
|
|
assert context.context_config.execution_environment == env
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Resolver steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I create an ExecutionEnvironmentResolver")
|
|
def step_create_resolver(context: Context) -> None:
|
|
from cleveragents.application.services.execution_environment_resolver import (
|
|
ExecutionEnvironmentResolver,
|
|
)
|
|
|
|
context.resolver = ExecutionEnvironmentResolver()
|
|
context.resolved_env = None
|
|
context.raised_error: Exception | None = None
|
|
|
|
|
|
@when("I resolve with no overrides")
|
|
def step_resolve_no_overrides(context: Context) -> None:
|
|
context.resolved_env = context.resolver.resolve()
|
|
|
|
|
|
@when('I resolve with default "{default}"')
|
|
def step_resolve_default(context: Context, default: str) -> None:
|
|
context.resolved_env = context.resolver.resolve(default=default)
|
|
|
|
|
|
@when('I resolve with project_env "{project_env}"')
|
|
def step_resolve_project(context: Context, project_env: str) -> None:
|
|
context.resolved_env = context.resolver.resolve(project_env=project_env)
|
|
|
|
|
|
@when('I resolve with plan_env "{plan_env}" and project_env "{project_env}"')
|
|
def step_resolve_plan_project(
|
|
context: Context, plan_env: str, project_env: str
|
|
) -> None:
|
|
context.resolved_env = context.resolver.resolve(
|
|
plan_env=plan_env, project_env=project_env
|
|
)
|
|
|
|
|
|
@when('I resolve with tool_env "{tool_env}" and plan_env "{plan_env}"')
|
|
def step_resolve_tool_plan(context: Context, tool_env: str, plan_env: str) -> None:
|
|
context.resolved_env = context.resolver.resolve(
|
|
tool_env=tool_env, plan_env=plan_env
|
|
)
|
|
|
|
|
|
@when(
|
|
'I resolve the full chain with tool_env "{tool_env}" and plan_env '
|
|
'"{plan_env}" and project_env "{project_env}"'
|
|
)
|
|
def step_resolve_full_chain(
|
|
context: Context, tool_env: str, plan_env: str, project_env: str
|
|
) -> None:
|
|
context.resolved_env = context.resolver.resolve(
|
|
tool_env=tool_env, plan_env=plan_env, project_env=project_env
|
|
)
|
|
|
|
|
|
@when('I resolve with tool_env "{tool_env}"')
|
|
def step_resolve_tool_only(context: Context, tool_env: str) -> None:
|
|
try:
|
|
context.resolved_env = context.resolver.resolve(tool_env=tool_env)
|
|
except ValueError as exc:
|
|
context.raised_error = exc
|
|
|
|
|
|
@then('the resolved environment should be "{expected}"')
|
|
def step_resolved_env(context: Context, expected: str) -> None:
|
|
assert context.resolved_env is not None
|
|
assert context.resolved_env.value == expected
|
|
|
|
|
|
@then("exec-env a ValueError should be raised")
|
|
def step_value_error_raised(context: Context) -> None:
|
|
assert isinstance(context.raised_error, ValueError)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Container availability steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I validate container available with types "{types_str}"')
|
|
def step_validate_container(context: Context, types_str: str) -> None:
|
|
from cleveragents.application.services.execution_environment_resolver import (
|
|
ContainerUnavailableError,
|
|
)
|
|
|
|
types = [t.strip() for t in types_str.split(",")]
|
|
try:
|
|
context.resolver.validate_container_available(types)
|
|
context.validation_ok = True
|
|
context.raised_error = None
|
|
except ContainerUnavailableError as exc:
|
|
context.validation_ok = False
|
|
context.raised_error = exc
|
|
|
|
|
|
@then("validation should succeed")
|
|
def step_validation_ok(context: Context) -> None:
|
|
assert context.validation_ok is True
|
|
|
|
|
|
@then("a ContainerUnavailableError should be raised")
|
|
def step_container_error_raised(context: Context) -> None:
|
|
from cleveragents.application.services.execution_environment_resolver import (
|
|
ContainerUnavailableError,
|
|
)
|
|
|
|
assert isinstance(context.raised_error, ContainerUnavailableError)
|
|
|
|
|
|
@when('I resolve_and_validate with project_env "{project_env}" and types "{types_str}"')
|
|
def step_resolve_validate_project(
|
|
context: Context, project_env: str, types_str: str
|
|
) -> None:
|
|
from cleveragents.application.services.execution_environment_resolver import (
|
|
ContainerUnavailableError,
|
|
)
|
|
|
|
types = [t.strip() for t in types_str.split(",")]
|
|
try:
|
|
context.resolved_env = context.resolver.resolve_and_validate(
|
|
linked_resource_types=types,
|
|
project_env=project_env,
|
|
)
|
|
context.raised_error = None
|
|
except ContainerUnavailableError as exc:
|
|
context.raised_error = exc
|
|
|
|
|
|
@when('I resolve_and_validate with plan_env "{plan_env}" and types "{types_str}"')
|
|
def step_resolve_validate_plan(context: Context, plan_env: str, types_str: str) -> None:
|
|
from cleveragents.application.services.execution_environment_resolver import (
|
|
ContainerUnavailableError,
|
|
)
|
|
|
|
types = [t.strip() for t in types_str.split(",")]
|
|
try:
|
|
context.resolved_env = context.resolver.resolve_and_validate(
|
|
linked_resource_types=types,
|
|
plan_env=plan_env,
|
|
)
|
|
context.raised_error = None
|
|
except ContainerUnavailableError as exc:
|
|
context.raised_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ToolRunner integration steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I create a ToolRunner with a mock registry")
|
|
def step_create_tool_runner(context: Context) -> None:
|
|
from cleveragents.tool.registry import ToolRegistry
|
|
from cleveragents.tool.runner import ToolRunner
|
|
from cleveragents.tool.runtime import ToolSpec
|
|
|
|
registry = ToolRegistry()
|
|
|
|
def echo_handler(inputs: dict[str, Any]) -> dict[str, Any]:
|
|
return {"echo": inputs}
|
|
|
|
spec = ToolSpec(
|
|
name="test/echo",
|
|
description="Echo tool",
|
|
input_schema={"type": "object", "properties": {}},
|
|
handler=echo_handler,
|
|
)
|
|
registry.register(spec)
|
|
context.runner = ToolRunner(registry)
|
|
context.tool_result = None
|
|
|
|
|
|
@when("I execute a tool with no environment overrides")
|
|
def step_execute_no_overrides(context: Context) -> None:
|
|
context.tool_result = context.runner.execute("test/echo", {"key": "value"})
|
|
|
|
|
|
@then("the tool should execute successfully on host")
|
|
def step_tool_success_host(context: Context) -> None:
|
|
assert context.tool_result is not None
|
|
assert context.tool_result.success is True
|
|
|
|
|
|
@when('I execute a tool with plan_env "{plan_env}" and no container resources')
|
|
def step_execute_container_no_resource(context: Context, plan_env: str) -> None:
|
|
context.tool_result = context.runner.execute(
|
|
"test/echo",
|
|
{"key": "value"},
|
|
plan_env=plan_env,
|
|
linked_resource_types=["git-checkout"],
|
|
)
|
|
|
|
|
|
@then("the tool result should have an error about container unavailable")
|
|
def step_tool_container_error(context: Context) -> None:
|
|
assert context.tool_result is not None
|
|
assert context.tool_result.success is False
|
|
assert "Container resource unavailable" in (context.tool_result.error or "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CONTAINER_RESOURCE_TYPES steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I import CONTAINER_RESOURCE_TYPES")
|
|
def step_import_types(context: Context) -> None:
|
|
from cleveragents.application.services.execution_environment_resolver import (
|
|
CONTAINER_RESOURCE_TYPES,
|
|
)
|
|
|
|
context.container_types = CONTAINER_RESOURCE_TYPES
|
|
|
|
|
|
@then('the container types should contain "{value}"')
|
|
def step_types_contain(context: Context, value: str) -> None:
|
|
assert value in context.container_types
|
|
|
|
|
|
@then('the container types should not contain "{value}"')
|
|
def step_types_not_contain(context: Context, value: str) -> None:
|
|
assert value not in context.container_types
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ContainerUnavailableError steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a ContainerUnavailableError for project "{name}"')
|
|
def step_create_error_with_project(context: Context, name: str) -> None:
|
|
from cleveragents.application.services.execution_environment_resolver import (
|
|
ContainerUnavailableError,
|
|
)
|
|
|
|
context.container_error = ContainerUnavailableError(name)
|
|
|
|
|
|
@when("I create a ContainerUnavailableError without project name")
|
|
def step_create_error_no_project(context: Context) -> None:
|
|
from cleveragents.application.services.execution_environment_resolver import (
|
|
ContainerUnavailableError,
|
|
)
|
|
|
|
context.container_error = ContainerUnavailableError()
|
|
|
|
|
|
@then('exec-env the error message should contain "{text}"')
|
|
def step_error_contains(context: Context, text: str) -> None:
|
|
assert text in str(context.container_error)
|
|
|
|
|
|
@then('exec-env the error message should not contain "{text}"')
|
|
def step_error_not_contains(context: Context, text: str) -> None:
|
|
assert text not in str(context.container_error)
|
|
|
|
|
|
# ── 6-Level Precedence Chain Steps (#877) ────────────────
|
|
|
|
|
|
def _resolver():
|
|
from cleveragents.application.services.execution_environment_resolver import (
|
|
ExecutionEnvironmentResolver,
|
|
)
|
|
|
|
return ExecutionEnvironmentResolver()
|
|
|
|
|
|
@when(
|
|
'I resolve with plan_env "{plan}" plan_priority "{pp}" '
|
|
'project_env "{proj}" project_priority "{projp}" '
|
|
"devcontainer {dc}"
|
|
)
|
|
def step_resolve_6level(
|
|
context: Context, plan: str, pp: str, proj: str, projp: str, dc: str
|
|
) -> None:
|
|
context.resolved_env = _resolver().resolve_with_precedence(
|
|
plan_env=plan,
|
|
plan_priority=pp,
|
|
project_env=proj,
|
|
project_priority=projp,
|
|
devcontainer_available=dc.lower() == "true",
|
|
)
|
|
|
|
|
|
@when(
|
|
'I resolve with tool_env "{tool}" plan_env "{plan}" '
|
|
'plan_priority "{pp}" devcontainer {dc}'
|
|
)
|
|
def step_resolve_tool_wins(
|
|
context: Context, tool: str, plan: str, pp: str, dc: str
|
|
) -> None:
|
|
context.resolved_env = _resolver().resolve_with_precedence(
|
|
tool_env=tool,
|
|
plan_env=plan,
|
|
plan_priority=pp,
|
|
devcontainer_available=dc.lower() == "true",
|
|
)
|
|
|
|
|
|
@when(
|
|
'I resolve with project_env "{proj}" project_priority "{projp}" devcontainer {dc}'
|
|
)
|
|
def step_resolve_project_prio(context: Context, proj: str, projp: str, dc: str) -> None:
|
|
context.resolved_env = _resolver().resolve_with_precedence(
|
|
project_env=proj,
|
|
project_priority=projp,
|
|
devcontainer_available=dc.lower() == "true",
|
|
)
|
|
|
|
|
|
@when("I resolve with devcontainer {dc}")
|
|
def step_resolve_dc_only(context: Context, dc: str) -> None:
|
|
context.resolved_env = _resolver().resolve_with_precedence(
|
|
devcontainer_available=dc.lower() == "true",
|
|
)
|
|
|
|
|
|
@when('I resolve with plan_env "{plan}" project_env "{proj}" devcontainer {dc}')
|
|
def step_resolve_both_fallback(context: Context, plan: str, proj: str, dc: str) -> None:
|
|
context.resolved_env = _resolver().resolve_with_precedence(
|
|
plan_env=plan,
|
|
project_env=proj,
|
|
devcontainer_available=dc.lower() == "true",
|
|
)
|
|
|
|
|
|
@when('I check has_devcontainer with types "{types}"')
|
|
def step_has_devcontainer(context: Context, types: str) -> None:
|
|
type_list = [t.strip() for t in types.split(",")]
|
|
context.has_dc_result = _resolver().has_devcontainer(type_list)
|
|
|
|
|
|
@then("exec-env has_devcontainer should be {expected}")
|
|
def step_has_dc_assert(context: Context, expected: str) -> None:
|
|
assert context.has_dc_result == (expected.lower() == "true")
|
|
|
|
|
|
@when('I create a ContextConfig with execution_env_priority "{priority}"')
|
|
def step_context_config_priority(context: Context, priority: str) -> None:
|
|
from cleveragents.domain.models.core.project import ContextConfig
|
|
|
|
config = ContextConfig(
|
|
execution_environment="container",
|
|
execution_env_priority=priority,
|
|
)
|
|
context.ctx_config = config
|
|
|
|
|
|
@then('exec-env the context config priority should be "{priority}"')
|
|
def step_ctx_config_priority_check(context: Context, priority: str) -> None:
|
|
assert context.ctx_config.execution_env_priority == priority
|
|
|
|
|
|
# ── DAG walk and container ref steps (#877 subtasks 2-4) ──
|
|
|
|
|
|
@when('I resolve with DAG walk from "{resource_id}" with devcontainer ancestor')
|
|
def step_dag_walk_with_dc(context: Context, resource_id: str) -> None:
|
|
resolver = _resolver()
|
|
# Simulate a DAG: res-file -> res-dir -> res-dc (devcontainer-instance)
|
|
linked_types = {
|
|
resource_id: "fs-file",
|
|
"res-dir": "fs-directory",
|
|
"res-dc": "devcontainer-instance",
|
|
}
|
|
parent_map = {
|
|
resource_id: ["res-dir"],
|
|
"res-dir": ["res-dc"],
|
|
}
|
|
context.resolved_env = resolver.resolve_with_dag(
|
|
resource_id=resource_id,
|
|
linked_resource_types=linked_types,
|
|
parent_map=parent_map,
|
|
)
|
|
|
|
|
|
@when('I resolve with DAG walk from "{resource_id}" without devcontainer ancestor')
|
|
def step_dag_walk_no_dc(context: Context, resource_id: str) -> None:
|
|
resolver = _resolver()
|
|
linked_types = {
|
|
resource_id: "fs-file",
|
|
"res-dir": "fs-directory",
|
|
"res-checkout": "git-checkout",
|
|
}
|
|
parent_map = {
|
|
resource_id: ["res-dir"],
|
|
"res-dir": ["res-checkout"],
|
|
}
|
|
context.resolved_env = resolver.resolve_with_dag(
|
|
resource_id=resource_id,
|
|
linked_resource_types=linked_types,
|
|
parent_map=parent_map,
|
|
)
|
|
|
|
|
|
@when('I parse container ref "{ref}"')
|
|
def step_parse_container_ref(context: Context, ref: str) -> None:
|
|
resolver = _resolver()
|
|
env, name = resolver.parse_container_ref(ref)
|
|
context.resolved_env = env
|
|
context.container_ref_name = name
|
|
|
|
|
|
@then('exec-env the container ref name should be "{name}"')
|
|
def step_container_ref_name(context: Context, name: str) -> None:
|
|
assert context.container_ref_name == name
|