HAL9000
c8596a44f0
Remove duplicate except ImportError blocks in cloud_resources_steps.py (B025 violations in step_sandbox_create, step_sandbox_commit, step_sandbox_rollback). Apply ruff format to 5 files flagged by the format check.
222 lines
9.2 KiB
Python
222 lines
9.2 KiB
Python
"""Step definitions for AWS resource discovery scenarios.
|
|
|
|
Tests discover_aws_resources() and CloudResourceHandler.discover_children()
|
|
with mocked boto3.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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 Resource
|
|
from cleveragents.resource.handlers.cloud import CloudResourceHandler
|
|
from cleveragents.resource.handlers.cloud_aws import (
|
|
_AWS_RESOURCE_MAP,
|
|
discover_aws_resources,
|
|
)
|
|
from features.steps.cloud_aws_helpers import (
|
|
_make_mock_session,
|
|
_make_resource,
|
|
_restore_cloud_env,
|
|
)
|
|
|
|
|
|
@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 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")
|
|
)
|
|
|
|
|
|
@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_aws._BOTO3_AVAILABLE", False),
|
|
patch("cleveragents.resource.handlers.cloud_aws.boto3", None),
|
|
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", False),
|
|
):
|
|
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_aws.boto3", mock_boto3),
|
|
patch("cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", True),
|
|
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)
|
|
|
|
|
|
@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 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())}"
|
|
)
|