chore: Document and validate settable status values in _ModelBackend.set_status

ops/__init__.py

@@ -162,6 +162,7 @@
     'SecretRotate',
     'ServiceInfoMapping',
     'StatusBase',
+    'StatusName',
     'Storage',
     'StorageMapping',
     'TooManyRelatedAppsError',
@@ -309,6 +310,7 @@
     SecretRotate,
     ServiceInfoMapping,
     StatusBase,
+    StatusName,
     Storage,
     StorageMapping,
     TooManyRelatedAppsError,

ops/_private/harness.py

@@ -61,7 +61,7 @@
 from ops._private import yaml
 from ops.charm import CharmBase, CharmMeta, RelationRole
 from ops.jujucontext import _JujuContext
-from ops.model import Container, RelationNotFoundError, _NetworkDict
+from ops.model import Container, RelationNotFoundError, StatusName, _NetworkDict
 from ops.pebble import ExecProcess
 
 ReadableBuffer = Union[bytes, str, StringIO, BytesIO, BinaryIO]
@@ -80,11 +80,10 @@
 
 _RelationEntities = TypedDict('_RelationEntities', {'app': str, 'units': List[str]})
 
-_StatusName = Literal['unknown', 'blocked', 'active', 'maintenance', 'waiting']
 _RawStatus = TypedDict(
     '_RawStatus',
     {
-        'status': _StatusName,
+        'status': StatusName,
         'message': str,
     },
 )
@@ -2491,7 +2490,7 @@ def status_get(self, *, is_app: bool = False):
         else:
             return self._unit_status
 
-    def status_set(self, status: '_StatusName', message: str = '', *, is_app: bool = False):
+    def status_set(self, status: 'StatusName', message: str = '', *, is_app: bool = False):
         if status in [model.ErrorStatus.name, model.UnknownStatus.name]:
             raise model.ModelError(
                 f'ERROR invalid status "{status}", expected one of'

ops/model.py

@@ -42,6 +42,7 @@
     Generator,
     Iterable,
     List,
+    Literal,
     Mapping,
     MutableMapping,
     Optional,
@@ -51,6 +52,7 @@
     Type,
     TypedDict,
     Union,
+    get_args,
 )
 
 import ops
@@ -69,7 +71,11 @@
 _StorageDictType = Dict[str, Optional[List['Storage']]]
 _BindingDictType = Dict[Union[str, 'Relation'], 'Binding']
 
-_StatusDict = TypedDict('_StatusDict', {'status': str, 'message': str})
+_ReadOnlyStatusName = Literal['error', 'unknown']
+_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)
 
 # mapping from relation name to a list of relation objects
 _RelationMapping_Raw = Dict[str, Optional[List['Relation']]]
@@ -424,9 +430,12 @@ def status(self, value: 'StatusBase'):
         if not self._backend.is_leader():
             raise RuntimeError('cannot set application status as a non-leader unit')
 
-        for _key in {'name', 'message'}:
-            assert isinstance(getattr(value, _key), str), f'status.{_key} must be a string'
-        self._backend.status_set(value.name, value.message, is_app=True)
+        self._backend.status_set(
+            typing.cast(_SettableStatusName, value.name),  # status_set will validate at runtime
+            value.message,
+            is_app=True,
+        )
+
         self._status = value
 
     def planned_units(self) -> int:
@@ -590,8 +599,11 @@ def status(self, value: 'StatusBase'):
         if not self._is_our_unit:
             raise RuntimeError(f'cannot set status for a remote unit {self}')
 
-        # fixme: if value.messages
-        self._backend.status_set(value.name, value.message, is_app=False)
+        self._backend.status_set(
+            typing.cast(_SettableStatusName, value.name),  # status_set will validate at runtime
+            value.message,
+            is_app=False,
+        )
         self._status = value
 
     def __repr__(self):
@@ -1885,10 +1897,10 @@ class StatusBase:
     directly use the child class such as :class:`ActiveStatus` to indicate their status.
     """
 
-    _statuses: Dict[str, Type['StatusBase']] = {}
+    _statuses: Dict[StatusName, Type['StatusBase']] = {}
 
-    # Subclasses should override this attribute
-    name = ''
+    # Subclasses must provide this attribute
+    name: StatusName
 
     def __init__(self, message: str = ''):
         if self.__class__ is StatusBase:
@@ -1911,7 +1923,8 @@ def from_name(cls, name: str, message: str):
         does not have an associated message.
 
         Args:
-            name: Name of the status, for example "active" or "blocked".
+            name: Name of the status, one of:
+                "active", "blocked", "maintenance", "waiting", "error", or "unknown".
             message: Message to include with the status.
 
         Raises:
@@ -1921,12 +1934,12 @@ def from_name(cls, name: str, message: str):
             # unknown is special
             return UnknownStatus()
         else:
-            return cls._statuses[name](message)
+            return cls._statuses[typing.cast(StatusName, name)](message)
 
     @classmethod
     def register(cls, child: Type['StatusBase']):
         """Register a Status for the child's name."""
-        if not isinstance(child.name, str):
+        if not (hasattr(child, 'name') and isinstance(child.name, str)):
             raise TypeError(
                 f"Can't register StatusBase subclass {child}: ",
                 'missing required `name: str` class attribute',
@@ -1960,7 +1973,7 @@ class UnknownStatus(StatusBase):
     charm has not called status-set yet.
 
     This status is read-only; trying to set unit or application status to
-    ``UnknownStatus`` will raise :class:`ModelError`.
+    ``UnknownStatus`` will raise :class:`InvalidStatusError`.
     """
 
     name = 'unknown'
@@ -1981,7 +1994,7 @@ class ErrorStatus(StatusBase):
     human intervention in order to operate correctly).
 
     This status is read-only; trying to set unit or application status to
-    ``ErrorStatus`` will raise :class:`ModelError`.
+    ``ErrorStatus`` will raise :class:`InvalidStatusError`.
     """
 
     name = 'error'
@@ -3415,13 +3428,15 @@ def status_get(self, *, is_app: bool = False) -> '_StatusDict':
         #       status-data: {}
 
         if is_app:
-            content = typing.cast(Dict[str, Dict[str, str]], content)
+            content = typing.cast(Dict[str, '_StatusDict'], content)
             app_status = content['application-status']
             return {'status': app_status['status'], 'message': app_status['message']}
         else:
             return typing.cast('_StatusDict', content)
 
-    def status_set(self, status: str, message: str = '', *, is_app: bool = False) -> None:
+    def status_set(
+        self, status: _SettableStatusName, message: str = '', *, is_app: bool = False
+    ) -> None:
         """Set a status of a unit or an application.
 
         Args:
@@ -3432,6 +3447,10 @@ def status_set(self, status: str, message: str = '', *, is_app: bool = False) ->
         """
         if not isinstance(is_app, bool):
             raise TypeError('is_app parameter must be boolean')
+        if not isinstance(message, str):
+            raise TypeError('message parameter must be a string')
+        if status not in _SETTABLE_STATUS_NAMES:
+            raise InvalidStatusError(f'status must be in {_SETTABLE_STATUS_NAMES}, not {status!r}')
         self._run('status-set', f'--application={is_app}', status, message)
 
     def storage_list(self, name: str) -> List[int]:

test/test_charm.py

@@ -22,7 +22,7 @@
 
 import ops
 import ops.charm
-from ops.model import ModelError
+from ops.model import ModelError, StatusName
 
 from .test_helpers import FakeScript, create_framework
 
@@ -956,22 +956,20 @@ def _on_collect_status(self, event: ops.CollectStatusEvent):
 @pytest.mark.parametrize(
     'statuses,expected',
     [
-        (['blocked', 'error'], 'error'),
         (['waiting', 'blocked'], 'blocked'),
         (['waiting', 'maintenance'], 'maintenance'),
         (['active', 'waiting'], 'waiting'),
         (['active', 'unknown'], 'active'),
-        (['unknown'], 'unknown'),
     ],
 )
-def test_collect_status_priority(
+def test_collect_status_priority_valid(
     request: pytest.FixtureRequest,
     fake_script: FakeScript,
-    statuses: typing.List[str],
+    statuses: typing.List[StatusName],
     expected: str,
 ):
     class MyCharm(ops.CharmBase):
-        def __init__(self, framework: ops.Framework, statuses: typing.List[str]):
+        def __init__(self, framework: ops.Framework, statuses: typing.List[StatusName]):
             super().__init__(framework)
             self.framework.observe(self.on.collect_app_status, self._on_collect_status)
             self.statuses = statuses
@@ -991,6 +989,36 @@ def _on_collect_status(self, event: ops.CollectStatusEvent):
     assert status_set_calls == [['status-set', '--application=True', expected, '']]
 
 
+@pytest.mark.parametrize(
+    'statuses',
+    [
+        ['blocked', 'error'],
+        ['unknown'],
+    ],
+)
+def test_collect_status_priority_invalid(
+    request: pytest.FixtureRequest,
+    fake_script: FakeScript,
+    statuses: typing.List[StatusName],
+):
+    class MyCharm(ops.CharmBase):
+        def __init__(self, framework: ops.Framework, statuses: typing.List[StatusName]):
+            super().__init__(framework)
+            self.framework.observe(self.on.collect_app_status, self._on_collect_status)
+            self.statuses = statuses
+
+        def _on_collect_status(self, event: ops.CollectStatusEvent):
+            for status in self.statuses:
+                event.add_status(ops.StatusBase.from_name(status, ''))
+
+    fake_script.write('is-leader', 'echo true')
+
+    framework = create_framework(request)
+    charm = MyCharm(framework, statuses=statuses)
+    with pytest.raises(ModelError):
+        ops.charm._evaluate_status(charm)
+
+
 def test_meta_links():
     # Each type of link can be a single item.
     meta = ops.CharmMeta.from_yaml("""

test/test_model.py

@@ -823,8 +823,8 @@ def test_base_status_instance_raises(self):
         class NoNameStatus(ops.StatusBase):
             pass
 
-        with pytest.raises(AttributeError):
-            ops.StatusBase.register_status(NoNameStatus)  # type: ignore
+        with pytest.raises(TypeError):
+            ops.StatusBase.register(NoNameStatus)
 
     def test_status_repr(self):
         test_cases = {
@@ -2878,34 +2878,28 @@ def test_status_is_app_forced_kwargs(self, fake_script: FakeScript):
                 case()
 
     def test_local_set_invalid_status(self, fake_script: FakeScript):
-        # juju returns exit code 1 if you ask to set status to 'unknown' or 'error'
+        # ops will directly raise InvalidStatusError if you try to set status to unknown or error
         meta = ops.CharmMeta.from_yaml("""
             name: myapp
         """)
         model = ops.Model(meta, self.backend)
-        fake_script.write('status-set', 'exit 1')
         fake_script.write('is-leader', 'echo true')
 
-        with pytest.raises(ops.ModelError):
+        with pytest.raises(ops.InvalidStatusError):
             model.unit.status = ops.UnknownStatus()
-        with pytest.raises(ops.ModelError):
+        with pytest.raises(ops.InvalidStatusError):
             model.unit.status = ops.ErrorStatus()
 
-        assert fake_script.calls(True) == [
-            ['status-set', '--application=False', 'unknown', ''],
-            ['status-set', '--application=False', 'error', ''],
-        ]
+        assert fake_script.calls(True) == []
 
-        with pytest.raises(ops.ModelError):
+        with pytest.raises(ops.InvalidStatusError):
             model.app.status = ops.UnknownStatus()
-        with pytest.raises(ops.ModelError):
+        with pytest.raises(ops.InvalidStatusError):
             model.app.status = ops.ErrorStatus()
 
         # A leadership check is needed for application status.
         assert fake_script.calls(True) == [
             ['is-leader', '--format=json'],
-            ['status-set', '--application=True', 'unknown', ''],
-            ['status-set', '--application=True', 'error', ''],
         ]
 
     @pytest.mark.parametrize('name', ['active', 'waiting', 'blocked', 'maintenance', 'error'])
@@ -2949,9 +2943,20 @@ def test_local_get_status(self, fake_script: FakeScript, name: str):
         assert model.app.status.message == 'bar'
 
     def test_status_set_is_app_not_bool_raises(self):
-        for is_app_v in [None, 1, 2.0, 'a', b'beef', object]:
+        for is_app_v in [None, 1, 2.0, 'a', b'beef', object()]:
             with pytest.raises(TypeError):
-                self.backend.status_set(ops.ActiveStatus, is_app=is_app_v)  # type: ignore
+                self.backend.status_set(
+                    'active',
+                    is_app=is_app_v,  # type: ignore[assignment]
+                )
+
+    def test_status_set_message_not_str_raises(self):
+        for message in [None, 1, 2.0, True, b'beef', object()]:
+            with pytest.raises(TypeError):
+                self.backend.status_set(
+                    'active',
+                    message=message,  # type: ignore[assignment]
+                )
 
     def test_storage_tool_errors(self, fake_script: FakeScript):
         fake_script.write('storage-list', 'echo fooerror >&2 ; exit 1')