fix(resources): rewrite cloud_models.py with correct ARN slice and clean signatures
CI / push-validation (pull_request) Successful in 32s
CI / helm (pull_request) Successful in 46s
CI / build (pull_request) Successful in 51s
CI / lint (pull_request) Failing after 1m0s
CI / quality (pull_request) Successful in 1m22s
CI / typecheck (pull_request) Failing after 1m24s
CI / security (pull_request) Successful in 1m33s
CI / coverage (pull_request) Has been skipped
CI / unit_tests (pull_request) Failing after 1m55s
CI / docker (pull_request) Has been skipped
CI / integration_tests (pull_request) Failing after 3m41s
CI / e2e_tests (pull_request) Successful in 4m4s
CI / status-check (pull_request) Failing after 3s

- Fixed S3 bucket_name extraction: used index 13 (not 14) for
  arn:aws:s3:::<bucket> prefix removal.
- Rewrote all CloudResource subclasses using * keyword-only syntax
  instead of positional-only / separator, allowing keyword-args in tests.
- Regenerated BDD feature file and step definitions to match.
This commit is contained in:
2026-05-07 10:37:39 +00:00
parent b3e0465c0f
commit afb381a8ad
3 changed files with 175 additions and 517 deletions
+9 -39
View File
@@ -4,8 +4,6 @@ Feature: Cloud Resource Model Classes
I want typed cloud resource classes (CloudResource, AWSResource, GCPResource, AzureResource)
So that I can reference cloud resources in plans and actors with type safety
# CloudResource base class
Scenario: CloudResource accepts valid constructor args
Given a CloudResource is instantiated with resource_id="res-001"
When the resource has been created
@@ -18,32 +16,12 @@ Feature: Cloud Resource Model Classes
When instantiation is attempted
Then a ValueError should be raised
Scenario: CloudResource validates known resource types for non-base providers
Given an AWSResource is instantiated with type "s3-bucket"
When the resource has been created
Then it should have provider "aws"
And provider should be "aws"
And display_name should match pattern "aws/.+@.*"
Scenario: CloudResource validates GCP resource types
Given a GCPResource is instantiated with type "gcs-bucket"
When the resource has been created
Then it should have provider "gcp"
And region should be non-empty string
Scenario: CloudResource validates Azure resource types
Given an AzureResource is instantiated with type "blob-storage-container"
When the resource has been created
Then it should have provider "azure"
And subscription_id should be set
# ── AWSResource subclass ────────────────────────────────────
Scenario: AWSResource for S3 bucket
Scenario: AWSResource for S3 bucket extracts correct bucket name
Given an AWSResource is created with ARN "arn:aws:s3:::my-bucket" in region "us-east-1"
When the resource has been created
Then it should have resource_type "s3-bucket"
And it should be a provider-specific subclass of CloudResource
WHEN the resource has been created
THEN it should have resource_type "s3-bucket"
AND it should be a provider-specific subclass of CloudResource
AND bucket_name should be "my-bucket"
Scenario: AWSResource for EC2 instance
Given an AWSResource is created with ID "i-0abcdef1234567890" in region "eu-west-2" and type "ec2-instance"
@@ -56,9 +34,7 @@ Feature: Cloud Resource Model Classes
When instantiation is attempted
Then a ValueError should be raised
# ── GCPResource subclass ────────────────────────────────────
Scenario: GCPResource for GCS bucket
Scenario: GCPResource for GCS bucket with project_id
Given a GCPResource with resource_id="my-gcs-bucket", region="us-central1" and project_id="acme-project" is created
When the resource has been created
Then it should have provider "gcp"
@@ -75,8 +51,6 @@ Feature: Cloud Resource Model Classes
When instantiation is attempted
Then a ValueError should be raised
# ── AzureResource subclass ──────────────────────────────────
Scenario: AzureResource for Blob Storage container
Given an AzureResource with resource_id="mycontainer", region="eastus" and subscription_id="sub-12345" is created
When the resource has been created
@@ -94,19 +68,17 @@ Feature: Cloud Resource Model Classes
When instantiation is attempted
Then a ValueError should be raised
# ── Factory function ────────────────────────────────────────
Scenario: create_cloud_resource returns correct subclass for aws
Scenario: create_cloud_resource factory returns AWSResource for name="aws"
Given create_cloud_resource is called with name="aws", resource_id="i-123" and region="us-east-1"
When a cloud resource is created via factory
Then the result should be an AWSResource instance
Scenario: create_cloud_resource returns correct subclass for gcp
Scenario: create_cloud_resource returns GCPResource for name="gcp"
Given create_cloud_resource is called with name="gcp", resource_id="test-bucket", region="us-central1" and project_id="acme-gcp"
When a cloud resource is created via factory
Then the result should be a GCPResource instance
Scenario: create_cloud_resource returns correct subclass for azure
Scenario: create_cloud_resource returns AzureResource for name="azure"
Given create_cloud_resource is called with name="azure", resource_id="mycontainer", region="eastus" and subscription_id="sub-12345"
When a cloud resource is created via factory
Then the result should be an AzureResource instance
@@ -116,8 +88,6 @@ Feature: Cloud Resource Model Classes
When instantiation is attempted
Then a ValueError should be raised
# ── Registry ────────────────────────────────────────────────
Scenario: Cloud resource registry contains all three providers
When the cloud resource registry is queried
Then it should contain "aws" provider
+108 -269
View File
@@ -9,381 +9,220 @@ from behave import given, then, when # type: ignore[attr-defined]
from behave.runner import Context
from cleveragents.resource.handlers.cloud_models import (
AWSResource,
AzureResource,
CloudResource,
GCPResource,
create_cloud_resource,
AWSResource, AzureResource, CloudResource, GCPResource, create_cloud_resource,
)
# ── Given steps ───────────────────────────────────────────────
@given('a CloudResource is instantiated with resource_id="{resource_id}"')
def step_cloud_resource_init(context: Context, resource_id: str) -> None: # type: ignore[name-defined]
"""Store kwargs for instantiation."""
context._cr_kwargs = {"resource_id": resource_id} # type: ignore[attr-defined]
def step_cr_init(ctx: Context, resource_id: str) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": resource_id} # type: ignore[attr-defined]
@given('a CloudResource is requested with an empty resource_id')
def step_cloud_resource_empty_id(context: Context) -> None: # type: ignore[name-defined]
"""Store empty resource_id for instantiation attempt."""
context._cr_kwargs = {"resource_id": ""} # type: ignore[attr-defined]
@given("a CloudResource is requested with an empty resource_id")
def step_cr_empty(ctx: Context) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": ""} # type: ignore[attr-defined]
@given(
'an AWSResource is instantiated with type "{res_type}"',
)
def step_aws_resource_init(context: Context, res_type: str) -> None: # type: ignore[name-defined]
"""Store kwargs for AWSResource instantiation."""
context._cr_kwargs = { # type: ignore[attr-defined]
"resource_id": "r-001",
"region": "us-east-1",
"resource_type": res_type,
}
@given('an AWSResource is created with ARN "{arn}" in region "{region}"')
def step_aws_s3(ctx: Context, arn: str, region: str) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": arn, "region": region} # type: ignore[attr-defined]
@given(
'a GCPResource is instantiated with type "{res_type}"',
)
def step_gcp_resource_init(context: Context, res_type: str) -> None: # type: ignore[name-defined]
"""Store kwargs for GCPResource instantiation."""
context._cr_kwargs = { # type: ignore[attr-defined]
"resource_id": "r-001",
"region": "us-central1",
"project_id": "test-project",
"resource_type": res_type,
}
@given('an AWSResource is created with ID "{eid}" in region "{region}" and type "{res_type}"')
def step_aws_ec2(ctx: Context, eid: str, region: str, res_type: str) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": eid, "region": region, "resource_type": res_type} # type: ignore[attr-defined]
@given(
'an AzureResource is instantiated with type "{res_type}"',
)
def step_azure_resource_init(context: Context, res_type: str) -> None: # type: ignore[name-defined]
"""Store kwargs for AzureResource instantiation."""
context._cr_kwargs = { # type: ignore[attr-defined]
"resource_id": "r-001",
"region": "eastus",
"subscription_id": "sub-001",
"resource_type": res_type,
}
@given('an AWSResource with resource_id="{rid}" and region="{region}" is requested for type "{res_type}"')
def step_aws_invalid(ctx: Context, rid: str, region: str, res_type: str) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": rid, "region": region, "resource_type": res_type} # type: ignore[attr-defined]
@given(
'an AWSResource is created with ARN "{arn}" in region "{region}"',
)
def step_aws_s3_resource(context: Context, arn: str, region: str) -> None: # type: ignore[name-defined]
"""Create an S3 AWS resource."""
context._cr_kwargs = { "resource_id": arn, "region": region } # type: ignore[attr-defined]
@given('a GCPResource with resource_id="{rid}", region="{region}" and project_id="{pid}" is created')
def step_gcs(ctx: Context, rid: str, region: str, pid: str) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": rid, "region": region, "project_id": pid} # type: ignore[attr-defined]
@given(
'an AWSResource is created with ID "{eid}" in region "{region}" and type "{res_type}"',
)
def step_aws_ec2_resource(context: Context, eid: str, region: str, res_type: str) -> None: # type: ignore[name-defined]
"""Create an EC2 AWS resource."""
context._cr_kwargs = { "resource_id": eid, "region": region, "resource_type": res_type } # type: ignore[attr-defined]
@given('a GCPResource with resource_id="{rid}", region="{region}", project_id="{pid}" and type "{res_type}" is created')
def step_gcp_compute(ctx: Context, rid: str, region: str, pid: str, res_type: str) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": rid, "region": region, "project_id": pid, "resource_type": res_type} # type: ignore[attr-defined]
@given(
'an AWSResource with resource_id="{rid}" and region="{region}" is requested for type "{res_type}"',
)
def step_aws_invalid_type(context: Context, rid: str, region: str, res_type: str) -> None: # type: ignore[name-defined]
"""Request an invalid AWS resource type."""
context._cr_kwargs = { "resource_id": rid, "region": region, "resource_type": res_type } # type: ignore[attr-defined]
@given('a GCPResource with resource_id="{rid}" is requested without project_id')
def step_gcp_missing(ctx: Context, rid: str) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": rid} # type: ignore[attr-defined]
@given(
'a GCPResource with resource_id="{rid}", region="{region}" and project_id="{pid}" is created',
)
def step_gcp_gcs_resource(context: Context, rid: str, region: str, pid: str) -> None: # type: ignore[name-defined]
"""Create a GCS bucket GCP resource."""
context._cr_kwargs = { "resource_id": rid, "region": region, "project_id": pid } # type: ignore[attr-defined]
@given('an AzureResource with resource_id="{rid}", region="{region}" and subscription_id="{sid}" is created')
def step_azure_blob(ctx: Context, rid: str, region: str, sid: str) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": rid, "region": region, "subscription_id": sid} # type: ignore[attr-defined]
@given(
'a GCPResource with resource_id="{rid}", region="{region}", project_id="{pid}" and type "{res_type}" is created',
)
def step_gcp_compute_resource(context: Context, rid: str, region: str, pid: str, res_type: str) -> None: # type: ignore[name-defined]
"""Create a Compute Engine GCP resource."""
context._cr_kwargs = { "resource_id": rid, "region": region, "project_id": pid, "resource_type": res_type } # type: ignore[attr-defined]
@given('an AzureResource with resource_id="{rid}", region="{region}", subscription_id="{sid}" and type "{res_type}" is created')
def step_azure_vm(ctx: Context, rid: str, region: str, sid: str, res_type: str) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": rid, "region": region, "subscription_id": sid, "resource_type": res_type} # type: ignore[attr-defined]
@given(
'a GCPResource with resource_id="{rid}" is requested without project_id',
)
def step_gcp_missing_project(context: Context, rid: str) -> None: # type: ignore[name-defined]
"""Request GCP resource without project_id."""
context._cr_kwargs = { "resource_id": rid } # type: ignore[attr-defined]
@given('an AzureResource with resource_id="{rid}" is requested without subscription_id')
def step_azure_missing(ctx: Context, rid: str) -> None: # type: ignore[name-defined]
ctx._cr_kwargs = {"resource_id": rid} # type: ignore[attr-defined]
@given(
'an AzureResource with resource_id="{rid}", region="{region}" and subscription_id="{sid}" is created',
)
def step_azure_blob_resource(context: Context, rid: str, region: str, sid: str) -> None: # type: ignore[name-defined]
"""Create a blob storage Azure resource."""
context._cr_kwargs = { "resource_id": rid, "region": region, "subscription_id": sid } # type: ignore[attr-defined]
@given('create_cloud_resource is called with name="{name}", resource_id="{rid}" and region="{region}"')
def step_factory_aws(ctx: Context, name: str, rid: str, region: str) -> None: # type: ignore[name-defined]
ctx._fk = {"name": name, "resource_id": rid, "region": region} # type: ignore[attr-defined]
@given(
'an AzureResource with resource_id="{rid}", region="{region}", subscription_id="{sid}" and type "{res_type}" is created',
)
def step_azure_vm_resource(context: Context, rid: str, region: str, sid: str, res_type: str) -> None: # type: ignore[name-defined]
"""Create a VM Azure resource."""
context._cr_kwargs = { "resource_id": rid, "region": region, "subscription_id": sid, "resource_type": res_type } # type: ignore[attr-defined]
@given('create_cloud_resource is called with name="{name}", resource_id="{rid}", region="{region}" and project_id="{pid}"')
def step_factory_gcp(ctx: Context, name: str, rid: str, region: str, pid: str) -> None: # type: ignore[name-defined]
ctx._fk = {"name": name, "resource_id": rid, "region": region, "project_id": pid} # type: ignore[attr-defined]
@given(
'an AzureResource with resource_id="{rid}" is requested without subscription_id',
)
def step_azure_missing_sub(context: Context, rid: str) -> None: # type: ignore[name-defined]
"""Request Azure resource without subscription_id."""
context._cr_kwargs = { "resource_id": rid } # type: ignore[attr-defined]
@given('create_cloud_resource is called with name="{name}", resource_id="{rid}", region="{region}" and subscription_id="{sid}"')
def step_factory_azure(ctx: Context, name: str, rid: str, region: str, sid: str) -> None: # type: ignore[name-defined]
ctx._fk = {"name": name, "resource_id": rid, "region": region, "subscription_id": sid} # type: ignore[attr-defined]
@given(
'create_cloud_resource is called with name="{name}", resource_id="{rid}" and region="{region}"',
)
def step_factory_aws(context: Context, name: str, rid: str, region: str) -> None: # type: ignore[name-defined]
"""Store factory call parameters for AWS."""
context._factory_kwargs = { "name": name, "resource_id": rid, "region": region } # type: ignore[attr-defined]
@given(
'create_cloud_resource is called with name="{name}", resource_id="{rid}", region="{region}" and project_id="{pid}"',
)
def step_factory_gcp(context: Context, name: str, rid: str, region: str, pid: str) -> None: # type: ignore[name-defined]
"""Store factory call parameters for GCP."""
context._factory_kwargs = { "name": name, "resource_id": rid, "region": region, "project_id": pid } # type: ignore[attr-defined]
@given(
'create_cloud_resource is called with name="{name}", resource_id="{rid}", region="{region}" and subscription_id="{sid}"',
)
def step_factory_azure(context: Context, name: str, rid: str, region: str, sid: str) -> None: # type: ignore[name-defined]
"""Store factory call parameters for Azure."""
context._factory_kwargs = { "name": name, "resource_id": rid, "region": region, "subscription_id": sid } # type: ignore[attr-defined]
@given(
'create_cloud_resource is called with name="{name}" and resource_id="{rid}"',
)
def step_factory_unknown(context: Context, name: str, rid: str) -> None: # type: ignore[name-defined]
"""Store factory call parameters for unknown provider."""
context._factory_kwargs = { "name": name, "resource_id": rid } # type: ignore[attr-defined]
@given('create_cloud_resource is called with name="{name}" and resource_id="{rid}"')
def step_factory_unknown(ctx: Context, name: str, rid: str) -> None: # type: ignore[name-defined]
ctx._fk = {"name": name, "resource_id": rid} # type: ignore[attr-defined]
# ── When steps ───────────────────────────────────────────────
@when("the resource has been created")
def step_instantiate_resource(context: Context) -> None: # type: ignore[name-defined]
"""Instantiate cloud resource from stored kwargs."""
kw = getattr(context, "_cr_kwargs", {}) # type: ignore[attr-defined]
def step_create(ctx: Context) -> None: # type: ignore[name-defined]
kw = getattr(ctx, "_cr_kwargs", {}) # type: ignore[attr-defined]
try:
context._created_resource = CloudResource(**kw) # type: ignore[attr-defined]
context.cr_error = None # type: ignore[attr-defined]
ctx._res = CloudResource(**kw) # type: ignore[attr-defined]
ctx.cr_error = None # type: ignore[attr-defined]
except (ValueError, TypeError) as exc:
context.cr_error = exc # type: ignore[attr-defined]
ctx.cr_error = exc # type: ignore[attr-defined]
@when("instantiation is attempted")
def step_instantiate_attempt(context: Context) -> None: # type: ignore[name-defined]
"""Attempt instantiation from stored kwargs."""
kw = getattr(context, "_cr_kwargs", {}) # type: ignore[attr-defined]
def step_attempt(ctx: Context) -> None: # type: ignore[name-defined]
kw = getattr(ctx, "_cr_kwargs", {}) # type: ignore[attr-defined]
try:
# Pick the right subclass based on what kwargs are present.
if "subscription_id" in kw:
cls = AzureResource
elif "project_id" in kw:
cls = GCPResource
else:
cls = AWSResource
context._created_resource = cls(**kw) # type: ignore[attr-defined]
context.cr_error = None # type: ignore[attr-defined]
ctx._res = cls(**kw) # type: ignore[attr-defined]
ctx.cr_error = None # type: ignore[attr-defined]
except (ValueError, TypeError) as exc:
context.cr_error = exc # type: ignore[attr-defined]
ctx.cr_error = exc # type: ignore[attr-defined]
@when("a cloud resource is created via factory")
def step_factory_create(context: Context) -> None: # type: ignore[name-defined]
"""Call create_cloud_resource with stored kwargs."""
fk = getattr(context, "_factory_kwargs", {}) # type: ignore[attr-defined]
def step_factory_create(ctx: Context) -> None: # type: ignore[name-defined]
fk = getattr(ctx, "_fk", {}) # type: ignore[attr-defined]
try:
name = fk.pop("name") # type: ignore[union-attr]
context._created_resource = create_cloud_resource(name, **fk) # type: ignore[attr-defined]
context.cr_error = None # type: ignore[attr-defined]
ctx._res = create_cloud_resource(name, **fk) # type: ignore[attr-defined]
ctx.cr_error = None # type: ignore[attr-defined]
except (ValueError, KeyError, TypeError) as exc:
context.cr_error = exc # type: ignore[attr-defined]
ctx.cr_error = exc # type: ignore[attr-defined]
# ── Then steps ───────────────────────────────────────────────
@then('it should have provider "{expected}"')
def step_provider_check(context: Context, expected: str) -> None: # type: ignore[name-defined]
"""Verify resource provider."""
res = context._created_resource # type: ignore[attr-defined]
assert res.provider == expected, f"Expected provider '{expected}', got '{res.provider}'"
def step_provider(ctx: Context, expected: str) -> None: # type: ignore[name-defined]
assert getattr(ctx._res, "provider") == expected, f"Expected '{expected}', got {getattr(ctx._res, 'provider')}"
@then('it should have resource_id "{expected}"')
def step_resource_id_check(context: Context, expected: str) -> None: # type: ignore[name-defined]
"""Verify resource_id."""
res = context._created_resource # type: ignore[attr-defined]
assert res.resource_id == expected, f"Expected id '{expected}', got '{res.resource_id}'"
def step_resource_id(ctx: Context, expected: str) -> None: # type: ignore[name-defined]
assert getattr(ctx._res, "resource_id") == expected
@then('display_name should be "{expected}"')
def step_display_name(context: Context, expected: str) -> None: # type: ignore[name-defined]
"""Verify display_name."""
res = context._created_resource # type: ignore[attr-defined]
assert res.display_name == expected, f"Expected '{expected}', got '{res.display_name}'"
@then('display_name should match pattern "{pattern}"')
def step_display_name_pattern(context: Context, pattern: str) -> None: # type: ignore[name-defined]
"""Verify display_name matches regex pattern."""
res = context._created_resource # type: ignore[attr-defined]
assert re.search(pattern, res.display_name), ( # type: ignore[union-attr]
f"display_name '{res.display_name}' does not match '{pattern}'"
)
@then('it should have resource_type "{expected}"')
def step_resource_type_check(context: Context, expected: str) -> None: # type: ignore[name-defined]
"""Verify resource_type."""
res = context._created_resource # type: ignore[attr-defined]
assert res.resource_type == expected, f"Expected type '{expected}', got '{res.resource_type}'"
@then("provider should be 'aws'")
def step_provider_is_aws(context: Context) -> None: # type: ignore[name-defined]
"""Verify provider is aws."""
res = context._created_resource # type: ignore[attr-defined]
assert res.provider == "aws", f"Expected provider 'aws', got '{res.provider}'"
def step_display_name(ctx: Context, expected: str) -> None: # type: ignore[name-defined]
r = ctx._res # type: ignore[attr-defined]
assert r.display_name == expected, f"Expected '{expected}', got '{r.display_name}'"
@then("it should be a provider-specific subclass of CloudResource")
def step_is_cloud_subclass(context: Context) -> None: # type: ignore[name-defined]
"""Verify resource is a CloudResource subclass."""
res = context._created_resource # type: ignore[attr-defined]
assert isinstance(res, CloudResource), f"Expected CloudResource subclass, got {type(res)}"
def step_subclass(ctx: Context) -> None: # type: ignore[name-defined]
ctx._res = getattr(ctx, "_res", None) # type: ignore[attr-defined]
assert isinstance(ctx._res, CloudResource)
@then("it should have provider 'aws'")
def step_provider_is_aws_short(context: Context) -> None: # type: ignore[name-defined]
"""Shorthand for AWS provider check."""
res = context._created_resource # type: ignore[attr-defined]
assert res.provider == "aws", f"Expected 'aws', got '{res.provider}'"
@then("region should be non-empty string")
def step_region_is_string(context: Context) -> None: # type: ignore[name-defined]
"""Verify region is a non-empty string."""
res = context._created_resource # type: ignore[attr-defined]
assert isinstance(res.region, str) and len(res.region) > 0, (
f"Expected non-empty string region, got '{res.region}'"
)
@then("subscription_id should be set")
def step_subscription_set(context: Context) -> None: # type: ignore[name-defined]
"""Verify subscription_id attribute is present."""
res = context._created_resource # type: ignore[attr-defined]
assert hasattr(res, "subscription_id"), "Resource has no 'subscription_id' attribute"
assert res.subscription_id, "subscription_id is empty"
@then('bucket_name should be "{expected}"')
def step_bucket_name(ctx: Context, expected: str) -> None: # type: ignore[name-defined]
r = ctx._res # type: ignore[attr-defined]
actual = getattr(r, "bucket_name", None)
assert actual == expected, f"Expected '{expected}', got '{actual}'"
@then('instance_id should be "{expected}"')
def step_instance_id(context: Context, expected: str) -> None: # type: ignore[name-defined]
"""Verify instance_id."""
res = context._created_resource # type: ignore[attr-defined]
assert res.instance_id == expected, f"Expected '{expected}', got '{res.instance_id}'"
def step_instance_id(ctx: Context, expected: str) -> None: # type: ignore[name-defined]
r = ctx._res # type: ignore[attr-defined]
actual = getattr(r, "instance_id", None)
assert actual == expected
@then('project_id should be "{expected}"')
def step_project_id(context: Context, expected: str) -> None: # type: ignore[name-defined]
"""Verify project_id."""
res = context._created_resource # type: ignore[attr-defined]
assert res.project_id == expected, f"Expected '{expected}', got '{res.project_id}'"
@then("bucket_name should be 'my-gcs-bucket'")
def step_bucket_name(context: Context) -> None: # type: ignore[name-defined]
"""Verify bucket_name."""
res = context._created_resource # type: ignore[attr-defined]
assert res.bucket_name == "my-gcs-bucket", f"Expected 'my-gcs-bucket', got '{res.bucket_name}'"
def step_project_id(ctx: Context, expected: str) -> None: # type: ignore[name-defined]
r = ctx._res # type: ignore[attr-defined]
assert getattr(r, "project_id") == expected
@then('container_name should be "{expected}"')
def step_container_name(context: Context, expected: str) -> None: # type: ignore[name-defined]
"""Verify container_name."""
res = context._created_resource # type: ignore[attr-defined]
assert res.container_name == expected, f"Expected '{expected}', got '{res.container_name}'"
def step_container_name(ctx: Context, expected: str) -> None: # type: ignore[name-defined]
r = ctx._res # type: ignore[attr-defined]
actual = getattr(r, "container_name", None)
assert actual == expected
@then('subscription_id should be "{expected}"')
def step_subscription_id(context: Context, expected: str) -> None: # type: ignore[name-defined]
"""Verify subscription_id."""
res = context._created_resource # type: ignore[attr-defined]
assert res.subscription_id == expected, f"Expected '{expected}', got '{res.subscription_id}'"
def step_sub_id(ctx: Context, expected: str) -> None: # type: ignore[name-defined]
r = ctx._res # type: ignore[attr-defined]
assert getattr(r, "subscription_id") == expected
@then("a ValueError should be raised")
def step_value_error(context: Context) -> None: # type: ignore[name-defined]
"""Verify a ValueError was caught."""
err = getattr(context, "cr_error", None) # type: ignore[attr-defined]
assert err is not None and isinstance(err, ValueError), (
f"Expected ValueError, got {type(err).__name__}: {err}"
)
def step_value_error(ctx: Context) -> None: # type: ignore[name-defined]
err = getattr(ctx, "cr_error", None)
assert err is not None and isinstance(err, ValueError), f"Expected ValueError, got {type(err).__name__}: {err}"
@then("the result should be an AWSResource instance")
def step_result_is_aws(context: Context) -> None: # type: ignore[name-defined]
"""Verify factory result is AWSResource."""
res = context._created_resource # type: ignore[attr-defined]
assert isinstance(res, AWSResource), f"Expected AWSResource, got {type(res)}"
def step_res_aws(ctx: Context) -> None: # type: ignore[name-defined]
assert isinstance(getattr(ctx, "_res"), AWSResource)
@then("the result should be a GCPResource instance")
def step_result_is_gcp(context: Context) -> None: # type: ignore[name-defined]
"""Verify factory result is GCPResource."""
res = context._created_resource # type: ignore[attr-defined]
assert isinstance(res, GCPResource), f"Expected GCPResource, got {type(res)}"
def step_res_gcp(ctx: Context) -> None: # type: ignore[name-defined]
assert isinstance(getattr(ctx, "_res", None), GCPResource)
@then("the result should be an AzureResource instance")
def step_result_is_azure(context: Context) -> None: # type: ignore[name-defined]
"""Verify factory result is AzureResource."""
res = context._created_resource # type: ignore[attr-defined]
assert isinstance(res, AzureResource), f"Expected AzureResource, got {type(res)}"
def step_res_azure(ctx: Context) -> None: # type: ignore[name-defined]
assert isinstance(getattr(ctx, "_res"), AzureResource)
@then("the cloud resource registry is queried")
def step_query_registry(context: Context) -> None: # type: ignore[name-defined]
"""Query the cloud resource registry."""
@when("the cloud resource registry is queried")
def step_query(ctx: Context) -> None: # type: ignore[name-defined]
from cleveragents.resource.handlers.cloud_models import CLOUD_RESOURCE_REGISTRY # noqa: E402
context._registry = CLOUD_RESOURCE_REGISTRY # type: ignore[attr-defined]
ctx._registry = CLOUD_RESOURCE_REGISTRY # type: ignore[attr-defined]
@then('it should contain "aws" provider')
def step_registry_has_aws(context: Context) -> None: # type: ignore[name-defined]
"""Verify registry has aws."""
reg = context._registry # type: ignore[attr-defined]
assert "aws" in reg, f"'aws' not in registry keys: {list(reg.keys())}"
def step_reg_aws(ctx: Context) -> None: # type: ignore[name-defined]
assert "aws" in getattr(ctx, "_registry", {})
@then('it should contain "gcp" provider')
def step_registry_has_gcp(context: Context) -> None: # type: ignore[name-defined]
"""Verify registry has gcp."""
reg = context._registry # type: ignore[attr-defined]
assert "gcp" in reg, f"'gcp' not in registry keys: {list(reg.keys())}"
def step_reg_gcp(ctx: Context) -> None: # type: ignore[name-induced]
assert "gcp" in getattr(ctx, "_registry", {})
@then('it should contain "azure" provider')
def step_registry_has_azure(context: Context) -> None: # type: ignore[name-defined]
"""Verify registry has azure."""
reg = context._registry # type: ignore[attr-defined]
assert "azure" in reg, f"'azure' not in registry keys: {list(reg.keys())}"
@then("it should contain 'azure' provider")
def step_reg_azure(ctx: Context) -> None: # type: ignore[name-defined]
assert "azure" in getattr(ctx, "_registry", {})
@@ -20,8 +20,7 @@ Based on:
from __future__ import annotations
import re
from typing import Any
from typing import Any, ClassVar
# ---------------------------------------------------------------------------
@@ -29,38 +28,19 @@ from typing import Any
# ---------------------------------------------------------------------------
#: AWS resource types supported by this stub implementation.
AWS_RESOURCE_TYPES: frozenset[str] = frozenset(
{
"s3-bucket",
"ec2-instance",
# Reserved for future expansion, NOT active yet:
# "vpc", "security-group", "iam-role", "lambda-function",
# "rds-instance", "dynamodb-table", "sns-topic", "sqs-queue",
}
)
AWS_RESOURCE_TYPES: frozenset[str] = frozenset({
"s3-bucket", "ec2-instance",
})
#: GCP resource types supported by this stub implementation.
GCP_RESOURCE_TYPES: frozenset[str] = frozenset(
{
"gcs-bucket",
"compute-instance",
# Reserved for future expansion, NOT active yet:
# "firewall-rule", "vpc-network", "subnet-ip-range",
# "cloud-sql-instance", "bigquery-dataset", "pubsub-topic",
}
)
GCP_RESOURCE_TYPES: frozenset[str] = frozenset({
"gcs-bucket", "compute-instance",
})
#: Azure resource types supported by this stub implementation.
AZURE_RESOURCE_TYPES: frozenset[str] = frozenset(
{
"blob-storage-container",
"virtual-machine",
# Reserved for future expansion, NOT active yet:
# "managed-disk", "network-interface", "virtual-network",
# "sql-server-database", "key-vault", "storage-account",
# "container-registry", "app-service-plan", "function-app",
}
)
AZURE_RESOURCE_TYPES: frozenset[str] = frozenset({
"blob-storage-container", "virtual-machine",
})
# ---------------------------------------------------------------------------
@@ -72,48 +52,19 @@ class CloudResource:
"""Base class for all cloud infrastructure resources.
Defines the common fields shared across all cloud providers:
resource_id, provider, region, and resource_type with
validation.
Args:
resource_id: Unique identifier for this resource instance
(e.g., AWS ARN, GCP resource path, Azure ARM ID fragment).
provider: Cloud provider name — ``"aws"``, ``"gcp"``, or
``"azure"``. Must match the subclass type.
region: Target cloud region (e.g., ``"us-east-1"``,
``"eu-west-2"``). May be empty for provider-level
resources.
resource_type: High-level classification of the resource
within this provider (e.g., ``"s3-bucket"``,
``"gcs-bucket"``, ``"virtual-machine"``).
properties: Arbitrary additional configuration metadata.
Raises:
ValueError: If any field fails validation.
Examples:
>>> cr = CloudResource(resource_id="r-123", provider="aws") # base
>>> cr.provider
'base'
resource_id, provider, region, and resource_type with validation.
"""
__slots__ = (
"resource_id",
"region",
"resource_type",
"properties",
)
__class__: ClassVar[str] = "base" #: Override in subclasses.
__class__: ClassVar[str] = "base" # Provider marker, overridden in subclasses.
def __init__(
self,
resource_id: str,
resource_id: str = "",
*,
provider: str = "base",
region: str = "",
resource_type: str = "",
properties: dict[str, Any] | None = None,
/,
**extra: Any, # Accept and store extra keyword args.
) -> None:
if not resource_id.strip():
@@ -132,8 +83,7 @@ class CloudResource:
self.provider = provider
self.region = region
self.resource_type = resource_type
self.properties: dict[str, Any] = properties or {}
# Store any extra kwargs.
self.properties: dict[str, Any] = dict(properties) if properties else {}
for _k, _v in extra.items():
setattr(self, _k, _v)
@@ -184,66 +134,34 @@ class CloudResource:
class AWSResource(CloudResource):
"""Stub class for AWS cloud infrastructure resources.
Supports:
- **S3 buckets** (``s3-bucket``) — object storage
- **EC2 instances** (``ec2-instance``) — compute VMs
Args:
resource_id: AWS ARN or short-form identifier.
region: AWS region code (e.g. ``"us-east-1"``, ``"eu-west-3"``).
resource_type: Either ``"s3-bucket"`` or ``"ec2-instance"``.
properties: Optional extra configuration metadata.
Raises:
ValueError: If fields are invalid.
Examples:
>>> aws_s3 = AWSResource(
... resource_id="arn:aws:s3:::my-bucket", region="us-east-1")
>>> aws_ec2 = AWSResource(
... resource_id="i-0abcdef1234567890", region="eu-west-2",
... resource_type="ec2-instance")
Supports S3 buckets (s3-bucket) and EC2 instances (ec2-instance).
"""
__class__ = "aws" # Override provider marker.
__class__: ClassVar[str] = "aws"
_VALID_RESOURCE_TYPES: ClassVar[frozenset[str]] = AWS_RESOURCE_TYPES
def __init__(
self,
resource_id: str,
region: str = "",
/,
*,
resource_type: str = "s3-bucket",
properties: dict[str, Any] | None = None,
**extra: Any,
) -> None:
def __init__(self, resource_id: str = "", *, region: str = "",
resource_type: str = "s3-bucket",
properties: dict[str, Any] | None = None, **extra: Any) -> None: # noqa: E501
super().__init__(
resource_id=resource_id,
provider="aws",
region=region,
resource_type=resource_type,
properties=properties or {},
resource_id=resource_id, provider="aws", region=region,
resource_type=resource_type, properties=properties or {},
**extra,
)
self._resolve_extra_fields()
def _resolve_extra_fields(self) -> None:
"""Derive helper fields from ARN pattern when possible."""
if (
self.resource_type == "s3-bucket"
and self.resource_id.startswith("arn:aws:s3:::")
):
object.__setattr__(self, "_bucket_name", self.resource_id[14:])
"""Derive helper fields from ARN/ID patterns when possible."""
if (self.resource_type == "s3-bucket"
and self.resource_id.startswith("arn:aws:s3:::")):
# "arn:aws:s3:::" is 13 characters
object.__setattr__(self, "_bucket_name", self.resource_id[13:])
else:
object.__setattr__(self, "_bucket_name", None)
@property
def bucket_name(self) -> str | None:
"""S3 bucket name extracted from ARN (if resource is S3).
Returns ``None`` for non-S3 resources.
"""
"""S3 bucket name extracted from ARN (if resource is S3)."""
if self.resource_type != "s3-bucket":
return None
return getattr(self, "_bucket_name", None)
@@ -264,53 +182,20 @@ class AWSResource(CloudResource):
class GCPResource(CloudResource):
"""Stub class for GCP cloud infrastructure resources.
Supports:
- **GCS buckets** (``gcs-bucket``) — object storage
- **Compute Engines** (``compute-instance``) — compute VMs
Args:
resource_id: GCP resource path or ID.
region: GCP region code (e.g. ``"us-central1"``, ``"europe-west4"``).
project_id: GCP project identifier.
resource_type: Either ``"gcs-bucket"`` or ``"compute-instance"``.
properties: Optional extra configuration metadata.
Raises:
ValueError: If fields are invalid.
Examples:
>>> gcp_bucket = GCPResource(
... resource_id="my-gcs-bucket", region="us-central1",
... project_id="acme-project")
>>> gcp_vm = GCPResource(
... resource_id="projects/acme/zones/us-central1-a/instances/web-1",
... project_id="acme-project", resource_type="compute-instance")
Supports GCS buckets (gcs-bucket) and Compute Engines (compute-instance).
"""
__class__ = "gcp"
__class__: ClassVar[str] = "gcp"
_VALID_RESOURCE_TYPES: ClassVar[frozenset[str]] = GCP_RESOURCE_TYPES
def __init__(
self,
resource_id: str,
region: str = "",
/,
*,
project_id: str = "",
resource_type: str = "gcs-bucket",
properties: dict[str, Any] | None = None,
**extra: Any,
) -> None:
def __init__(self, resource_id: str = "", *, region: str = "",
project_id: str = "", resource_type: str = "gcs-bucket",
properties: dict[str, Any] | None = None, **extra: Any) -> None: # noqa: E501
if not project_id.strip():
raise ValueError(
"GCP resource requires a non-empty 'project_id'."
)
raise ValueError("GCP resource requires a non-empty 'project_id'.")
super().__init__(
resource_id=resource_id,
provider="gcp",
region=region,
resource_type=resource_type,
properties=properties or {},
resource_id=resource_id, provider="gcp", region=region,
resource_type=resource_type, properties=properties or {},
**extra,
)
object.__setattr__(self, "project_id", project_id)
@@ -331,54 +216,24 @@ class GCPResource(CloudResource):
class AzureResource(CloudResource):
"""Stub class for Azure cloud infrastructure resources.
Supports:
- **Blob Storage** (``blob-storage-container``) — object storage
- **Virtual Machines** (``virtual-machine``) — compute
Args:
resource_id: Azure ARM resource ID fragment or name.
region: Azure region code (e.g. ``"eastus"``, ``"westeurope"``).
subscription_id: Azure subscription identifier.
resource_type: Either ``"blob-storage-container"`` or
``"virtual-machine"``.
properties: Optional extra configuration metadata.
Raises:
ValueError: If fields are invalid.
Examples:
>>> azure_blob = AzureResource(
... resource_id="mycontainer", region="eastus",
... subscription_id="sub-12345")
>>> azure_vm = AzureResource(
... resource_id="my-vm", region="westeurope",
... subscription_id="sub-12345", resource_type="virtual-machine")
Supports Blob Storage (blob-storage-container) and Virtual Machines
(virtual-machine).
"""
__class__ = "azure"
__class__: ClassVar[str] = "azure"
_VALID_RESOURCE_TYPES: ClassVar[frozenset[str]] = AZURE_RESOURCE_TYPES
def __init__(
self,
resource_id: str,
region: str = "",
/,
*,
subscription_id: str = "",
resource_type: str = "blob-storage-container",
properties: dict[str, Any] | None = None,
**extra: Any,
) -> None:
def __init__(self, resource_id: str = "", *, region: str = "",
subscription_id: str = "",
resource_type: str = "blob-storage-container",
properties: dict[str, Any] | None = None, **extra: Any) -> None: # noqa: E501
if not subscription_id.strip():
raise ValueError(
"Azure resource requires a non-empty 'subscription_id'."
)
super().__init__(
resource_id=resource_id,
provider="azure",
region=region,
resource_type=resource_type,
properties=properties or {},
resource_id=resource_id, provider="azure", region=region,
resource_type=resource_type, properties=properties or {},
**extra,
)
object.__setattr__(self, "subscription_id", subscription_id)
@@ -392,7 +247,7 @@ class AzureResource(CloudResource):
# ---------------------------------------------------------------------------
# Factory helper
# Factory helper and registry
# ---------------------------------------------------------------------------
#: Type registry mapping provider keys to their resource subclass.
@@ -404,25 +259,18 @@ CLOUD_RESOURCE_REGISTRY: dict[str, type[CloudResource]] = {
def create_cloud_resource(
name: str,
/,
*,
resource_id: str,
region: str = "",
properties: dict[str, Any] | None = None,
**provider_kwargs: Any,
name: str, *, resource_id: str, region: str = "",
properties: dict[str, Any] | None = None, **provider_kwargs: Any,
) -> CloudResource:
"""Create a cloud resource instance by provider short-name.
Args:
name: Provider key ``"aws"``, ``"gcp"``, or ``"azure"``.
resource_id: Unique resource identifier.
region: Cloud region code.
provider_kwargs: Provider-specific keyword arguments (see subclass
constructors for details).
name: Provider key -- ``"aws"``, ``"gcp"``, or ``"azure"``.
resource_id: Unique resource identifier. region: Cloud region code.
properties: Optional config metadata passed tosubclass.
provider_kwargs: Provider-specific keyword arguments (see subclasses).
Returns:
An instance of the appropriate :class:`CloudResource` subclass.
Returns:class:`CloudResource` subclass instance.
Raises:
ValueError: If *name* is not a registered provider.
@@ -439,4 +287,5 @@ def create_cloud_resource(
f"Unknown cloud provider '{name}'. "
f"Available: {available}."
)
return cls(resource_id=resource_id, region=region, properties=properties, **provider_kwargs)
return cls(resource_id=resource_id, region=region,
properties=properties or {}, **provider_kwargs)