node: fix get_monitor result type and cast explicitly

Author: cdleonard

jenkinsapi/node.py

@@ -7,7 +7,7 @@
 import json
 import logging
 
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Optional, Union
 import xml.etree.ElementTree as ET
 
 import time
@@ -496,7 +496,7 @@ def set_config_element(self, el_name: str, value: str) -> None:
         xml_str = ET.tostring(self._et)
         self.upload_config(xml_str)
 
-    def get_monitor(self, monitor_name: str, poll_monitor=True) -> str:
+    def get_monitor(self, monitor_name: str, poll_monitor=True) -> object:
         """
         Polls the node returning one of the monitors in the monitorData
         branch of the returned node api tree.
@@ -514,68 +514,85 @@ def get_monitor(self, monitor_name: str, poll_monitor=True) -> str:
 
         return monitor_data[full_monitor_name]
 
+    def get_monitor_dict(
+        self,
+        monitor_name: str,
+        poll_monitor: bool = True,
+    ) -> dict:
+        value = self.get_monitor(monitor_name, poll_monitor)
+        if not isinstance(value, dict):
+            raise JenkinsAPIException(
+                f"Monitor {monitor_name!r} did not return a dictionary"
+            )
+        return value
+
     def get_available_physical_memory(self) -> int:
         """
         Returns the node's available physical memory in bytes.
         """
-        monitor_data = self.get_monitor("SwapSpaceMonitor")
+        monitor_data = self.get_monitor_dict("SwapSpaceMonitor")
         return monitor_data["availablePhysicalMemory"]
 
     def get_available_swap_space(self) -> int:
         """
         Returns the node's available swap space in bytes.
         """
-        monitor_data = self.get_monitor("SwapSpaceMonitor")
+        monitor_data = self.get_monitor_dict("SwapSpaceMonitor")
         return monitor_data["availableSwapSpace"]
 
     def get_total_physical_memory(self) -> int:
         """
         Returns the node's total physical memory in bytes.
         """
-        monitor_data = self.get_monitor("SwapSpaceMonitor")
+        monitor_data = self.get_monitor_dict("SwapSpaceMonitor")
         return monitor_data["totalPhysicalMemory"]
 
     def get_total_swap_space(self) -> int:
         """
         Returns the node's total swap space in bytes.
         """
-        monitor_data = self.get_monitor("SwapSpaceMonitor")
+        monitor_data = self.get_monitor_dict("SwapSpaceMonitor")
         return monitor_data["totalSwapSpace"]
 
     def get_workspace_path(self) -> str:
         """
         Returns the local path to the node's Jenkins workspace directory.
         """
-        monitor_data = self.get_monitor("DiskSpaceMonitor")
+        monitor_data = self.get_monitor_dict("DiskSpaceMonitor")
         return monitor_data["path"]
 
     def get_workspace_size(self) -> int:
         """
         Returns the size in bytes of the node's Jenkins workspace directory.
         """
-        monitor_data = self.get_monitor("DiskSpaceMonitor")
+        monitor_data = self.get_monitor_dict("DiskSpaceMonitor")
         return monitor_data["size"]
 
     def get_temp_path(self) -> str:
         """
         Returns the local path to the node's temp directory.
         """
-        monitor_data = self.get_monitor("TemporarySpaceMonitor")
+        monitor_data = self.get_monitor_dict("TemporarySpaceMonitor")
         return monitor_data["path"]
 
     def get_temp_size(self) -> int:
         """
         Returns the size in bytes of the node's temp directory.
         """
-        monitor_data = self.get_monitor("TemporarySpaceMonitor")
+        monitor_data = self.get_monitor_dict("TemporarySpaceMonitor")
         return monitor_data["size"]
 
     def get_architecture(self) -> str:
         """
         Returns the system architecture of the node eg. "Linux (amd64)".
         """
         # no need to poll as the architecture will never change
-        return self.get_monitor("ArchitectureMonitor", poll_monitor=False)
+        value = self.get_monitor("ArchitectureMonitor", poll_monitor=False)
+        if not isinstance(value, str):
+            raise JenkinsAPIException(
+                "Monitor ArchitectureMonitor did not return a string"
+            )
+        return value
 
     def block_until_idle(self, timeout: int, poll_time: int = 5) -> None:
         """
@@ -603,7 +620,7 @@ def get_response_time(self) -> int:
         """
         Returns the node's average response time.
         """
-        monitor_data = self.get_monitor("ResponseTimeMonitor")
+        monitor_data = self.get_monitor_dict("ResponseTimeMonitor")
         return monitor_data["average"]
 
     def get_clock_difference(self) -> int:
@@ -612,5 +629,5 @@ def get_clock_difference(self) -> int:
         the master Jenkins clock.
         Used to detect out of sync clocks.
         """
-        monitor_data = self.get_monitor("ClockMonitor")
+        monitor_data = self.get_monitor_dict("ClockMonitor")
         return monitor_data["diff"]