Refactor get hypervisor state

Update HypervisorSate enum so it's easier to understand and use
Refactor check valid sate into separate function
Return HypervisorState instead of string from get_hypervisor_state
Update tests
Update to work with new version of query library

lib/enums/hypervisor_states.py

@@ -6,39 +6,35 @@ class HypervisorState(Enum):
     Class of states used for automating maintenace of hypervisors
     """
 
-    RUNNING = auto()
-    PENDING_MAINTENANCE = auto()
-    DRAINING = auto()
-    DRAINED = auto()
+    RUNNING = {
+        "uptime": True,
+        "enabled": True,
+        "state": True,
+        "servers": True,
+    }
+    PENDING_MAINTENANCE = {
+        "uptime": False,
+        "enabled": True,
+        "state": True,
+        "servers": True,
+    }
+    DRAINING = {
+        "uptime": False,
+        "enabled": False,
+        "state": True,
+        "servers": True,
+    }
+    DRAINED = {
+        "uptime": False,
+        "enabled": False,
+        "state": True,
+        "servers": False,
+    }
+    UNKNOWN = auto()
 
-    def properties(self):
+    @classmethod
+    def _missing_(cls, value):
         """
-        Return hypervisor state for given variables
+        Return UNKNOWN if state not found in class
         """
-        state_properties = {
-            HypervisorState.RUNNING: {
-                "uptime": True,
-                "enabled": True,
-                "state": True,
-                "servers": True,
-            },
-            HypervisorState.PENDING_MAINTENANCE: {
-                "uptime": False,
-                "enabled": True,
-                "state": True,
-                "servers": True,
-            },
-            HypervisorState.DRAINING: {
-                "uptime": False,
-                "enabled": False,
-                "state": True,
-                "servers": True,
-            },
-            HypervisorState.DRAINED: {
-                "uptime": False,
-                "enabled": False,
-                "state": True,
-                "servers": False,
-            },
-        }
-        return state_properties[self]
+        return cls.UNKNOWN

lib/openstack_api/openstack_hypervisor.py

@@ -1,46 +1,42 @@
-import re
 from typing import Dict
 from enums.hypervisor_states import HypervisorState
 
 
-def get_hypervisor_state(hypervisor: Dict, uptime_limit: int) -> str:
+def get_hypervisor_state(hypervisor: Dict, uptime_limit: int) -> HypervisorState:
     """
     Returns a hypervisor state given a set of hypervisor variables
     :param hypervisor: Dictionary containing hypervisor: uptime, state, status and server count
     :param uptime_limit: Number of days of uptime before hypervisor requires maintenance
     :return: Hypervisor state
     """
-    uptime_pattern = re.compile(r"up\s+(\d+)\s+days,?\s+(\d+):(\d+)")
-    hypervisor_uptime = hypervisor.get("hypervisor_uptime")
+    if not valid_state(hypervisor):
+        return HypervisorState.UNKNOWN
 
-    if not (
-        isinstance(hypervisor_uptime, str)
-        and (match := uptime_pattern.search(hypervisor_uptime))
-    ):
-        return "UNKNOWN"
-
-    hypervisor_status = hypervisor.get("hypervisor_status")
-    if hypervisor_status not in ["enabled", "disabled"]:
-        return "UNKNOWN"
-
-    hypervisor_state = hypervisor.get("hypervisor_state")
-    if hypervisor_state not in ["up", "down"]:
-        return "UNKNOWN"
-
-    hypervisor_server_count = hypervisor.get("hypervisor_server_count")
-    if not isinstance(hypervisor_server_count, int) or hypervisor_server_count < 0:
-        return "UNKNOWN"
-
-    days = int(match.group(1))
     hv_state = {
-        "uptime": days < uptime_limit,
-        "enabled": hypervisor_status == "enabled",
-        "state": hypervisor_state == "up",
-        "servers": hypervisor_server_count > 0,
+        "uptime": hypervisor["hypervisor_uptime_days"] < uptime_limit,
+        "enabled": hypervisor["hypervisor_status"] == "enabled",
+        "state": hypervisor["hypervisor_state"] == "up",
+        "servers": hypervisor["hypervisor_server_count"] > 0,
     }
 
-    for state in HypervisorState:
-        if state.properties() == hv_state:
-            return state.name
+    return HypervisorState(hv_state)
 
-    return "UNKNOWN"
+
+def valid_state(state) -> bool:
+    """
+    Validates the hypervisor state
+    :param state: Dictionary containing hypervisor state
+    :return: True for valid state
+    """
+    if not isinstance(state["hypervisor_uptime_days"], int):
+        return False
+    hypervisor_status = state["hypervisor_status"]
+    if hypervisor_status not in ["enabled", "disabled"]:
+        return False
+    hypervisor_state = state["hypervisor_state"]
+    if hypervisor_state not in ["up", "down"]:
+        return False
+    hypervisor_server_count = state["hypervisor_server_count"]
+    if not isinstance(hypervisor_server_count, int) or hypervisor_server_count < 0:
+        return False
+    return True

lib/openstack_query_api/hypervisor_queries.py

@@ -16,7 +16,7 @@ def query_hypervisor_state(cloud_account: str):
         "hypervisor_name",
         "hypervisor_state",
         "hypervisor_status",
-        "hypervisor_uptime",
+        "hypervisor_uptime_days",
         "hypervisor_server_count",
     )
 

sensors/src/hypervisor_state_sensor.py

@@ -35,8 +35,9 @@ def poll(self):
         """
 
         data = query_hypervisor_state(self.cloud_account)
-
         for hypervisor in data:
+            if not isinstance(hypervisor, dict):
+                continue
             current_state = get_hypervisor_state(
                 hypervisor, uptime_limit=self.uptime_limit
             )

tests/lib/openstack_api/test_openstack_hypervisor.py

@@ -1,3 +1,4 @@
+from enums.hypervisor_states import HypervisorState
 from openstack_api.openstack_hypervisor import get_hypervisor_state
 
 
@@ -6,119 +7,119 @@ def test_get_state_running():
     Test hypervisor state is running for given variables
     """
     hypervisor = {
-        "hypervisor_uptime": "13:37:30 up 7 days,  4:29, 19 users,  load average: 0.04, 0.04, 0.0",
+        "hypervisor_uptime_days": 7,
         "hypervisor_status": "enabled",
         "hypervisor_state": "up",
         "hypervisor_server_count": 5,
     }
     state = get_hypervisor_state(hypervisor, 60)
-    assert state == "RUNNING"
+    assert state == HypervisorState.RUNNING
 
 
 def test_get_state_pending_maintenance():
     """
     Test hypervisor state is pending maintenace for given variables
     """
     hypervisor = {
-        "hypervisor_uptime": "13:37:30 up 100 days,  4:29, 19 users,  load average: 0.04, 0.04, 0.0",
+        "hypervisor_uptime_days": 100,
         "hypervisor_status": "enabled",
         "hypervisor_state": "up",
         "hypervisor_server_count": 5,
     }
     state = get_hypervisor_state(hypervisor, 60)
-    assert state == "PENDING_MAINTENANCE"
+    assert state == HypervisorState.PENDING_MAINTENANCE
 
 
 def test_get_state_draining():
     """
     Test hypervisor state is draining for given variables
     """
     hypervisor = {
-        "hypervisor_uptime": "13:37:30 up 100 days,  4:29, 19 users,  load average: 0.04, 0.04, 0.0",
+        "hypervisor_uptime_days": 100,
         "hypervisor_status": "disabled",
         "hypervisor_state": "up",
         "hypervisor_server_count": 5,
     }
     state = get_hypervisor_state(hypervisor, 60)
-    assert state == "DRAINING"
+    assert state == HypervisorState.DRAINING
 
 
 def test_get_state_drained():
     """
     Test hypervisor state is drained for given variables
     """
     hypervisor = {
-        "hypervisor_uptime": "13:37:30 up 100 days,  4:29, 19 users,  load average: 0.04, 0.04, 0.0",
+        "hypervisor_uptime_days": 100,
         "hypervisor_status": "disabled",
         "hypervisor_state": "up",
         "hypervisor_server_count": 0,
     }
     state = get_hypervisor_state(hypervisor, 60)
-    assert state == "DRAINED"
+    assert state == HypervisorState.DRAINED
 
 
 def test_get_state_unkown_status():
     """
     Test hypervisor state is unknown when missing status
     """
     hypervisor = {
-        "hypervisor_uptime": "13:37:30 up 100 days,  4:29, 19 users,  load average: 0.04, 0.04, 0.0",
+        "hypervisor_uptime_days": 100,
         "hypervisor_status": None,
         "hypervisor_state": "up",
         "hypervisor_server_count": 0,
     }
     state = get_hypervisor_state(hypervisor, 60)
-    assert state == "UNKNOWN"
+    assert state == HypervisorState.UNKNOWN
 
 
 def test_get_state_unkown_state():
     """
     Test hypervisor state is unknown when missing state
     """
     hypervisor = {
-        "hypervisor_uptime": "13:37:30 up 100 days,  4:29, 19 users,  load average: 0.04, 0.04, 0.0",
+        "hypervisor_uptime_days": 100,
         "hypervisor_status": "enabled",
         "hypervisor_state": None,
         "hypervisor_server_count": 0,
     }
     state = get_hypervisor_state(hypervisor, 60)
-    assert state == "UNKNOWN"
+    assert state == HypervisorState.UNKNOWN
 
 
 def test_get_state_unkown_uptime():
     """
     Test hypervisor state is unknown when missing uptime
     """
     hypervisor = {
-        "hypervisor_uptime": None,
+        "hypervisor_uptime_days": None,
         "hypervisor_status": "enabled",
         "hypervisor_state": "up",
         "hypervisor_server_count": 0,
     }
     state = get_hypervisor_state(hypervisor, 60)
-    assert state == "UNKNOWN"
+    assert state == HypervisorState.UNKNOWN
 
 
 def test_get_state_unkown_server_count():
     """
     Test hypervisor state is unknown when missing server count
     """
     hypervisor = {
-        "hypervisor_uptime": "13:37:30 up 100 days,  4:29, 19 users,  load average: 0.04, 0.04, 0.0",
+        "hypervisor_uptime_days": 100,
         "hypervisor_status": "enabled",
         "hypervisor_state": "up",
         "hypervisor_server_count": None,
     }
     state = get_hypervisor_state(hypervisor, 60)
-    assert state == "UNKNOWN"
+    assert state == HypervisorState.UNKNOWN
 
 
 def test_get_state_no_matching_state():
     hypervisor = {
-        "hypervisor_uptime": "up 60 days, 12:34",
+        "hypervisor_uptime_days": 60,
         "hypervisor_status": "disabled",
         "hypervisor_state": "down",
         "hypervisor_server_count": 10,
     }
     result = get_hypervisor_state(hypervisor, 60)
-    assert result == "UNKNOWN"
+    assert result == HypervisorState.UNKNOWN