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
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:
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user