feat: add support for cloud spec

scenario/mocking.py

@@ -35,6 +35,7 @@
 
 from scenario.logger import logger as scenario_logger
 from scenario.state import (
+    CloudSpec,
     JujuLogLine,
     Mount,
     Network,
@@ -631,6 +632,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=...)`",
+            )
+        return self._state.cloud_spec
+
 
 class _MockPebbleClient(_TestingPebbleClient):
     def __init__(

scenario/state.py

@@ -140,6 +140,83 @@ 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."""
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "CloudCredential":
+        """Create a new CloudCredential object from a dictionary."""
+        return cls(
+            auth_type=d["auth-type"],
+            attributes=d.get("attrs") or {},
+            redacted=d.get("redacted") or [],
+        )
+
+
+@dataclasses.dataclass(frozen=True)
+class CloudSpec:
+    type: str
+    """Type of the cloud."""
+
+    name: str
+    """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, defaults to False."""
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "CloudSpec":
+        """Create a new CloudSpec object from a dict parsed from JSON."""
+        return cls(
+            type=d["type"],
+            name=d["name"],
+            region=d.get("region") or None,
+            endpoint=d.get("endpoint") or None,
+            identity_endpoint=d.get("identity-endpoint") or None,
+            storage_endpoint=d.get("storage-endpoint") or None,
+            credential=CloudCredential.from_dict(d["credential"])
+            if d.get("credential")
+            else None,
+            ca_certificates=d.get("cacertificates") or [],
+            skip_tls_verify=d.get("skip-tls-verify") or False,
+            is_controller_cloud=d.get("is-controller-cloud") or False,
+        )
+
+
 @dataclasses.dataclass(frozen=True)
 class Secret(_DCBase):
     id: str
@@ -919,6 +996,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_e2e/test_cloud_spec.py

@@ -0,0 +1,48 @@
+import ops
+import pytest
+
+import scenario
+from scenario.state import CloudSpec
+
+
+@pytest.fixture(scope="function")
+def mycharm():
+    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
+
+    return MyCharm
+
+
+def test_get_cloud_spec(mycharm):
+    cloud_spec = CloudSpec.from_dict(
+        {
+            "name": "localhost",
+            "type": "lxd",
+            "endpoint": "https://127.0.0.1:8443",
+            "credential": {
+                "auth-type": "certificate",
+                "attrs": {
+                    "client-cert": "foo",
+                    "client-key": "bar",
+                    "server-cert": "baz",
+                },
+            },
+        }
+    )
+
+    ctx = scenario.Context(mycharm, meta={"name": "foo"})
+    with ctx.manager("start", scenario.State(cloud_spec=cloud_spec)) as mgr:
+        assert mgr.charm.model.get_cloud_spec() == cloud_spec
+
+
+def test_get_cloud_spec(mycharm):
+    ctx = scenario.Context(mycharm, meta={"name": "foo"})
+    with ctx.manager("start", scenario.State()) as mgr:
+        with pytest.raises(ops.ModelError):
+            mgr.charm.model.get_cloud_spec()