feat: add support for cloud spec

README.md

@@ -948,6 +948,46 @@ assert out.model.name == "my-model"
 assert out.model.uuid == state_in.model.uuid
 ```
 
+## CloudSpec
+
+You can set CloudSpec information in the state (only `type` and `name` are required).
+
+Example:
+
+```python
+import scenario
+
+state = scenario.State(
+    cloud_spec=scenario.CloudSpec(
+        type="lxd",
+        name="localhost",
+        endpoint="https://127.0.0.1:8443",
+        credential=scenario.CloudCredential(
+            auth_type="clientcertificate",
+            attributes={
+                "client-cert": "foo",
+                "client-key": "bar",
+                "server-cert": "baz",
+            },
+        ),
+    ),
+    model=scenario.Model(name="my-vm-model", type="lxd"),
+)
+```
+
+Then you can access it by `Model.get_cloud_spec()`:
+
+```python
+# charm.py
+class MyVMCharm(ops.CharmBase):
+    def __init__(self, framework: ops.Framework):
+        super().__init__(framework)
+        framework.observe(self.on.start, self._on_start)
+
+    def _on_start(self, event: ops.StartEvent):
+        self.cloud_spec = self.model.get_cloud_spec()
+```
+
 # Actions
 
 An action is a special sort of event, even though `ops` handles them almost identically.

scenario/__init__.py

@@ -6,6 +6,8 @@
     Action,
     Address,
     BindAddress,
+    CloudCredential,
+    CloudSpec,
     Container,
     DeferredEvent,
     Event,
@@ -29,6 +31,8 @@
 __all__ = [
     "Action",
     "ActionOutput",
+    "CloudCredential",
+    "CloudSpec",
     "Context",
     "deferred",
     "StateValidationError",

scenario/consistency_checker.py

@@ -70,6 +70,7 @@ def check_consistency(
         check_storages_consistency,
         check_relation_consistency,
         check_network_consistency,
+        check_cloudspec_consistency,
     ):
         results = check(
             state=state,
@@ -565,3 +566,23 @@ def check_containers_consistency(
         errors.append(f"Duplicate container name(s): {dupes}.")
 
     return Results(errors, [])
+
+
+def check_cloudspec_consistency(
+    *,
+    state: "State",
+    event: "Event",
+    charm_spec: "_CharmSpec",
+    **_kwargs,  # noqa: U101
+) -> Results:
+    """Check that Kubernetes charms/models don't have `state.cloud_spec`."""
+
+    errors = []
+    warnings = []
+
+    if state.model.type == "kubernetes" and state.cloud_spec:
+        errors.append(
+            "CloudSpec is only available for machine charms, not Kubernetes charms.",
+        )
+
+    return Results(errors, warnings)

scenario/mocking.py

@@ -20,7 +20,7 @@
     cast,
 )
 
-from ops import JujuVersion, pebble
+from ops import CloudSpec, JujuVersion, pebble
 from ops.model import ModelError, RelationNotFoundError
 from ops.model import Secret as Secret_Ops  # lol
 from ops.model import (
@@ -631,6 +631,13 @@ def resource_get(self, resource_name: str) -> str:
                 f"resource {resource_name} not found in State. please pass it.",
             )
 
+    def credential_get(self) -> CloudSpec:
+        if not self._state.cloud_spec:
+            raise ModelError(
+                "ERROR cloud spec is empty, initialise it with `scenario.State(cloud_spec=scenario.CloudSpec(...))`",
+            )
+        return self._state.cloud_spec._to_ops()
+
 
 class _MockPebbleClient(_TestingPebbleClient):
     def __init__(

scenario/state.py

@@ -29,6 +29,7 @@
 )
 from uuid import uuid4
 
+import ops
 import yaml
 from ops import pebble
 from ops.charm import CharmBase, CharmEvents
@@ -140,6 +141,77 @@ def copy(self) -> "Self":
         return copy.deepcopy(self)
 
 
+@dataclasses.dataclass(frozen=True)
+class CloudCredential:
+    auth_type: str
+    """Authentication type."""
+
+    attributes: Dict[str, str] = dataclasses.field(default_factory=dict)
+    """A dictionary containing cloud credentials.
+
+    For example, for AWS, it contains `access-key` and `secret-key`;
+    for Azure, `application-id`, `application-password` and `subscription-id`
+    can be found here.
+    """
+
+    redacted: List[str] = dataclasses.field(default_factory=list)
+    """A list of redacted secrets."""
+
+    def _to_ops(self) -> ops.CloudCredential:
+        return ops.CloudCredential(
+            auth_type=self.auth_type,
+            attributes=self.attributes,
+            redacted=self.redacted,
+        )
+
+
+@dataclasses.dataclass(frozen=True)
+class CloudSpec:
+    type: str
+    """Type of the cloud."""
+
+    name: str = "localhost"
+    """Juju cloud name."""
+
+    region: Optional[str] = None
+    """Region of the cloud."""
+
+    endpoint: Optional[str] = None
+    """Endpoint of the cloud."""
+
+    identity_endpoint: Optional[str] = None
+    """Identity endpoint of the cloud."""
+
+    storage_endpoint: Optional[str] = None
+    """Storage endpoint of the cloud."""
+
+    credential: Optional[CloudCredential] = None
+    """Cloud credentials with key-value attributes."""
+
+    ca_certificates: List[str] = dataclasses.field(default_factory=list)
+    """A list of CA certificates."""
+
+    skip_tls_verify: bool = False
+    """Whether to skip TLS verfication."""
+
+    is_controller_cloud: bool = False
+    """If this is the cloud used by the controller."""
+
+    def _to_ops(self) -> ops.CloudSpec:
+        return ops.CloudSpec(
+            type=self.type,
+            name=self.name,
+            region=self.region,
+            endpoint=self.endpoint,
+            identity_endpoint=self.identity_endpoint,
+            storage_endpoint=self.storage_endpoint,
+            credential=self.credential._to_ops() if self.credential else None,
+            ca_certificates=self.ca_certificates,
+            skip_tls_verify=self.skip_tls_verify,
+            is_controller_cloud=self.is_controller_cloud,
+        )
+
+
 @dataclasses.dataclass(frozen=True)
 class Secret(_DCBase):
     id: str
@@ -919,6 +991,8 @@ class State(_DCBase):
     """Status of the unit."""
     workload_version: str = ""
     """Workload version."""
+    cloud_spec: Optional[CloudSpec] = None
+    """Cloud specification information (metadata) including credentials."""
 
     def __post_init__(self):
         for name in ["app_status", "unit_status"]:

tests/test_consistency_checker.py

@@ -1,11 +1,14 @@
 import pytest
 from ops.charm import CharmBase
 
+from scenario import Model
 from scenario.consistency_checker import check_consistency
 from scenario.runtime import InconsistentScenarioError
 from scenario.state import (
     RELATION_EVENTS_SUFFIX,
     Action,
+    CloudCredential,
+    CloudSpec,
     Container,
     Event,
     Network,
@@ -568,3 +571,36 @@ def test_networks_consistency():
             },
         ),
     )
+
+
+def test_cloudspec_consistency():
+    cloud_spec = CloudSpec(
+        type="lxd",
+        endpoint="https://127.0.0.1:8443",
+        credential=CloudCredential(
+            auth_type="clientcertificate",
+            attributes={
+                "client-cert": "foo",
+                "client-key": "bar",
+                "server-cert": "baz",
+            },
+        ),
+    )
+
+    assert_consistent(
+        State(cloud_spec=cloud_spec, model=Model(name="lxd-model", type="lxd")),
+        Event("start"),
+        _CharmSpec(
+            MyCharm,
+            meta={"name": "MyVMCharm"},
+        ),
+    )
+
+    assert_inconsistent(
+        State(cloud_spec=cloud_spec, model=Model(name="k8s-model", type="kubernetes")),
+        Event("start"),
+        _CharmSpec(
+            MyCharm,
+            meta={"name": "MyK8sCharm"},
+        ),
+    )

tests/test_e2e/test_cloud_spec.py

@@ -0,0 +1,58 @@
+import ops
+import pytest
+
+import scenario
+
+
+class MyCharm(ops.CharmBase):
+    def __init__(self, framework: ops.Framework):
+        super().__init__(framework)
+        for evt in self.on.events().values():
+            self.framework.observe(evt, self._on_event)
+
+    def _on_event(self, event):
+        pass
+
+
+def test_get_cloud_spec():
+    scenario_cloud_spec = scenario.CloudSpec(
+        type="lxd",
+        name="localhost",
+        endpoint="https://127.0.0.1:8443",
+        credential=scenario.CloudCredential(
+            auth_type="clientcertificate",
+            attributes={
+                "client-cert": "foo",
+                "client-key": "bar",
+                "server-cert": "baz",
+            },
+        ),
+    )
+    expected_cloud_spec = ops.CloudSpec(
+        type="lxd",
+        name="localhost",
+        endpoint="https://127.0.0.1:8443",
+        credential=ops.CloudCredential(
+            auth_type="clientcertificate",
+            attributes={
+                "client-cert": "foo",
+                "client-key": "bar",
+                "server-cert": "baz",
+            },
+        ),
+    )
+    ctx = scenario.Context(MyCharm, meta={"name": "foo"})
+    state = scenario.State(
+        cloud_spec=scenario_cloud_spec,
+        model=scenario.Model(name="lxd-model", type="lxd"),
+    )
+    with ctx.manager("start", state=state) as mgr:
+        assert mgr.charm.model.get_cloud_spec() == expected_cloud_spec
+
+
+def test_get_cloud_spec_error():
+    ctx = scenario.Context(MyCharm, meta={"name": "foo"})
+    state = scenario.State(model=scenario.Model(name="lxd-model", type="lxd"))
+    with ctx.manager("start", state) as mgr:
+        with pytest.raises(ops.ModelError):
+            mgr.charm.model.get_cloud_spec()