forked from cleveragents/cleveragents-core
freemo
9c4655fd5c
Implement server-side container and devcontainer lifecycle management for the CleverAgents server infrastructure. New modules in src/cleveragents/infrastructure/server/containers/: - DevcontainerSpec: Pydantic model for devcontainer.json parsing and validation (image, build, features, mounts, env vars, ports, hostRequirements, postCreateCommand, remoteUser, workspaceFolder). - ResourceLimits: CPU, memory, storage, and PID limits model with Docker CLI argument generation and hostRequirements conversion. - ContainerManager: Server-side lifecycle orchestration with create/start/stop/destroy operations using pluggable Docker CLI runner (mocked for tests). - HealthMonitor: Periodic container health probing via docker inspect with background thread management and probe history. - ContainerStatus enum: pending/created/running/stopped/destroyed/error with validated state transitions. Tests: - Behave BDD: 67 scenarios covering spec parsing, resource limits, lifecycle operations, health monitoring, and error handling (features/container_lifecycle.feature). - Robot Framework: 13 integration test cases with mock Docker runner (robot/container_lifecycle_server.robot). ISSUES CLOSED: #865
930 lines
31 KiB
Python
930 lines
31 KiB
Python
"""Step definitions for container lifecycle management (issue #865).
|
|
|
|
Covers: devcontainer.json parsing, resource limits, ContainerManager
|
|
lifecycle (create/start/stop/destroy), health monitoring, and status
|
|
transitions. All Docker interactions are mocked.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from dataclasses import dataclass, field
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.infrastructure.server.containers.container_manager import (
|
|
ContainerManager,
|
|
ContainerStatus,
|
|
ManagedContainer,
|
|
_validate_status_transition,
|
|
)
|
|
from cleveragents.infrastructure.server.containers.devcontainer_spec import (
|
|
DevcontainerSpec,
|
|
parse_devcontainer_json,
|
|
parse_devcontainer_json_string,
|
|
)
|
|
from cleveragents.infrastructure.server.containers.health_monitor import (
|
|
HealthMonitor,
|
|
HealthProbeResult,
|
|
)
|
|
from cleveragents.infrastructure.server.containers.resource_limits import (
|
|
ResourceLimits,
|
|
_parse_memory_string,
|
|
resource_limits_from_host_requirements,
|
|
)
|
|
|
|
# ── Mock runner ──────────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class MockContainerResult:
|
|
"""Minimal subprocess.CompletedProcess stand-in."""
|
|
|
|
returncode: int = 0
|
|
stdout: str = ""
|
|
stderr: str = ""
|
|
args: list[str] = field(default_factory=list)
|
|
|
|
|
|
class MockContainerRunner:
|
|
"""Configurable mock for subprocess.run in container operations."""
|
|
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[list[str], dict[str, object]]] = []
|
|
self.up_result: MockContainerResult = MockContainerResult()
|
|
self.start_result: MockContainerResult = MockContainerResult()
|
|
self.stop_result: MockContainerResult = MockContainerResult()
|
|
self.rm_result: MockContainerResult = MockContainerResult()
|
|
self.inspect_result: MockContainerResult = MockContainerResult()
|
|
|
|
def __call__(self, args: list[str], **kwargs: object) -> MockContainerResult:
|
|
self.calls.append((list(args), dict(kwargs)))
|
|
|
|
if len(args) >= 2 and args[0] == "devcontainer" and args[1] == "up":
|
|
return self.up_result
|
|
if len(args) >= 2 and args[0] == "docker" and args[1] == "start":
|
|
return self.start_result
|
|
if len(args) >= 2 and args[0] == "docker" and args[1] == "stop":
|
|
return self.stop_result
|
|
if len(args) >= 2 and args[0] == "docker" and args[1] == "rm":
|
|
return self.rm_result
|
|
if len(args) >= 2 and args[0] == "docker" and args[1] == "inspect":
|
|
return self.inspect_result
|
|
return MockContainerResult(returncode=0)
|
|
|
|
def set_up_success(
|
|
self,
|
|
container_id: str = "aabbccddee0011223344",
|
|
) -> None:
|
|
output = json.dumps(
|
|
{
|
|
"outcome": "success",
|
|
"containerId": container_id,
|
|
"remoteWorkspaceFolder": "/workspaces/project",
|
|
}
|
|
)
|
|
self.up_result = MockContainerResult(returncode=0, stdout=output)
|
|
|
|
def set_up_failure(self) -> None:
|
|
self.up_result = MockContainerResult(
|
|
returncode=1, stderr="container build failed"
|
|
)
|
|
|
|
def set_start_failure(self) -> None:
|
|
self.start_result = MockContainerResult(returncode=1, stderr="start failed")
|
|
|
|
def set_stop_failure(self) -> None:
|
|
self.stop_result = MockContainerResult(returncode=1, stderr="stop failed")
|
|
|
|
def set_inspect_healthy(self) -> None:
|
|
self.inspect_result = MockContainerResult(returncode=0, stdout="true")
|
|
|
|
def set_inspect_unhealthy(self) -> None:
|
|
self.inspect_result = MockContainerResult(returncode=0, stdout="false")
|
|
|
|
|
|
# ── DevcontainerSpec parsing steps ───────────────────────────
|
|
|
|
|
|
@given('a devcontainer.json string with image "{image}"')
|
|
def step_spec_string_with_image(context: Context, image: str) -> None:
|
|
context.spec_json = json.dumps({"image": image})
|
|
|
|
|
|
@given('a devcontainer.json string with build dockerfile "{dockerfile}"')
|
|
def step_spec_string_with_build(context: Context, dockerfile: str) -> None:
|
|
context.spec_json = json.dumps(
|
|
{"build": {"dockerfile": dockerfile, "context": "."}}
|
|
)
|
|
|
|
|
|
@given("a devcontainer.json string with features")
|
|
def step_spec_string_with_features(context: Context) -> None:
|
|
context.spec_json = json.dumps(
|
|
{"features": {"ghcr.io/devcontainers/features/python:1": {}}}
|
|
)
|
|
|
|
|
|
@given("a devcontainer.json string with mounts")
|
|
def step_spec_string_with_mounts(context: Context) -> None:
|
|
context.spec_json = json.dumps(
|
|
{
|
|
"mounts": [
|
|
{
|
|
"source": "/host/data",
|
|
"target": "/container/data",
|
|
"type": "bind",
|
|
}
|
|
]
|
|
}
|
|
)
|
|
|
|
|
|
@given("a devcontainer.json string with container environment variables")
|
|
def step_spec_string_with_env(context: Context) -> None:
|
|
context.spec_json = json.dumps({"containerEnv": {"MY_VAR": "hello"}})
|
|
|
|
|
|
@given("a devcontainer.json string with forwarded ports {port1:d} and {port2:d}")
|
|
def step_spec_string_with_ports(context: Context, port1: int, port2: int) -> None:
|
|
context.spec_json = json.dumps({"forwardPorts": [port1, port2]})
|
|
|
|
|
|
@given(
|
|
'a devcontainer.json string with host requirements cpus {cpus:d} and memory "{memory}"'
|
|
)
|
|
def step_spec_string_with_host_reqs(context: Context, cpus: int, memory: str) -> None:
|
|
context.spec_json = json.dumps(
|
|
{"hostRequirements": {"cpus": cpus, "memory": memory}}
|
|
)
|
|
|
|
|
|
@given('a temporary devcontainer.json file with image "{image}"')
|
|
def step_spec_file_with_image(context: Context, image: str) -> None:
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w",
|
|
suffix=".json",
|
|
delete=False,
|
|
prefix="devcontainer_",
|
|
) as tmp:
|
|
json.dump({"image": image}, tmp)
|
|
context.spec_file_path = tmp.name
|
|
|
|
|
|
@given('a devcontainer.json string with string mount "{mount_str}"')
|
|
def step_spec_string_with_string_mount(context: Context, mount_str: str) -> None:
|
|
context.spec_json = json.dumps({"mounts": [mount_str]})
|
|
|
|
|
|
@given('a devcontainer.json string with postCreateCommand "{cmd}"')
|
|
def step_spec_string_with_post_create(context: Context, cmd: str) -> None:
|
|
context.spec_json = json.dumps({"postCreateCommand": cmd})
|
|
|
|
|
|
@given('a devcontainer.json string with remoteUser "{user}"')
|
|
def step_spec_string_with_remote_user(context: Context, user: str) -> None:
|
|
context.spec_json = json.dumps({"remoteUser": user})
|
|
|
|
|
|
@given('a devcontainer.json string with workspaceFolder "{folder}"')
|
|
def step_spec_string_with_workspace(context: Context, folder: str) -> None:
|
|
context.spec_json = json.dumps({"workspaceFolder": folder})
|
|
|
|
|
|
@when("I parse the devcontainer spec string")
|
|
def step_parse_spec_string(context: Context) -> None:
|
|
context.spec = parse_devcontainer_json_string(context.spec_json)
|
|
|
|
|
|
@when("I parse the devcontainer spec file")
|
|
def step_parse_spec_file(context: Context) -> None:
|
|
context.spec = parse_devcontainer_json(context.spec_file_path)
|
|
os.unlink(context.spec_file_path)
|
|
|
|
|
|
@when("I attempt to parse devcontainer spec from empty path")
|
|
def step_parse_empty_path(context: Context) -> None:
|
|
try:
|
|
parse_devcontainer_json("")
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I attempt to parse devcontainer spec from relative path "{path}"')
|
|
def step_parse_relative_path(context: Context, path: str) -> None:
|
|
try:
|
|
parse_devcontainer_json(path)
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I attempt to parse devcontainer spec from missing file "{path}"')
|
|
def step_parse_missing_file(context: Context, path: str) -> None:
|
|
try:
|
|
parse_devcontainer_json(path)
|
|
context.raised = None
|
|
except FileNotFoundError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when("I attempt to parse devcontainer spec from empty string")
|
|
def step_parse_empty_string(context: Context) -> None:
|
|
try:
|
|
parse_devcontainer_json_string("")
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@then('the spec image should be "{expected}"')
|
|
def step_spec_image(context: Context, expected: str) -> None:
|
|
assert context.spec.image == expected
|
|
|
|
|
|
@then("the spec should have image set")
|
|
def step_spec_has_image(context: Context) -> None:
|
|
assert context.spec.has_image()
|
|
|
|
|
|
@then("the spec should not have image set")
|
|
def step_spec_no_image(context: Context) -> None:
|
|
assert not context.spec.has_image()
|
|
|
|
|
|
@then("the spec should have build set")
|
|
def step_spec_has_build(context: Context) -> None:
|
|
assert context.spec.has_build()
|
|
|
|
|
|
@then("the spec should not have build set")
|
|
def step_spec_no_build(context: Context) -> None:
|
|
assert not context.spec.has_build()
|
|
|
|
|
|
@then('the build dockerfile should be "{expected}"')
|
|
def step_build_dockerfile(context: Context, expected: str) -> None:
|
|
assert context.spec.build is not None
|
|
assert context.spec.build.dockerfile == expected
|
|
|
|
|
|
@then("the spec should have features")
|
|
def step_spec_has_features(context: Context) -> None:
|
|
assert len(context.spec.features) > 0
|
|
|
|
|
|
@then('the feature "{feature}" should be present')
|
|
def step_feature_present(context: Context, feature: str) -> None:
|
|
assert feature in context.spec.features
|
|
|
|
|
|
@then("the spec should have {count:d} mount")
|
|
def step_spec_mount_count(context: Context, count: int) -> None:
|
|
assert len(context.spec.mounts) == count
|
|
|
|
|
|
@then('mount {index:d} source should be "{expected}"')
|
|
def step_mount_source(context: Context, index: int, expected: str) -> None:
|
|
assert context.spec.mounts[index].source == expected
|
|
|
|
|
|
@then('mount {index:d} target should be "{expected}"')
|
|
def step_mount_target(context: Context, index: int, expected: str) -> None:
|
|
assert context.spec.mounts[index].target == expected
|
|
|
|
|
|
@then('the container env should contain "{key}" with value "{value}"')
|
|
def step_container_env(context: Context, key: str, value: str) -> None:
|
|
assert context.spec.container_env.get(key) == value
|
|
|
|
|
|
@then("the forwarded ports should contain {port:d}")
|
|
def step_forwarded_port(context: Context, port: int) -> None:
|
|
assert port in context.spec.forwarded_ports
|
|
|
|
|
|
@then("the resource limits cpu should be {cpu:g}")
|
|
def step_resource_limits_cpu(context: Context, cpu: float) -> None:
|
|
assert context.spec.resource_limits.cpu_limit == cpu
|
|
|
|
|
|
@then("the resource limits memory should be {memory:d}")
|
|
def step_resource_limits_memory(context: Context, memory: int) -> None:
|
|
assert context.spec.resource_limits.memory_limit_mb == memory
|
|
|
|
|
|
@then('the post create command should be "{expected}"')
|
|
def step_post_create_command(context: Context, expected: str) -> None:
|
|
assert context.spec.post_create_command == expected
|
|
|
|
|
|
@then('the remote user should be "{expected}"')
|
|
def step_remote_user(context: Context, expected: str) -> None:
|
|
assert context.spec.remote_user == expected
|
|
|
|
|
|
@then('the workspace folder should be "{expected}"')
|
|
def step_workspace_folder(context: Context, expected: str) -> None:
|
|
assert context.spec.workspace_folder == expected
|
|
|
|
|
|
# ── ResourceLimits steps ─────────────────────────────────────
|
|
|
|
|
|
@when("I create default resource limits")
|
|
def step_default_limits(context: Context) -> None:
|
|
context.limits = ResourceLimits()
|
|
|
|
|
|
@when("I create resource limits with cpu {cpu:g} and memory {memory:d}")
|
|
def step_custom_limits(context: Context, cpu: float, memory: int) -> None:
|
|
context.limits = ResourceLimits(cpu_limit=cpu, memory_limit_mb=memory)
|
|
|
|
|
|
@given('host requirements with cpus {cpus:d} and memory "{memory}"')
|
|
def step_host_requirements(context: Context, cpus: int, memory: str) -> None:
|
|
context.host_reqs = {"cpus": cpus, "memory": memory}
|
|
|
|
|
|
@when("I create resource limits from host requirements")
|
|
def step_limits_from_host_reqs(context: Context) -> None:
|
|
context.limits = resource_limits_from_host_requirements(context.host_reqs)
|
|
|
|
|
|
@then("the cpu limit should be {cpu:g}")
|
|
def step_check_cpu(context: Context, cpu: float) -> None:
|
|
assert context.limits.cpu_limit == cpu
|
|
|
|
|
|
@then("the memory limit should be {memory:d}")
|
|
def step_check_memory(context: Context, memory: int) -> None:
|
|
assert context.limits.memory_limit_mb == memory
|
|
|
|
|
|
@then('the docker args should contain "{arg}"')
|
|
def step_docker_args_contain(context: Context, arg: str) -> None:
|
|
args = context.limits.to_docker_args()
|
|
assert arg in args, f"Expected {arg!r} in {args}"
|
|
|
|
|
|
@when("I attempt to create resource limits with cpu {cpu:g}")
|
|
def step_invalid_cpu(context: Context, cpu: float) -> None:
|
|
try:
|
|
ResourceLimits(cpu_limit=cpu)
|
|
context.raised = None
|
|
except ValidationError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when("I attempt to create resource limits with memory {memory:d}")
|
|
def step_invalid_memory(context: Context, memory: int) -> None:
|
|
try:
|
|
ResourceLimits(memory_limit_mb=memory)
|
|
context.raised = None
|
|
except ValidationError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@then("a container validation error should be raised")
|
|
def step_validation_error(context: Context) -> None:
|
|
assert context.raised is not None
|
|
assert isinstance(context.raised, (ValidationError, ValueError))
|
|
|
|
|
|
@when('I parse memory string "{value}"')
|
|
def step_parse_memory(context: Context, value: str) -> None:
|
|
context.parsed_memory = _parse_memory_string(value)
|
|
|
|
|
|
@then("the parsed memory should be {expected:d}")
|
|
def step_check_parsed_memory(context: Context, expected: int) -> None:
|
|
assert context.parsed_memory == expected
|
|
|
|
|
|
@when("I attempt to parse empty memory string")
|
|
def step_parse_empty_memory(context: Context) -> None:
|
|
try:
|
|
_parse_memory_string("")
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when("I attempt to create resource limits from non-dict host requirements")
|
|
def step_limits_from_non_dict(context: Context) -> None:
|
|
try:
|
|
resource_limits_from_host_requirements("not a dict") # type: ignore[arg-type]
|
|
context.raised = None
|
|
except TypeError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when("I create resource limits with pids limit {pids:d}")
|
|
def step_limits_with_pids(context: Context, pids: int) -> None:
|
|
context.limits = ResourceLimits(pids_limit=pids)
|
|
|
|
|
|
@when('I create resource limits from string values cpu "{cpu}" and memory "{memory}"')
|
|
def step_limits_from_strings(context: Context, cpu: str, memory: str) -> None:
|
|
context.limits = ResourceLimits(**{"cpu_limit": cpu, "memory_limit_mb": memory})
|
|
|
|
|
|
# ── Common error assertion steps ─────────────────────────────
|
|
|
|
|
|
@then('a container ValueError should be raised with "{message}"')
|
|
def step_value_error_with_msg(context: Context, message: str) -> None:
|
|
assert context.raised is not None, "Expected ValueError but none raised"
|
|
assert isinstance(context.raised, ValueError), (
|
|
f"Expected ValueError, got {type(context.raised).__name__}"
|
|
)
|
|
assert message in str(context.raised), f"Expected '{message}' in '{context.raised}'"
|
|
|
|
|
|
@then("a container FileNotFoundError should be raised")
|
|
def step_file_not_found(context: Context) -> None:
|
|
assert context.raised is not None
|
|
assert isinstance(context.raised, FileNotFoundError)
|
|
|
|
|
|
@then("a container TypeError should be raised")
|
|
def step_type_error(context: Context) -> None:
|
|
assert context.raised is not None
|
|
assert isinstance(context.raised, TypeError)
|
|
|
|
|
|
@then("a container RuntimeError should be raised")
|
|
def step_runtime_error(context: Context) -> None:
|
|
assert context.raised is not None
|
|
assert isinstance(context.raised, RuntimeError)
|
|
|
|
|
|
# ── ContainerManager steps ───────────────────────────────────
|
|
|
|
|
|
@given("a mock container command runner")
|
|
def step_mock_container_runner(context: Context) -> None:
|
|
context.container_runner = MockContainerRunner()
|
|
context.container_manager = ContainerManager(run_command=context.container_runner)
|
|
|
|
|
|
@given("the container runner configured for successful creation")
|
|
def step_runner_success(context: Context) -> None:
|
|
context.container_runner.set_up_success()
|
|
|
|
|
|
@given("the container runner configured for failed creation")
|
|
def step_runner_fail_create(context: Context) -> None:
|
|
context.container_runner.set_up_failure()
|
|
|
|
|
|
@given("the container runner configured for failed start")
|
|
def step_runner_fail_start(context: Context) -> None:
|
|
context.container_runner.set_start_failure()
|
|
|
|
|
|
@given("the container runner configured for failed stop")
|
|
def step_runner_fail_stop(context: Context) -> None:
|
|
context.container_runner.set_stop_failure()
|
|
|
|
|
|
@given('a created container "{name}" with container ID "{ctr_id}"')
|
|
def step_created_container(context: Context, name: str, ctr_id: str) -> None:
|
|
mgr = context.container_manager
|
|
container = ManagedContainer(
|
|
name=name,
|
|
status=ContainerStatus.CREATED,
|
|
container_id=ctr_id,
|
|
)
|
|
mgr._containers[name] = container
|
|
|
|
|
|
@given('a running container "{name}" with container ID "{ctr_id}"')
|
|
def step_running_container(context: Context, name: str, ctr_id: str) -> None:
|
|
mgr = context.container_manager
|
|
container = ManagedContainer(
|
|
name=name,
|
|
status=ContainerStatus.RUNNING,
|
|
container_id=ctr_id,
|
|
)
|
|
mgr._containers[name] = container
|
|
|
|
|
|
@given('a stopped managed container "{name}" with container ID "{ctr_id}"')
|
|
def step_stopped_container(context: Context, name: str, ctr_id: str) -> None:
|
|
mgr = context.container_manager
|
|
container = ManagedContainer(
|
|
name=name,
|
|
status=ContainerStatus.STOPPED,
|
|
container_id=ctr_id,
|
|
)
|
|
mgr._containers[name] = container
|
|
|
|
|
|
@given('an errored managed container "{name}" with container ID "{ctr_id}"')
|
|
def step_errored_managed_container(context: Context, name: str, ctr_id: str) -> None:
|
|
mgr = context.container_manager
|
|
container = ManagedContainer(
|
|
name=name,
|
|
status=ContainerStatus.ERROR,
|
|
container_id=ctr_id,
|
|
)
|
|
mgr._containers[name] = container
|
|
|
|
|
|
@given('an existing container "{name}" in the manager')
|
|
def step_existing_container(context: Context, name: str) -> None:
|
|
mgr = context.container_manager
|
|
container = ManagedContainer(
|
|
name=name,
|
|
status=ContainerStatus.PENDING,
|
|
)
|
|
mgr._containers[name] = container
|
|
|
|
|
|
@given("a devcontainer spec with cpu limit {cpu:g} and memory {memory:d}")
|
|
def step_spec_with_limits(context: Context, cpu: float, memory: int) -> None:
|
|
limits = ResourceLimits(cpu_limit=cpu, memory_limit_mb=memory)
|
|
context.test_spec = DevcontainerSpec(resource_limits=limits)
|
|
|
|
|
|
@when('I create container "{name}" with workspace "{workspace}"')
|
|
def step_create_container(context: Context, name: str, workspace: str) -> None:
|
|
context.container_result = context.container_manager.create(
|
|
name, workspace_folder=workspace
|
|
)
|
|
|
|
|
|
@when('I start container "{name}"')
|
|
def step_start_container(context: Context, name: str) -> None:
|
|
context.container_result = context.container_manager.start(name)
|
|
|
|
|
|
@when('I stop managed container "{name}"')
|
|
def step_stop_container(context: Context, name: str) -> None:
|
|
context.container_result = context.container_manager.stop(name)
|
|
|
|
|
|
@when('I destroy container "{name}"')
|
|
def step_destroy_container(context: Context, name: str) -> None:
|
|
context.container_result = context.container_manager.destroy(name)
|
|
|
|
|
|
@when("I attempt to create container with empty name")
|
|
def step_create_empty_name(context: Context) -> None:
|
|
try:
|
|
context.container_manager.create("")
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I attempt to create duplicate container "{name}"')
|
|
def step_attempt_create_container(context: Context, name: str) -> None:
|
|
try:
|
|
context.container_manager.create(name)
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I attempt to create container "{name}" with workspace "{workspace}"')
|
|
def step_attempt_create_with_ws(context: Context, name: str, workspace: str) -> None:
|
|
try:
|
|
context.container_manager.create(name, workspace_folder=workspace)
|
|
context.raised = None
|
|
except RuntimeError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I attempt to start non-existent container "{name}"')
|
|
def step_start_nonexistent(context: Context, name: str) -> None:
|
|
try:
|
|
context.container_manager.start(name)
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I attempt to stop managed container "{name}"')
|
|
def step_attempt_stop(context: Context, name: str) -> None:
|
|
try:
|
|
context.container_manager.stop(name)
|
|
context.raised = None
|
|
except (ValueError, RuntimeError) as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I attempt to destroy container "{name}"')
|
|
def step_attempt_destroy(context: Context, name: str) -> None:
|
|
try:
|
|
context.container_manager.destroy(name)
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I attempt to start container "{name}"')
|
|
def step_attempt_start(context: Context, name: str) -> None:
|
|
try:
|
|
context.container_manager.start(name)
|
|
context.raised = None
|
|
except (ValueError, RuntimeError) as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I list containers with status "{status}"')
|
|
def step_list_by_status(context: Context, status: str) -> None:
|
|
context.container_list = context.container_manager.list_containers(
|
|
status=ContainerStatus(status)
|
|
)
|
|
|
|
|
|
@when("I list all containers")
|
|
def step_list_all(context: Context) -> None:
|
|
context.container_list = context.container_manager.list_containers()
|
|
|
|
|
|
@when('I get container "{name}"')
|
|
def step_get_container(context: Context, name: str) -> None:
|
|
context.get_result = context.container_manager.get_container(name)
|
|
|
|
|
|
@when("I attempt to get container with empty name")
|
|
def step_get_empty_name(context: Context) -> None:
|
|
try:
|
|
context.container_manager.get_container("")
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I create container "{name}" with spec and limits')
|
|
def step_create_with_spec(context: Context, name: str) -> None:
|
|
context.container_result = context.container_manager.create(
|
|
name,
|
|
spec=context.test_spec,
|
|
workspace_folder="/workspace",
|
|
)
|
|
|
|
|
|
@then('the container "{name}" should have status "{status}"')
|
|
def step_container_status(context: Context, name: str, status: str) -> None:
|
|
container = context.container_manager.get_container(name)
|
|
assert container is not None, f"Container '{name}' not found"
|
|
assert container.status.value == status, (
|
|
f"Expected status '{status}', got '{container.status.value}'"
|
|
)
|
|
|
|
|
|
@then('the container "{name}" should have a container ID')
|
|
def step_container_has_id(context: Context, name: str) -> None:
|
|
container = context.container_manager.get_container(name)
|
|
assert container is not None
|
|
assert container.container_id != ""
|
|
|
|
|
|
@then("the list should contain {count:d} container")
|
|
def step_list_count_singular(context: Context, count: int) -> None:
|
|
assert len(context.container_list) == count
|
|
|
|
|
|
@then("the list should contain {count:d} containers")
|
|
def step_list_count_plural(context: Context, count: int) -> None:
|
|
assert len(context.container_list) == count
|
|
|
|
|
|
@then('the list should contain container "{name}"')
|
|
def step_list_contains(context: Context, name: str) -> None:
|
|
names = [c.name for c in context.container_list]
|
|
assert name in names, f"Expected '{name}' in {names}"
|
|
|
|
|
|
@then("the container result should be None")
|
|
def step_result_none(context: Context) -> None:
|
|
assert context.get_result is None
|
|
|
|
|
|
@then('the container "{name}" resource limits cpu should be {cpu:g}')
|
|
def step_container_limits_cpu(context: Context, name: str, cpu: float) -> None:
|
|
container = context.container_manager.get_container(name)
|
|
assert container is not None
|
|
assert container.resource_limits.cpu_limit == cpu
|
|
|
|
|
|
@then("the ContainerStatus enum should have {count:d} values")
|
|
def step_status_enum_count(context: Context, count: int) -> None:
|
|
assert len(ContainerStatus) == count
|
|
|
|
|
|
@then('ContainerStatus should include "{value}"')
|
|
def step_status_includes(context: Context, value: str) -> None:
|
|
values = {s.value for s in ContainerStatus}
|
|
assert value in values, f"Expected '{value}' in {values}"
|
|
|
|
|
|
@then('container transition from "{source}" to "{target}" should be valid')
|
|
def step_valid_transition(context: Context, source: str, target: str) -> None:
|
|
assert _validate_status_transition(ContainerStatus(source), ContainerStatus(target))
|
|
|
|
|
|
@then('container transition from "{source}" to "{target}" should be invalid')
|
|
def step_invalid_transition(context: Context, source: str, target: str) -> None:
|
|
assert not _validate_status_transition(
|
|
ContainerStatus(source), ContainerStatus(target)
|
|
)
|
|
|
|
|
|
# ── HealthMonitor steps ──────────────────────────────────────
|
|
|
|
|
|
@given("a health monitor with interval {interval:g}")
|
|
def step_health_monitor(context: Context, interval: float) -> None:
|
|
context.health_monitor = HealthMonitor(
|
|
context.container_manager,
|
|
interval=interval,
|
|
run_command=context.container_runner,
|
|
)
|
|
|
|
|
|
@given('a running monitored container "{name}" with container ID "{ctr_id}"')
|
|
def step_running_monitored(context: Context, name: str, ctr_id: str) -> None:
|
|
mgr = context.container_manager
|
|
container = ManagedContainer(
|
|
name=name,
|
|
status=ContainerStatus.RUNNING,
|
|
container_id=ctr_id,
|
|
)
|
|
mgr._containers[name] = container
|
|
|
|
|
|
@given('a running monitored container "{name}" without container ID')
|
|
def step_running_no_id(context: Context, name: str) -> None:
|
|
mgr = context.container_manager
|
|
container = ManagedContainer(
|
|
name=name,
|
|
status=ContainerStatus.RUNNING,
|
|
container_id="",
|
|
)
|
|
mgr._containers[name] = container
|
|
|
|
|
|
@given('a stopped monitored container "{name}"')
|
|
def step_stopped_monitored(context: Context, name: str) -> None:
|
|
mgr = context.container_manager
|
|
container = ManagedContainer(
|
|
name=name,
|
|
status=ContainerStatus.STOPPED,
|
|
)
|
|
mgr._containers[name] = container
|
|
|
|
|
|
@given("the container runner configured for healthy probe")
|
|
def step_runner_healthy_probe(context: Context) -> None:
|
|
context.container_runner.set_inspect_healthy()
|
|
|
|
|
|
@given("the container runner configured for unhealthy probe")
|
|
def step_runner_unhealthy_probe(context: Context) -> None:
|
|
context.container_runner.set_inspect_unhealthy()
|
|
|
|
|
|
@given('I start monitoring "{name}"')
|
|
def step_start_monitoring_given(context: Context, name: str) -> None:
|
|
context.health_monitor.start_monitoring(name)
|
|
|
|
|
|
@when('I probe container "{name}"')
|
|
def step_probe_container(context: Context, name: str) -> None:
|
|
context.probe_result = context.health_monitor.probe(name)
|
|
|
|
|
|
@when("I attempt to probe container with empty name")
|
|
def step_probe_empty_name(context: Context) -> None:
|
|
try:
|
|
context.health_monitor.probe("")
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I attempt to start monitoring "{name}"')
|
|
def step_attempt_start_monitoring(context: Context, name: str) -> None:
|
|
try:
|
|
context.health_monitor.start_monitoring(name)
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when("I attempt to create health monitor with interval {interval:g}")
|
|
def step_attempt_invalid_interval(context: Context, interval: float) -> None:
|
|
try:
|
|
HealthMonitor(
|
|
context.container_manager,
|
|
interval=interval,
|
|
run_command=context.container_runner,
|
|
)
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when("I attempt to create health monitor with non-manager")
|
|
def step_attempt_non_manager(context: Context) -> None:
|
|
try:
|
|
HealthMonitor("not a manager") # type: ignore[arg-type]
|
|
context.raised = None
|
|
except TypeError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when('I stop monitoring "{name}"')
|
|
def step_stop_monitoring(context: Context, name: str) -> None:
|
|
context.health_monitor.stop_monitoring(name)
|
|
|
|
|
|
@when("I stop all monitoring")
|
|
def step_stop_all(context: Context) -> None:
|
|
context.health_monitor.stop_all()
|
|
|
|
|
|
@then("the probe result should be healthy")
|
|
def step_probe_healthy(context: Context) -> None:
|
|
assert context.probe_result.healthy
|
|
|
|
|
|
@then("the probe result should be unhealthy")
|
|
def step_probe_unhealthy(context: Context) -> None:
|
|
assert not context.probe_result.healthy
|
|
|
|
|
|
@then('the probe message should contain "{message}"')
|
|
def step_probe_message(context: Context, message: str) -> None:
|
|
assert message in context.probe_result.message
|
|
|
|
|
|
@then("no container error should be raised")
|
|
def step_no_error(context: Context) -> None:
|
|
pass # If we got here, no error was raised
|
|
|
|
|
|
@then("no containers should be monitored")
|
|
def step_no_monitored(context: Context) -> None:
|
|
assert len(context.health_monitor.monitored_containers) == 0
|
|
|
|
|
|
@then("the probe history should have {count:d} entry")
|
|
def step_probe_history(context: Context, count: int) -> None:
|
|
assert len(context.health_monitor.probe_history) == count
|
|
|
|
|
|
@given('a health probe result for "{name}" that is healthy')
|
|
def step_health_probe_result(context: Context, name: str) -> None:
|
|
context.probe_result = HealthProbeResult(
|
|
container_name=name, healthy=True, message="ok"
|
|
)
|
|
|
|
|
|
@then('the probe repr should contain "{text}"')
|
|
def step_probe_repr(context: Context, text: str) -> None:
|
|
assert text in repr(context.probe_result)
|
|
|
|
|
|
@when("I attempt to create health probe with empty name")
|
|
def step_probe_empty_container_name(context: Context) -> None:
|
|
try:
|
|
HealthProbeResult(container_name="", healthy=True)
|
|
context.raised = None
|
|
except ValueError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when("I attempt to validate transition with non-enum current")
|
|
def step_invalid_transition_current(context: Context) -> None:
|
|
try:
|
|
_validate_status_transition(
|
|
"not_an_enum",
|
|
ContainerStatus.RUNNING, # type: ignore[arg-type]
|
|
)
|
|
context.raised = None
|
|
except TypeError as exc:
|
|
context.raised = exc
|
|
|
|
|
|
@when("I attempt to validate transition with non-enum target")
|
|
def step_invalid_transition_target(context: Context) -> None:
|
|
try:
|
|
_validate_status_transition(
|
|
ContainerStatus.PENDING,
|
|
"not_an_enum", # type: ignore[arg-type]
|
|
)
|
|
context.raised = None
|
|
except TypeError as exc:
|
|
context.raised = exc
|