fix: clear all mypy errors

Annotate untyped containers and Optional attributes, type the stdlib-shaped
metric dicts in the client plugins, and add types-PyYAML/types-psutil to the
dev extras plus a mypy override for nio/mattermostdriver (no stubs).

Behaviour fixes found on the way: enable_kernel_timestamps() now returns False
when SO_TIMESTAMP is unknown instead of raising TypeError, send_shutdown is
defined at module level so cleanup() cannot hit a NameError, and a failed
subprocess returncode falls back to UNKNOWN.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-28 05:04:54 -04:00
co-authored by Claude Sonnet 5.5
parent 748349a7b9
commit 445ff6b474
20 changed files with 63 additions and 49 deletions
+1 -1
View File
@@ -28,7 +28,7 @@ class CPUMonitorPlugin(MonitorPlugin):
def __init__(self, config: Optional[Dict[str, Any]] = None):
super().__init__(config)
self.psutil = None
self.psutil: Any = None
self.per_core = config.get("per_core", False) if config else False
self.interval = config.get("interval", 300) if config else 300
+3 -3
View File
@@ -10,7 +10,7 @@ from typing import Dict, Any, Optional
try:
import psutil
except ImportError:
psutil = None
psutil = None # type: ignore[assignment]
from hbd.client.plugin import MonitorPlugin
@@ -58,7 +58,7 @@ class DiskMonitorPlugin(MonitorPlugin):
raise ImportError("psutil library is required for disk_monitor plugin")
# Store previous I/O counters for delta calculation
self._prev_io = {}
self._prev_io: Dict[str, Any] = {}
async def initialize(self):
"""Initialize the plugin (check psutil availability)."""
@@ -114,7 +114,7 @@ class DiskMonitorPlugin(MonitorPlugin):
async def _collect_metrics(self) -> Dict[str, Any]:
"""Collect disk metrics from psutil."""
metrics = {}
metrics: Dict[str, Any] = {}
# Collect partition usage
partitions_data = {}
+4 -4
View File
@@ -10,7 +10,7 @@ from typing import Dict, Any, Optional
try:
import psutil
except ImportError:
psutil = None
psutil = None # type: ignore[assignment]
from hbd.client.plugin import InfoPlugin
@@ -98,8 +98,8 @@ class FilesystemInfoPlugin(InfoPlugin):
async def _collect_info(self) -> Dict[str, Any]:
"""Collect filesystem information from psutil."""
info = {}
filesystems = []
info: Dict[str, Any] = {}
filesystems: list = []
filesystem_types = set()
# Get mounted disk partitions
@@ -112,7 +112,7 @@ class FilesystemInfoPlugin(InfoPlugin):
if partition.fstype in self.exclude_types:
continue
fs_info = {
fs_info: Dict[str, Any] = {
'device': partition.device,
'mountpoint': partition.mountpoint,
'fstype': partition.fstype,
+2 -2
View File
@@ -10,7 +10,7 @@ from typing import Dict, Any, Optional
try:
import psutil
except ImportError:
psutil = None
psutil = None # type: ignore[assignment]
from hbd.client.plugin import MonitorPlugin
@@ -116,7 +116,7 @@ class MemoryMonitorPlugin(MonitorPlugin):
async def _collect_metrics(self) -> Dict[str, Any]:
"""Collect memory metrics from psutil."""
metrics = {}
metrics: Dict[str, Any] = {}
# Virtual (physical) memory statistics
vmem = psutil.virtual_memory()
+4 -3
View File
@@ -124,7 +124,7 @@ class NagiosRunnerPlugin(MonitorPlugin):
Returns:
Dictionary with results from all plugins
"""
results = {}
results: Dict[str, Any] = {}
for cmd_config in self.commands:
name = cmd_config.get("name")
@@ -181,7 +181,8 @@ class NagiosRunnerPlugin(MonitorPlugin):
self.logger.error(f"Command timed out: {command}")
return NAGIOS_UNKNOWN, f"Command timed out after {self.timeout}s", {}
status_code = proc.returncode
# communicate() has finished, so the process has exited and returncode is set
status_code = proc.returncode if proc.returncode is not None else NAGIOS_UNKNOWN
if status_code < 0:
return NAGIOS_UNKNOWN, f"Process killed by signal {-status_code}", {}
@@ -225,7 +226,7 @@ class NagiosRunnerPlugin(MonitorPlugin):
Returns:
Dictionary of metric_name: value
"""
perfdata = {}
perfdata: Dict[str, Any] = {}
# Performance data comes after the pipe character
if '|' not in output:
+4 -4
View File
@@ -10,7 +10,7 @@ from typing import Dict, Any, Optional
try:
import psutil
except ImportError:
psutil = None
psutil = None # type: ignore[assignment]
from hbd.client.plugin import MonitorPlugin
from hbd.client.connectivity import get_shared
@@ -63,7 +63,7 @@ class NetworkMonitorPlugin(MonitorPlugin):
raise ImportError("psutil library is required for network_monitor plugin")
# Store previous I/O counters for delta calculation
self._prev_io = {}
self._prev_io: Dict[str, Any] = {}
async def initialize(self):
"""Initialize the plugin (check psutil availability)."""
@@ -124,7 +124,7 @@ class NetworkMonitorPlugin(MonitorPlugin):
async def _collect_metrics(self) -> Dict[str, Any]:
"""Collect network metrics from psutil."""
metrics = {}
metrics: Dict[str, Any] = {}
# Collect per-interface I/O counters
try:
@@ -169,7 +169,7 @@ class NetworkMonitorPlugin(MonitorPlugin):
if self.include_connections:
try:
connections = psutil.net_connections(kind='inet')
conn_stats = {}
conn_stats: Dict[str, int] = {}
# Count connections by state
for conn in connections:
+1
View File
@@ -67,6 +67,7 @@ class PingMonitorPlugin(MonitorPlugin):
self.timeout = int(cfg.get("timeout", 5))
# hosts: dict of {hostname: {warning: x, critical: y}} or list of hostnames
raw_hosts = cfg.get("hosts", {})
self.hosts: Dict[str, Any]
if isinstance(raw_hosts, list):
self.hosts = {h: {} for h in raw_hosts}
else: