feat!: simplify testing secret management

README.md

@@ -804,22 +804,26 @@ Scenario has secrets. Here's how you use them.
 state = scenario.State(
     secrets={
         scenario.Secret(
-            {0: {'key': 'public'}},
-            id='foo',
-        ),
-    },
+            current={'key': 'public'},
+            latest={'key': 'public', 'cert': 'private'},
+        )
+    ]
 )
 ```
 
-The only mandatory arguments to Secret are its secret ID (which should be unique) and its 'contents': that is, a mapping
-from revision numbers (integers) to a `str:str` dict representing the payload of the revision.
+The only mandatory arguments to Secret is the `latest` contents: that is, a `str:str` mapping
+representing the payload of the revision. If the unit that's handling the event
+is tracking a different revision of the content, then `current` should also be
+provided - if it's not, then Scenario assumes that `current` is the `latest`.
+If there are other revisions of the content, simply don't include them: the
+unit has no way of knowing about these.
 
 There are three cases:
 - the secret is owned by this app but not this unit, in which case this charm can only manage it if we are the leader
 - the secret is owned by this unit, in which case this charm can always manage it (leader or not)
-- (default) the secret is not owned by this app nor unit, which means we can't manage it but only view it
+- (default) the secret is not owned by this app nor unit, which means we can't manage it but only view it (this includes user secrets)
 
-Thus by default, the secret is not owned by **this charm**, but, implicitly, by some unknown 'other charm', and that other charm has granted us view rights.
+Thus by default, the secret is not owned by **this charm**, but, implicitly, by some unknown 'other charm' (or a user), and that other has granted us view rights.
 
 The presence of the secret in `State.secrets` entails that we have access to it, either as owners or as grantees. Therefore, if we're not owners, we must be grantees. Absence of a Secret from the known secrets list means we are not entitled to obtaining it in any way. The charm, indeed, shouldn't even know it exists.
 
@@ -833,29 +837,48 @@ To specify a secret owned by this unit (or app):
 state = scenario.State(
     secrets={
         scenario.Secret(
-            {0: {'key': 'private'}},
-            id='foo',
+            latest={'key': 'private'},
             owner='unit',  # or 'app'
+            # The secret owner has granted access to the "remote" app over some relation with ID 0:
             remote_grants={0: {"remote"}}
-            # the secret owner has granted access to the "remote" app over some relation with ID 0
-        ),
-    },
+        )
+    }
 )
 ```
 
-To specify a secret owned by some other application and give this unit (or app) access to it:
+To specify a secret owned by some other application, or a user secret, and give this unit (or app) access to it:
 
 ```python
 state = scenario.State(
     secrets={
         scenario.Secret(
-            {0: {'key': 'public'}},
-            id='foo',
+            latest={'key': 'public'},
             # owner=None, which is the default
-            revision=0,  # the revision that this unit (or app) is currently tracking
-        ),
-    },
+        )
+    }
+)
+```
+
+When handling the `secret-expired` and `secret-remove` events, the charm must remove the specified revision of the secret. For `secret-remove`, the revision will no longer be in the `State`, because it's no longer in use (which is why the `secret-remove` event was triggered). To ensure that the charm is removing the secret, check the context for the history of secret removal:
+
+```python
+class SecretCharm(ops.CharmBase):
+    def __init__(self, framework):
+        super().__init__(framework)
+        self.framework.observe(self.on.secret_remove, self._on_secret_remove)
+
+    def _on_secret_remove(self, event):
+        event.secret.remove_revision(event.revision)
+
+
+ctx = Context(SecretCharm, meta={"name": "foo"})
+secret = Secret({"password": "xxxxxxxx"}, owner="app")
+old_revision = 42
+state = ctx.run(
+    ctx.on.secret_remove(secret, revision=old_revision),
+    State(leader=True, secrets={secret})
 )
+assert ctx.secret_removal_history == [old_revision]
 ```
 
 ## StoredState

scenario/context.py

@@ -376,6 +376,7 @@ class Context:
     - :attr:`app_status_history`: record of the app statuses the charm has set
     - :attr:`unit_status_history`: record of the unit statuses the charm has set
     - :attr:`workload_version_history`: record of the workload versions the charm has set
+    - :attr:`secret_removal_history`: record of the secret revisions the charm has removed
     - :attr:`emitted_events`: record of the events (including custom) that the charm has processed
 
     This allows you to write assertions not only on the output state, but also, to some
@@ -551,6 +552,7 @@ def __init__(
         self.app_status_history: List["_EntityStatus"] = []
         self.unit_status_history: List["_EntityStatus"] = []
         self.workload_version_history: List[str] = []
+        self.secret_removal_history: List[int] = []
         self.emitted_events: List[EventBase] = []
         self.requested_storages: Dict[str, int] = {}
 

scenario/mocking.py

@@ -2,7 +2,6 @@
 # Copyright 2023 Canonical Ltd.
 # See LICENSE file for licensing details.
 import datetime
-import random
 import shutil
 from io import StringIO
 from pathlib import Path
@@ -198,25 +197,20 @@ def _get_secret(self, id=None, label=None):
                 if canonicalize_id(s.id) == canonicalize_id(id)
             ]
             if not secrets:
-                raise SecretNotFoundError(id)
+                raise SecretNotFoundError(id) from None
             return secrets[0]
 
         elif label:
-            secrets = [s for s in self._state.secrets if s.label == label]
-            if not secrets:
-                raise SecretNotFoundError(label)
-            return secrets[0]
+            try:
+                return self._state.get_secret(label=label)
+            except KeyError:
+                raise SecretNotFoundError(label) from None
 
         else:
             # if all goes well, this should never be reached. ops.model.Secret will check upon
             # instantiation that either an id or a label are set, and raise a TypeError if not.
             raise RuntimeError("need id or label.")
 
-    @staticmethod
-    def _generate_secret_id():
-        id = "".join(map(str, [random.choice(list(range(10))) for _ in range(20)]))
-        return f"secret:{id}"
-
     def _check_app_data_access(self, is_app: bool):
         if not isinstance(is_app, bool):
             raise TypeError("is_app parameter to relation_get must be a boolean")
@@ -368,10 +362,8 @@ def secret_add(
     ) -> str:
         from scenario.state import Secret
 
-        secret_id = self._generate_secret_id()
         secret = Secret(
-            id=secret_id,
-            contents={0: content},
+            latest=content,
             label=label,
             description=description,
             expire=expire,
@@ -381,7 +373,7 @@ def secret_add(
         secrets = set(self._state.secrets)
         secrets.add(secret)
         self._state._update_secrets(frozenset(secrets))
-        return secret_id
+        return secret.id
 
     def _check_can_manage_secret(
         self,
@@ -411,19 +403,19 @@ def secret_get(
         secret = self._get_secret(id, label)
         juju_version = self._context.juju_version
         if not (juju_version == "3.1.7" or juju_version >= "3.3.1"):
-            # in this medieval juju chapter,
+            # In this medieval Juju chapter,
             # secret owners always used to track the latest revision.
             # ref: https://bugs.launchpad.net/juju/+bug/2037120
             if secret.owner is not None:
                 refresh = True
 
-        revision = secret.revision
         if peek or refresh:
-            revision = max(secret.contents.keys())
             if refresh:
-                secret._set_revision(revision)
+                secret._track_latest_revision()
+            return secret.latest
 
-        return secret.contents[revision]
+        assert secret.current is not None
+        return secret.current
 
     def secret_info_get(
         self,
@@ -439,7 +431,7 @@ def secret_info_get(
         return SecretInfo(
             id=secret.id,
             label=secret.label,
-            revision=max(secret.contents),
+            revision=secret._latest_revision,
             expires=secret.expire,
             rotation=secret.rotate,
             rotates=None,  # not implemented yet.
@@ -458,6 +450,15 @@ def secret_set(
         secret = self._get_secret(id, label)
         self._check_can_manage_secret(secret)
 
+        if content == secret.latest:
+            # In Juju 3.6 and higher, this is a no-op, but it's good to warn
+            # charmers if they are doing this, because it's not generally good
+            # practice.
+            # https://bugs.launchpad.net/juju/+bug/2069238
+            logger.warning(
+                f"secret {id} contents set to the existing value: new revision not needed",
+            )
+
         secret._update_metadata(
             content=content,
             label=label,
@@ -496,10 +497,32 @@ def secret_remove(self, id: str, *, revision: Optional[int] = None):
         secret = self._get_secret(id)
         self._check_can_manage_secret(secret)
 
-        if revision:
-            del secret.contents[revision]
-        else:
-            secret.contents.clear()
+        # Removing all revisions means that the secret is removed, so is no
+        # longer in the state.
+        if revision is None:
+            secrets = set(self._state.secrets)
+            secrets.remove(secret)
+            self._state._update_secrets(frozenset(secrets))
+            return
+
+        # Juju does not prevent removing the tracked or latest revision, but it
+        # is always a mistake to do this. Rather than having the state model a
+        # secret where the tracked/latest revision cannot be retrieved but the
+        # secret still exists, we raise instead, so that charms know that there
+        # is a problem with their code.
+        if revision in (secret._tracked_revision, secret._latest_revision):
+            raise ValueError(
+                "Charms should not remove the latest revision of a secret. "
+                "Add a new revision with `set_content()` instead, and the previous "
+                "revision will be garbage collected when no longer used.",
+            )
+
+        # For all other revisions, the content is not visible to the charm
+        # (this is as designed: the secret is being removed, so it should no
+        # longer be in use). That means that the state does not need to be
+        # modified - however, unit tests should verify that the remove call was
+        # executed, so we track that in a history list in the context.
+        self._context.secret_removal_history.append(revision)
 
     def relation_remote_app_name(
         self,

scenario/state.py

@@ -5,7 +5,9 @@
 import dataclasses
 import datetime
 import inspect
+import random
 import re
+import string
 from collections import namedtuple
 from enum import Enum
 from itertools import chain
@@ -268,24 +270,26 @@ def _to_ops(self) -> CloudSpec_Ops:
         )
 
 
+def _generate_secret_id():
+    # This doesn't account for collisions, but the odds are so low that it
+    # should not be possible in any realistic test run.
+    secret_id = "".join(
+        random.choice(string.ascii_lowercase + string.digits) for _ in range(20)
+    )
+    return f"secret:{secret_id}"
+
+
 @dataclasses.dataclass(frozen=True)
 class Secret(_max_posargs(1)):
-    # mapping from revision IDs to each revision's contents
-    contents: Dict[int, "RawSecretRevisionContents"]
+    latest: "RawSecretRevisionContents"
+    current: Optional["RawSecretRevisionContents"] = None
 
-    id: str
-    # CAUTION: ops-created Secrets (via .add_secret()) will have a canonicalized
-    #  secret id (`secret:` prefix)
-    #  but user-created ones will not. Using post-init to patch it in feels bad, but requiring the user to
-    #  add the prefix manually every time seems painful as well.
+    id: str = dataclasses.field(default_factory=_generate_secret_id)
 
     # indicates if the secret is owned by THIS unit, THIS app or some other app/unit.
     # if None, the implication is that the secret has been granted to this unit.
     owner: Literal["unit", "app", None] = None
 
-    # what revision is currently tracked by this charm. Only meaningful if owner=False
-    revision: int = 0
-
     # mapping from relation IDs to remote unit/apps to which this secret has been granted.
     # Only applicable if owner
     remote_grants: Dict[int, Set[str]] = dataclasses.field(default_factory=dict)
@@ -295,13 +299,25 @@ class Secret(_max_posargs(1)):
     expire: Optional[datetime.datetime] = None
     rotate: Optional[SecretRotate] = None
 
+    # what revision is currently tracked by this charm. Only meaningful if owner=False
+    _tracked_revision: int = 1
+
+    # what revision is the latest for this secret.
+    _latest_revision: int = 1
+
     def __hash__(self) -> int:
         return hash(self.id)
 
-    def _set_revision(self, revision: int):
-        """Set a new tracked revision."""
+    def __post_init__(self):
+        if self.current is None:
+            # bypass frozen dataclass
+            object.__setattr__(self, "current", self.latest)
+
+    def _track_latest_revision(self):
+        """Set the current revision to the tracked revision."""
         # bypass frozen dataclass
-        object.__setattr__(self, "revision", revision)
+        object.__setattr__(self, "_tracked_revision", self._latest_revision)
+        object.__setattr__(self, "current", self.latest)
 
     def _update_metadata(
         self,
@@ -312,11 +328,12 @@ def _update_metadata(
         rotate: Optional[SecretRotate] = None,
     ):
         """Update the metadata."""
-        revision = max(self.contents.keys())
-        if content:
-            self.contents[revision + 1] = content
-
         # bypass frozen dataclass
+        object.__setattr__(self, "_latest_revision", self._latest_revision + 1)
+        # TODO: if this is done twice in the same hook, then Juju ignores the
+        # first call, it doesn't continue to update like this does.
+        if content:
+            object.__setattr__(self, "latest", content)
         if label:
             object.__setattr__(self, "label", label)
         if description: