feat(monitors): add gmirror monitor

Add a monitor which can check the status of a gmirror RAID array.

docs/monitors/gmirror.rst

@@ -0,0 +1,20 @@
+.. _gmirror:
+
+gmirror - check gmirror array status
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Shells out to ``gmirror`` to check array status
+
+.. confval:: array_device
+
+   :type: string
+   :required: true
+
+   The device to check (e.g., ``gm0``).
+
+.. confval:: expected_disks
+
+   :type: int
+   :required: true
+
+   Number of expected members of the given array.

simplemonitor/Monitors/__init__.py

@@ -5,6 +5,7 @@
 from .arlo import MonitorArloCamera
 from .compound import CompoundMonitor, RemoteHostsMonitor
 from .file import MonitorBackup
+from .gmirror import MonitorGmirrorStatus
 from .hass import MonitorSensor
 from .host import (
     MonitorApcupsd,
@@ -49,6 +50,7 @@
     "MonitorDiskSpace",
     "MonitorEximQueue",
     "MonitorFileStat",
+    "MonitorGmirrorStatus",
     "MonitorHTTP",
     "MonitorHost",
     "MonitorLoadAvg",

simplemonitor/Monitors/gmirror.py

@@ -0,0 +1,56 @@
+"""
+Gmirror array checks for simplemonitor.
+"""
+
+import subprocess
+
+from .monitor import Monitor, register
+
+
+@register
+class MonitorGmirrorStatus(Monitor):
+    """
+    Check gmirror status for specified device.
+    """
+
+    monitor_type = "gmirror_status"
+
+    def __init__(self, name: str, config_options: dict) -> None:
+        super().__init__(name, config_options)
+        self.array_device = self.get_config_option(
+            "array_device", required_type="str", required=True
+        )
+        self.expected_disks = self.get_config_option(
+            "expected_disks", required_type="int", required=True
+        )
+
+    def run_test(self) -> bool:
+        """
+        Run `gmirror status` for the specified device. Keep the logic simpler
+        by requiring each device and the # of expected disks to be specified
+        separately.
+        """
+
+        run_cmd = ["gmirror", "status", "-gs", self.array_device]
+        try:
+            result = subprocess.run(run_cmd, capture_output=True, check=True)
+        except subprocess.CalledProcessError:
+            return self.record_fail("gmirror command failed")
+
+        status_lines = result.stdout.decode("utf-8").rstrip("\n").split("\n")
+
+        # Status should be same for the array, so just grab first line
+        status = status_lines[0].split()[1]
+
+        # Infer # of disks based on the number of lines
+        disk_count = len(status_lines)
+
+        msg = f"Array {self.array_device} is in state {status} with {disk_count} disks"
+
+        if status == "COMPLETE" and disk_count == self.expected_disks:
+            return self.record_success(msg)
+
+        return self.record_fail(msg)
+
+    def describe(self) -> str:
+        return f"Check RAID status of {self.array_device}."

tests/test_gmirror.py

@@ -0,0 +1,113 @@
+# type: ignore
+import subprocess
+import unittest
+from unittest.mock import MagicMock, patch
+
+from simplemonitor.Monitors import gmirror
+
+DEFAULT_CONFIG_OPTIONS = {"array_device": "gm0", "expected_disks": 2}
+
+MOCK_OUTPUT_GOOD = """gm0  COMPLETE  ada0 (ACTIVE)
+gm0  COMPLETE  ada1 (ACTIVE)
+"""
+MOCK_OUTPUT_SYNCHRONIZING = """gm0  DEGRADED  ada0 (ACTIVE)
+gm0  DEGRADED  ada1 (SYNCHRONIZING, 7%)
+"""
+MOCK_OUTPUT_BAD = """gm0  DEGRADED  ada0 (ACTIVE)
+"""
+
+
+class TestGmirrorStatusMonitors(unittest.TestCase):
+    @patch("subprocess.run")
+    def test_GmirrorStatus_success(self, mock_run):
+        """Success / happy path tests."""
+
+        mock_stdout = MagicMock()
+        mock_stdout.configure_mock(**{"stdout.decode.return_value": MOCK_OUTPUT_GOOD})
+        mock_run.return_value = mock_stdout
+
+        m = gmirror.MonitorGmirrorStatus("test", DEFAULT_CONFIG_OPTIONS)
+
+        m.run_test()
+
+        mock_run.assert_called_with(
+            ["gmirror", "status", "-gs", DEFAULT_CONFIG_OPTIONS.get("array_device")],
+            capture_output=True,
+            check=True,
+        )
+        self.assertEqual("Array gm0 is in state COMPLETE with 2 disks", m.get_result())
+        self.assertTrue(m.test_success())
+        self.assertEqual(m.error_count, 0)
+
+        m.run_test()
+
+        self.assertTrue(m.test_success())
+        self.assertEqual(m.error_count, 0)
+
+    @patch("subprocess.run")
+    def test_GmirrorStatus_failedSynchronizing(self, mock_run):
+        """Check failure test cases."""
+
+        mock_stdout = MagicMock()
+        mock_stdout.configure_mock(
+            **{"stdout.decode.return_value": MOCK_OUTPUT_SYNCHRONIZING}
+        )
+        mock_run.return_value = mock_stdout
+
+        m = gmirror.MonitorGmirrorStatus("test", DEFAULT_CONFIG_OPTIONS)
+
+        m.run_test()
+
+        mock_run.assert_called()
+        self.assertEqual(m.get_result(), "Array gm0 is in state DEGRADED with 2 disks")
+        self.assertFalse(m.test_success())
+        self.assertEqual(m.error_count, 1)
+
+        m.run_test()
+
+        self.assertFalse(m.test_success())
+        self.assertEqual(m.error_count, 2)
+
+    @patch("subprocess.run")
+    def test_GmirrorStatus_failedMissingDisk(self, mock_run):
+        """Check failure test cases."""
+
+        mock_stdout = MagicMock()
+        mock_stdout.configure_mock(**{"stdout.decode.return_value": MOCK_OUTPUT_BAD})
+        mock_run.return_value = mock_stdout
+
+        m = gmirror.MonitorGmirrorStatus("test", DEFAULT_CONFIG_OPTIONS)
+
+        m.run_test()
+
+        mock_run.assert_called()
+        self.assertEqual(m.get_result(), "Array gm0 is in state DEGRADED with 1 disks")
+        self.assertFalse(m.test_success())
+        self.assertEqual(m.error_count, 1)
+
+    @patch("subprocess.run")
+    def test_GmirrorStatus_raiseFileNotFoundError(self, mock_run):
+        """Make sure the program raises if the binary isn't present at all."""
+
+        mock_run.side_effect = FileNotFoundError(
+            "[Errno 2] No such file or directory: 'gmirror'"
+        )
+        m = gmirror.MonitorGmirrorStatus("test", DEFAULT_CONFIG_OPTIONS)
+        self.assertRaises(FileNotFoundError, m.run_test)
+
+    @patch("subprocess.run")
+    def test_GmirrorStatus_failureSubprocessFailure(self, mock_run):
+        """Handle failure based on command failing with non-0 exit code."""
+
+        mock_run.side_effect = subprocess.CalledProcessError(
+            1,
+            ["gmirror", "status", "-gs", DEFAULT_CONFIG_OPTIONS.get("array_device")],
+        )
+        m = gmirror.MonitorGmirrorStatus("test", DEFAULT_CONFIG_OPTIONS)
+        m.run_test()
+        self.assertFalse(m.test_success())
+        self.assertEqual(m.error_count, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()