forked from HAL9000/cleveragents-core
freemo
31472b5413
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate. ISSUES CLOSED: #1232
500 lines
19 KiB
Python
500 lines
19 KiB
Python
"""Step definitions for devcontainer_content_ops_coverage.feature.
|
|
|
|
Exercises uncovered lines 223-348 in devcontainer.py:
|
|
- read() (lines 223-238)
|
|
- write() (lines 254-280)
|
|
- discover_children() (lines 294-328)
|
|
- _exec_ls() (lines 333-348)
|
|
|
|
All subprocess.run calls are mocked via unittest.mock.patch.
|
|
Step prefix: dccov3 (devcontainer coverage round 3).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.domain.models.core.resource import PhysVirt, Resource
|
|
from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
|
|
|
|
# Valid 26-char Crockford Base32 ULID for test resources.
|
|
_TEST_ULID = "01JTESTDCC0V3RES0000000000"
|
|
|
|
|
|
def _make_resource(
|
|
location: str | None = "/ws/project",
|
|
resource_id: str = _TEST_ULID,
|
|
) -> Resource:
|
|
"""Create a minimal devcontainer-instance Resource for testing."""
|
|
return Resource(
|
|
resource_id=resource_id,
|
|
name="test-devcontainer",
|
|
resource_type_name="devcontainer-instance",
|
|
classification=PhysVirt.PHYSICAL,
|
|
description="Test devcontainer resource",
|
|
location=location,
|
|
)
|
|
|
|
|
|
# ── Given steps ──────────────────────────────────────────────
|
|
|
|
|
|
@given("dccov3 a devcontainer handler")
|
|
def step_dccov3_create_handler(context: Context) -> None:
|
|
context.dccov3_handler = DevcontainerHandler()
|
|
|
|
|
|
@given('dccov3 a devcontainer-instance resource with location "{loc}"')
|
|
def step_dccov3_resource_with_location(context: Context, loc: str) -> None:
|
|
context.dccov3_resource = _make_resource(location=loc)
|
|
|
|
|
|
@given("dccov3 a devcontainer-instance resource with no location")
|
|
def step_dccov3_resource_no_location(context: Context) -> None:
|
|
context.dccov3_resource = _make_resource(location=None)
|
|
|
|
|
|
@given('dccov3 subprocess returns successful cat output "{text}"')
|
|
def step_dccov3_subprocess_cat_success(context: Context, text: str) -> None:
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
mock_result.stdout = text.encode("utf-8")
|
|
mock_result.stderr = b""
|
|
context.dccov3_mock_result = mock_result
|
|
context.dccov3_mock_mode = "cat_success"
|
|
|
|
|
|
@given('dccov3 subprocess returns successful ls output "{text}"')
|
|
def step_dccov3_subprocess_ls_success(context: Context, text: str) -> None:
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
# For ls via read (bytes), and for ls via discover_children (text mode)
|
|
raw = text.replace("\\n", "\n")
|
|
mock_result.stdout = raw.encode("utf-8")
|
|
mock_result.stderr = b""
|
|
context.dccov3_mock_result = mock_result
|
|
context.dccov3_mock_mode = "ls_success"
|
|
|
|
|
|
@given('dccov3 subprocess returns failed cat output with stderr "{stderr}"')
|
|
def step_dccov3_subprocess_cat_failure(context: Context, stderr: str) -> None:
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 1
|
|
mock_result.stdout = b""
|
|
mock_result.stderr = stderr.encode("utf-8")
|
|
context.dccov3_mock_result = mock_result
|
|
context.dccov3_mock_mode = "cat_failure"
|
|
|
|
|
|
@given("dccov3 subprocess returns successful tee output")
|
|
def step_dccov3_subprocess_tee_success(context: Context) -> None:
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
mock_result.stdout = b""
|
|
mock_result.stderr = b""
|
|
context.dccov3_mock_result = mock_result
|
|
context.dccov3_mock_mode = "tee_success"
|
|
|
|
|
|
@given('dccov3 subprocess returns failed tee output with stderr "{stderr}"')
|
|
def step_dccov3_subprocess_tee_failure(context: Context, stderr: str) -> None:
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 1
|
|
mock_result.stdout = b""
|
|
mock_result.stderr = stderr.encode("utf-8")
|
|
context.dccov3_mock_result = mock_result
|
|
context.dccov3_mock_mode = "tee_failure"
|
|
|
|
|
|
@given('dccov3 subprocess returns successful ls listing "{text}"')
|
|
def step_dccov3_subprocess_ls_listing(context: Context, text: str) -> None:
|
|
"""Configure mock for discover_children which uses text=True."""
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
raw = text.replace("\\n", "\n")
|
|
mock_result.stdout = raw
|
|
mock_result.stderr = ""
|
|
context.dccov3_mock_result = mock_result
|
|
context.dccov3_mock_mode = "ls_listing"
|
|
|
|
|
|
@given("dccov3 subprocess returns failed ls output")
|
|
def step_dccov3_subprocess_ls_failure(context: Context) -> None:
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 1
|
|
mock_result.stdout = ""
|
|
mock_result.stderr = "ls: cannot access"
|
|
context.dccov3_mock_result = mock_result
|
|
context.dccov3_mock_mode = "ls_failure"
|
|
|
|
|
|
@given("dccov3 subprocess returns failed ls for exec_ls")
|
|
def step_dccov3_subprocess_ls_failure_for_exec_ls(context: Context) -> None:
|
|
"""Configure mock for _exec_ls failure path (bytes mode, not text mode)."""
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 1
|
|
mock_result.stdout = b""
|
|
mock_result.stderr = b"ls: error"
|
|
context.dccov3_mock_result = mock_result
|
|
context.dccov3_mock_mode = "exec_ls_failure"
|
|
|
|
|
|
# ── When steps ───────────────────────────────────────────────
|
|
|
|
|
|
def _do_read(context: Context, path: str) -> None:
|
|
"""Shared implementation for read steps."""
|
|
context.dccov3_error = None
|
|
context.dccov3_read_result = None
|
|
context.dccov3_subprocess_calls = []
|
|
|
|
if not hasattr(context, "dccov3_mock_result"):
|
|
# No subprocess mock configured — expect _require_location to fail
|
|
try:
|
|
context.dccov3_read_result = context.dccov3_handler.read(
|
|
resource=context.dccov3_resource,
|
|
path=path,
|
|
)
|
|
except (ValueError, FileNotFoundError, NotImplementedError) as exc:
|
|
context.dccov3_error = exc
|
|
return
|
|
|
|
def _capture_run(*args, **kwargs):
|
|
context.dccov3_subprocess_calls.append((args, kwargs))
|
|
return context.dccov3_mock_result
|
|
|
|
patcher = patch(
|
|
"cleveragents.resource.handlers.devcontainer.subprocess.run",
|
|
side_effect=_capture_run,
|
|
)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
|
|
try:
|
|
context.dccov3_read_result = context.dccov3_handler.read(
|
|
resource=context.dccov3_resource,
|
|
path=path,
|
|
)
|
|
except (ValueError, FileNotFoundError, NotImplementedError) as exc:
|
|
context.dccov3_error = exc
|
|
|
|
|
|
@when("dccov3 I read with an empty path from the resource")
|
|
def step_dccov3_read_empty_path(context: Context) -> None:
|
|
"""Dedicated step for reading with empty path (Behave can't match empty string)."""
|
|
_do_read(context, path="")
|
|
|
|
|
|
@when('dccov3 I read path "{path}" from the resource')
|
|
def step_dccov3_read_path(context: Context, path: str) -> None:
|
|
_do_read(context, path=path)
|
|
|
|
|
|
@when('dccov3 I write path "{path}" with data "{data}" to the resource')
|
|
def step_dccov3_write_path(context: Context, path: str, data: str) -> None:
|
|
context.dccov3_error = None
|
|
context.dccov3_write_result = None
|
|
context.dccov3_subprocess_calls = []
|
|
data_bytes = data.encode("utf-8")
|
|
|
|
if not hasattr(context, "dccov3_mock_result"):
|
|
try:
|
|
context.dccov3_write_result = context.dccov3_handler.write(
|
|
resource=context.dccov3_resource,
|
|
path=path,
|
|
data=data_bytes,
|
|
)
|
|
except (ValueError, NotImplementedError) as exc:
|
|
context.dccov3_error = exc
|
|
return
|
|
|
|
def _capture_run(*args, **kwargs):
|
|
context.dccov3_subprocess_calls.append((args, kwargs))
|
|
return context.dccov3_mock_result
|
|
|
|
patcher = patch(
|
|
"cleveragents.resource.handlers.devcontainer.subprocess.run",
|
|
side_effect=_capture_run,
|
|
)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
|
|
try:
|
|
context.dccov3_write_result = context.dccov3_handler.write(
|
|
resource=context.dccov3_resource,
|
|
path=path,
|
|
data=data_bytes,
|
|
)
|
|
except (ValueError, NotImplementedError) as exc:
|
|
context.dccov3_error = exc
|
|
|
|
|
|
@when('dccov3 I write path "{path}" with bytes to the resource')
|
|
def step_dccov3_write_bytes(context: Context, path: str) -> None:
|
|
"""Write binary data to verify subprocess receives bytes as input."""
|
|
context.dccov3_error = None
|
|
context.dccov3_write_result = None
|
|
context.dccov3_subprocess_calls = []
|
|
context.dccov3_write_bytes = b"\x00\x01\x02\xff"
|
|
|
|
def _capture_run(*args, **kwargs):
|
|
context.dccov3_subprocess_calls.append((args, kwargs))
|
|
return context.dccov3_mock_result
|
|
|
|
patcher = patch(
|
|
"cleveragents.resource.handlers.devcontainer.subprocess.run",
|
|
side_effect=_capture_run,
|
|
)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
|
|
try:
|
|
context.dccov3_write_result = context.dccov3_handler.write(
|
|
resource=context.dccov3_resource,
|
|
path=path,
|
|
data=context.dccov3_write_bytes,
|
|
)
|
|
except (ValueError, NotImplementedError) as exc:
|
|
context.dccov3_error = exc
|
|
|
|
|
|
@when("dccov3 I discover children of the resource")
|
|
def step_dccov3_discover_children(context: Context) -> None:
|
|
context.dccov3_error = None
|
|
context.dccov3_children = None
|
|
context.dccov3_subprocess_calls = []
|
|
|
|
if not hasattr(context, "dccov3_mock_result"):
|
|
try:
|
|
context.dccov3_children = context.dccov3_handler.discover_children(
|
|
resource=context.dccov3_resource,
|
|
)
|
|
except (ValueError, NotImplementedError) as exc:
|
|
context.dccov3_error = exc
|
|
return
|
|
|
|
def _capture_run(*args, **kwargs):
|
|
context.dccov3_subprocess_calls.append((args, kwargs))
|
|
return context.dccov3_mock_result
|
|
|
|
patcher = patch(
|
|
"cleveragents.resource.handlers.devcontainer.subprocess.run",
|
|
side_effect=_capture_run,
|
|
)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
|
|
try:
|
|
context.dccov3_children = context.dccov3_handler.discover_children(
|
|
resource=context.dccov3_resource,
|
|
)
|
|
except (ValueError, NotImplementedError) as exc:
|
|
context.dccov3_error = exc
|
|
|
|
|
|
# ── Then steps ───────────────────────────────────────────────
|
|
|
|
|
|
@then('dccov3 the read should succeed with data "{expected}"')
|
|
def step_dccov3_read_success(context: Context, expected: str) -> None:
|
|
assert context.dccov3_error is None, (
|
|
f"Expected no error, got {context.dccov3_error!r}"
|
|
)
|
|
assert context.dccov3_read_result is not None, "No read result returned"
|
|
raw_expected = expected.replace("\\n", "\n")
|
|
actual = context.dccov3_read_result.data.decode("utf-8")
|
|
assert actual == raw_expected, f"Expected data {raw_expected!r}, got {actual!r}"
|
|
|
|
|
|
@then("dccov3 the read should succeed with empty data")
|
|
def step_dccov3_read_success_empty(context: Context) -> None:
|
|
assert context.dccov3_error is None, (
|
|
f"Expected no error, got {context.dccov3_error!r}"
|
|
)
|
|
assert context.dccov3_read_result is not None, "No read result returned"
|
|
assert context.dccov3_read_result.data == b"", (
|
|
f"Expected empty data, got {context.dccov3_read_result.data!r}"
|
|
)
|
|
|
|
|
|
@then('dccov3 the read should raise FileNotFoundError containing "{text}"')
|
|
def step_dccov3_read_fnf_error(context: Context, text: str) -> None:
|
|
assert context.dccov3_error is not None, "Expected FileNotFoundError"
|
|
assert isinstance(context.dccov3_error, FileNotFoundError), (
|
|
f"Expected FileNotFoundError, got {type(context.dccov3_error).__name__}"
|
|
)
|
|
assert text in str(context.dccov3_error), (
|
|
f"Expected '{text}' in error message: {context.dccov3_error}"
|
|
)
|
|
|
|
|
|
@then('dccov3 the read should raise ValueError containing "{text}"')
|
|
def step_dccov3_read_value_error(context: Context, text: str) -> None:
|
|
assert context.dccov3_error is not None, "Expected ValueError"
|
|
assert isinstance(context.dccov3_error, ValueError), (
|
|
f"Expected ValueError, got {type(context.dccov3_error).__name__}"
|
|
)
|
|
assert text in str(context.dccov3_error), (
|
|
f"Expected '{text}' in error message: {context.dccov3_error}"
|
|
)
|
|
|
|
|
|
@then(
|
|
'dccov3 the subprocess should have been called with command containing "{cmd}" and "{arg}"'
|
|
)
|
|
def step_dccov3_check_subprocess_args(context: Context, cmd: str, arg: str) -> None:
|
|
assert len(context.dccov3_subprocess_calls) > 0, "No subprocess calls recorded"
|
|
# args[0] is the positional arg (the command list)
|
|
call_args, _call_kwargs = context.dccov3_subprocess_calls[0]
|
|
cmd_list = call_args[0] # First positional argument is the command list
|
|
assert cmd in cmd_list, f"Expected '{cmd}' in command {cmd_list}"
|
|
assert arg in cmd_list, f"Expected '{arg}' in command {cmd_list}"
|
|
|
|
|
|
@then("dccov3 the write should succeed with bytes_written {count:d}")
|
|
def step_dccov3_write_success(context: Context, count: int) -> None:
|
|
assert context.dccov3_error is None, (
|
|
f"Expected no error, got {context.dccov3_error!r}"
|
|
)
|
|
assert context.dccov3_write_result is not None, "No write result returned"
|
|
assert context.dccov3_write_result.success is True, (
|
|
f"Expected success, got failure: {context.dccov3_write_result.message}"
|
|
)
|
|
assert context.dccov3_write_result.bytes_written == count, (
|
|
f"Expected {count} bytes written, got {context.dccov3_write_result.bytes_written}"
|
|
)
|
|
|
|
|
|
@then('dccov3 the write message should contain "{text}"')
|
|
def step_dccov3_write_msg_contains(context: Context, text: str) -> None:
|
|
assert context.dccov3_write_result is not None
|
|
assert text in context.dccov3_write_result.message, (
|
|
f"Expected '{text}' in message: {context.dccov3_write_result.message}"
|
|
)
|
|
|
|
|
|
@then("dccov3 the write should fail")
|
|
def step_dccov3_write_failure(context: Context) -> None:
|
|
assert context.dccov3_error is None, (
|
|
f"Unexpected exception: {context.dccov3_error!r}"
|
|
)
|
|
assert context.dccov3_write_result is not None, "No write result returned"
|
|
assert context.dccov3_write_result.success is False, (
|
|
"Expected write to fail but it succeeded"
|
|
)
|
|
|
|
|
|
@then('dccov3 the write failure message should contain "{text}"')
|
|
def step_dccov3_write_fail_msg(context: Context, text: str) -> None:
|
|
assert context.dccov3_write_result is not None
|
|
assert context.dccov3_write_result.success is False
|
|
assert text in context.dccov3_write_result.message, (
|
|
f"Expected '{text}' in failure message: {context.dccov3_write_result.message}"
|
|
)
|
|
|
|
|
|
@then('dccov3 the write should raise ValueError containing "{text}"')
|
|
def step_dccov3_write_value_error(context: Context, text: str) -> None:
|
|
assert context.dccov3_error is not None, "Expected ValueError"
|
|
assert isinstance(context.dccov3_error, ValueError), (
|
|
f"Expected ValueError, got {type(context.dccov3_error).__name__}"
|
|
)
|
|
assert text in str(context.dccov3_error), (
|
|
f"Expected '{text}' in error message: {context.dccov3_error}"
|
|
)
|
|
|
|
|
|
@then("dccov3 the children list should have {count:d} entries")
|
|
def step_dccov3_children_count(context: Context, count: int) -> None:
|
|
assert context.dccov3_error is None, (
|
|
f"Expected no error, got {context.dccov3_error!r}"
|
|
)
|
|
assert context.dccov3_children is not None, "No children result"
|
|
assert len(context.dccov3_children) == count, (
|
|
f"Expected {count} children, got {len(context.dccov3_children)}"
|
|
)
|
|
|
|
|
|
@then('dccov3 the first child should have name "{name}"')
|
|
def step_dccov3_first_child_name(context: Context, name: str) -> None:
|
|
assert len(context.dccov3_children) >= 1, "No children found"
|
|
assert context.dccov3_children[0].name == name, (
|
|
f"Expected first child name '{name}', got '{context.dccov3_children[0].name}'"
|
|
)
|
|
|
|
|
|
@then('dccov3 the second child should have name "{name}"')
|
|
def step_dccov3_second_child_name(context: Context, name: str) -> None:
|
|
assert len(context.dccov3_children) >= 2, "Fewer than 2 children found"
|
|
assert context.dccov3_children[1].name == name, (
|
|
f"Expected second child name '{name}', got '{context.dccov3_children[1].name}'"
|
|
)
|
|
|
|
|
|
@then('dccov3 the third child should have name "{name}"')
|
|
def step_dccov3_third_child_name(context: Context, name: str) -> None:
|
|
assert len(context.dccov3_children) >= 3, "Fewer than 3 children found"
|
|
assert context.dccov3_children[2].name == name, (
|
|
f"Expected third child name '{name}', got '{context.dccov3_children[2].name}'"
|
|
)
|
|
|
|
|
|
@then('dccov3 each child should have type "{type_name}"')
|
|
def step_dccov3_children_type(context: Context, type_name: str) -> None:
|
|
for child in context.dccov3_children:
|
|
assert child.resource_type_name == type_name, (
|
|
f"Expected type '{type_name}', got '{child.resource_type_name}'"
|
|
)
|
|
|
|
|
|
@then('dccov3 each child location should start with "{prefix}"')
|
|
def step_dccov3_children_location_prefix(context: Context, prefix: str) -> None:
|
|
for child in context.dccov3_children:
|
|
assert child.location is not None, f"Child '{child.name}' has no location"
|
|
assert child.location.startswith(prefix), (
|
|
f"Expected location to start with '{prefix}', got '{child.location}'"
|
|
)
|
|
|
|
|
|
@then("dccov3 each child should reference the parent resource id")
|
|
def step_dccov3_children_parent_ref(context: Context) -> None:
|
|
parent_id = context.dccov3_resource.resource_id
|
|
for child in context.dccov3_children:
|
|
assert parent_id in child.parents, (
|
|
f"Expected parent {parent_id} in child.parents {child.parents}"
|
|
)
|
|
|
|
|
|
@then('dccov3 discover children should raise ValueError containing "{text}"')
|
|
def step_dccov3_discover_value_error(context: Context, text: str) -> None:
|
|
assert context.dccov3_error is not None, "Expected ValueError"
|
|
assert isinstance(context.dccov3_error, ValueError), (
|
|
f"Expected ValueError, got {type(context.dccov3_error).__name__}"
|
|
)
|
|
assert text in str(context.dccov3_error), (
|
|
f"Expected '{text}' in error message: {context.dccov3_error}"
|
|
)
|
|
|
|
|
|
@then("dccov3 the subprocess should have received input bytes")
|
|
def step_dccov3_subprocess_input_bytes(context: Context) -> None:
|
|
assert len(context.dccov3_subprocess_calls) > 0, "No subprocess calls recorded"
|
|
_call_args, call_kwargs = context.dccov3_subprocess_calls[0]
|
|
assert "input" in call_kwargs, "subprocess.run was not called with 'input' kwarg"
|
|
assert call_kwargs["input"] == context.dccov3_write_bytes, (
|
|
f"Expected input {context.dccov3_write_bytes!r}, got {call_kwargs['input']!r}"
|
|
)
|
|
|
|
|
|
@then("dccov3 each child description should mention the parent resource id")
|
|
def step_dccov3_children_description_parent(context: Context) -> None:
|
|
parent_id = context.dccov3_resource.resource_id
|
|
for child in context.dccov3_children:
|
|
assert child.description is not None, f"Child '{child.name}' has no description"
|
|
assert parent_id in child.description, (
|
|
f"Expected parent id '{parent_id}' in description '{child.description}'"
|
|
)
|