Merge branch 'main' into chore/invalid_status

ops/model.py

@@ -2761,7 +2761,6 @@ def get_notices(
         user_id: Optional[int] = None,
         types: Optional[Iterable[Union[pebble.NoticeType, str]]] = None,
         keys: Optional[Iterable[str]] = None,
-        after: Optional[datetime.datetime] = None,
     ) -> List[pebble.Notice]:
         """Query for notices that match all of the provided filters.
 
@@ -2773,7 +2772,6 @@ def get_notices(
             user_id=user_id,
             types=types,
             keys=keys,
-            after=after,
         )
 
     # Define this last to avoid clashes with the imported "pebble" module

ops/pebble.py

@@ -2744,6 +2744,33 @@ def get_checks(
         resp = self._request('GET', '/v1/checks', query)
         return [CheckInfo.from_dict(info) for info in resp['result']]
 
+    def notify(self, type: NoticeType, key: str, *,
+               data: Optional[Dict[str, str]] = None,
+               repeat_after: Optional[datetime.timedelta] = None) -> str:
+        """Record an occurrence of a notice with the specified options.
+
+        Args:
+            type: Notice type (currently only "custom" notices are supported).
+            key: Notice key; must be in "example.com/path" format.
+            data: Data fields for this notice.
+            repeat_after: Only allow this notice to repeat after this duration
+                has elapsed (the default is to always repeat).
+
+        Returns:
+            The notice's ID.
+        """
+        body: Dict[str, Any] = {
+            'action': 'add',
+            'type': type.value,
+            'key': key,
+        }
+        if data is not None:
+            body['data'] = data
+        if repeat_after is not None:
+            body['repeat-after'] = _format_timeout(repeat_after.total_seconds())
+        resp = self._request('POST', '/v1/notices', body=body)
+        return resp['result']['id']
+
     def get_notice(self, id: str) -> Notice:
         """Get details about a single notice by ID.
 
@@ -2760,7 +2787,6 @@ def get_notices(
         user_id: Optional[int] = None,
         types: Optional[Iterable[Union[NoticeType, str]]] = None,
         keys: Optional[Iterable[str]] = None,
-        after: Optional[datetime.datetime] = None,
     ) -> List[Notice]:
         """Query for notices that match all of the provided filters.
 
@@ -2772,14 +2798,18 @@ def get_notices(
         user (notices whose ``user_id`` matches the requester UID as well as
         public notices).
 
+        Note that the "after" filter is not yet implemented, as it's not
+        needed right now and it's hard to implement correctly with Python's
+        datetime type, which only has microsecond precision (and Go's Time
+        type has nanosecond precision).
+
         Args:
-            select: select which notices to return (instead of returning
-                notices for the current user)
-            user_id: filter for notices for the specified user, including
-                public notices (only works for Pebble admins)
-            types: filter for notices with any of the specified types
-            keys: filter for notices with any of the specified keys
-            after: filter for notices that were last repeated after this time
+            select: Select which notices to return (instead of returning
+                notices for the current user).
+            user_id: Filter for notices for the specified user, including
+                public notices (only works for Pebble admins).
+            types: Filter for notices with any of the specified types.
+            keys: Filter for notices with any of the specified keys.
         """
         query: Dict[str, Union[str, List[str]]] = {}
         if select is not None:
@@ -2790,8 +2820,6 @@ def get_notices(
             query['types'] = [(t.value if isinstance(t, NoticeType) else t) for t in types]
         if keys is not None:
             query['keys'] = list(keys)
-        if after is not None:
-            query['after'] = after.isoformat()
         resp = self._request('GET', '/v1/notices', query)
         return [Notice.from_dict(info) for info in resp['result']]
 

ops/testing.py

@@ -18,6 +18,7 @@
 import dataclasses
 import datetime
 import fnmatch
+import http
 import inspect
 import io
 import ipaddress
@@ -1098,6 +1099,37 @@ def container_pebble_ready(self, container_name: str):
         self.set_can_connect(container, True)
         self.charm.on[container_name].pebble_ready.emit(container)
 
+    def pebble_notify(self, container_name: str, key: str, *,
+                      data: Optional[Dict[str, str]] = None,
+                      repeat_after: Optional[datetime.timedelta] = None,
+                      type: pebble.NoticeType = pebble.NoticeType.CUSTOM) -> str:
+        """Record a Pebble notice with the specified key and data.
+
+        If :meth:`begin` has been called and the notice is new or was repeated,
+        this will trigger a notice event of the appropriate type, for example
+        :class:`ops.PebbleCustomNoticeEvent`.
+
+        Args:
+            container_name: Name of workload container.
+            key: Notice key; must be in "example.com/path" format.
+            data: Data fields for this notice.
+            repeat_after: Only allow this notice to repeat after this duration
+                has elapsed (the default is to always repeat).
+            type: Notice type (currently only "custom" notices are supported).
+
+        Returns:
+            The notice's ID.
+        """
+        container = self.model.unit.get_container(container_name)
+        client = self._backend._pebble_clients[container.name]
+
+        id, new_or_repeated = client._notify(type, key, data=data, repeat_after=repeat_after)
+
+        if self._charm is not None and type == pebble.NoticeType.CUSTOM and new_or_repeated:
+            self.charm.on[container_name].pebble_custom_notice.emit(container, id, type.value, key)
+
+        return id
+
     def get_workload_version(self) -> str:
         """Read the workload version that was set by the unit."""
         return self._backend._workload_version
@@ -2733,6 +2765,8 @@ def __init__(self, backend: _TestingModelBackend, container_root: pathlib.Path):
         self._root = container_root
         self._backend = backend
         self._exec_handlers: Dict[Tuple[str, ...], ExecHandler] = {}
+        self._notices: Dict[Tuple[str, str], pebble.Notice] = {}
+        self._last_notice_id = 0
 
     def _handle_exec(self, command_prefix: Sequence[str], handler: ExecHandler):
         prefix = tuple(command_prefix)
@@ -3012,9 +3046,7 @@ def list_files(self, path: str, *, pattern: Optional[str] = None,
         self._check_absolute_path(path)
         file_path = self._root / path[1:]
         if not file_path.exists():
-            raise pebble.APIError(
-                body={}, code=404, status='Not Found',
-                message=f"stat {path}: no such file or directory")
+            raise self._api_error(404, f"stat {path}: no such file or directory")
         files = [file_path]
         if not itself:
             try:
@@ -3143,19 +3175,13 @@ def exec(
         handler = self._find_exec_handler(command)
         if handler is None:
             message = "execution handler not found, please register one using Harness.handle_exec"
-            raise pebble.APIError(
-                body={}, code=500, status='Internal Server Error', message=message
-            )
+            raise self._api_error(500, message)
         environment = {} if environment is None else environment
         if service_context is not None:
             plan = self.get_plan()
             if service_context not in plan.services:
                 message = f'context service "{service_context}" not found'
-                body = {'type': 'error', 'status-code': 500, 'status': 'Internal Server Error',
-                        'result': {'message': message}}
-                raise pebble.APIError(
-                    body=body, code=500, status='Internal Server Error', message=message
-                )
+                raise self._api_error(500, message)
             service = plan.services[service_context]
             environment = {**service.environment, **environment}
             working_dir = service.working_dir if working_dir is None else working_dir
@@ -3246,11 +3272,7 @@ def send_signal(self, sig: Union[int, str], service_names: Iterable[str]):
             if service not in plan.services or not self.get_services([service])[0].is_running():
                 # conform with the real pebble api
                 message = f'cannot send signal to "{service}": service is not running'
-                body = {'type': 'error', 'status-code': 500, 'status': 'Internal Server Error',
-                        'result': {'message': message}}
-                raise pebble.APIError(
-                    body=body, code=500, status='Internal Server Error', message=message
-                )
+                raise self._api_error(500, message)
 
         # Check if signal name is valid
         try:
@@ -3259,19 +3281,86 @@ def send_signal(self, sig: Union[int, str], service_names: Iterable[str]):
             # conform with the real pebble api
             first_service = next(iter(service_names))
             message = f'cannot send signal to "{first_service}": invalid signal name "{sig}"'
-            body = {'type': 'error', 'status-code': 500, 'status': 'Internal Server Error',
-                    'result': {'message': message}}
-            raise pebble.APIError(
-                body=body,
-                code=500,
-                status='Internal Server Error',
-                message=message) from None
+            raise self._api_error(500, message)
 
     def get_checks(self, level=None, names=None):  # type:ignore
         raise NotImplementedError(self.get_checks)  # type:ignore
 
+    def notify(self, type: pebble.NoticeType, key: str, *,
+               data: Optional[Dict[str, str]] = None,
+               repeat_after: Optional[datetime.timedelta] = None) -> str:
+        notice_id, _ = self._notify(type, key, data=data, repeat_after=repeat_after)
+        return notice_id
+
+    def _notify(self, type: pebble.NoticeType, key: str, *,
+                data: Optional[Dict[str, str]] = None,
+                repeat_after: Optional[datetime.timedelta] = None) -> Tuple[str, bool]:
+        """Record an occurrence of a notice with the specified details.
+
+        Return a tuple of (notice_id, new_or_repeated).
+        """
+        if type != pebble.NoticeType.CUSTOM:
+            message = f'invalid type "{type.value}" (can only add "custom" notices)'
+            raise self._api_error(400, message)
+
+        # The shape of the code below is taken from State.AddNotice in Pebble.
+        now = datetime.datetime.now(tz=datetime.timezone.utc)
+
+        new_or_repeated = False
+        unique_key = (type.value, key)
+        notice = self._notices.get(unique_key)
+        if notice is None:
+            # First occurrence of this notice uid+type+key
+            self._last_notice_id += 1
+            notice = pebble.Notice(
+                id=str(self._last_notice_id),
+                user_id=0,  # Charm should always be able to read pebble_notify notices.
+                type=type,
+                key=key,
+                first_occurred=now,
+                last_occurred=now,
+                last_repeated=now,
+                expire_after=datetime.timedelta(days=7),
+                occurrences=1,
+                last_data=data or {},
+                repeat_after=repeat_after,
+            )
+            self._notices[unique_key] = notice
+            new_or_repeated = True
+        else:
+            # Additional occurrence, update existing notice
+            last_repeated = notice.last_repeated
+            if repeat_after is None or now > notice.last_repeated + repeat_after:
+                # Update last repeated time if repeat-after time has elapsed (or is None)
+                last_repeated = now
+                new_or_repeated = True
+            notice = dataclasses.replace(
+                notice,
+                last_occurred=now,
+                last_repeated=last_repeated,
+                occurrences=notice.occurrences + 1,
+                last_data=data or {},
+                repeat_after=repeat_after,
+            )
+            self._notices[unique_key] = notice
+
+        return notice.id, new_or_repeated
+
+    def _api_error(self, code: int, message: str) -> pebble.APIError:
+        status = http.HTTPStatus(code).phrase
+        body = {
+            'type': 'error',
+            'status-code': code,
+            'status': status,
+            'result': {'message': message},
+        }
+        return pebble.APIError(body, code, status, message)
+
     def get_notice(self, id: str) -> pebble.Notice:
-        raise NotImplementedError(self.get_notice)
+        for notice in self._notices.values():
+            if notice.id == id:
+                return notice
+        raise self._api_error(404, f'cannot find notice with ID "{id}"')
 
     def get_notices(
         self,
@@ -3280,6 +3369,38 @@ def get_notices(
         user_id: Optional[int] = None,
         types: Optional[Iterable[Union[pebble.NoticeType, str]]] = None,
         keys: Optional[Iterable[str]] = None,
-        after: Optional[datetime.datetime] = None,
     ) -> List[pebble.Notice]:
-        raise NotImplementedError(self.get_notices)
+        # Similar logic as api_notices.go:v1GetNotices in Pebble.
+
+        filter_user_id = 0  # default is to filter by request UID (root)
+        if user_id is not None:
+            filter_user_id = user_id
+        if select is not None:
+            if user_id is not None:
+                raise self._api_error(400, 'cannot use both "select" and "user_id"')
+            filter_user_id = None
+
+        if types is not None:
+            types = {(t.value if isinstance(t, pebble.NoticeType) else t) for t in types}
+        if keys is not None:
+            keys = set(keys)
+
+        notices = [notice for notice in self._notices.values() if
+                   self._notice_matches(notice, filter_user_id, types, keys)]
+        notices.sort(key=lambda notice: notice.last_repeated)
+        return notices
+
+    @staticmethod
+    def _notice_matches(notice: pebble.Notice,
+                        user_id: Optional[int] = None,
+                        types: Optional[Set[str]] = None,
+                        keys: Optional[Set[str]] = None) -> bool:
+        # Same logic as NoticeFilter.matches in Pebble.
+        # For example: if user_id filter is set and it doesn't match, return False.
+        if user_id is not None and not (notice.user_id is None or user_id == notice.user_id):
+            return False
+        if types is not None and notice.type not in types:
+            return False
+        if keys is not None and notice.key not in keys:
+            return False
+        return True

requirements-dev.txt

@@ -11,6 +11,4 @@ pytest-operator~=0.23
 coverage[toml]~=7.0
 typing_extensions~=4.2
 
-macaroonbakery != 1.3.3  # TODO: remove once this is fixed: https://github.com/go-macaroon-bakery/py-macaroon-bakery/issues/94
-
 -r requirements.txt

test/test_model.py

@@ -1980,7 +1980,6 @@ def test_get_notices(self):
             select=pebble.NoticesSelect.ALL,
             types=[pebble.NoticeType.CUSTOM],
             keys=['example.com/a', 'example.com/b'],
-            after=datetime.datetime(2023, 12, 1, 2, 3, 4, 5, tzinfo=datetime.timezone.utc),
         )
         self.assertEqual(len(notices), 1)
         self.assertEqual(notices[0].id, '124')
@@ -1992,7 +1991,6 @@ def test_get_notices(self):
             select=pebble.NoticesSelect.ALL,
             types=[pebble.NoticeType.CUSTOM],
             keys=['example.com/a', 'example.com/b'],
-            after=datetime.datetime(2023, 12, 1, 2, 3, 4, 5, tzinfo=datetime.timezone.utc),
         ))])