forked from HAL9000/cleveragents-core
freemo
3556481681
Implements real AWS SDK integration for CloudResourceHandler using boto3 as an optional dependency. Key changes: - Add boto3/botocore as optional [aws] dependency in pyproject.toml - Implement CloudResourceHandler.resolve() for AWS: builds boto3 session, verifies credentials via STS get_caller_identity for account-level types, and returns a BoundResource with the resource ARN as sandbox_path - Implement discover_aws_resources() to enumerate VPCs, subnets, instances, S3 buckets, IAM roles, RDS instances, ECS clusters, Lambda functions, and EKS clusters via the AWS API - Implement CloudResourceHandler.discover_children() for AWS resource types using the new discovery function - Implement CloudSandboxStrategy.create/commit/rollback for AWS using a tag-based isolation strategy (CleverAgents:PlanId tag) - GCP and Azure providers still raise NotImplementedError (pending) - boto3 is optional: handler raises ImportError with helpful install message when boto3 is not installed - Credentials are never logged (existing redaction infrastructure preserved) - Update cloud_resources.feature to reflect new AWS behavior - Add comprehensive cloud_aws_sdk.feature with 47 BDD scenarios covering all new code paths with mocked boto3 Closes #1021
756 lines
31 KiB
Python
756 lines
31 KiB
Python
"""Step definitions for cloud_aws_sdk.feature.
|
|
|
|
Tests AWS SDK integration for CloudResourceHandler using mocked boto3.
|
|
All steps use the "awssdk" prefix to avoid conflicts with existing
|
|
step definitions in other feature files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when # type: ignore[attr-defined]
|
|
|
|
from cleveragents.domain.models.core.resource import (
|
|
PhysVirt,
|
|
Resource,
|
|
ResourceCapabilities,
|
|
)
|
|
from cleveragents.resource.handlers.cloud import (
|
|
CloudResourceHandler,
|
|
CloudSandboxStrategy,
|
|
_AWS_RESOURCE_MAP,
|
|
_BOTO3_AVAILABLE,
|
|
_build_aws_session,
|
|
discover_aws_resources,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_CLOUD_ENV_VARS = [
|
|
"AWS_ACCESS_KEY_ID",
|
|
"AWS_SECRET_ACCESS_KEY",
|
|
"AWS_SESSION_TOKEN",
|
|
"AWS_REGION",
|
|
"AWS_PROFILE",
|
|
"GOOGLE_APPLICATION_CREDENTIALS",
|
|
"GCLOUD_PROJECT",
|
|
"GCP_REGION",
|
|
"AZURE_SUBSCRIPTION_ID",
|
|
"AZURE_TENANT_ID",
|
|
"AZURE_CLIENT_ID",
|
|
"AZURE_CLIENT_SECRET",
|
|
"AZURE_REGION",
|
|
]
|
|
|
|
|
|
def _clear_cloud_env() -> dict[str, str]:
|
|
saved: dict[str, str] = {}
|
|
for var in _CLOUD_ENV_VARS:
|
|
val = os.environ.pop(var, None)
|
|
if val is not None:
|
|
saved[var] = val
|
|
return saved
|
|
|
|
|
|
def _restore_cloud_env(saved: dict[str, str]) -> None:
|
|
for var in _CLOUD_ENV_VARS:
|
|
if var in saved:
|
|
os.environ[var] = saved[var]
|
|
else:
|
|
os.environ.pop(var, None)
|
|
|
|
|
|
def _make_resource(
|
|
type_name: str,
|
|
properties: dict[str, Any] | None = None,
|
|
location: str | None = None,
|
|
) -> Resource:
|
|
return Resource(
|
|
resource_id="01KJ5C5TPMP8GGX3QC83E2MAQS",
|
|
resource_type_name=type_name,
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=location,
|
|
properties=properties or {},
|
|
capabilities=ResourceCapabilities(
|
|
readable=True,
|
|
writable=False,
|
|
sandboxable=False,
|
|
checkpointable=False,
|
|
),
|
|
)
|
|
|
|
|
|
def _make_mock_session(
|
|
account_id: str = "123456789012",
|
|
sts_error: Exception | None = None,
|
|
ec2_vpcs: list[dict[str, Any]] | None = None,
|
|
s3_buckets: list[dict[str, Any]] | None = None,
|
|
ecs_clusters: list[str] | None = None,
|
|
client_error: Exception | None = None,
|
|
) -> MagicMock:
|
|
"""Build a mock boto3 Session with configurable responses."""
|
|
session = MagicMock()
|
|
|
|
def _make_client(service: str, **kwargs: Any) -> MagicMock: # noqa: ARG001
|
|
client = MagicMock()
|
|
|
|
if service == "sts":
|
|
if sts_error is not None:
|
|
client.get_caller_identity.side_effect = sts_error
|
|
else:
|
|
client.get_caller_identity.return_value = {
|
|
"Account": account_id,
|
|
"UserId": "AIDAIOSFODNN7EXAMPLE",
|
|
"Arn": f"arn:aws:iam::{account_id}:user/test",
|
|
}
|
|
|
|
elif service == "ec2":
|
|
if client_error is not None:
|
|
client.describe_vpcs.side_effect = client_error
|
|
client.describe_subnets.side_effect = client_error
|
|
client.describe_instances.side_effect = client_error
|
|
client.describe_security_groups.side_effect = client_error
|
|
else:
|
|
vpcs = ec2_vpcs or []
|
|
client.describe_vpcs.return_value = {"Vpcs": vpcs}
|
|
client.describe_subnets.return_value = {"Subnets": []}
|
|
client.describe_instances.return_value = {"Reservations": []}
|
|
client.describe_security_groups.return_value = {"SecurityGroups": []}
|
|
|
|
elif service == "s3":
|
|
if client_error is not None:
|
|
client.list_buckets.side_effect = client_error
|
|
else:
|
|
buckets = s3_buckets or []
|
|
client.list_buckets.return_value = {"Buckets": buckets}
|
|
|
|
elif service == "ecs":
|
|
if client_error is not None:
|
|
client.list_clusters.side_effect = client_error
|
|
else:
|
|
arns = ecs_clusters or []
|
|
client.list_clusters.return_value = {"clusterArns": arns}
|
|
|
|
else:
|
|
if client_error is not None:
|
|
for attr in dir(client):
|
|
if not attr.startswith("_"):
|
|
try:
|
|
getattr(client, attr).side_effect = client_error
|
|
except AttributeError:
|
|
pass
|
|
|
|
return client
|
|
|
|
session.client.side_effect = _make_client
|
|
return session
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps (all prefixed with "awssdk")
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("awssdk the cloud handler module is imported")
|
|
def step_module_imported(context: Any) -> None:
|
|
"""Ensure the cloud handler module is importable."""
|
|
import cleveragents.resource.handlers.cloud as _mod # noqa: PLC0415
|
|
|
|
context.cloud_module = _mod # type: ignore[attr-defined]
|
|
|
|
|
|
@given("awssdk boto3 is not available")
|
|
def step_boto3_not_available(context: Any) -> None:
|
|
"""Simulate boto3 not being installed."""
|
|
context.boto3_available = False # type: ignore[attr-defined]
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
|
|
|
|
@given("awssdk boto3 is available via mock")
|
|
def step_boto3_available_mock(context: Any) -> None:
|
|
"""Mark boto3 as available (mocked)."""
|
|
context.boto3_available = True # type: ignore[attr-defined]
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
os.environ["AWS_ACCESS_KEY_ID"] = "AKIAIOSFODNN7EXAMPLE"
|
|
os.environ["AWS_SECRET_ACCESS_KEY"] = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
|
|
os.environ["AWS_REGION"] = "us-east-1"
|
|
|
|
|
|
@given("awssdk a mock boto3 session")
|
|
def step_mock_session(context: Any) -> None:
|
|
"""Create a generic mock boto3 session."""
|
|
context.mock_session = _make_mock_session() # type: ignore[attr-defined]
|
|
|
|
|
|
@given("awssdk a mock boto3 session that raises an exception")
|
|
def step_mock_session_error(context: Any) -> None:
|
|
"""Create a mock boto3 session that raises on all client calls."""
|
|
context.mock_session = _make_mock_session( # type: ignore[attr-defined]
|
|
client_error=RuntimeError("Simulated AWS API error")
|
|
)
|
|
|
|
|
|
@given('awssdk AWS STS returns account id "{account_id}"')
|
|
def step_sts_account_id(context: Any, account_id: str) -> None:
|
|
"""Configure mock STS to return a specific account ID."""
|
|
context.mock_account_id = account_id # type: ignore[attr-defined]
|
|
|
|
|
|
@given("awssdk AWS STS raises a ClientError")
|
|
def step_sts_client_error(context: Any) -> None:
|
|
"""Configure mock STS to raise a ClientError."""
|
|
context.sts_error = RuntimeError("AWS ClientError: InvalidClientTokenId") # type: ignore[attr-defined]
|
|
|
|
|
|
@given('awssdk a valid AWS cloud resource of type "{type_name}"')
|
|
def step_aws_resource_type(context: Any, type_name: str) -> None:
|
|
"""Create an AWS resource of the given type."""
|
|
context.cloud_resource = _make_resource( # type: ignore[attr-defined]
|
|
type_name,
|
|
properties={
|
|
"access-key-id": "AKIAIOSFODNN7EXAMPLE",
|
|
"secret-access-key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
|
|
"region": "us-east-1",
|
|
},
|
|
)
|
|
|
|
|
|
@given("awssdk a valid GCP cloud resource with credentials")
|
|
def step_gcp_resource_with_creds(context: Any) -> None:
|
|
"""Create a GCP resource with credentials."""
|
|
context.cloud_resource = _make_resource( # type: ignore[attr-defined]
|
|
"gcp",
|
|
properties={
|
|
"service-account-json-path": "/path/to/sa.json",
|
|
"project-id": "my-project",
|
|
},
|
|
)
|
|
|
|
|
|
@given("awssdk a valid Azure cloud resource with credentials")
|
|
def step_azure_resource_with_creds(context: Any) -> None:
|
|
"""Create an Azure resource with credentials."""
|
|
context.cloud_resource = _make_resource( # type: ignore[attr-defined]
|
|
"azure",
|
|
properties={
|
|
"subscription-id": "sub-123",
|
|
"tenant-id": "tenant-456",
|
|
"client-id": "client-789",
|
|
"client-secret": "secret-abc",
|
|
},
|
|
)
|
|
|
|
|
|
@given('awssdk a cloud resource of type "{type_name}"')
|
|
def step_cloud_resource_type(context: Any, type_name: str) -> None:
|
|
"""Create a cloud resource of the given type."""
|
|
context.cloud_resource = _make_resource(type_name) # type: ignore[attr-defined]
|
|
|
|
|
|
@given("awssdk EC2 describe_vpcs returns {count:d} VPCs")
|
|
def step_ec2_vpcs(context: Any, count: int) -> None:
|
|
"""Configure mock EC2 to return N VPCs."""
|
|
vpcs = [
|
|
{
|
|
"VpcId": f"vpc-{i:08x}",
|
|
"State": "available",
|
|
"CidrBlock": f"10.{i}.0.0/16",
|
|
}
|
|
for i in range(count)
|
|
]
|
|
context.mock_session = _make_mock_session(ec2_vpcs=vpcs) # type: ignore[attr-defined]
|
|
|
|
|
|
@given("awssdk S3 list_buckets returns {count:d} buckets")
|
|
def step_s3_buckets(context: Any, count: int) -> None:
|
|
"""Configure mock S3 to return N buckets."""
|
|
buckets = [
|
|
{"Name": f"my-bucket-{i}", "CreationDate": "2024-01-01"} for i in range(count)
|
|
]
|
|
context.mock_session = _make_mock_session(s3_buckets=buckets) # type: ignore[attr-defined]
|
|
|
|
|
|
@given("awssdk ECS list_clusters returns {count:d} cluster ARNs")
|
|
def step_ecs_clusters(context: Any, count: int) -> None:
|
|
"""Configure mock ECS to return N cluster ARNs."""
|
|
arns = [
|
|
f"arn:aws:ecs:us-east-1:123456789012:cluster/cluster-{i}" for i in range(count)
|
|
]
|
|
context.mock_session = _make_mock_session(ecs_clusters=arns) # type: ignore[attr-defined]
|
|
|
|
|
|
@given("awssdk a CloudResourceHandler instance")
|
|
def step_handler_instance(context: Any) -> None:
|
|
"""Create a CloudResourceHandler instance."""
|
|
context.cloud_handler = CloudResourceHandler() # type: ignore[attr-defined]
|
|
|
|
|
|
@given('awssdk a cloud sandbox strategy for "{provider}"')
|
|
def step_sandbox_strategy(context: Any, provider: str) -> None:
|
|
"""Create a CloudSandboxStrategy for the given provider."""
|
|
context.cloud_sandbox = CloudSandboxStrategy(provider) # type: ignore[attr-defined]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps (all prefixed with "awssdk")
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("awssdk I try to build an AWS session with valid credentials")
|
|
def step_try_build_session_no_boto3(context: Any) -> None:
|
|
"""Try to build an AWS session when boto3 is unavailable."""
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
context.raised_error_type = None # type: ignore[attr-defined]
|
|
with (
|
|
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", False),
|
|
patch("cleveragents.resource.handlers.cloud.boto3", None),
|
|
):
|
|
try:
|
|
_build_aws_session(
|
|
{
|
|
"access-key-id": "AKIAIOSFODNN7EXAMPLE",
|
|
"secret-access-key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
|
|
}
|
|
)
|
|
except ImportError as exc:
|
|
context.raised_error = exc # type: ignore[attr-defined]
|
|
context.raised_error_type = "ImportError" # type: ignore[attr-defined]
|
|
|
|
|
|
@when("awssdk I build an AWS session with explicit credentials")
|
|
def step_build_session_explicit(context: Any) -> None:
|
|
"""Build an AWS session with explicit credentials using mocked boto3."""
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
mock_boto3 = MagicMock()
|
|
mock_session = MagicMock()
|
|
mock_boto3.Session.return_value = mock_session
|
|
with (
|
|
patch("cleveragents.resource.handlers.cloud.boto3", mock_boto3),
|
|
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", True),
|
|
):
|
|
result = _build_aws_session(
|
|
{
|
|
"access-key-id": "AKIAIOSFODNN7EXAMPLE",
|
|
"secret-access-key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
|
|
"region": "us-east-1",
|
|
}
|
|
)
|
|
context.session_result = result # type: ignore[attr-defined]
|
|
context.mock_boto3 = mock_boto3 # type: ignore[attr-defined]
|
|
|
|
|
|
@when('awssdk I build an AWS session with a profile name "{profile}"')
|
|
def step_build_session_profile(context: Any, profile: str) -> None:
|
|
"""Build an AWS session with a profile name."""
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
mock_boto3 = MagicMock()
|
|
mock_session = MagicMock()
|
|
mock_boto3.Session.return_value = mock_session
|
|
with (
|
|
patch("cleveragents.resource.handlers.cloud.boto3", mock_boto3),
|
|
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", True),
|
|
):
|
|
result = _build_aws_session({"profile": profile})
|
|
context.session_result = result # type: ignore[attr-defined]
|
|
context.mock_boto3 = mock_boto3 # type: ignore[attr-defined]
|
|
context.expected_profile = profile # type: ignore[attr-defined]
|
|
|
|
|
|
@when("awssdk I call resolve on the cloud handler with mocked boto3")
|
|
def step_resolve_with_mock(context: Any) -> None:
|
|
"""Call resolve on CloudResourceHandler with mocked boto3."""
|
|
handler = CloudResourceHandler()
|
|
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
|
|
mock_manager = MagicMock()
|
|
|
|
account_id = getattr(context, "mock_account_id", "123456789012")
|
|
sts_error = getattr(context, "sts_error", None)
|
|
mock_session = _make_mock_session(account_id=account_id, sts_error=sts_error)
|
|
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
context.raised_error_type = None # type: ignore[attr-defined]
|
|
context.bound_resource = None # type: ignore[attr-defined]
|
|
|
|
mock_boto3 = MagicMock()
|
|
mock_boto3.Session.return_value = mock_session
|
|
|
|
with (
|
|
patch("cleveragents.resource.handlers.cloud.boto3", mock_boto3),
|
|
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", True),
|
|
):
|
|
try:
|
|
result = handler.resolve(
|
|
resource=resource,
|
|
plan_id="PLAN-TEST-001",
|
|
slot_name="cloud-slot",
|
|
sandbox_manager=mock_manager,
|
|
)
|
|
context.bound_resource = result # type: ignore[attr-defined]
|
|
except (ValueError, NotImplementedError, ImportError) as exc:
|
|
context.raised_error = exc # type: ignore[attr-defined]
|
|
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
|
|
saved = getattr(context, "saved_env", {})
|
|
_restore_cloud_env(saved)
|
|
|
|
|
|
@when("awssdk I call resolve on the cloud handler without boto3")
|
|
def step_resolve_without_boto3(context: Any) -> None:
|
|
"""Call resolve on CloudResourceHandler without boto3."""
|
|
handler = CloudResourceHandler()
|
|
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
|
|
mock_manager = MagicMock()
|
|
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
context.raised_error_type = None # type: ignore[attr-defined]
|
|
context.bound_resource = None # type: ignore[attr-defined]
|
|
|
|
with (
|
|
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", False),
|
|
patch("cleveragents.resource.handlers.cloud.boto3", None),
|
|
):
|
|
try:
|
|
result = handler.resolve(
|
|
resource=resource,
|
|
plan_id="PLAN-TEST-001",
|
|
slot_name="cloud-slot",
|
|
sandbox_manager=mock_manager,
|
|
)
|
|
context.bound_resource = result # type: ignore[attr-defined]
|
|
except (ValueError, NotImplementedError, ImportError) as exc:
|
|
context.raised_error = exc # type: ignore[attr-defined]
|
|
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
|
|
saved = getattr(context, "saved_env", {})
|
|
_restore_cloud_env(saved)
|
|
|
|
|
|
@when("awssdk I call discover_aws_resources")
|
|
def step_call_discover_aws(context: Any) -> None:
|
|
"""Call discover_aws_resources with the mock session."""
|
|
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
|
|
session = getattr(context, "mock_session", _make_mock_session())
|
|
|
|
context.discovery_result = discover_aws_resources(resource, session) # type: ignore[attr-defined]
|
|
|
|
|
|
@when("awssdk I call discover_children on the cloud handler")
|
|
def step_call_discover_children_no_boto3(context: Any) -> None:
|
|
"""Call discover_children on the handler (non-AWS provider)."""
|
|
handler: CloudResourceHandler = context.cloud_handler # type: ignore[attr-defined]
|
|
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
|
|
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
context.raised_error_type = None # type: ignore[attr-defined]
|
|
context.discovery_result = None # type: ignore[attr-defined]
|
|
|
|
try:
|
|
result = handler.discover_children(resource=resource)
|
|
context.discovery_result = result # type: ignore[attr-defined]
|
|
except (NotImplementedError, ImportError) as exc:
|
|
context.raised_error = exc # type: ignore[attr-defined]
|
|
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
|
|
|
|
|
|
@when("awssdk I call discover_children on the cloud handler without boto3")
|
|
def step_call_discover_children_no_boto3_explicit(context: Any) -> None:
|
|
"""Call discover_children without boto3 available."""
|
|
handler: CloudResourceHandler = context.cloud_handler # type: ignore[attr-defined]
|
|
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
|
|
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
context.raised_error_type = None # type: ignore[attr-defined]
|
|
context.discovery_result = None # type: ignore[attr-defined]
|
|
|
|
with (
|
|
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", False),
|
|
patch("cleveragents.resource.handlers.cloud.boto3", None),
|
|
):
|
|
try:
|
|
result = handler.discover_children(resource=resource)
|
|
context.discovery_result = result # type: ignore[attr-defined]
|
|
except (NotImplementedError, ImportError) as exc:
|
|
context.raised_error = exc # type: ignore[attr-defined]
|
|
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
|
|
|
|
|
|
@when("awssdk I call discover_children on the cloud handler with mocked boto3")
|
|
def step_call_discover_children_mocked(context: Any) -> None:
|
|
"""Call discover_children with mocked boto3."""
|
|
handler: CloudResourceHandler = context.cloud_handler # type: ignore[attr-defined]
|
|
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
|
|
mock_session = getattr(context, "mock_session", _make_mock_session())
|
|
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
context.raised_error_type = None # type: ignore[attr-defined]
|
|
context.discovery_result = None # type: ignore[attr-defined]
|
|
|
|
mock_boto3 = MagicMock()
|
|
mock_boto3.Session.return_value = mock_session
|
|
|
|
with (
|
|
patch("cleveragents.resource.handlers.cloud.boto3", mock_boto3),
|
|
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", True),
|
|
):
|
|
try:
|
|
result = handler.discover_children(resource=resource)
|
|
context.discovery_result = result # type: ignore[attr-defined]
|
|
except (NotImplementedError, ImportError) as exc:
|
|
context.raised_error = exc # type: ignore[attr-defined]
|
|
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
|
|
saved = getattr(context, "saved_env", {})
|
|
_restore_cloud_env(saved)
|
|
|
|
|
|
@when('awssdk I call create on the AWS sandbox strategy with plan "{plan_id}"')
|
|
def step_sandbox_create_aws(context: Any, plan_id: str) -> None:
|
|
"""Call create on the CloudSandboxStrategy."""
|
|
strategy: CloudSandboxStrategy = context.cloud_sandbox # type: ignore[attr-defined]
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
context.raised_error_type = None # type: ignore[attr-defined]
|
|
|
|
with patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", True):
|
|
try:
|
|
strategy.create("res-test-001", plan_id)
|
|
except (NotImplementedError, ImportError, ValueError) as exc:
|
|
context.raised_error = exc # type: ignore[attr-defined]
|
|
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
|
|
|
|
|
|
@when('awssdk I call commit on the AWS sandbox strategy with plan "{plan_id}"')
|
|
def step_sandbox_commit_aws(context: Any, plan_id: str) -> None:
|
|
"""Call commit on the CloudSandboxStrategy."""
|
|
strategy: CloudSandboxStrategy = context.cloud_sandbox # type: ignore[attr-defined]
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
context.raised_error_type = None # type: ignore[attr-defined]
|
|
|
|
with patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", True):
|
|
try:
|
|
strategy.commit("res-test-001", plan_id)
|
|
except (NotImplementedError, ImportError, ValueError) as exc:
|
|
context.raised_error = exc # type: ignore[attr-defined]
|
|
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
|
|
|
|
|
|
@when('awssdk I call rollback on the AWS sandbox strategy with plan "{plan_id}"')
|
|
def step_sandbox_rollback_aws(context: Any, plan_id: str) -> None:
|
|
"""Call rollback on the CloudSandboxStrategy."""
|
|
strategy: CloudSandboxStrategy = context.cloud_sandbox # type: ignore[attr-defined]
|
|
context.raised_error = None # type: ignore[attr-defined]
|
|
context.raised_error_type = None # type: ignore[attr-defined]
|
|
|
|
with patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", True):
|
|
try:
|
|
strategy.rollback("res-test-001", plan_id)
|
|
except (NotImplementedError, ImportError, ValueError) as exc:
|
|
context.raised_error = exc # type: ignore[attr-defined]
|
|
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps (all prefixed with "awssdk")
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("awssdk the boto3 availability flag should be a boolean")
|
|
def step_boto3_flag_bool(context: Any) -> None:
|
|
"""Check that _BOTO3_AVAILABLE is a boolean."""
|
|
assert isinstance(_BOTO3_AVAILABLE, bool), (
|
|
f"Expected bool, got {type(_BOTO3_AVAILABLE)}"
|
|
)
|
|
|
|
|
|
@then('awssdk an ImportError should be raised mentioning "{text}"')
|
|
def step_import_error_raised(context: Any, text: str) -> None:
|
|
"""Check that an ImportError was raised with the expected text."""
|
|
assert context.raised_error_type == "ImportError", ( # type: ignore[attr-defined]
|
|
f"Expected ImportError, got {context.raised_error_type}: " # type: ignore[attr-defined]
|
|
f"{context.raised_error}" # type: ignore[attr-defined]
|
|
)
|
|
msg = str(context.raised_error) # type: ignore[attr-defined]
|
|
assert text in msg, f"'{text}' not found in ImportError: {msg}"
|
|
|
|
|
|
@then("awssdk the session should be created successfully")
|
|
def step_session_created(context: Any) -> None:
|
|
"""Check that a session was created."""
|
|
assert context.session_result is not None # type: ignore[attr-defined]
|
|
mock_boto3: MagicMock = context.mock_boto3 # type: ignore[attr-defined]
|
|
assert mock_boto3.Session.called, "boto3.Session was not called"
|
|
|
|
|
|
@then('awssdk the session should be created with profile "{profile}"')
|
|
def step_session_created_with_profile(context: Any, profile: str) -> None:
|
|
"""Check that the session was created with the expected profile."""
|
|
assert context.session_result is not None # type: ignore[attr-defined]
|
|
mock_boto3: MagicMock = context.mock_boto3 # type: ignore[attr-defined]
|
|
call_kwargs = mock_boto3.Session.call_args[1]
|
|
assert call_kwargs.get("profile_name") == profile, (
|
|
f"Expected profile_name='{profile}', got {call_kwargs}"
|
|
)
|
|
|
|
|
|
@then("awssdk a BoundResource should be returned")
|
|
def step_bound_resource_returned(context: Any) -> None:
|
|
"""Check that a BoundResource was returned."""
|
|
from cleveragents.tool.context import BoundResource # noqa: PLC0415
|
|
|
|
assert context.raised_error is None, ( # type: ignore[attr-defined]
|
|
f"Expected BoundResource but got error: "
|
|
f"{context.raised_error_type}: {context.raised_error}" # type: ignore[attr-defined]
|
|
)
|
|
assert isinstance(context.bound_resource, BoundResource), ( # type: ignore[attr-defined]
|
|
f"Expected BoundResource, got {type(context.bound_resource)}" # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
@then('awssdk the BoundResource slot_name should be "{slot_name}"')
|
|
def step_bound_resource_slot(context: Any, slot_name: str) -> None:
|
|
"""Check BoundResource slot_name."""
|
|
from cleveragents.tool.context import BoundResource # noqa: PLC0415
|
|
|
|
br: BoundResource = context.bound_resource # type: ignore[attr-defined]
|
|
assert br.slot_name == slot_name, (
|
|
f"Expected slot_name='{slot_name}', got '{br.slot_name}'"
|
|
)
|
|
|
|
|
|
@then('awssdk the BoundResource resource_type should be "{resource_type}"')
|
|
def step_bound_resource_type(context: Any, resource_type: str) -> None:
|
|
"""Check BoundResource resource_type."""
|
|
from cleveragents.tool.context import BoundResource # noqa: PLC0415
|
|
|
|
br: BoundResource = context.bound_resource # type: ignore[attr-defined]
|
|
assert br.resource_type == resource_type, (
|
|
f"Expected resource_type='{resource_type}', got '{br.resource_type}'"
|
|
)
|
|
|
|
|
|
@then('awssdk the BoundResource sandbox_path should contain "{text}"')
|
|
def step_bound_resource_sandbox_path(context: Any, text: str) -> None:
|
|
"""Check BoundResource sandbox_path contains expected text."""
|
|
from cleveragents.tool.context import BoundResource # noqa: PLC0415
|
|
|
|
br: BoundResource = context.bound_resource # type: ignore[attr-defined]
|
|
assert br.sandbox_path is not None, "Expected sandbox_path to be set"
|
|
assert text in br.sandbox_path, (
|
|
f"Expected '{text}' in sandbox_path='{br.sandbox_path}'"
|
|
)
|
|
|
|
|
|
@then('awssdk a ValueError should be raised mentioning "{text}"')
|
|
def step_value_error_raised(context: Any, text: str) -> None:
|
|
"""Check that a ValueError was raised with the expected text."""
|
|
assert context.raised_error_type == "ValueError", ( # type: ignore[attr-defined]
|
|
f"Expected ValueError, got {context.raised_error_type}: " # type: ignore[attr-defined]
|
|
f"{context.raised_error}" # type: ignore[attr-defined]
|
|
)
|
|
msg = str(context.raised_error) # type: ignore[attr-defined]
|
|
assert text in msg, f"'{text}' not found in ValueError: {msg}"
|
|
|
|
|
|
@then('awssdk a NotImplementedError should be raised mentioning "{text}"')
|
|
def step_not_implemented_raised(context: Any, text: str) -> None:
|
|
"""Check that a NotImplementedError was raised with the expected text."""
|
|
assert context.raised_error_type == "NotImplementedError", ( # type: ignore[attr-defined]
|
|
f"Expected NotImplementedError, got {context.raised_error_type}: " # type: ignore[attr-defined]
|
|
f"{context.raised_error}" # type: ignore[attr-defined]
|
|
)
|
|
msg = str(context.raised_error) # type: ignore[attr-defined]
|
|
assert text in msg, f"'{text}' not found in NotImplementedError: {msg}"
|
|
|
|
|
|
@then("awssdk a NotImplementedError should be raised")
|
|
def step_not_implemented_raised_any(context: Any) -> None:
|
|
"""Check that a NotImplementedError was raised."""
|
|
assert context.raised_error_type == "NotImplementedError", ( # type: ignore[attr-defined]
|
|
f"Expected NotImplementedError, got {context.raised_error_type}: " # type: ignore[attr-defined]
|
|
f"{context.raised_error}" # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
@then("awssdk the discovery result should be an empty list")
|
|
def step_discovery_empty(context: Any) -> None:
|
|
"""Check that the discovery result is empty."""
|
|
result = context.discovery_result # type: ignore[attr-defined]
|
|
assert result == [], f"Expected empty list, got {result}"
|
|
|
|
|
|
@then("awssdk the discovery result should have {count:d} items")
|
|
def step_discovery_count(context: Any, count: int) -> None:
|
|
"""Check the discovery result count."""
|
|
result = context.discovery_result # type: ignore[attr-defined]
|
|
assert len(result) == count, f"Expected {count} items, got {len(result)}: {result}"
|
|
|
|
|
|
@then('awssdk each discovery item should have an "{key}" key')
|
|
def step_discovery_item_key(context: Any, key: str) -> None:
|
|
"""Check that each discovery item has the expected key."""
|
|
result = context.discovery_result # type: ignore[attr-defined]
|
|
for item in result:
|
|
assert key in item, f"Key '{key}' not found in item: {item}"
|
|
|
|
|
|
@then('awssdk each discovery item arn should start with "{prefix}"')
|
|
def step_discovery_item_arn_prefix(context: Any, prefix: str) -> None:
|
|
"""Check that each discovery item ARN starts with the expected prefix."""
|
|
result = context.discovery_result # type: ignore[attr-defined]
|
|
for item in result:
|
|
arn = item.get("arn", "")
|
|
assert arn.startswith(prefix), (
|
|
f"Expected ARN to start with '{prefix}', got '{arn}'"
|
|
)
|
|
|
|
|
|
@then("awssdk the aws discovery result should be a list of Resource objects")
|
|
def step_result_resource_list(context: Any) -> None:
|
|
"""Check that the result is a list of Resource objects."""
|
|
result = context.discovery_result # type: ignore[attr-defined]
|
|
assert isinstance(result, list), f"Expected list, got {type(result)}"
|
|
for item in result:
|
|
assert isinstance(item, Resource), (
|
|
f"Expected Resource, got {type(item)}: {item}"
|
|
)
|
|
|
|
|
|
@then("awssdk the aws discovery result should have {count:d} items")
|
|
def step_result_count(context: Any, count: int) -> None:
|
|
"""Check the result count."""
|
|
result = context.discovery_result # type: ignore[attr-defined]
|
|
assert len(result) == count, f"Expected {count} items, got {len(result)}"
|
|
|
|
|
|
@then("awssdk no exception should be raised")
|
|
def step_no_exception(context: Any) -> None:
|
|
"""Check that no exception was raised."""
|
|
assert context.raised_error is None, ( # type: ignore[attr-defined]
|
|
f"Expected no exception, got {context.raised_error_type}: " # type: ignore[attr-defined]
|
|
f"{context.raised_error}" # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
@then("awssdk no raw credential values should appear in log output")
|
|
def step_no_raw_creds_in_log(context: Any) -> None:
|
|
"""Check that no raw credential values appear in log output."""
|
|
# Verify the handler ran
|
|
assert (
|
|
context.bound_resource is not None or context.raised_error is not None # type: ignore[attr-defined]
|
|
), "Handler did not run"
|
|
# Verify the known secret value is not in the error message (if any)
|
|
if context.raised_error is not None: # type: ignore[attr-defined]
|
|
msg = str(context.raised_error) # type: ignore[attr-defined]
|
|
assert "wJalrXUtnFEMI" not in msg, f"Secret found in error: {msg}"
|
|
assert "AKIAIOSFODNN7EXAMPLE" not in msg, f"Secret found in error: {msg}"
|
|
|
|
|
|
@then('awssdk the AWS resource map should contain "{type_name}"')
|
|
def step_aws_resource_map_contains(context: Any, type_name: str) -> None:
|
|
"""Check that the AWS resource map contains the expected type."""
|
|
assert type_name in _AWS_RESOURCE_MAP, (
|
|
f"'{type_name}' not found in _AWS_RESOURCE_MAP. "
|
|
f"Available: {sorted(_AWS_RESOURCE_MAP.keys())}"
|
|
)
|