freemo
80ebed962e
- Remove mix_stderr parameter from CliRunner (removed in Click 8.2+) - Fix import sorting issues (2 auto-fixed) - Fix line length issues (3 manual fixes) - Format 38 files with ruff - Fix ambiguous step definitions in tool_runtime_steps.py - Update feature files to use non-ambiguous step definitions All quality gates now pass: - Lint: ✓ - Format: ✓ - Typecheck: ✓ (only optional dependency warnings) - Security scan: ✓
314 lines
11 KiB
Python
314 lines
11 KiB
Python
"""Step definitions for devcontainer_handler.feature.
|
|
|
|
Tests devcontainer resource registration, auto-discovery, handler
|
|
protocol conformance, and resource type registry integration.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.domain.models.core.resource_type import ResourceTypeSpec
|
|
from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
|
|
from cleveragents.resource.handlers.discovery import (
|
|
DevcontainerDiscoveryResult,
|
|
discover_devcontainers,
|
|
is_trigger_type,
|
|
)
|
|
from cleveragents.resource.handlers.protocol import ResourceHandler
|
|
|
|
_VALID_DC_JSON: str = json.dumps(
|
|
{"name": "test-devcontainer", "image": "ubuntu:latest"}
|
|
)
|
|
|
|
|
|
# ── Given steps ──────────────────────────────────────────────
|
|
|
|
|
|
@given("a temporary directory with a valid devcontainer.json")
|
|
def step_tmp_dir_valid_dc(context: Context) -> None:
|
|
context.tmp_dir_obj = tempfile.TemporaryDirectory()
|
|
tmp = Path(context.tmp_dir_obj.name)
|
|
dc_dir = tmp / ".devcontainer"
|
|
dc_dir.mkdir()
|
|
(dc_dir / "devcontainer.json").write_text(_VALID_DC_JSON, encoding="utf-8")
|
|
context.tmp_path = str(tmp)
|
|
|
|
|
|
@given("a temporary directory simulating a git checkout")
|
|
def step_tmp_dir_git(context: Context) -> None:
|
|
context.tmp_dir_obj = tempfile.TemporaryDirectory()
|
|
context.tmp_path = context.tmp_dir_obj.name
|
|
|
|
|
|
@given("a temporary directory simulating a filesystem mount")
|
|
def step_tmp_dir_fs(context: Context) -> None:
|
|
context.tmp_dir_obj = tempfile.TemporaryDirectory()
|
|
context.tmp_path = context.tmp_dir_obj.name
|
|
|
|
|
|
@given("the directory has a .devcontainer/devcontainer.json file")
|
|
def step_add_dc_subdir(context: Context) -> None:
|
|
tmp = Path(context.tmp_path)
|
|
dc_dir = tmp / ".devcontainer"
|
|
dc_dir.mkdir(exist_ok=True)
|
|
(dc_dir / "devcontainer.json").write_text(_VALID_DC_JSON, encoding="utf-8")
|
|
|
|
|
|
@given("the directory has a root .devcontainer.json file")
|
|
def step_add_dc_root(context: Context) -> None:
|
|
tmp = Path(context.tmp_path)
|
|
(tmp / ".devcontainer.json").write_text(_VALID_DC_JSON, encoding="utf-8")
|
|
|
|
|
|
@given("a temporary directory with an invalid devcontainer.json")
|
|
def step_tmp_dir_invalid_dc(context: Context) -> None:
|
|
context.tmp_dir_obj = tempfile.TemporaryDirectory()
|
|
tmp = Path(context.tmp_dir_obj.name)
|
|
dc_dir = tmp / ".devcontainer"
|
|
dc_dir.mkdir()
|
|
(dc_dir / "devcontainer.json").write_text("not valid json {{", encoding="utf-8")
|
|
context.tmp_path = str(tmp)
|
|
|
|
|
|
@given("a temporary directory with a non-object devcontainer.json")
|
|
def step_tmp_dir_nonobj_dc(context: Context) -> None:
|
|
context.tmp_dir_obj = tempfile.TemporaryDirectory()
|
|
tmp = Path(context.tmp_dir_obj.name)
|
|
dc_dir = tmp / ".devcontainer"
|
|
dc_dir.mkdir()
|
|
(dc_dir / "devcontainer.json").write_text(
|
|
'["not", "an", "object"]', encoding="utf-8"
|
|
)
|
|
context.tmp_path = str(tmp)
|
|
|
|
|
|
@given("a temporary directory with no devcontainer configuration")
|
|
def step_tmp_dir_no_dc(context: Context) -> None:
|
|
context.tmp_dir_obj = tempfile.TemporaryDirectory()
|
|
context.tmp_path = context.tmp_dir_obj.name
|
|
|
|
|
|
@given("the built-in resource type registry")
|
|
def step_builtin_registry(context: Context) -> None:
|
|
context.builtin_names = ResourceTypeSpec.BUILTIN_NAMES
|
|
|
|
|
|
# ── When steps ───────────────────────────────────────────────
|
|
|
|
|
|
@when("I create a devcontainer-instance resource at that path")
|
|
def step_create_dc_resource(context: Context) -> None:
|
|
# Verify the type is actually recognised by the built-in registry
|
|
# instead of just assigning a string and asserting it back (U7).
|
|
context.resource_type = "devcontainer-instance"
|
|
assert context.resource_type in ResourceTypeSpec.BUILTIN_NAMES, (
|
|
f"'{context.resource_type}' is not a built-in resource type"
|
|
)
|
|
# Verify discovery finds the devcontainer in our temp directory
|
|
results = discover_devcontainers(
|
|
resource_location=context.tmp_path,
|
|
resource_type="git-checkout",
|
|
)
|
|
assert len(results) > 0, "Expected at least one discovery result"
|
|
context.resource_id = f"01TEST{results[0].config_path.name}"
|
|
context.resource_created = True
|
|
|
|
|
|
@when('I create a container-instance resource with image "{image}"')
|
|
def step_create_ctr_resource(context: Context, image: str) -> None:
|
|
# Verify the type is actually recognised by the built-in registry
|
|
# instead of just assigning a string and asserting it back (U7).
|
|
context.container_type = "container-instance"
|
|
assert context.container_type in ResourceTypeSpec.BUILTIN_NAMES, (
|
|
f"'{context.container_type}' is not a built-in resource type"
|
|
)
|
|
context.container_image = image
|
|
context.container_created = True
|
|
|
|
|
|
@when('I run devcontainer discovery on the directory as "{rtype}"')
|
|
def step_run_discovery(context: Context, rtype: str) -> None:
|
|
context.discovery_results = discover_devcontainers(
|
|
resource_location=context.tmp_path,
|
|
resource_type=rtype,
|
|
)
|
|
|
|
|
|
@when("I list all built-in type names")
|
|
def step_list_builtin_names(context: Context) -> None:
|
|
context.type_name_list = sorted(context.builtin_names)
|
|
|
|
|
|
@when("I instantiate DevcontainerHandler")
|
|
def step_instantiate_handler(context: Context) -> None:
|
|
context.handler = DevcontainerHandler()
|
|
|
|
|
|
@when("I create a discovery result with valid parameters")
|
|
def step_create_valid_result(context: Context) -> None:
|
|
context.tmp_dir_obj = tempfile.TemporaryDirectory()
|
|
tmp = Path(context.tmp_dir_obj.name)
|
|
dc_file = tmp / "devcontainer.json"
|
|
dc_file.write_text(_VALID_DC_JSON, encoding="utf-8")
|
|
context.discovery_result = DevcontainerDiscoveryResult(
|
|
config_path=dc_file,
|
|
config_data=json.loads(_VALID_DC_JSON),
|
|
parent_location=str(tmp),
|
|
)
|
|
|
|
|
|
@when("I create a discovery result with empty parent_location")
|
|
def step_create_invalid_result(context: Context) -> None:
|
|
context.tmp_dir_obj = tempfile.TemporaryDirectory()
|
|
tmp = Path(context.tmp_dir_obj.name)
|
|
dc_file = tmp / "devcontainer.json"
|
|
dc_file.write_text(_VALID_DC_JSON, encoding="utf-8")
|
|
context.result_error = None
|
|
try:
|
|
DevcontainerDiscoveryResult(
|
|
config_path=dc_file,
|
|
config_data=json.loads(_VALID_DC_JSON),
|
|
parent_location="",
|
|
)
|
|
except ValueError as exc:
|
|
context.result_error = exc
|
|
|
|
|
|
# ── Then steps ───────────────────────────────────────────────
|
|
|
|
|
|
@then('the resource should have type "{expected_type}"')
|
|
def step_check_resource_type(context: Context, expected_type: str) -> None:
|
|
assert context.resource_type == expected_type
|
|
|
|
|
|
@then("the resource should have a non-empty resource_id")
|
|
def step_check_resource_id(context: Context) -> None:
|
|
assert context.resource_id
|
|
assert len(context.resource_id) > 0
|
|
|
|
|
|
@then('the container resource should have type "{expected_type}"')
|
|
def step_check_container_type(context: Context, expected_type: str) -> None:
|
|
assert context.container_type == expected_type
|
|
|
|
|
|
@then("discovery should return {count:d} result")
|
|
def step_check_discovery_count_singular(context: Context, count: int) -> None:
|
|
assert len(context.discovery_results) == count
|
|
|
|
|
|
@then("discovery should return {count:d} results")
|
|
def step_check_discovery_count_plural(context: Context, count: int) -> None:
|
|
assert len(context.discovery_results) == count
|
|
|
|
|
|
@then('the discovered config path should end with "{suffix}"')
|
|
def step_check_config_path_suffix(context: Context, suffix: str) -> None:
|
|
assert len(context.discovery_results) > 0
|
|
result = context.discovery_results[0]
|
|
assert str(result.config_path).endswith(suffix)
|
|
|
|
|
|
@then('the type list should contain "{type_name}"')
|
|
def step_check_type_in_list(context: Context, type_name: str) -> None:
|
|
assert type_name in context.builtin_names
|
|
|
|
|
|
@then("it should satisfy the ResourceHandler protocol")
|
|
def step_check_protocol(context: Context) -> None:
|
|
assert isinstance(context.handler, ResourceHandler)
|
|
|
|
|
|
@then('its default strategy should be "{strategy}"')
|
|
def step_check_default_strategy(context: Context, strategy: str) -> None:
|
|
assert context.handler._default_strategy.value == strategy
|
|
|
|
|
|
@then('its type label should be "{label}"')
|
|
def step_check_type_label(context: Context, label: str) -> None:
|
|
assert context.handler._type_label == label
|
|
|
|
|
|
@then("the result should have a config_path attribute")
|
|
def step_check_result_attr(context: Context) -> None:
|
|
assert hasattr(context.discovery_result, "config_path")
|
|
assert context.discovery_result.config_path is not None
|
|
|
|
|
|
@then("it should raise a ValueError")
|
|
def step_check_value_error(context: Context) -> None:
|
|
assert context.result_error is not None
|
|
assert isinstance(context.result_error, ValueError)
|
|
|
|
|
|
@then('"{rtype}" should be a trigger type')
|
|
def step_check_is_trigger(context: Context, rtype: str) -> None:
|
|
assert is_trigger_type(rtype) is True
|
|
|
|
|
|
@then('"{rtype}" should not be a trigger type')
|
|
def step_check_not_trigger(context: Context, rtype: str) -> None:
|
|
assert is_trigger_type(rtype) is False
|
|
|
|
|
|
# ── Given steps for named configurations ─────────────────────
|
|
|
|
|
|
@given('the directory has a named devcontainer configuration "{name}"')
|
|
def step_add_named_dc(context: Context, name: str) -> None:
|
|
tmp = Path(context.tmp_path)
|
|
named_dir = tmp / ".devcontainer" / name
|
|
named_dir.mkdir(parents=True, exist_ok=True)
|
|
(named_dir / "devcontainer.json").write_text(_VALID_DC_JSON, encoding="utf-8")
|
|
|
|
|
|
@given(
|
|
'the directory has a named devcontainer configuration "{name}" with invalid JSON'
|
|
)
|
|
def step_add_named_dc_invalid(context: Context, name: str) -> None:
|
|
tmp = Path(context.tmp_path)
|
|
named_dir = tmp / ".devcontainer" / name
|
|
named_dir.mkdir(parents=True, exist_ok=True)
|
|
(named_dir / "devcontainer.json").write_text("not valid json {{", encoding="utf-8")
|
|
|
|
|
|
@given("the directory has an empty .devcontainer directory")
|
|
def step_add_empty_dc_dir(context: Context) -> None:
|
|
tmp = Path(context.tmp_path)
|
|
dc_dir = tmp / ".devcontainer"
|
|
dc_dir.mkdir(exist_ok=True)
|
|
|
|
|
|
# ── Then steps for named configurations ──────────────────────
|
|
|
|
|
|
@then('the first result should have config name "{name}"')
|
|
def step_check_first_result_config_name(context: Context, name: str) -> None:
|
|
assert len(context.discovery_results) > 0
|
|
result = context.discovery_results[0]
|
|
assert result.config_name == name, (
|
|
f"Expected config_name={name!r}, got {result.config_name!r}"
|
|
)
|
|
|
|
|
|
@then("the first result should have no config name")
|
|
def step_check_first_result_no_config_name(context: Context) -> None:
|
|
assert len(context.discovery_results) > 0
|
|
result = context.discovery_results[0]
|
|
assert result.config_name is None, (
|
|
f"Expected config_name=None, got {result.config_name!r}"
|
|
)
|
|
|
|
|
|
@then('the result config names should include "{name}"')
|
|
def step_check_result_config_names_include(context: Context, name: str) -> None:
|
|
names = {r.config_name for r in context.discovery_results}
|
|
assert name in names, f"Expected {name!r} in config names, got {names!r}"
|