"""Step definitions for binding_resolution_coverage.feature. Covers the uncovered lines in binding_resolution_service.py: - Static binding with unknown resource (lines 147-148) - Parameter binding with unknown ref (lines 198-201) - Parameter binding with valid ref (lines 209, 214) - Contextual binding, slot not required, no match (line 274) - _find_matching_resources skips NotFoundError (lines 298-299) - _disambiguate via resource name suffix (lines 325-329) - _disambiguate ambiguous raises ValidationError (lines 337-338) - _is_type_compatible returns False on NotFoundError (lines 369-370) All Background / shared steps are inherited from binding_resolution_steps.py. Only NEW step patterns are defined here to avoid duplication. """ from __future__ import annotations from behave import given, then, when from behave.runner import Context from cleveragents.core.exceptions import ValidationError from cleveragents.domain.models.core.project import ( LinkedResource, NamespacedProject, ) from cleveragents.domain.models.core.resource import ( PhysVirt, Resource, ) from cleveragents.domain.models.core.tool import ( BindingMode, ResourceAccessMode, ResourceSlot, Tool, ToolSource, ToolType, ) # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def _make_resource( resource_id: str, type_name: str, name: str | None = None, ) -> Resource: return Resource( resource_id=resource_id, name=name, resource_type_name=type_name, classification=PhysVirt.PHYSICAL, description=f"Test resource {resource_id}", ) # ------------------------------------------------------------------ # Given: named resource in registry (with both id and name keys) # ------------------------------------------------------------------ @given( 'a named resource "{rid}" called "{res_name}" of type "{type_name}" in the registry' ) def step_named_resource_in_registry( context: Context, rid: str, res_name: str, type_name: str, ) -> None: resource = _make_resource(rid, type_name, name=res_name) context.mock_registry._resources[res_name] = resource context.mock_registry._resources[rid] = resource # ------------------------------------------------------------------ # Given: optional contextual slot # ------------------------------------------------------------------ @given( 'a tool "{tool_name}" with an optional contextual slot "{slot}" of type "{rtype}"' ) def step_tool_optional_contextual( context: Context, tool_name: str, slot: str, rtype: str, ) -> None: context.tool = Tool( name=tool_name, description=f"Tool {tool_name}", source=ToolSource.BUILTIN, tool_type=ToolType.TOOL, resource_slots=[ ResourceSlot( name=slot, resource_type=rtype, access=ResourceAccessMode.READ_WRITE, binding=BindingMode.CONTEXTUAL, required=False, ), ], ) # ------------------------------------------------------------------ # Given: project with linked resources including missing ones # ------------------------------------------------------------------ @given('a project "{name}" with linked resources including a missing one:') def step_project_with_missing_resource( context: Context, name: str, ) -> None: parts = name.split("/", 1) namespace = parts[0] if len(parts) == 2 else "local" proj_name = parts[1] if len(parts) == 2 else parts[0] links: list[LinkedResource] = [] for row in context.table: rid = row["resource_id"] tname = row["type_name"] alias = row["alias"] if row["alias"] else None exists = row["exists"].lower() == "true" if exists: resource = _make_resource(rid, tname, name=None) context.mock_registry._resources[rid] = resource # Always create the link, even for non-existent resources links.append( LinkedResource( resource_id=rid, alias=alias, ) ) context.project = NamespacedProject( name=proj_name, namespace=namespace, linked_resources=links, ) # ------------------------------------------------------------------ # Given: project with named linked resources (resources have names) # ------------------------------------------------------------------ @given('a project "{name}" with named linked resources:') def step_project_named_resources( context: Context, name: str, ) -> None: parts = name.split("/", 1) namespace = parts[0] if len(parts) == 2 else "local" proj_name = parts[1] if len(parts) == 2 else parts[0] links: list[LinkedResource] = [] for row in context.table: rid = row["resource_id"] tname = row["type_name"] alias = row["alias"] if row["alias"] else None res_name = row["resource_name"] if row["resource_name"] else None resource = _make_resource(rid, tname, name=res_name) context.mock_registry._resources[rid] = resource links.append( LinkedResource( resource_id=rid, alias=alias, ) ) context.project = NamespacedProject( name=proj_name, namespace=namespace, linked_resources=links, ) # ------------------------------------------------------------------ # Given: remove resource type from registry # ------------------------------------------------------------------ @given('the resource type "{tname}" is removed from the registry') def step_remove_resource_type(context: Context, tname: str) -> None: context.mock_registry._types.pop(tname, None) # ------------------------------------------------------------------ # Given: mark slot as required (for post-hoc override) # ------------------------------------------------------------------ @given('the slot "{slot_name}" is required') def step_slot_is_required(context: Context, slot_name: str) -> None: # The tool already has required=True by default; this is a # documentation step that confirms the slot is required. for slot in context.tool.resource_slots: if slot.name == slot_name: assert slot.required, f"Expected slot '{slot_name}' to be required" return raise AssertionError(f"No slot named '{slot_name}' on tool") # ------------------------------------------------------------------ # When: resolve with invocation params # ------------------------------------------------------------------ @when("the bindings are resolved with invocation params:") def step_resolve_with_params(context: Context) -> None: params: dict[str, str] = {} for row in context.table: params[row["slot"]] = row["ref"] try: context.binding_results = context.binding_service.resolve( context.tool, context.project, invocation_params=params, ) context.binding_error = None except ValidationError as exc: context.binding_error = exc context.binding_results = None # ------------------------------------------------------------------ # Then: check resource_name on binding result # ------------------------------------------------------------------ @then('the binding for slot "{slot}" should have resource_name "{rname}"') def step_check_resource_name( context: Context, slot: str, rname: str, ) -> None: result = _find_result(context, slot) assert result.resource_name == rname, ( f"Expected resource_name '{rname}', got '{result.resource_name}'" ) # ------------------------------------------------------------------ # Internal helper # ------------------------------------------------------------------ def _find_result(context: Context, slot_name: str): """Find a binding result by slot name.""" for result in context.binding_results: if result.slot_name == slot_name: return result raise AssertionError(f"No binding result for slot '{slot_name}'")