feat: add the ability to compare two Plan objects with == and create a Plan from a dict

CHANGES.md

@@ -2,6 +2,8 @@
 
 * `StopEvent`, `RemoveEvent`, and all `LifeCycleEvent`s are no longer deferrable, and will raise a `RuntimeError` if `defer()` is called on the event object.
 * Added `ActionEvent.id`, exposing the JUJU_ACTION_UUID environment variable.
+* Added support for creating `pebble.Plan` objects by passing in a `pebble.PlanDict`, and the
+  ability to compare two `Plan` objects with `==`.
 
 # 2.10.0
 

docs/conf.py

@@ -22,12 +22,14 @@
 
 _old_compute_navigation_tree = furo._compute_navigation_tree
 
+
 def _compute_navigation_tree(context):
     tree_html = _old_compute_navigation_tree(context)
     if not tree_html and context.get("toc"):
         tree_html = furo.navigation.get_navigation_tree(context["toc"])
     return tree_html
 
+
 furo._compute_navigation_tree = _compute_navigation_tree
 
 

ops/pebble.py

@@ -179,13 +179,17 @@ def __call__(self, data: bytes, done: bool = False) -> None: ...
 
 class _Tempfile(Protocol):
     name = ''
+
     def write(self, data: bytes): ...
+
     def close(self): ...
 
 
 class _FileLikeIO(Protocol[typing.AnyStr]):  # That also covers TextIO and BytesIO
     def read(self, __n: int = ...) -> typing.AnyStr: ...  # for BinaryIO
+
     def write(self, __s: typing.AnyStr) -> int: ...
+
     def __enter__(self) -> typing.IO[typing.AnyStr]: ...
 
 
@@ -276,9 +280,13 @@ def __enter__(self) -> typing.IO[typing.AnyStr]: ...
 
 class _WebSocket(Protocol):
     def connect(self, url: str, socket: socket.socket): ...
+
     def shutdown(self): ...
+
     def send(self, payload: str): ...
+
     def send_binary(self, payload: bytes): ...
+
     def recv(self) -> Union[str, bytes]: ...
 
 
@@ -735,8 +743,11 @@ class Plan:
     documented at https://github.com/canonical/pebble/#layer-specification.
     """
 
-    def __init__(self, raw: str):
-        d = yaml.safe_load(raw) or {}  # type: ignore
+    def __init__(self, raw: Optional[Union[str, 'PlanDict']]):
+        if isinstance(raw, str):  # noqa: SIM108
+            d = yaml.safe_load(raw) or {}  # type: ignore
+        else:
+            d = raw or {}
         d = typing.cast('PlanDict', d)
 
         self._raw = raw
@@ -788,6 +799,13 @@ def to_yaml(self) -> str:
 
     __str__ = to_yaml
 
+    def __eq__(self, __value: Union['PlanDict', 'Plan']) -> bool:
+        if isinstance(__value, dict):
+            return self.to_dict() == __value
+        elif isinstance(__value, Plan):
+            return self.to_dict() == __value.to_dict()
+        return super().__eq__(__value)
+
 
 class Layer:
     """Represents a Pebble configuration layer.

test/test_pebble.py

@@ -589,6 +589,35 @@ def test_yaml(self):
         self.assertEqual(plan.to_yaml(), reformed)
         self.assertEqual(str(plan), reformed)
 
+    def test_plandict(self):
+        # Starting with nothing, we get the empty result.
+        plan = pebble.Plan(pebble.PlanDict({}))
+        self.assertEqual(plan.to_dict(), {})
+
+        # With a service, we return validated yaml content.
+        raw = pebble.PlanDict({
+            "services": {
+                "foo": pebble.ServiceDict({
+                    "override": "replace",
+                    "command": "echo foo",
+                }),
+            },
+            "checks": {
+                "bar": pebble.CheckDict({
+                    "http": pebble.HttpDict({"url": "https://example.com/"}),
+                }),
+            },
+            "log-targets": {
+                "baz": pebble.LogTargetDict({
+                    "override": "replace",
+                    "type": "loki",
+                    "location": "https://example.com:3100/loki/api/v1/push",
+                }),
+            }
+        })
+        plan = pebble.Plan(raw)
+        self.assertEqual(plan.to_dict(), raw)
+
     def test_service_equality(self):
         plan = pebble.Plan("""
 services:
@@ -610,6 +639,110 @@ def test_service_equality(self):
         }
         self.assertEqual(plan.services, services_as_dict)
 
+    def test_plan_equality(self):
+        plan1 = pebble.Plan('''
+services:
+  foo:
+    override: replace
+    command: echo foo
+''')
+        self.assertFalse(plan1 == "foo")
+        plan2 = pebble.Plan('''
+services:
+  foo:
+    command: echo foo
+    override: replace
+''')
+        self.assertTrue(plan1 == plan2)
+        plan1_as_dict = pebble.PlanDict({
+            "services": {
+                "foo": pebble.ServiceDict({
+                    "command": "echo foo",
+                    "override": "replace",
+                })
+            },
+        })
+        self.assertTrue(plan1, plan1_as_dict)
+        plan3 = pebble.Plan('''
+services:
+  foo:
+    override: replace
+    command: echo bar
+''')
+        self.assertFalse(plan1 == plan3)
+        plan4 = pebble.Plan('''
+services:
+ foo:
+  override: replace
+  command: echo foo
+
+checks:
+ bar:
+  http:
+   https://example.com/
+
+log-targets:
+ baz:
+  override: replace
+  type: loki
+  location: https://example.com:3100/loki/api/v1/push
+''')
+        plan5 = pebble.Plan('''
+services:
+ foo:
+  override: replace
+  command: echo foo
+
+checks:
+ bar:
+  http:
+   https://example.org/
+
+log-targets:
+ baz:
+  override: replace
+  type: loki
+  location: https://example.com:3100/loki/api/v1/push
+''')
+        self.assertFalse(plan4 == plan5)
+        plan6 = pebble.Plan('''
+services:
+ foo:
+  override: replace
+  command: echo foo
+
+checks:
+ bar:
+  http:
+   https://example.com/
+
+log-targets:
+ baz:
+  override: replace
+  type: loki
+  location: https://example.com:3200/loki/api/v1/push
+''')
+        self.assertFalse(plan4 == plan6)
+        plan7 = pebble.Plan('''
+services:
+ foo:
+  command: echo foo
+  override: replace
+
+log-targets:
+ baz:
+  type: loki
+  override: replace
+  location: https://example.com:3100/loki/api/v1/push
+
+checks:
+ bar:
+  http:
+   https://example.com/
+
+''')
+        self.assertTrue(plan4 == plan7)
+
 
 class TestLayer(unittest.TestCase):
     def _assert_empty(self, layer: pebble.Layer):