feat: remove pebble enum | str unions

ops/_private/harness.py

@@ -3863,7 +3863,7 @@ def _notice_matches(
         # 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:
+        if types is not None and notice.type.value not in types:
             return False
         if keys is not None and notice.key not in keys:
             return False

ops/model.py

@@ -3047,7 +3047,7 @@ def get_notices(
         *,
         users: Optional[pebble.NoticesUsers] = None,
         user_id: Optional[int] = None,
-        types: Optional[Iterable[Union[pebble.NoticeType, str]]] = None,
+        types: Optional[Iterable[pebble.NoticeType]] = None,
         keys: Optional[Iterable[str]] = None,
     ) -> List[pebble.Notice]:
         """Query for notices that match all of the provided filters.

ops/pebble.py

@@ -134,7 +134,7 @@
     'CheckDict',
     {
         'override': str,
-        'level': Union['CheckLevel', str],
+        'level': str,
         'period': Optional[str],
         'timeout': Optional[str],
         'http': Optional[HttpDict],
@@ -197,7 +197,7 @@
 
 _ServiceInfoDict = TypedDict(
     '_ServiceInfoDict',
-    {'startup': Union['ServiceStartup', str], 'current': Union['ServiceStatus', str], 'name': str},
+    {'startup': str, 'current': str, 'name': str},
 )
 
 # Callback types for _MultiParser header and body handlers
@@ -242,8 +242,8 @@ def __enter__(self) -> typing.IO[typing.AnyStr]: ...
         '_CheckInfoDict',
         {
             'name': str,
-            'level': NotRequired[Optional[Union['CheckLevel', str]]],
-            'status': Union['CheckStatus', str],
+            'level': NotRequired[Optional[str]],
+            'status': str,
             'failures': NotRequired[int],
             'threshold': int,
             'change-id': NotRequired[str],
@@ -261,7 +261,7 @@ def __enter__(self) -> typing.IO[typing.AnyStr]: ...
             'user': NotRequired[Optional[str]],
             'group-id': NotRequired[Optional[int]],
             'group': NotRequired[Optional[str]],
-            'type': Union['FileType', str],
+            'type': str,
         },
     )
 
@@ -1038,6 +1038,7 @@ class ServiceStartup(enum.Enum):
 
     ENABLED = 'enabled'
     DISABLED = 'disabled'
+    UNKNOWN = 'unknown'
 
 
 class ServiceStatus(enum.Enum):
@@ -1046,6 +1047,7 @@ class ServiceStatus(enum.Enum):
     ACTIVE = 'active'
     INACTIVE = 'inactive'
     ERROR = 'error'
+    UNKNOWN = 'unknown'
 
 
 class ServiceInfo:
@@ -1054,15 +1056,20 @@ class ServiceInfo:
     def __init__(
         self,
         name: str,
-        startup: Union[ServiceStartup, str],
-        current: Union[ServiceStatus, str],
+        startup: ServiceStartup,
+        current: ServiceStatus,
     ):
         self.name = name
         self.startup = startup
         self.current = current
 
     def is_running(self) -> bool:
         """Return True if this service is running (in the active state)."""
+        if not isinstance(self.current, ServiceStatus):
+            raise ValueError(
+                'self.current must be a ServiceStatus member (e.g. ServiceStatus.ACTIVE)'
+                f', not {self.current}'
+            )
         return self.current == ServiceStatus.ACTIVE
 
     @classmethod
@@ -1071,11 +1078,13 @@ def from_dict(cls, d: _ServiceInfoDict) -> ServiceInfo:
         try:
             startup = ServiceStartup(d['startup'])
         except ValueError:
-            startup = d['startup']
+            warnings.warn(f'Unknown ServiceStartup value {d["startup"]!r}')
+            startup = ServiceStartup.UNKNOWN
         try:
             current = ServiceStatus(d['current'])
         except ValueError:
-            current = d['current']
+            warnings.warn(f'Unknown ServiceStatus value {d["current"]!r}')
+            current = ServiceStatus.UNKNOWN
         return cls(
             name=d['name'],
             startup=startup,
@@ -1098,11 +1107,16 @@ def __init__(self, name: str, raw: Optional[CheckDict] = None):
         self.name = name
         dct: CheckDict = raw or {}
         self.override: str = dct.get('override', '')
+        self.level: CheckLevel
+        dct_level = dct.get('level', '')
         try:
-            level: Union[CheckLevel, str] = CheckLevel(dct.get('level', ''))
+            self.level = CheckLevel(dct_level)  # CheckLevel('') -> CheckLevel.UNSET
         except ValueError:
-            level = dct.get('level', '')
-        self.level = level
+            warnings.warn(f'Unknown CheckLevel value {dct_level!r}')
+            self.level = CheckLevel.UNKNOWN
+        #
+        # these are all Optional in CheckDict and here, why '' instead of None?
+        # note that all falsey values are filtered out in to_dict
         self.period: Optional[str] = dct.get('period', '')
         self.timeout: Optional[str] = dct.get('timeout', '')
         self.threshold: Optional[int] = dct.get('threshold')
@@ -1124,7 +1138,20 @@ def __init__(self, name: str, raw: Optional[CheckDict] = None):
 
     def to_dict(self) -> CheckDict:
         """Convert this check object to its dict representation."""
-        level: str = self.level.value if isinstance(self.level, CheckLevel) else self.level
+        # is this the behaviour we want for unknown?
+        level: str = '' if self.level is CheckLevel.UNKNOWN else self.level.value
+        # it seems unfortunate that we'd have
+        #
+        #  d = {'level'='unknown-level'}
+        # c1 = Check.from_dict(d)
+        # c2 = Check.from_dict(c1.to_dict())
+        # c1.level == CheckLevel.UNKNOWN
+        # c2.level == CheckLevel.UNSET
+        #
+        # so do we want to just use 'unknown' for CheckLevel.UNKNOWN here?
+        # we still get a warning on making `c1` but not on `c2`, but maybe that's ok
+        # but is it useful to have 'unknown' in the dict?
+        # It'll error if that value makes it to pebble, right?
         fields = [
             ('override', self.override),
             ('level', level),
@@ -1205,9 +1232,20 @@ def _merge(self, other: Check):
         For attributes present in both objects, the passed in check
         attributes take precedence.
         """
+        # wouldn't it be better to use other.to_dict?
+        # it would let us skip these checks
         for name, value in other.__dict__.items():
             if not value or name == 'name':
                 continue
+            # how to handle CheckLevel.UNKNOWN?
+            #
+            # it would be skipped here if we used other.to_dict
+            #
+            # previously it would have been <some-unknown-value-string>
+            # while we might want <some-unknown-value-string> to overwrite,
+            # would we ever want Checklevel.UNKNOWN to overwrite?
+            if value is CheckLevel.UNKNOWN:
+                continue
             if name == 'http':
                 self._merge_http(value)
             elif name == 'tcp':
@@ -1224,13 +1262,15 @@ class CheckLevel(enum.Enum):
     UNSET = ''
     ALIVE = 'alive'
     READY = 'ready'
+    UNKNOWN = 'unknown'
 
 
 class CheckStatus(enum.Enum):
     """Enum of check statuses."""
 
     UP = 'up'
     DOWN = 'down'
+    UNKNOWN = 'unknown'
 
 
 class LogTarget:
@@ -1308,8 +1348,8 @@ class FileInfo:
     name: str
     """Base name of the file."""
 
-    type: Union[FileType, str]
-    """Type of the file ("file", "directory", "symlink", etc)."""
+    type: FileType
+    """Type of the file (FileType.FILE, FileType.DIRECTORY, FileType.SYMLINK, etc)."""
 
     size: Optional[int]
     """Size of the file (will be 0 if ``type`` is not "file")."""
@@ -1336,7 +1376,7 @@ def __init__(
         self,
         path: str,
         name: str,
-        type: Union[FileType, str],
+        type: FileType,
         size: Optional[int],
         permissions: int,
         last_modified: datetime.datetime,
@@ -1362,7 +1402,8 @@ def from_dict(cls, d: _FileInfoDict) -> FileInfo:
         try:
             file_type = FileType(d['type'])
         except ValueError:
-            file_type = d['type']
+            warnings.warn(f'Unknown FileType value {d["type"]!r}')
+            file_type = FileType.UNKNOWN
         return cls(
             path=d['path'],
             name=d['name'],
@@ -1401,13 +1442,16 @@ class CheckInfo:
     name: str
     """Name of the check."""
 
-    level: Optional[Union[CheckLevel, str]]
+    level: Optional[CheckLevel]
     """Check level.
 
     This can be :attr:`CheckLevel.ALIVE`, :attr:`CheckLevel.READY`, or None (level not set).
     """
+    # suspicious that we're documenting None as the value for unset ...
+    # from_dict will use CheckLevel.UNSET if no value is provided for level
+    # but it will pass along None, since that isn't defined in CheckLevel
 
-    status: Union[CheckStatus, str]
+    status: CheckStatus
     """Status of the check.
 
     :attr:`CheckStatus.UP` means the check is healthy (the number of failures
@@ -1437,8 +1481,8 @@ class CheckInfo:
     def __init__(
         self,
         name: str,
-        level: Optional[Union[CheckLevel, str]],
-        status: Union[CheckStatus, str],
+        level: Optional[CheckLevel],
+        status: CheckStatus,
         failures: int = 0,
         threshold: int = 0,
         change_id: Optional[ChangeID] = None,
@@ -1453,14 +1497,25 @@ def __init__(
     @classmethod
     def from_dict(cls, d: _CheckInfoDict) -> CheckInfo:
         """Create new :class:`CheckInfo` object from dict parsed from JSON."""
-        try:
-            level = CheckLevel(d.get('level', ''))
-        except ValueError:
-            level = d.get('level')
+        d_level = d.get('level', '')  # CheckLevel('') -> CheckLevel.UNSET
+        if d_level is None:
+            # previously None would get a ValueError and end up with level = None
+            # the same way that we'd have handled 'my-weird-new-value-for-level'
+            # but it's documented as a valid argument for CheckInfo, indicating not set
+            # so we shouldn't turn it into CheckLevel.UNKNOWN
+            # TODO: look into turning it into CheckLevel.UNSET instead? Or forbid it?
+            level = None
+        else:
+            try:
+                level = CheckLevel(d_level)
+            except ValueError:
+                warnings.warn(f'Unknown CheckLevel value {d_level!r}')
+                level = CheckLevel.UNKNOWN
         try:
             status = CheckStatus(d['status'])
         except ValueError:
-            status = d['status']
+            warnings.warn(f'Unknown CheckStatus value {d["status"]!r}')
+            status = CheckStatus.UNKNOWN
         return cls(
             name=d['name'],
             level=level,
@@ -1496,6 +1551,9 @@ class NoticeType(enum.Enum):
     notices is the change ID.
     """
 
+    UNKNOWN = 'unknown'
+    """Used if we receive an unrecognised notice type (e.g. from future Pebble versions)."""
+
 
 class NoticesUsers(enum.Enum):
     """Enum of :meth:`Client.get_notices` ``users`` values."""
@@ -1562,7 +1620,7 @@ class Notice:
     user_id: Optional[int]
     """UID of the user who may view this notice (None means notice is public)."""
 
-    type: Union[NoticeType, str]
+    type: NoticeType
     """Type of the notice."""
 
     key: str
@@ -1602,7 +1660,8 @@ def from_dict(cls, d: _NoticeDict) -> Notice:
         try:
             notice_type = NoticeType(d['type'])
         except ValueError:
-            notice_type = d['type']
+            warnings.warn(f'Unknown NoticeType value {d["type"]!r}')
+            notice_type = NoticeType.UNKNOWN
         return cls(
             id=d['id'],
             user_id=d.get('user-id'),
@@ -3046,6 +3105,9 @@ def get_checks(
         """
         query: Dict[str, Any] = {}
         if level is not None:
+            # is this bad now that value could be 'unknown'?
+            # should we explicitly avoid it here?
+            # if level is not None and level.value is not CheckLevel.UNKNOWN:
             query['level'] = level.value
         if names:
             query['names'] = list(names)
@@ -3098,7 +3160,7 @@ def get_notices(
         *,
         users: Optional[NoticesUsers] = None,
         user_id: Optional[int] = None,
-        types: Optional[Iterable[Union[NoticeType, str]]] = None,
+        types: Optional[Iterable[NoticeType]] = None,
         keys: Optional[Iterable[str]] = None,
     ) -> List[Notice]:
         """Query for notices that match all of the provided filters.

test/test_charm.py

@@ -346,7 +346,7 @@ def mock_check_info(
         return [
             ops.pebble.CheckInfo.from_dict({
                 'name': names[0],
-                'status': ops.pebble.CheckStatus.DOWN,
+                'status': ops.pebble.CheckStatus.DOWN.value,
                 'failures': 3,
                 'threshold': 3,
             })