Allow to enter troubleshooting by raising a CheckError exception

src/lvmopstools/actor.py

@@ -29,6 +29,7 @@
 
 __all__ = [
     "LVMActor",
+    "CheckError",
     "ActorState",
     "ErrorCodes",
     "ErrorData",
@@ -122,6 +123,16 @@ class ErrorCodes(enum.Enum):
 
     UNKNOWN = ErrorData(9999, True, "Unknown error.")
 
+    @classmethod
+    def get_error_code(cls, error_code: int):
+        """Returns the :obj:`.ErrorCodes` that matches the ``error_code`` value."""
+
+        for error in cls:
+            if error.value.code == error_code:
+                return error
+
+        raise ValueError(f"Error code {error_code} not found.")
+
 
 @click.command(cls=CluCommand, name="actor-state")
 async def actor_state(command: Command[LVMActor], *args, **kwargs):
@@ -141,6 +152,24 @@ async def actor_state(command: Command[LVMActor], *args, **kwargs):
     return command.finish(state={"code": code, "flags": flags, "error": None})
 
 
+class CheckError(Exception):
+    """An exception raised when the :obj:`.LVMActor` check fails."""
+
+    def __init__(
+        self,
+        message: str = "",
+        error_code: ErrorCodes | int = ErrorCodes.UNKNOWN,
+    ):
+        if isinstance(error_code, int):
+            self.error_code = ErrorCodes.get_error_code(error_code)
+        else:
+            self.error_code = error_code
+
+        self.message = message
+
+        super().__init__(message)
+
+
 class LVMActor(AMQPActor):
     """Base class for LVM actors."""
 
@@ -208,6 +237,8 @@ async def _check_loop(self):
 
             try:
                 await self._check_internal()
+            except CheckError as err:
+                await self.troubleshoot(error_code=err.error_code, exception=err)
             except Exception as err:
                 await self.troubleshoot(exception=err, traceback_frame=1)
             else:
@@ -320,6 +351,8 @@ async def _check_internal(self):
         :obj:`.LVMActor`. It is called by :obj:`.check_loop` every ``check_interval``
         seconds. The method must perform any necessary checks and, if a problem
         is found, call :obj:`.troubleshoot` with the appropriate error code.
+        Alternatively if the check fails it can raise a :obj:`.CheckError` with
+        the appropriate error code.
 
         """
 

tests/test_actor.py

@@ -20,6 +20,7 @@
 import lvmopstools.actor
 from lvmopstools.actor import (
     ActorState,
+    CheckError,
     ErrorCodes,
     ErrorData,
     LVMActor,
@@ -70,8 +71,21 @@ async def test_actor_state_no_model(lvm_actor: LVMActor):
     assert cmd.replies[-1].body["state"]["error"] is None
 
 
-async def test_actor_check_fails(lvm_actor: LVMActor, mocker):
-    lvm_actor._check_internal = mocker.AsyncMock(side_effect=ValueError("Test error"))
+def test_get_error_codes():
+    assert ErrorCodes.UNKNOWN == ErrorCodes.get_error_code(9999)
+
+
+def test_get_error_codes_not_valid():
+    with pytest.raises(ValueError):
+        ErrorCodes.get_error_code(999999)
+
+
+@pytest.mark.parametrize(
+    "side_effect",
+    [ValueError("Test error"), CheckError("Test error", ErrorCodes.UNKNOWN)],
+)
+async def test_actor_check_fails(lvm_actor: LVMActor, mocker, side_effect: Exception):
+    lvm_actor._check_internal = mocker.AsyncMock(side_effect=side_effect)
 
     # Restart the check loop
     await cancel_task(lvm_actor._check_task)