Merge branch 'main' into pebble-notices

README.md

@@ -20,7 +20,7 @@ union of an `Event` (why am I, charm, being executed), a `State` (am I leader? w
 config?...) and the charm's execution `Context` (what relations can I have? what containers can I have?...). The output is another `State`: the state after the charm has had a chance to interact with the
 mocked Juju model and affect the initial state back.
 
-![state transition model depiction](resources/state-transition-model.png)
+![state transition model depiction](https://raw.githubusercontent.com/canonical/ops-scenario/main/resources/state-transition-model.png)
 
 For example: a charm currently in `unknown` status is executed with a `start` event, and based on whether it has leadership or not (according to its input state), it will decide to set `active` or `blocked` status (which will be reflected in the output state).
 
@@ -982,6 +982,45 @@ assert out.model.name == "my-model"
 assert out.model.uuid == state_in.model.uuid
 ```
 
+### CloudSpec
+
+You can set CloudSpec information in the model (only `type` and `name` are required).
+
+Example:
+
+```python
+import scenario
+
+cloud_spec=scenario.CloudSpec(
+    type="lxd",
+    endpoint="https://127.0.0.1:8443",
+    credential=scenario.CloudCredential(
+        auth_type="clientcertificate",
+        attributes={
+            "client-cert": "foo",
+            "client-key": "bar",
+            "server-cert": "baz",
+        },
+    ),
+)
+state = scenario.State(
+    model=scenario.Model(name="my-vm-model", type="lxd", cloud_spec=cloud_spec),
+)
+```
+
+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,
@@ -30,6 +32,8 @@
 __all__ = [
     "Action",
     "ActionOutput",
+    "CloudCredential",
+    "CloudSpec",
     "Context",
     "deferred",
     "StateValidationError",

scenario/consistency_checker.py

@@ -2,10 +2,11 @@
 # Copyright 2023 Canonical Ltd.
 # See LICENSE file for licensing details.
 import os
+import re
 from collections import Counter
 from collections.abc import Sequence
 from numbers import Number
-from typing import TYPE_CHECKING, Iterable, List, NamedTuple, Tuple
+from typing import TYPE_CHECKING, Iterable, List, NamedTuple, Tuple, Union
 
 from scenario.runtime import InconsistentScenarioError
 from scenario.runtime import logger as scenario_logger
@@ -69,6 +70,7 @@ def check_consistency(
         check_storages_consistency,
         check_relation_consistency,
         check_network_consistency,
+        check_cloudspec_consistency,
     ):
         results = check(
             state=state,
@@ -326,10 +328,17 @@ def check_storages_consistency(
     return Results(errors, [])
 
 
+def _is_secret_identifier(value: Union[str, int, float, bool]) -> bool:
+    """Return true iff the value is in the form `secret:{secret id}`."""
+    # cf. https://github.com/juju/juju/blob/13eb9df3df16a84fd471af8a3c95ddbd04389b71/core/secrets/secret.go#L48
+    return bool(re.match(r"secret:[0-9a-z]{20}$", str(value)))
+
+
 def check_config_consistency(
     *,
     state: "State",
     charm_spec: "_CharmSpec",
+    juju_version: Tuple[int, ...],
     **_kwargs,  # noqa: U101
 ) -> Results:
     """Check the consistency of the state.config with the charm_spec.config (config.yaml)."""
@@ -348,16 +357,21 @@ def check_config_consistency(
         converters = {
             "string": str,
             "int": int,
-            "integer": int,  # fixme: which one is it?
-            "number": float,
+            "float": float,
             "boolean": bool,
-            # "attrs": NotImplemented,  # fixme: wot?
+        }
+        if juju_version >= (3, 4):
+            converters["secret"] = str
+
+        validators = {
+            "secret": _is_secret_identifier,
         }
 
         expected_type_name = meta_config[key].get("type", None)
         if not expected_type_name:
             errors.append(f"config.yaml invalid; option {key!r} has no 'type'.")
             continue
+        validator = validators.get(expected_type_name)
 
         expected_type = converters.get(expected_type_name)
         if not expected_type:
@@ -371,6 +385,11 @@ def check_config_consistency(
                 f"but is of type {type(value)}.",
             )
 
+        elif validator and not validator(value):
+            errors.append(
+                f"config invalid: option {key!r} value {value!r} is not valid.",
+            )
+
     return Results(errors, [])
 
 
@@ -553,3 +572,25 @@ 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.model.cloud_spec:
+        errors.append(
+            "CloudSpec is only available for machine charms, not Kubernetes charms. "
+            "Tell Scenario to simulate a machine substrate with: "
+            "`scenario.State(..., model=scenario.Model(type='lxd'))`.",
+        )
+
+    return Results(errors, warnings)

scenario/context.py

@@ -169,6 +169,7 @@ def __init__(
         capture_framework_events: bool = False,
         app_name: Optional[str] = None,
         unit_id: Optional[int] = 0,
+        app_trusted: bool = False,
     ):
         """Represents a simulated charm's execution context.
 
@@ -225,6 +226,8 @@ def __init__(
         :arg app_name: App name that this charm is deployed as. Defaults to the charm name as
             defined in metadata.yaml.
         :arg unit_id: Unit ID that this charm is deployed as. Defaults to 0.
+        :arg app_trusted: whether the charm has Juju trust (deployed with ``--trust`` or added with
+            ``juju trust``). Defaults to False
         :arg charm_root: virtual charm root the charm will be executed with.
             If the charm, say, expects a `./src/foo/bar.yaml` file present relative to the
             execution cwd, you need to use this. E.g.:
@@ -268,6 +271,7 @@ def __init__(
 
         self._app_name = app_name
         self._unit_id = unit_id
+        self.app_trusted = app_trusted
         self._tmp = tempfile.TemporaryDirectory()
 
         # config for what events to be captured in emitted_events.

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,18 @@ 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._context.app_trusted:
+            raise ModelError(
+                "ERROR charm is not trusted, initialise Context with `app_trusted=True`",
+            )
+        if not self._state.model.cloud_spec:
+            raise ModelError(
+                "ERROR cloud spec is empty, initialise it with "
+                "`State(model=Model(..., cloud_spec=ops.CloudSpec(...)))`",
+            )
+        return self._state.model.cloud_spec._to_ops()
+
 
 class _MockPebbleClient(_TestingPebbleClient):
     def __init__(

scenario/ops_main_mock.py

@@ -3,6 +3,7 @@
 # See LICENSE file for licensing details.
 import inspect
 import os
+import sys
 from typing import TYPE_CHECKING, Any, Optional, Sequence, cast
 
 import ops.charm
@@ -96,6 +97,9 @@ def setup_framework(
     )
     debug = "JUJU_DEBUG" in os.environ
     setup_root_logging(model_backend, debug=debug)
+    # ops sets sys.excepthook to go to Juju's debug-log, but that's not useful
+    # in a testing context, so reset it.
+    sys.excepthook = sys.__excepthook__
     ops_logger.debug(
         "Operator Framework %s up and running.",
         ops.__version__,

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
@@ -141,6 +142,76 @@ 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 generic cloud API 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
@@ -570,6 +641,9 @@ class Model(_DCBase):
     # TODO: make this exhaustive.
     type: Literal["kubernetes", "lxd"] = "kubernetes"
 
+    cloud_spec: Optional[CloudSpec] = None
+    """Cloud specification information (metadata) including credentials."""
+
 
 # for now, proc mock allows you to map one command to one mocked output.
 # todo extend: one input -> multiple outputs, at different times