feat: add support for Pebble custom notices

README.md

@@ -714,6 +714,40 @@ def test_pebble_exec():
     )
 ```
 
+### Pebble Notices
+
+Pebble can generate notices, which Juju will detect, and wake up the charm to
+let it know that something has happened in the container. The most common
+use-case is Pebble custom notices, which is a mechanism for the workload
+application to trigger a charm event.
+
+When the charm is notified, there might be a queue of existing notices, or just
+the one that has triggered the event:
+
+```python
+import ops
+import scenario
+
+class MyCharm(ops.CharmBase):
+    def __init__(self, framework):
+        super().__init__(framework)
+        framework.observe(self.on["cont"].pebble_custom_notice, self._on_notice)
+
+    def _on_notice(self, event):
+        event.notice.key  # == "example.com/c"
+        for notice in self.unit.get_container("cont").get_notices():
+            ...
+
+ctx = scenario.Context(MyCharm, meta={"name": "foo", "containers": {"my-container": {}}})
+notices = [
+    scenario.Notice(key="example.com/a", occurences=10),
+    scenario.Notice(key="example.com/b", last_data={"bar": "baz"}),
+    scenario.Notice(key="example.com/c"),
+]
+cont = scenario.Container(notices=notices)
+ctx.run(container.get_notice("example.com/c").event, scenario.State(containers=[cont]))
+```
+
 ## Storage
 
 If your charm defines `storage` in its metadata, you can use `scenario.Storage` to instruct Scenario to make (mocked) filesystem storage available to the charm at runtime.

pyproject.toml

@@ -18,7 +18,7 @@ license.text = "Apache-2.0"
 keywords = ["juju", "test"]
 
 dependencies = [
-    "ops>=2.6",
+    "ops>=2.10",
     "PyYAML>=6.0.1",
 ]
 readme = "README.md"

scenario/__init__.py

@@ -13,6 +13,7 @@
     Model,
     Mount,
     Network,
+    Notice,
     PeerRelation,
     Port,
     Relation,
@@ -41,6 +42,7 @@
     "ExecOutput",
     "Mount",
     "Container",
+    "Notice",
     "Address",
     "BindAddress",
     "Network",

scenario/consistency_checker.py

@@ -513,14 +513,15 @@ def check_containers_consistency(
     meta = charm_spec.meta
     meta_containers = list(map(normalize_name, meta.get("containers", {})))
     state_containers = [normalize_name(c.name) for c in state.containers]
+    all_notices = {notice.id for c in state.containers for notice in c.notices}
     errors = []
 
     # it's fine if you have containers in meta that are not in state.containers (yet), but it's
     # not fine if:
-    # - you're processing a pebble-ready event and that container is not in state.containers or
+    # - you're processing a Pebble event and that container is not in state.containers or
     #   meta.containers
     if event._is_workload_event:
-        evt_container_name = event.name[: -len("-pebble-ready")]
+        evt_container_name = event.name.split("_pebble_")[0]
         if evt_container_name not in meta_containers:
             errors.append(
                 f"the event being processed concerns container {evt_container_name!r}, but a "
@@ -529,8 +530,13 @@ def check_containers_consistency(
         if evt_container_name not in state_containers:
             errors.append(
                 f"the event being processed concerns container {evt_container_name!r}, but a "
-                f"container with that name is not present in the state. It's odd, but consistent, "
-                f"if it cannot connect; but it should at least be there.",
+                f"container with that name is not present in the state. It's odd, but "
+                f"consistent, if it cannot connect; but it should at least be there.",
+            )
+        if event.notice and event.notice.id not in all_notices:
+            errors.append(
+                f"the event being processed concerns notice {event.notice!r}, but that "
+                "notice is not in any of the containers present in the state.",
             )
 
     # - a container in state.containers is not in meta.containers

scenario/mocking.py

@@ -663,6 +663,16 @@ def __init__(
 
         self._root = container_root
 
+        # load any existing notices from the state
+        self._notices: Dict[Tuple[str, str], pebble.Notice] = {}
+        for container in state.containers:
+            for notice in container.notices:
+                if hasattr(notice.type, "value"):
+                    notice_type = cast(pebble.NoticeType, notice.type).value
+                else:
+                    notice_type = str(notice.type)
+                self._notices[notice_type, notice.key] = notice._to_ops()
+
     def get_plan(self) -> pebble.Plan:
         return self._container.plan
 

scenario/runtime.py

@@ -11,6 +11,7 @@
 from typing import TYPE_CHECKING, Dict, List, Optional, Type, Union
 
 import yaml
+from ops import pebble
 from ops.framework import _event_regex
 from ops.storage import NoSnapshotError, SQLiteStorage
 
@@ -248,6 +249,19 @@ def _get_event_env(self, state: "State", event: "Event", charm_root: Path):
         if container := event.container:
             env.update({"JUJU_WORKLOAD_NAME": container.name})
 
+        if notice := event.notice:
+            if hasattr(notice.type, "value"):
+                notice_type = typing.cast(pebble.NoticeType, notice.type).value
+            else:
+                notice_type = str(notice.type)
+            env.update(
+                {
+                    "JUJU_NOTICE_ID": notice.id,
+                    "JUJU_NOTICE_TYPE": notice_type,
+                    "JUJU_NOTICE_KEY": notice.key,
+                },
+            )
+
         if storage := event.storage:
             env.update({"JUJU_STORAGE_ID": f"{storage.name}/{storage.index}"})
 

scenario/state.py

@@ -87,6 +87,7 @@
     "collect_unit_status",
 }
 PEBBLE_READY_EVENT_SUFFIX = "_pebble_ready"
+PEBBLE_CUSTOM_NOTICE_EVENT_SUFFIX = "_pebble_custom_notice"
 RELATION_EVENTS_SUFFIX = {
     "_relation_changed",
     "_relation_broken",
@@ -609,6 +610,96 @@ class Mount(_DCBase):
     src: Union[str, Path]
 
 
+def _now_utc():
+    return datetime.datetime.now(tz=datetime.timezone.utc)
+
+
+_next_notice_id_counter = 1
+
+
+def next_notice_id(update=True):
+    global _next_notice_id_counter
+    cur = _next_notice_id_counter
+    if update:
+        _next_notice_id_counter += 1
+    return str(cur)
+
+
+@dataclasses.dataclass(frozen=True)
+class Notice(_DCBase):
+    key: str
+    """The notice key, a string that differentiates notices of this type.
+
+    This is in the format ``domain/path``; for example:
+    ``canonical.com/postgresql/backup`` or ``example.com/mycharm/notice``.
+    """
+
+    id: str = dataclasses.field(default_factory=next_notice_id)
+    """Unique ID for this notice."""
+
+    user_id: Optional[int] = None
+    """UID of the user who may view this notice (None means notice is public)."""
+
+    type: Union[pebble.NoticeType, str] = pebble.NoticeType.CUSTOM
+    """Type of the notice."""
+
+    first_occurred: datetime.datetime = dataclasses.field(default_factory=_now_utc)
+    """The first time one of these notices (type and key combination) occurs."""
+
+    last_occurred: datetime.datetime = dataclasses.field(default_factory=_now_utc)
+    """The last time one of these notices occurred."""
+
+    last_repeated: datetime.datetime = dataclasses.field(default_factory=_now_utc)
+    """The time this notice was last repeated.
+
+    See Pebble's `Notices documentation <https://github.com/canonical/pebble/#notices>`_
+    for an explanation of what "repeated" means.
+    """
+
+    occurrences: int = 1
+    """The number of times one of these notices has occurred."""
+
+    last_data: Dict[str, str] = dataclasses.field(default_factory=dict)
+    """Additional data captured from the last occurrence of one of these notices."""
+
+    repeat_after: Optional[datetime.timedelta] = None
+    """Minimum time after one of these was last repeated before Pebble will repeat it again."""
+
+    expire_after: Optional[datetime.timedelta] = None
+    """How long since one of these last occurred until Pebble will drop the notice."""
+
+    def _to_ops(self) -> pebble.Notice:
+        return pebble.Notice(
+            id=self.id,
+            user_id=self.user_id,
+            type=self.type,
+            key=self.key,
+            first_occurred=self.first_occurred,
+            last_occurred=self.last_occurred,
+            last_repeated=self.last_repeated,
+            occurrences=self.occurrences,
+            last_data=self.last_data,
+            repeat_after=self.repeat_after,
+            expire_after=self.expire_after,
+        )
+
+
+@dataclasses.dataclass(frozen=True)
+class BoundNotice(_DCBase):
+    notice: Notice
+    container: "Container"
+
+    @property
+    def event(self):
+        """Sugar to generate a <container's name>-pebble-custom-notice event for this notice."""
+        suffix = PEBBLE_CUSTOM_NOTICE_EVENT_SUFFIX.replace("_", "-")
+        return Event(
+            path=normalize_name(self.container.name + suffix),
+            container=self.container,
+            notice=self.notice,
+        )
+
+
 @dataclasses.dataclass(frozen=True)
 class Container(_DCBase):
     name: str
@@ -646,6 +737,8 @@ class Container(_DCBase):
 
     exec_mock: _ExecMock = dataclasses.field(default_factory=dict)
 
+    notices: List[Notice] = dataclasses.field(default_factory=list)
+
     def _render_services(self):
         # copied over from ops.testing._TestingPebbleClient._render_services()
         services = {}  # type: Dict[str, pebble.Service]
@@ -713,6 +806,23 @@ def pebble_ready_event(self):
             )
         return Event(path=normalize_name(self.name + "-pebble-ready"), container=self)
 
+    def get_notice(
+        self,
+        key: str,
+        notice_type: pebble.NoticeType = pebble.NoticeType.CUSTOM,
+    ) -> BoundNotice:
+        """Get a Pebble notice by key and type.
+
+        Raises:
+            KeyError: if the notice is not found.
+        """
+        for notice in self.notices:
+            if notice.key == key and notice.type == notice_type:
+                return BoundNotice(notice, self)
+        raise KeyError(
+            f"{self.name} does not have a notice with key {key} and type {notice_type}",
+        )
+
 
 _RawStatusLiteral = Literal[
     "waiting",
@@ -1191,6 +1301,8 @@ def _get_suffix_and_type(s: str) -> Tuple[str, _EventType]:
         # Whether the event name indicates that this is a workload event.
         if s.endswith(PEBBLE_READY_EVENT_SUFFIX):
             return PEBBLE_READY_EVENT_SUFFIX, _EventType.workload
+        if s.endswith(PEBBLE_CUSTOM_NOTICE_EVENT_SUFFIX):
+            return PEBBLE_CUSTOM_NOTICE_EVENT_SUFFIX, _EventType.workload
 
         if s in BUILTIN_EVENTS:
             return "", _EventType.builtin
@@ -1217,6 +1329,9 @@ class Event(_DCBase):
     # if this is a workload (container) event, the container it refers to
     container: Optional[Container] = None
 
+    # if this is a Pebble notice event, the notice it refers to
+    notice: Optional[Notice] = None
+
     # if this is an action event, the Action instance
     action: Optional["Action"] = None
 
@@ -1397,6 +1512,18 @@ def deferred(self, handler: Callable, event_id: int = 1) -> DeferredEvent:
             snapshot_data = {
                 "container_name": container.name,
             }
+            if self.notice:
+                if hasattr(self.notice.type, "value"):
+                    notice_type = cast(pebble.NoticeType, self.notice.type).value
+                else:
+                    notice_type = str(self.notice.type)
+                snapshot_data.update(
+                    {
+                        "notice_id": self.notice.id,
+                        "notice_key": self.notice.key,
+                        "notice_type": notice_type,
+                    },
+                )
 
         elif self._is_relation_event:
             # this is a RelationEvent.
@@ -1460,8 +1587,9 @@ def deferred(
     event_id: int = 1,
     relation: Optional["Relation"] = None,
     container: Optional["Container"] = None,
+    notice: Optional["Notice"] = None,
 ):
     """Construct a DeferredEvent from an Event or an event name."""
     if isinstance(event, str):
-        event = Event(event, relation=relation, container=container)
+        event = Event(event, relation=relation, container=container, notice=notice)
     return event.deferred(handler=handler, event_id=event_id)

tests/test_consistency_checker.py

@@ -9,6 +9,7 @@
     Container,
     Event,
     Network,
+    Notice,
     PeerRelation,
     Relation,
     Secret,
@@ -60,6 +61,22 @@ def test_workload_event_without_container():
         Event("foo-pebble-ready", container=Container("foo")),
         _CharmSpec(MyCharm, {"containers": {"foo": {}}}),
     )
+    assert_inconsistent(
+        State(),
+        Event("foo-pebble-custom-notice", container=Container("foo")),
+        _CharmSpec(MyCharm, {}),
+    )
+    notice = Notice("example.com/foo")
+    assert_consistent(
+        State(containers=[Container("foo", notices=[notice])]),
+        Event("foo-pebble-custom-notice", container=Container("foo"), notice=notice),
+        _CharmSpec(MyCharm, {"containers": {"foo": {}}}),
+    )
+    assert_inconsistent(
+        State(containers=[Container("foo")]),
+        Event("foo-pebble-custom-notice", container=Container("foo"), notice=notice),
+        _CharmSpec(MyCharm, {"containers": {"foo": {}}}),
+    )
 
 
 def test_container_meta_mismatch():

tests/test_e2e/test_deferred.py

@@ -12,7 +12,7 @@
 from ops.framework import Framework
 
 from scenario import Context
-from scenario.state import Container, DeferredEvent, Relation, State, deferred
+from scenario.state import Container, DeferredEvent, Notice, Relation, State, deferred
 from tests.helpers import trigger
 
 CHARM_CALLED = 0
@@ -97,6 +97,20 @@ def test_deferred_workload_evt(mycharm):
     assert asdict(evt2) == asdict(evt1)
 
 
+def test_deferred_notice_evt(mycharm):
+    notice = Notice(key="example.com/bar")
+    ctr = Container("foo", notices=[notice])
+    evt1 = ctr.get_notice("example.com/bar").event.deferred(handler=mycharm._on_event)
+    evt2 = deferred(
+        event="foo_pebble_custom_notice",
+        handler=mycharm._on_event,
+        container=ctr,
+        notice=notice,
+    )
+
+    assert asdict(evt2) == asdict(evt1)
+
+
 def test_deferred_relation_event(mycharm):
     mycharm.defer_next = 2