Make it an error to call CollectStatusEvent.add_status with error or …

…unknown

ops/__init__.py

@@ -172,6 +172,8 @@
     'SecretNotFoundError',
     'SecretRotate',
     'ServiceInfoMapping',
+    'SettableStatus',
+    'SettableStatusName',
     'StatusBase',
     'StatusName',
     'Storage',
@@ -320,6 +322,8 @@
     SecretNotFoundError,
     SecretRotate,
     ServiceInfoMapping,
+    SettableStatus,
+    SettableStatusName,
     StatusBase,
     StatusName,
     Storage,

ops/charm.py

@@ -1105,12 +1105,10 @@ class CollectStatusEvent(LifecycleEvent):
 
     The order of priorities is as follows, from highest to lowest:
 
-    * error
     * blocked
     * maintenance
     * waiting
     * active
-    * unknown
 
     If there are multiple statuses with the same priority, the first one added
     wins (and if an event is observed multiple times, the handlers are called
@@ -1144,13 +1142,19 @@ def _on_collect_status(self, event: ops.CollectStatusEvent):
                 event.add_status(ops.ActiveStatus())
     """  # noqa: D405, D214, D411, D416  Final return confuses docstyle.
 
-    def add_status(self, status: model.StatusBase):
+    def add_status(self, status: model.SettableStatus) -> None:
         """Add a status for evaluation.
 
         See :class:`CollectStatusEvent` for a description of how to use this.
         """
         if not isinstance(status, model.StatusBase):
-            raise TypeError(f'status should be a StatusBase, not {type(status).__name__}')
+            raise TypeError(
+                f'status should be one of {model._SETTABLE_STATUSES}, not {type(status).__name__}'
+            )
+        if not isinstance(status, model._SETTABLE_STATUSES):
+            raise model.InvalidStatusError(
+                f'status should be one of {model._SETTABLE_STATUSES}, not {type(status).__name__}'
+            )
         model_ = self.framework.model
         if self.handle.kind == 'collect_app_status':
             if not isinstance(status, model.ActiveStatus):

ops/model.py

@@ -72,10 +72,10 @@
 _BindingDictType = Dict[Union[str, 'Relation'], 'Binding']
 
 _ReadOnlyStatusName = Literal['error', 'unknown']
-_SettableStatusName = Literal['active', 'blocked', 'maintenance', 'waiting']
-StatusName = Union[_SettableStatusName, _ReadOnlyStatusName]
+SettableStatusName = Literal['active', 'blocked', 'maintenance', 'waiting']
+StatusName = Union[SettableStatusName, _ReadOnlyStatusName]
 _StatusDict = TypedDict('_StatusDict', {'status': StatusName, 'message': str})
-_SETTABLE_STATUS_NAMES: Tuple[_SettableStatusName, ...] = get_args(_SettableStatusName)
+_SETTABLE_STATUS_NAMES: Tuple[SettableStatusName, ...] = get_args(SettableStatusName)
 
 # mapping from relation name to a list of relation objects
 _RelationMapping_Raw = Dict[str, Optional[List['Relation']]]
@@ -431,7 +431,7 @@ def status(self, value: 'StatusBase'):
             raise RuntimeError('cannot set application status as a non-leader unit')
 
         self._backend.status_set(
-            typing.cast(_SettableStatusName, value.name),  # status_set will validate at runtime
+            typing.cast(SettableStatusName, value.name),  # status_set will validate at runtime
             value.message,
             is_app=True,
         )
@@ -600,7 +600,7 @@ def status(self, value: 'StatusBase'):
             raise RuntimeError(f'cannot set status for a remote unit {self}')
 
         self._backend.status_set(
-            typing.cast(_SettableStatusName, value.name),  # status_set will validate at runtime
+            typing.cast(SettableStatusName, value.name),  # status_set will validate at runtime
             value.message,
             is_app=False,
         )
@@ -1918,8 +1918,26 @@ def __eq__(self, other: 'StatusBase') -> bool:
     def __repr__(self):
         return f'{self.__class__.__name__}({self.message!r})'
 
+    @typing.overload
+    @classmethod
+    def from_name(cls, name: Literal['active'], message: str) -> 'ActiveStatus': ...
+    @typing.overload
+    @classmethod
+    def from_name(cls, name: Literal['blocked'], message: str) -> 'BlockedStatus': ...
+    @typing.overload
+    @classmethod
+    def from_name(cls, name: Literal['error'], message: str) -> 'ErrorStatus': ...
+    @typing.overload
     @classmethod
-    def from_name(cls, name: str, message: str):
+    def from_name(cls, name: Literal['maintenance'], message: str) -> 'MaintenanceStatus': ...
+    @typing.overload
+    @classmethod
+    def from_name(cls, name: Literal['unknown'], message: str) -> 'UnknownStatus': ...
+    @typing.overload
+    @classmethod
+    def from_name(cls, name: Literal['waiting'], message: str) -> 'WaitingStatus': ...
+    @classmethod
+    def from_name(cls, name: str, message: str) -> 'StatusBase':
         """Create a status instance from a name and message.
 
         If ``name`` is "unknown", ``message`` is ignored, because unknown status
@@ -2059,6 +2077,10 @@ class WaitingStatus(StatusBase):
     name = 'waiting'
 
 
+SettableStatus = Union[ActiveStatus, BlockedStatus, MaintenanceStatus, WaitingStatus]
+_SETTABLE_STATUSES: Tuple[Type[SettableStatus], ...] = get_args(SettableStatus)
+
+
 class Resources:
     """Object representing resources for the charm."""
 
@@ -3443,7 +3465,7 @@ def status_get(self, *, is_app: bool = False) -> '_StatusDict':
             return typing.cast('_StatusDict', content)
 
     def status_set(
-        self, status: _SettableStatusName, message: str = '', *, is_app: bool = False
+        self, status: SettableStatusName, message: str = '', *, is_app: bool = False
     ) -> None:
         """Set a status of a unit or an application.
 

test/test_charm.py

@@ -22,7 +22,7 @@
 
 import ops
 import ops.charm
-from ops.model import ModelError, StatusName
+from ops.model import ModelError, SettableStatusName, StatusName
 
 from .test_helpers import FakeScript, create_framework
 
@@ -959,17 +959,16 @@ def _on_collect_status(self, event: ops.CollectStatusEvent):
         (['waiting', 'blocked'], 'blocked'),
         (['waiting', 'maintenance'], 'maintenance'),
         (['active', 'waiting'], 'waiting'),
-        (['active', 'unknown'], 'active'),
     ],
 )
 def test_collect_status_priority_valid(
     request: pytest.FixtureRequest,
     fake_script: FakeScript,
-    statuses: typing.List[StatusName],
+    statuses: typing.List[SettableStatusName],
     expected: str,
 ):
     class MyCharm(ops.CharmBase):
-        def __init__(self, framework: ops.Framework, statuses: typing.List[StatusName]):
+        def __init__(self, framework: ops.Framework, statuses: typing.List[SettableStatusName]):
             super().__init__(framework)
             self.framework.observe(self.on.collect_app_status, self._on_collect_status)
             self.statuses = statuses
@@ -994,6 +993,7 @@ def _on_collect_status(self, event: ops.CollectStatusEvent):
     [
         ['blocked', 'error'],
         ['unknown'],
+        ['active', 'unknown'],
     ],
 )
 def test_collect_status_priority_invalid(
@@ -1009,13 +1009,13 @@ def __init__(self, framework: ops.Framework, statuses: typing.List[StatusName]):
 
         def _on_collect_status(self, event: ops.CollectStatusEvent):
             for status in self.statuses:
-                event.add_status(ops.StatusBase.from_name(status, ''))
+                event.add_status(ops.StatusBase.from_name(status, ''))  # pyright: ignore[reportArgumentType]
 
     fake_script.write('is-leader', 'echo true')
 
     framework = create_framework(request)
     charm = MyCharm(framework, statuses=statuses)
-    with pytest.raises(ModelError):
+    with pytest.raises(ops.InvalidStatusError):
         ops.charm._evaluate_status(charm)
 
 

test/test_testing.py

@@ -3232,8 +3232,8 @@ def test_get_filesystem_root(self):
 
     def test_evaluate_status(self):
         class TestCharm(ops.CharmBase):
-            app_status_to_add: ops.StatusBase
-            unit_status_to_add: ops.StatusBase
+            app_status_to_add: ops.SettableStatus
+            unit_status_to_add: ops.SettableStatus
 
             def __init__(self, framework: ops.Framework):
                 super().__init__(framework)