feat: add support for Pebble check events

README.md

@@ -690,7 +690,25 @@ notices = [
     scenario.Notice(key="example.com/c"),
 ]
 container = scenario.Container("my-container", notices=notices)
-ctx.run(container.get_notice("example.com/c").event, scenario.State(containers=[container]))
+state = scenario.State(containers={container})
+ctx.run(ctx.on.pebble_custom_notice(container=container, notice=notices[-1]), state)
+```
+
+### Pebble Checks
+
+A Pebble plan can contain checks, and when those checks exceed the configured
+failure threshold, or start succeeding again after, Juju will emit a
+pebble-check-failed or pebble-check-recovered event. In order to simulate these
+events, you need to add a `Check` to the container. Note that the status of the
+check doesn't have to match the event being generated: by the time that Juju
+sends a pebble-check-failed event the check might have started passing again.
+
+```python
+ctx = scenario.Context(MyCharm, meta={"name": "foo", "containers": {"my-container": {}}})
+check = scenario.Check("http-check", failures=7, status=ops.pebble.CheckStatus.DOWN)
+container = scenario.Container("my-container", checks={check})
+state = scenario.State(containers={container})
+ctx.run(ctx.on.pebble_check_failed(check=check, container=container), state=state)
 ```
 
 ## Storage

pyproject.toml

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

scenario/__init__.py

@@ -6,6 +6,7 @@
     Action,
     Address,
     BindAddress,
+    Check,
     CloudCredential,
     CloudSpec,
     Container,
@@ -33,6 +34,7 @@
 __all__ = [
     "Action",
     "ActionOutput",
+    "Check",
     "CloudCredential",
     "CloudSpec",
     "Context",

scenario/consistency_checker.py

@@ -560,6 +560,7 @@ def check_containers_consistency(
     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}
+    all_checks = {check.name for c in state.containers for check in c.checks}
     errors = []
 
     # it's fine if you have containers in meta that are not in state.containers (yet), but it's
@@ -584,6 +585,11 @@ def check_containers_consistency(
                 f"the event being processed concerns notice {event.notice!r}, but that "
                 "notice is not in any of the containers present in the state.",
             )
+        if event.check and event.check.name not in all_checks:
+            errors.append(
+                f"the event being processed concerns check {event.check!r}, but that "
+                "check is not in any of the containers present in the state.",
+            )
 
     # - a container in state.containers is not in meta.containers
     if diff := (set(state_containers).difference(set(meta_containers))):

scenario/context.py

@@ -13,8 +13,10 @@
 from scenario.runtime import Runtime
 from scenario.state import (
     Action,
+    Check,
     Container,
     MetadataNotFoundError,
+    Notice,
     Secret,
     Storage,
     _CharmSpec,
@@ -312,6 +314,30 @@ def storage_detaching(storage: Storage):
     def pebble_ready(container: Container):
         return _Event(f"{container.name}_pebble_ready", container=container)
 
+    @staticmethod
+    def pebble_custom_notice(container: Container, notice: Notice):
+        return _Event(
+            f"{container.name}_pebble_custom_notice",
+            container=container,
+            notice=notice,
+        )
+
+    @staticmethod
+    def pebble_check_failed(container: Container, check: Check):
+        return _Event(
+            f"{container.name}_pebble_check_failed",
+            container=container,
+            check=check,
+        )
+
+    @staticmethod
+    def pebble_check_recovered(container: Container, check: Check):
+        return _Event(
+            f"{container.name}_pebble_check_recovered",
+            container=container,
+            check=check,
+        )
+
 
 class Context:
     """Represents a simulated charm's execution context.

scenario/mocking.py

@@ -689,15 +689,18 @@ def __init__(
 
         self._root = container_root
 
-        # load any existing notices from the state
+        # load any existing notices and check information from the state
         self._notices: Dict[Tuple[str, str], pebble.Notice] = {}
+        self._check_infos: Dict[str, pebble.CheckInfo] = {}
         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()
+            for check in container.checks:
+                self._check_infos[check.name] = check._to_ops()
 
     def get_plan(self) -> pebble.Plan:
         return self._container.plan

scenario/runtime.py

@@ -271,6 +271,9 @@ def _get_event_env(self, state: "State", event: "_Event", charm_root: Path):
                 },
             )
 
+        if check := event.check:
+            env["JUJU_PEBBLE_CHECK_NAME"] = check.name
+
         if storage := event.storage:
             env.update({"JUJU_STORAGE_ID": f"{storage.name}/{storage.index}"})
 

scenario/state.py

@@ -86,6 +86,8 @@
 }
 PEBBLE_READY_EVENT_SUFFIX = "_pebble_ready"
 PEBBLE_CUSTOM_NOTICE_EVENT_SUFFIX = "_pebble_custom_notice"
+PEBBLE_CHECK_FAILED_EVENT_SUFFIX = "_pebble_check_failed"
+PEBBLE_CHECK_RECOVERED_EVENT_SUFFIX = "_pebble_check_recovered"
 RELATION_EVENTS_SUFFIX = {
     "_relation_changed",
     "_relation_broken",
@@ -765,18 +767,37 @@ def _to_ops(self) -> pebble.Notice:
 
 
 @dataclasses.dataclass(frozen=True)
-class _BoundNotice(_max_posargs(0)):
-    notice: Notice
-    container: "Container"
+class Check(_max_posargs(1)):
+    name: str
+    """Name of the check."""
 
-    @property
-    def event(self):
-        """Sugar to generate a <container's name>-pebble-custom-notice event for this notice."""
-        suffix = PEBBLE_CUSTOM_NOTICE_EVENT_SUFFIX
-        return _Event(
-            path=normalize_name(self.container.name) + suffix,
-            container=self.container,
-            notice=self.notice,
+    level: Optional[pebble.CheckLevel] = None
+    """Level of the check."""
+
+    status: pebble.CheckStatus = pebble.CheckStatus.UP
+    """Status of the check.
+
+    CheckStatus.UP means the check is healthy (the number of failures is less
+    than the threshold), CheckStatus.DOWN means the check is unhealthy
+    (the number of failures has reached the threshold).
+    """
+
+    failures: int = 0
+    """Number of failures since the check last succeeded."""
+
+    threshold: int = 3
+    """Failure threshold.
+
+    This is how many consecutive failures for the check to be considered “down”.
+    """
+
+    def _to_ops(self) -> pebble.CheckInfo:
+        return pebble.CheckInfo(
+            name=self.name,
+            level=self.level,
+            status=self.status,
+            failures=self.failures,
+            threshold=self.threshold,
         )
 
 
@@ -846,6 +867,8 @@ class Container(_max_posargs(1)):
 
     notices: List[Notice] = dataclasses.field(default_factory=list)
 
+    checks: FrozenSet[Check] = frozenset()
+
     def __hash__(self) -> int:
         return hash(self.name)
 
@@ -911,23 +934,6 @@ def get_filesystem(self, ctx: "Context") -> Path:
         """
         return ctx._get_container_root(self.name)
 
-    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=notice, container=self)
-        raise KeyError(
-            f"{self.name} does not have a notice with key {key} and type {notice_type}",
-        )
-
 
 _RawStatusLiteral = Literal[
     "waiting",
@@ -1519,6 +1525,10 @@ def _get_suffix_and_type(s: str) -> Tuple[str, _EventType]:
             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.endswith(PEBBLE_CHECK_FAILED_EVENT_SUFFIX):
+            return PEBBLE_CHECK_FAILED_EVENT_SUFFIX, _EventType.workload
+        if s.endswith(PEBBLE_CHECK_RECOVERED_EVENT_SUFFIX):
+            return PEBBLE_CHECK_RECOVERED_EVENT_SUFFIX, _EventType.workload
 
         if s in BUILTIN_EVENTS:
             return "", _EventType.builtin
@@ -1557,6 +1567,9 @@ class _Event:
     notice: Optional[Notice] = None
     """If this is a Pebble notice event, the notice it refers to."""
 
+    check: Optional[Check] = None
+    """If this is a Pebble check event, the check it refers to."""
+
     action: Optional["Action"] = None
     """If this is an action event, the :class:`Action` it refers to."""
 
@@ -1674,6 +1687,8 @@ def deferred(self, handler: Callable, event_id: int = 1) -> DeferredEvent:
                         "notice_type": notice_type,
                     },
                 )
+            elif self.check:
+                snapshot_data["check_name"] = self.check.name
 
         elif self._is_relation_event:
             # this is a RelationEvent.
@@ -1750,8 +1765,15 @@ def deferred(
     relation: Optional["Relation"] = None,
     container: Optional["Container"] = None,
     notice: Optional["Notice"] = None,
+    check: Optional["Check"] = None,
 ):
     """Construct a DeferredEvent from an Event or an event name."""
     if isinstance(event, str):
-        event = _Event(event, relation=relation, container=container, notice=notice)
+        event = _Event(
+            event,
+            relation=relation,
+            container=container,
+            notice=notice,
+            check=check,
+        )
     return event.deferred(handler=handler, event_id=event_id)

tests/test_consistency_checker.py

@@ -8,6 +8,7 @@
 from scenario.state import (
     RELATION_EVENTS_SUFFIX,
     Action,
+    Check,
     CloudCredential,
     CloudSpec,
     Container,
@@ -84,6 +85,27 @@ def test_workload_event_without_container():
         _Event("foo-pebble-custom-notice", container=Container("foo"), notice=notice),
         _CharmSpec(MyCharm, {"containers": {"foo": {}}}),
     )
+    check = Check("http-check")
+    assert_consistent(
+        State(containers={Container("foo", checks={check})}),
+        _Event("foo-pebble-check-failed", container=Container("foo"), check=check),
+        _CharmSpec(MyCharm, {"containers": {"foo": {}}}),
+    )
+    assert_inconsistent(
+        State(containers={Container("foo")}),
+        _Event("foo-pebble-check-failed", container=Container("foo"), check=check),
+        _CharmSpec(MyCharm, {"containers": {"foo": {}}}),
+    )
+    assert_consistent(
+        State(containers={Container("foo", checks={check})}),
+        _Event("foo-pebble-check-recovered", container=Container("foo"), check=check),
+        _CharmSpec(MyCharm, {"containers": {"foo": {}}}),
+    )
+    assert_inconsistent(
+        State(containers={Container("foo")}),
+        _Event("foo-pebble-check-recovered", container=Container("foo"), check=check),
+        _CharmSpec(MyCharm, {"containers": {"foo": {}}}),
+    )
 
 
 def test_container_meta_mismatch():

tests/test_e2e/test_deferred.py

@@ -102,7 +102,9 @@ def test_deferred_workload_evt(mycharm):
 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)
+    evt1 = _Event("foo_pebble_custom_notice", notice=notice, container=ctr).deferred(
+        handler=mycharm._on_event
+    )
     evt2 = deferred(
         event="foo_pebble_custom_notice",
         handler=mycharm._on_event,

tests/test_e2e/test_pebble.py

@@ -10,7 +10,7 @@
 from ops.pebble import ExecError, ServiceStartup, ServiceStatus
 
 from scenario import Context
-from scenario.state import Container, ExecOutput, Mount, Notice, State
+from scenario.state import Check, Container, ExecOutput, Mount, Notice, State
 from tests.helpers import jsonpatch_delta, trigger
 
 
@@ -381,7 +381,9 @@ def test_pebble_custom_notice(charm_cls):
 
     state = State(containers=[container])
     ctx = Context(charm_cls, meta={"name": "foo", "containers": {"foo": {}}})
-    with ctx.manager(container.get_notice("example.com/baz").event, state) as mgr:
+    with ctx.manager(
+        ctx.on.pebble_custom_notice(container=container, notice=notices[-1]), state
+    ) as mgr:
         container = mgr.charm.unit.get_container("foo")
         assert container.get_notices() == [n._to_ops() for n in notices]
 
@@ -437,4 +439,46 @@ def _on_custom_notice(self, event: PebbleCustomNoticeEvent):
     )
     state = State(containers=[container])
     ctx = Context(MyCharm, meta={"name": "foo", "containers": {"foo": {}}})
-    ctx.run(container.get_notice(key).event, state)
+    ctx.run(ctx.on.pebble_custom_notice(container=container, notice=notices[-1]), state)
+
+
+def test_pebble_check_failed():
+    class MyCharm(CharmBase):
+        def __init__(self, framework):
+            super().__init__(framework)
+            framework.observe(self.on.foo_pebble_check_failed, self._on_check_failed)
+
+        def _on_check_failed(self, event):
+            info = event.info
+            assert info.name == "http-check"
+            assert info.status == pebble.CheckStatus.DOWN
+            assert info.failures == 7
+
+    ctx = Context(MyCharm, meta={"name": "foo", "containers": {"foo": {}}})
+    check = Check("http-check", failures=7, status=pebble.CheckStatus.DOWN)
+    container = Container("foo", checks={check})
+    state = State(containers={container})
+    ctx.run(ctx.on.pebble_check_failed(check=check, container=container), state=state)
+
+
+def test_pebble_check_recovered():
+    class MyCharm(CharmBase):
+        def __init__(self, framework):
+            super().__init__(framework)
+            framework.observe(
+                self.on.foo_pebble_check_recovered, self._on_check_recovered
+            )
+
+        def _on_check_recovered(self, event):
+            info = event.info
+            assert info.name == "http-check"
+            assert info.status == pebble.CheckStatus.UP
+            assert info.failures == 0
+
+    ctx = Context(MyCharm, meta={"name": "foo", "containers": {"foo": {}}})
+    check = Check("http-check")
+    container = Container("foo", checks={check})
+    state = State(containers={container})
+    ctx.run(
+        ctx.on.pebble_check_recovered(check=check, container=container), state=state
+    )