fix: add pebble log targets and checks to testing plan

ops/testing.py

@@ -2931,23 +2931,49 @@ def add_layer(
         else:
             self._layers[label] = layer_obj
 
-    def _render_services(self) -> Dict[str, pebble.Service]:
-        services: Dict[str, pebble.Service] = {}
+    def _render_services(self) -> Dict[str, pebble.ServiceDict]:
+        services: Dict[str, pebble.ServiceDict] = {}
         for key in sorted(self._layers.keys()):
             layer = self._layers[key]
             for name, service in layer.services.items():
                 # TODO: (jam) 2021-04-07 have a way to merge existing services
-                services[name] = service
+                services[name] = service.to_dict()
         return services
 
+    def _render_checks(self) -> Dict[str, pebble.CheckDict]:
+        services: Dict[str, pebble.CheckDict] = {}
+        for key in sorted(self._layers.keys()):
+            layer = self._layers[key]
+            for name, check in layer.checks.items():
+                services[name] = check.to_dict()
+        return services
+
+    def _render_log_targets(self) -> Dict[str, pebble.LogTargetDict]:
+        services: Dict[str, pebble.LogTargetDict] = {}
+        for key in sorted(self._layers.keys()):
+            layer = self._layers[key]
+            for name, log_target in layer.log_targets.items():
+                services[name] = log_target.to_dict()
+        return services
+
+    def _render_plan(self) -> str:
+        raw_plan = {}
+        services = self._render_services()
+        if services:
+            raw_plan["services"] = services
+
+        checks = self._render_checks()
+        if checks:
+            raw_plan["checks"] = checks
+
+        log_targets = self._render_log_targets()
+        if log_targets:
+            raw_plan["log-targets"] = log_targets
+        return yaml.safe_dump(raw_plan)
+
     def get_plan(self) -> pebble.Plan:
         self._check_connection()
-        plan = pebble.Plan('{}')
-        services = self._render_services()
-        if not services:
-            return plan
-        for name in sorted(services.keys()):
-            plan.services[name] = services[name]
+        plan = pebble.Plan(self._render_plan())
         return plan
 
     def get_services(self, names: Optional[List[str]] = None) -> List[pebble.ServiceInfo]:

test/test_pebble.py

@@ -27,8 +27,10 @@
 
 import pytest
 import websocket  # type: ignore
+from testing import Harness
 
-from ops import pebble
+import ops
+from ops import Framework, pebble
 from ops._private import yaml
 
 # Ensure unittest diffs don't get truncated like "[17 chars]"
@@ -3599,3 +3601,42 @@ def recv():
     if hasattr(pytest, 'PytestUnhandledThreadExceptionWarning'):
         test_websocket_recv_raises = pytest.mark.filterwarnings(
             'ignore::pytest.PytestUnhandledThreadExceptionWarning')(test_websocket_recv_raises)
+
+
+class MyCharm(ops.CharmBase):
+    def __init__(self, framework: Framework):
+        super().__init__(framework)
+
+
+def test_add_layer_with_log_targets_to_plan():
+    raw = '''\
+    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
+    '''
+    h = Harness(MyCharm, meta=yaml.safe_dump(
+        {'name': 'foo', "containers": {"consumer": {"type": "oci-image"}}}))
+    h.begin()
+    h.set_can_connect('consumer', True)
+
+    c = h.charm.unit.containers["consumer"]
+    layer = pebble.Layer(raw)
+    c.add_layer('foo', layer)
+
+    c_plan = c.get_plan()
+
+    assert c_plan.services['foo']
+    assert c_plan.checks['bar']
+    assert c_plan.log_targets['baz']