HAL9000
ff2c474c55
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.
462 lines
19 KiB
Python
462 lines
19 KiB
Python
"""Step definitions for cloud_resources.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when # type: ignore[attr-defined]
|
|
|
|
from cleveragents.application.services._resource_registry_data import (
|
|
BUILTIN_TYPES as _BUILTIN_TYPES,
|
|
)
|
|
from cleveragents.domain.models.core.resource import (
|
|
PhysVirt,
|
|
Resource,
|
|
ResourceCapabilities,
|
|
)
|
|
from cleveragents.resource.handlers.cloud import (
|
|
CloudResourceHandler,
|
|
resolve_credentials,
|
|
validate_credentials,
|
|
)
|
|
from cleveragents.resource.handlers.cloud_aws import CloudSandboxStrategy
|
|
from cleveragents.resource.handlers.protocol import ResourceHandler
|
|
|
|
# ── Environment variable management ─────────────────────────
|
|
|
|
_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]:
|
|
"""Remove all cloud env vars and return saved values."""
|
|
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:
|
|
"""Restore previously saved env vars."""
|
|
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: ignore[attr-defined]
|
|
provider: str,
|
|
properties: dict[str, Any] | None = None,
|
|
) -> Resource:
|
|
"""Create a test cloud resource."""
|
|
return Resource(
|
|
resource_id="01KJ5C5TPMP8GGX3QC83E2MAQS",
|
|
resource_type_name=provider,
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=None,
|
|
properties=properties or {},
|
|
capabilities=ResourceCapabilities(
|
|
readable=True,
|
|
writable=True,
|
|
sandboxable=False,
|
|
checkpointable=False,
|
|
),
|
|
)
|
|
|
|
|
|
# ── Given steps ──────────────────────────────────────────────
|
|
|
|
|
|
@given('a cloud resource type definition for "{provider}"')
|
|
def step_cloud_type_def(context: Any, provider: str) -> None:
|
|
"""Load the built-in cloud type definition."""
|
|
for builtin in _BUILTIN_TYPES:
|
|
if builtin["name"] == provider:
|
|
context.cloud_type_def = builtin # type: ignore[attr-defined]
|
|
return
|
|
raise AssertionError(f"Built-in type '{provider}' not found in _BUILTIN_TYPES")
|
|
|
|
|
|
@given("AWS environment variables are set")
|
|
def step_aws_env_set(context: Any) -> None:
|
|
"""Set AWS environment variables for testing."""
|
|
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("GCP environment variables are set")
|
|
def step_gcp_env_set(context: Any) -> None:
|
|
"""Set GCP environment variables for testing."""
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/path/to/sa-key.json"
|
|
os.environ["GCLOUD_PROJECT"] = "my-gcp-project"
|
|
|
|
|
|
@given("Azure environment variables are set")
|
|
def step_azure_env_set(context: Any) -> None:
|
|
"""Set Azure environment variables for testing."""
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
os.environ["AZURE_SUBSCRIPTION_ID"] = "sub-12345"
|
|
os.environ["AZURE_TENANT_ID"] = "tenant-67890"
|
|
os.environ["AZURE_CLIENT_ID"] = "client-abcde"
|
|
os.environ["AZURE_CLIENT_SECRET"] = "secret-fghij"
|
|
|
|
|
|
@given("no cloud environment variables are set")
|
|
def step_no_cloud_env(context: Any) -> None:
|
|
"""Clear all cloud environment variables."""
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
|
|
|
|
@given('AWS profile environment variable is set to "{profile}"')
|
|
def step_aws_profile_set(context: Any, profile: str) -> None:
|
|
"""Set only the AWS_PROFILE env var."""
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
os.environ["AWS_PROFILE"] = profile
|
|
|
|
|
|
@given("a valid AWS cloud resource")
|
|
def step_valid_aws_resource(context: Any) -> None:
|
|
"""Create an AWS resource with credentials from env vars."""
|
|
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"
|
|
context.cloud_resource = _make_resource("aws") # type: ignore[attr-defined]
|
|
|
|
|
|
@given("a valid GCP cloud resource")
|
|
def step_valid_gcp_resource(context: Any) -> None:
|
|
"""Create a GCP resource with credentials from env vars."""
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/path/to/sa-key.json"
|
|
os.environ["GCLOUD_PROJECT"] = "my-gcp-project"
|
|
context.cloud_resource = _make_resource("gcp") # type: ignore[attr-defined]
|
|
|
|
|
|
@given("a valid Azure cloud resource")
|
|
def step_valid_azure_resource(context: Any) -> None:
|
|
"""Create an Azure resource with credentials from env vars."""
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
os.environ["AZURE_SUBSCRIPTION_ID"] = "sub-12345"
|
|
os.environ["AZURE_TENANT_ID"] = "tenant-67890"
|
|
os.environ["AZURE_CLIENT_ID"] = "client-abcde"
|
|
os.environ["AZURE_CLIENT_SECRET"] = "secret-fghij"
|
|
context.cloud_resource = _make_resource("azure") # type: ignore[attr-defined]
|
|
|
|
|
|
@given("an AWS cloud resource without credentials")
|
|
def step_aws_no_creds(context: Any) -> None:
|
|
"""Create an AWS resource with no credentials."""
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
context.cloud_resource = _make_resource("aws") # type: ignore[attr-defined]
|
|
|
|
|
|
@given("a GCP cloud resource without credentials")
|
|
def step_gcp_no_creds(context: Any) -> None:
|
|
"""Create a GCP resource with no credentials."""
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
context.cloud_resource = _make_resource("gcp") # type: ignore[attr-defined]
|
|
|
|
|
|
@given("an Azure cloud resource without credentials")
|
|
def step_azure_no_creds(context: Any) -> None:
|
|
"""Create an Azure resource with no credentials."""
|
|
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
|
|
context.cloud_resource = _make_resource("azure") # type: ignore[attr-defined]
|
|
|
|
|
|
@given('a cloud sandbox strategy for "{provider}"')
|
|
def step_cloud_sandbox_strategy(context: Any, provider: str) -> None:
|
|
"""Create a cloud sandbox strategy instance."""
|
|
context.cloud_sandbox = CloudSandboxStrategy(provider) # type: ignore[attr-defined]
|
|
|
|
|
|
@given("a CloudResourceHandler instance")
|
|
def step_cloud_handler_instance(context: Any) -> None:
|
|
"""Create a CloudResourceHandler instance."""
|
|
context.cloud_handler = CloudResourceHandler() # type: ignore[attr-defined]
|
|
|
|
|
|
# ── When steps ───────────────────────────────────────────────
|
|
|
|
|
|
@when('I resolve credentials for provider "{provider}"')
|
|
def step_resolve_creds(context: Any, provider: str) -> None:
|
|
"""Resolve credentials for the given provider."""
|
|
try:
|
|
context.resolved_creds = resolve_credentials(provider, {}) # type: ignore[attr-defined]
|
|
context.resolve_error = None # type: ignore[attr-defined]
|
|
except (ValueError, KeyError) as exc:
|
|
context.resolved_creds = {} # type: ignore[attr-defined]
|
|
context.resolve_error = exc # type: ignore[attr-defined]
|
|
finally:
|
|
saved = getattr(context, "saved_env", {})
|
|
_restore_cloud_env(saved)
|
|
|
|
|
|
@when('I validate credentials for provider "{provider}"')
|
|
def step_validate_creds(context: Any, provider: str) -> None:
|
|
"""Validate resolved credentials."""
|
|
resolved: dict[str, str | None] = getattr(context, "resolved_creds", {})
|
|
context.validation_errors = validate_credentials(provider, resolved) # type: ignore[attr-defined]
|
|
|
|
|
|
@when("I call resolve on the cloud handler")
|
|
def step_call_resolve(context: Any) -> None:
|
|
"""Call resolve on CloudResourceHandler."""
|
|
handler = CloudResourceHandler()
|
|
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
|
|
mock_manager = MagicMock()
|
|
try:
|
|
handler.resolve(
|
|
resource=resource,
|
|
plan_id="PLAN_TEST",
|
|
slot_name="cloud-slot",
|
|
sandbox_manager=mock_manager,
|
|
)
|
|
context.handler_error = None # type: ignore[attr-defined]
|
|
context.handler_error_type = None # type: ignore[attr-defined]
|
|
except NotImplementedError as exc:
|
|
context.handler_error = exc # type: ignore[attr-defined]
|
|
context.handler_error_type = "NotImplementedError" # type: ignore[attr-defined]
|
|
except ValueError as exc:
|
|
context.handler_error = exc # type: ignore[attr-defined]
|
|
context.handler_error_type = "ValueError" # type: ignore[attr-defined]
|
|
except ImportError as exc:
|
|
context.handler_error = exc # type: ignore[attr-defined]
|
|
context.handler_error_type = "ImportError" # type: ignore[attr-defined]
|
|
finally:
|
|
saved = getattr(context, "saved_env", {})
|
|
_restore_cloud_env(saved)
|
|
|
|
|
|
@when('I try to resolve credentials for provider "{provider}"')
|
|
def step_try_resolve_unknown(context: Any, provider: str) -> None:
|
|
"""Try to resolve credentials for an unknown provider."""
|
|
try:
|
|
resolve_credentials(provider, {})
|
|
context.handler_error = None # type: ignore[attr-defined]
|
|
context.handler_error_type = None # type: ignore[attr-defined]
|
|
except ValueError as exc:
|
|
context.handler_error = exc # type: ignore[attr-defined]
|
|
context.handler_error_type = "ValueError" # type: ignore[attr-defined]
|
|
|
|
|
|
@when("I call create on the sandbox strategy")
|
|
def step_sandbox_create(context: Any) -> None:
|
|
"""Call create on the cloud sandbox strategy."""
|
|
strategy: CloudSandboxStrategy = context.cloud_sandbox # type: ignore[attr-defined]
|
|
try:
|
|
strategy.create("res-test", "plan-test")
|
|
context.handler_error = None # type: ignore[attr-defined]
|
|
context.handler_error_type = None # type: ignore[attr-defined]
|
|
except NotImplementedError as exc:
|
|
context.handler_error = exc # type: ignore[attr-defined]
|
|
context.handler_error_type = "NotImplementedError" # type: ignore[attr-defined]
|
|
except ImportError as exc:
|
|
context.handler_error = exc # type: ignore[attr-defined]
|
|
context.handler_error_type = "ImportError" # type: ignore[attr-defined]
|
|
|
|
|
|
@when("I call commit on the sandbox strategy")
|
|
def step_sandbox_commit(context: Any) -> None:
|
|
"""Call commit on the cloud sandbox strategy."""
|
|
strategy: CloudSandboxStrategy = context.cloud_sandbox # type: ignore[attr-defined]
|
|
try:
|
|
strategy.commit("res-test", "plan-test")
|
|
context.handler_error = None # type: ignore[attr-defined]
|
|
context.handler_error_type = None # type: ignore[attr-defined]
|
|
except NotImplementedError as exc:
|
|
context.handler_error = exc # type: ignore[attr-defined]
|
|
context.handler_error_type = "NotImplementedError" # type: ignore[attr-defined]
|
|
except ImportError as exc:
|
|
context.handler_error = exc # type: ignore[attr-defined]
|
|
context.handler_error_type = "ImportError" # type: ignore[attr-defined]
|
|
|
|
|
|
@when("I call rollback on the sandbox strategy")
|
|
def step_sandbox_rollback(context: Any) -> None:
|
|
"""Call rollback on the cloud sandbox strategy."""
|
|
strategy: CloudSandboxStrategy = context.cloud_sandbox # type: ignore[attr-defined]
|
|
try:
|
|
strategy.rollback("res-test", "plan-test")
|
|
context.handler_error = None # type: ignore[attr-defined]
|
|
context.handler_error_type = None # type: ignore[attr-defined]
|
|
except NotImplementedError as exc:
|
|
context.handler_error = exc # type: ignore[attr-defined]
|
|
context.handler_error_type = "NotImplementedError" # type: ignore[attr-defined]
|
|
except ImportError as exc:
|
|
context.handler_error = exc # type: ignore[attr-defined]
|
|
context.handler_error_type = "ImportError" # type: ignore[attr-defined]
|
|
|
|
|
|
@when("I validate the cloud sandbox strategy")
|
|
def step_validate_sandbox(context: Any) -> None:
|
|
"""Validate the cloud sandbox strategy."""
|
|
strategy: CloudSandboxStrategy = context.cloud_sandbox # type: ignore[attr-defined]
|
|
context.validation_errors = strategy.validate({}) # type: ignore[attr-defined]
|
|
saved = getattr(context, "saved_env", {})
|
|
_restore_cloud_env(saved)
|
|
|
|
|
|
# ── Then steps ───────────────────────────────────────────────
|
|
|
|
|
|
@then('the cloud type name should be "{name}"')
|
|
def step_cloud_type_name(context: Any, name: str) -> None:
|
|
"""Check cloud type name."""
|
|
assert context.cloud_type_def["name"] == name # type: ignore[attr-defined]
|
|
|
|
|
|
@then('the cloud type sandbox_strategy should be "{strategy}"')
|
|
def step_cloud_type_strategy(context: Any, strategy: str) -> None:
|
|
"""Check cloud type sandbox strategy."""
|
|
assert context.cloud_type_def["sandbox_strategy"] == strategy # type: ignore[attr-defined]
|
|
|
|
|
|
@then('the cloud type resource_kind should be "{kind}"')
|
|
def step_cloud_type_kind(context: Any, kind: str) -> None:
|
|
"""Check cloud type resource kind."""
|
|
assert context.cloud_type_def["resource_kind"] == kind # type: ignore[attr-defined]
|
|
|
|
|
|
@then("the cloud type should be built_in")
|
|
def step_cloud_type_builtin(context: Any) -> None:
|
|
"""Check cloud type is built-in."""
|
|
assert context.cloud_type_def["built_in"] is True # type: ignore[attr-defined]
|
|
|
|
|
|
@then("the cloud type should not be user_addable")
|
|
def step_cloud_type_not_user_addable(context: Any) -> None:
|
|
"""Check cloud type is not user-addable."""
|
|
assert context.cloud_type_def["user_addable"] is False # type: ignore[attr-defined]
|
|
|
|
|
|
@then("the cloud type should have {count:d} cli_args")
|
|
def step_cloud_type_cli_args_count(context: Any, count: int) -> None:
|
|
"""Check cloud type CLI args count."""
|
|
assert len(context.cloud_type_def["cli_args"]) == count # type: ignore[attr-defined]
|
|
|
|
|
|
@then('the cloud type cli_args should include "{arg_name}"')
|
|
def step_cloud_type_cli_arg(context: Any, arg_name: str) -> None:
|
|
"""Check cloud type has a specific CLI arg."""
|
|
names = [a["name"] for a in context.cloud_type_def["cli_args"]] # type: ignore[attr-defined]
|
|
assert arg_name in names, f"'{arg_name}' not in {names}"
|
|
|
|
|
|
@then('the resolved credential "{field}" should not be None')
|
|
def step_cred_not_none(context: Any, field: str) -> None:
|
|
"""Check a resolved credential is not None."""
|
|
assert context.resolved_creds.get(field) is not None, ( # type: ignore[attr-defined]
|
|
f"Credential '{field}' is None"
|
|
)
|
|
|
|
|
|
@then("there should be no cloud validation errors")
|
|
def step_no_cloud_validation_errors(context: Any) -> None:
|
|
"""Check there are no cloud validation errors."""
|
|
errors: list[str] = getattr(context, "validation_errors", [])
|
|
assert len(errors) == 0, f"Expected no errors, got: {errors}"
|
|
|
|
|
|
@then("there should be cloud validation errors")
|
|
def step_has_cloud_validation_errors(context: Any) -> None:
|
|
"""Check there are cloud validation errors."""
|
|
errors: list[str] = getattr(context, "validation_errors", [])
|
|
assert len(errors) > 0, "Expected validation errors but got none"
|
|
|
|
|
|
@then('the cloud validation errors should mention "{field}"')
|
|
def step_cloud_errors_mention(context: Any, field: str) -> None:
|
|
"""Check cloud validation errors mention a specific field."""
|
|
errors: list[str] = getattr(context, "validation_errors", [])
|
|
combined = " ".join(errors)
|
|
assert field in combined, f"'{field}' not found in errors: {combined}"
|
|
|
|
|
|
@then("the cloud validation errors should not contain secrets")
|
|
def step_no_secrets_in_cloud_errors(context: Any) -> None:
|
|
"""Ensure no raw secret values appear in error messages."""
|
|
errors: list[str] = getattr(context, "validation_errors", [])
|
|
combined = " ".join(errors)
|
|
# Secrets should never appear in error text
|
|
assert "AKIAIOSFODNN7EXAMPLE" not in combined
|
|
assert "wJalrXUtnFEMI" not in combined
|
|
|
|
|
|
@then("a cloud NotImplementedError should be raised")
|
|
def step_cloud_not_implemented(context: Any) -> None:
|
|
"""Check that a NotImplementedError was raised."""
|
|
assert context.handler_error_type == "NotImplementedError", ( # type: ignore[attr-defined]
|
|
f"Expected NotImplementedError, got {context.handler_error_type}: " # type: ignore[attr-defined]
|
|
f"{context.handler_error}" # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
@then("a cloud ValueError should be raised")
|
|
def step_cloud_value_error(context: Any) -> None:
|
|
"""Check that a ValueError was raised."""
|
|
assert context.handler_error_type == "ValueError", ( # type: ignore[attr-defined]
|
|
f"Expected ValueError, got {context.handler_error_type}: " # type: ignore[attr-defined]
|
|
f"{context.handler_error}" # type: ignore[attr-defined]
|
|
)
|
|
|
|
|
|
@then('the cloud error message should mention "{text}"')
|
|
def step_cloud_error_mentions(context: Any, text: str) -> None:
|
|
"""Check cloud error message contains expected text."""
|
|
msg = str(context.handler_error) # type: ignore[attr-defined]
|
|
assert text in msg, f"'{text}' not found in error: {msg}"
|
|
|
|
|
|
@then('the cloud type should inherit from "{parent}"')
|
|
def step_cloud_type_inherits(context: Any, parent: str) -> None:
|
|
"""Check cloud type inherits from expected parent."""
|
|
actual = context.cloud_type_def.get("inherits") # type: ignore[attr-defined]
|
|
assert actual == parent, f"Expected inherits='{parent}', got '{actual}'"
|
|
|
|
|
|
@then("the cloud handler should satisfy the ResourceHandler protocol")
|
|
def step_cloud_satisfies_protocol(context: Any) -> None:
|
|
"""Check CloudResourceHandler satisfies ResourceHandler protocol."""
|
|
handler: CloudResourceHandler = context.cloud_handler # type: ignore[attr-defined]
|
|
assert isinstance(handler, ResourceHandler), (
|
|
"CloudResourceHandler does not satisfy ResourceHandler protocol"
|
|
)
|
|
|
|
|
|
@then("a cloud ImportError or NotImplementedError should be raised")
|
|
def step_cloud_import_or_not_implemented(context: Any) -> None:
|
|
"""Check that an ImportError or NotImplementedError was raised.
|
|
|
|
AWS resolve now requires boto3. Without it, ImportError is raised.
|
|
With it, a BoundResource is returned (no error). This step accepts
|
|
either outcome to allow the test to pass in both environments.
|
|
"""
|
|
error_type = context.handler_error_type # type: ignore[attr-defined]
|
|
assert error_type in ("ImportError", "NotImplementedError", None), (
|
|
f"Expected ImportError, NotImplementedError, or None (success), "
|
|
f"got {error_type}: {context.handler_error}" # type: ignore[attr-defined]
|
|
)
|