forked from HAL9000/cleveragents-core
a808c395f9
Add 53 new .feature files and corresponding step definition files targeting uncovered lines identified in build/coverage.xml. Fix AmbiguousStep conflicts in 7 pre-existing step files by disambiguating step text. New tests cover: ACP clients/facade, actor CLI/config, application container, ACMS service/strategies, async worker, automation profile CLI, autonomy guardrail, bridge, change model, config CLI/service, context service, cross-plan correction, database models, decision service, decomposition clustering/service, discovery handler, langchain chat provider, langgraph nodes, materializers, multi-project service, plan apply/CLI/lifecycle/model/ preflight/resume/service, PostgreSQL analyzer, project CLI/context CLI, provider registry, reactive application/route, repositories, resolver handler, resource registry service, resume model, retry patterns, sandbox protocol, server CLI, skill CLI/service, skills registry, subplan execution/service, system CLI, UKO loader, UoW, and YAML template engine. Closes #645
221 lines
8.0 KiB
Python
221 lines
8.0 KiB
Python
"""Step definitions for discovery_handler_coverage_boost.feature.
|
|
|
|
Targets uncovered lines in src/cleveragents/resource/handlers/discovery.py:
|
|
- Lines 57-58: TypeError for non-Path config_path
|
|
- Lines 61-62: TypeError for non-dict config_data
|
|
- Line 94: ValueError for empty resource_location
|
|
- Line 96: ValueError for empty resource_type
|
|
- Lines 103-107: non-directory resource location branch
|
|
- Line 140: ValueError for empty resource_type in is_trigger_type
|
|
- Lines 155-157: OSError reading devcontainer.json
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.resource.handlers.discovery import (
|
|
DevcontainerDiscoveryResult,
|
|
discover_devcontainers,
|
|
is_trigger_type,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the devcontainer discovery module is imported")
|
|
def step_module_imported(context: Context) -> None:
|
|
"""Verify imports are accessible."""
|
|
assert DevcontainerDiscoveryResult is not None
|
|
assert discover_devcontainers is not None
|
|
assert is_trigger_type is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DevcontainerDiscoveryResult - TypeError for non-Path config_path (L57-58)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a DevcontainerDiscoveryResult with a string config_path")
|
|
def step_create_result_string_config_path(context: Context) -> None:
|
|
context.discovery_caught_error = None
|
|
try:
|
|
DevcontainerDiscoveryResult(
|
|
config_path="/some/string/path", # type: ignore[arg-type]
|
|
config_data={"name": "test"},
|
|
parent_location="/parent",
|
|
)
|
|
except TypeError as exc:
|
|
context.discovery_caught_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DevcontainerDiscoveryResult - TypeError for non-dict config_data (L61-62)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create a DevcontainerDiscoveryResult with a list config_data")
|
|
def step_create_result_list_config_data(context: Context) -> None:
|
|
context.discovery_caught_error = None
|
|
try:
|
|
DevcontainerDiscoveryResult(
|
|
config_path=Path("/some/path/devcontainer.json"),
|
|
config_data=["not", "a", "dict"], # type: ignore[arg-type]
|
|
parent_location="/parent",
|
|
)
|
|
except TypeError as exc:
|
|
context.discovery_caught_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# discover_devcontainers - empty resource_location (L94)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I call discover_devcontainers with an empty resource_location")
|
|
def step_discover_empty_location(context: Context) -> None:
|
|
context.discovery_caught_error = None
|
|
try:
|
|
discover_devcontainers(resource_location="", resource_type="git-checkout")
|
|
except ValueError as exc:
|
|
context.discovery_caught_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# discover_devcontainers - empty resource_type (L96)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I call discover_devcontainers with an empty resource_type")
|
|
def step_discover_empty_type(context: Context) -> None:
|
|
context.discovery_caught_error = None
|
|
try:
|
|
discover_devcontainers(resource_location="/some/path", resource_type="")
|
|
except ValueError as exc:
|
|
context.discovery_caught_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# discover_devcontainers - non-directory location (L103-107)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a temporary file that is not a directory")
|
|
def step_create_temp_file(context: Context) -> None:
|
|
fd, path = tempfile.mkstemp(suffix=".txt")
|
|
os.write(fd, b"just a file, not a directory")
|
|
os.close(fd)
|
|
context.temp_file_path = path
|
|
context.add_cleanup(lambda: os.unlink(path))
|
|
|
|
|
|
@when(
|
|
'I call discover_devcontainers with that file path as location and type "{rtype}"'
|
|
)
|
|
def step_discover_non_dir(context: Context, rtype: str) -> None:
|
|
context.discovery_result_list = discover_devcontainers(
|
|
resource_location=context.temp_file_path,
|
|
resource_type=rtype,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# is_trigger_type - empty resource_type (L140)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I call is_trigger_type with an empty string")
|
|
def step_trigger_type_empty(context: Context) -> None:
|
|
context.discovery_caught_error = None
|
|
try:
|
|
is_trigger_type("")
|
|
except ValueError as exc:
|
|
context.discovery_caught_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _load_devcontainer_json - OSError (L155-157)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a temporary directory with an unreadable devcontainer.json")
|
|
def step_create_unreadable_dc(context: Context) -> None:
|
|
tmp_obj = tempfile.TemporaryDirectory()
|
|
tmp = Path(tmp_obj.name)
|
|
context.add_cleanup(tmp_obj.cleanup)
|
|
|
|
dc_dir = tmp / ".devcontainer"
|
|
dc_dir.mkdir()
|
|
dc_file = dc_dir / "devcontainer.json"
|
|
dc_file.write_text('{"name": "test"}', encoding="utf-8")
|
|
|
|
# Patch read_text to raise OSError so the test is reliable even when
|
|
# running as root (where chmod 0o000 is ignored).
|
|
from unittest.mock import patch
|
|
|
|
original_read_text = Path.read_text
|
|
|
|
def _patched_read_text(self: Path, *args: object, **kwargs: object) -> str:
|
|
if self == dc_file:
|
|
raise OSError("Permission denied (mocked)")
|
|
return original_read_text(self, *args, **kwargs) # type: ignore[arg-type]
|
|
|
|
patcher = patch.object(Path, "read_text", _patched_read_text)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
|
|
context.tmp_path_str = str(tmp)
|
|
|
|
|
|
@when('I call discover_devcontainers on that directory as "{rtype}"')
|
|
def step_discover_on_dir(context: Context, rtype: str) -> None:
|
|
context.discovery_result_list = discover_devcontainers(
|
|
resource_location=context.tmp_path_str,
|
|
resource_type=rtype,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared Then steps - unique names to avoid collisions with other step files
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('a discovery TypeError should be raised with message containing "{text}"')
|
|
def step_check_discovery_type_error(context: Context, text: str) -> None:
|
|
assert context.discovery_caught_error is not None, (
|
|
"Expected TypeError was not raised"
|
|
)
|
|
assert isinstance(context.discovery_caught_error, TypeError), (
|
|
f"Expected TypeError, got {type(context.discovery_caught_error).__name__}"
|
|
)
|
|
assert text in str(context.discovery_caught_error), (
|
|
f"Expected '{text}' in error message, got: {context.discovery_caught_error}"
|
|
)
|
|
|
|
|
|
@then('a discovery ValueError should be raised with message containing "{text}"')
|
|
def step_check_discovery_value_error(context: Context, text: str) -> None:
|
|
assert context.discovery_caught_error is not None, (
|
|
"Expected ValueError was not raised"
|
|
)
|
|
assert isinstance(context.discovery_caught_error, ValueError), (
|
|
f"Expected ValueError, got {type(context.discovery_caught_error).__name__}"
|
|
)
|
|
assert text in str(context.discovery_caught_error), (
|
|
f"Expected '{text}' in error message, got: {context.discovery_caught_error}"
|
|
)
|
|
|
|
|
|
@then("the discovery result list should be empty")
|
|
def step_check_empty_results(context: Context) -> None:
|
|
assert context.discovery_result_list == [], (
|
|
f"Expected empty list, got {context.discovery_result_list}"
|
|
)
|