feat(PDRIVE-687): add mypy-based pre-commit hook for type checking

.claude/rules/code-style.md

@@ -195,6 +195,10 @@ def has_resource(self) -> bool:
 
 **Check `src/in_cluster_checks/utils` for suitable functions before executing shell commands or implementing new logic.**
 
+## Mypy Type Errors
+
+**Fix mypy errors by correcting type annotations — never suppress them with `# noqa`, `# type: ignore`, or by modifying the filter script (`check_mypy.py`).**
+
 ## Debug Logging
 
 See [@.claude/rules/debug-rule.md](debug-rule.md) for debug logging guidelines.

.claude/skills/new-rule/SKILL.md

@@ -107,7 +107,7 @@ class MyOrchestratorRule(OrchestratorRule):
 ```python
 # Use existing oc_api methods when available
 pods = self.oc_api.get_pods(namespace="openshift-etcd")
-network = self.oc_api.select_resources("network.operator/cluster", single=True)
+network = self.oc_api.select_single_resource("network.operator/cluster")
 
 # Run commands inside pods
 cmd = SafeCmdString("etcdctl version")

.pre-commit-config.yaml

@@ -58,6 +58,14 @@ repos:
         types: [python]
         files: ^src/in_cluster_checks/
 
+      - id: mypy-type-check
+        name: check types with mypy
+        entry: python tests/linters/check_mypy.py
+        language: system
+        types: [python]
+        files: ^src/in_cluster_checks/
+        pass_filenames: false
+
       - id: pytest-coverage
         name: pytest with coverage
         entry: python -m pytest

CONTRIBUTING.md

@@ -298,7 +298,7 @@ This validates variables to prevent shell injection.
 ```python
 # Use existing oc_api methods when available
 pods = self.oc_api.get_pods(namespace="openshift-etcd")
-network = self.oc_api.select_resources("network.operator/cluster", single=True)
+network = self.oc_api.select_single_resource("network.operator/cluster")
 
 # Use run_oc_command for other oc commands
 rc, out, err = self.oc_api.run_oc_command("get", ["nodes", "-o", "json"])

mypy.ini

@@ -0,0 +1,19 @@
+[mypy]
+python_version = 3.12
+check_untyped_defs = True
+show_error_codes = True
+
+# Only report SafeCmdString type errors
+disable_error_code = var-annotated,no-untyped-def,import-untyped,no-any-return,override
+
+# Ignore third-party modules without stubs
+ignore_missing_imports = True
+
+# Files to check
+files = src/in_cluster_checks/
+
+[mypy-openshift_client.*]
+ignore_missing_imports = True
+
+[mypy-dateutil.*]
+ignore_missing_imports = True

pyproject.toml

@@ -34,6 +34,7 @@ dev = [
     "black>=24.1.1",
     "flake8>=7.0.0",
     "isort>=5.13.2",
+    "mypy>=1.7.0",
 ]
 
 [project.scripts]

src/in_cluster_checks/core/data_collector_runner.py

@@ -13,6 +13,7 @@
 
 from in_cluster_checks.core.exceptions import UnExpectedSystemOutput
 from in_cluster_checks.core.executor import OrchestratorExecutor
+from in_cluster_checks.core.operations import DataCollector
 from in_cluster_checks.core.parallel_runner import ParallelRunner
 from in_cluster_checks.utils.dict_utils import convert_dict_to_sorted_json_str
 from in_cluster_checks.utils.enums import ORCHESTRATOR_HOST_NAME, Objectives
@@ -37,7 +38,7 @@ class DataCollectorRunner:
 
     @classmethod
     def execute_data_collector(
-        cls, rule_instance, collector_class: type, use_parallel: bool = True, **kwargs
+        cls, rule_instance, collector_class: type[DataCollector], use_parallel: bool = True, **kwargs
     ) -> Dict[str, Any]:
         """
         Execute a DataCollector on all hosts and return aggregated results.
@@ -193,7 +194,12 @@ def clear_data_collector_cache(cls):
 
     @classmethod
     def run_collectors(
-        cls, collector_instances: list, use_parallel: bool, collector_class: type, rule_instance, **kwargs
+        cls,
+        collector_instances: list,
+        use_parallel: bool,
+        collector_class: type[DataCollector],
+        rule_instance,
+        **kwargs,
     ) -> Dict[str, Dict]:
         """
         Run collectors with caching for many-to-one relationships.
@@ -391,7 +397,7 @@ def aggregate_collector_results(rule_instance, results_dict: Dict[str, Dict]) ->
 
     @staticmethod
     def handle_collector_failures(
-        rule_instance, collector_class: type, host_exceptions: Dict[str, str], hosts_dict: Dict[str, Any]
+        rule_instance, collector_class: type[DataCollector], host_exceptions: Dict[str, str], hosts_dict: Dict[str, Any]
     ):
         """
         Handle collector failures and track success.
@@ -428,7 +434,7 @@ def is_collector_failed_on_all_hosts(host_exceptions_dict: Dict[str, str], hosts
         Returns:
             True if all hosts failed, False otherwise
         """
-        return host_exceptions_dict and set(host_exceptions_dict.keys()) == set(hosts_dict.keys())
+        return bool(host_exceptions_dict) and set(host_exceptions_dict.keys()) == set(hosts_dict.keys())
 
     @staticmethod
     def format_collector_exceptions(collector_name: str, host_exceptions_dict: Dict[str, str]) -> list:

src/in_cluster_checks/core/domain.py

@@ -43,7 +43,7 @@ def domain_name(self) -> str:
         raise NotImplementedError(f"domain_name() must be implemented in {self.__class__.__name__}")
 
     @abc.abstractmethod
-    def get_rule_classes(self) -> List[type]:
+    def get_rule_classes(self) -> List[type[Rule]]:
         """
         Get list of rule classes to run in this domain.
 
@@ -59,7 +59,7 @@ def get_rule_classes(self) -> List[type]:
         """
         raise NotImplementedError(f"get_rule_classes() must be implemented in {self.__class__.__name__}")
 
-    def _filter_rules_for_light_run(self, rule_classes: List[type]) -> List[type]:
+    def _filter_rules_for_light_run(self, rule_classes: List[type[Rule]]) -> List[type[Rule]]:
         """
         Filter rules based on light_run mode.
 
@@ -111,7 +111,9 @@ def clean_domain(self):
         """Clean up after domain execution — e.g. free cached command outputs."""
         pass
 
-    def _create_rule_groups(self, rule_classes: List[type], host_executors_dict: Dict[str, Any]) -> List[List[Rule]]:
+    def _create_rule_groups(
+        self, rule_classes: List[type[Rule]], host_executors_dict: Dict[str, Any]
+    ) -> List[List[Rule]]:
         """
         Create grouped rule instances following HC's pattern.
 
@@ -139,7 +141,7 @@ def _create_rule_groups(self, rule_classes: List[type], host_executors_dict: Dic
 
         return rule_groups
 
-    def _create_instances_for_rule(self, rule_class: type, host_executors_dict: Dict[str, Any]) -> List[Rule]:
+    def _create_instances_for_rule(self, rule_class: type[Rule], host_executors_dict: Dict[str, Any]) -> List[Rule]:
         """
         Create rule instances for a single rule class.
 
@@ -163,7 +165,7 @@ def _create_instances_for_rule(self, rule_class: type, host_executors_dict: Dict
         # Create instances for matching nodes (handles both ONE_* and multi-type)
         return self._create_per_node_instances(rule_class, host_executors_dict)
 
-    def _create_orchestrator_instance(self, rule_class: type, host_executors_dict: Dict[str, Any]) -> List[Rule]:
+    def _create_orchestrator_instance(self, rule_class: type[Rule], host_executors_dict: Dict[str, Any]) -> List[Rule]:
         """
         Create single orchestrator instance.
 
@@ -186,7 +188,7 @@ def _create_orchestrator_instance(self, rule_class: type, host_executors_dict: D
             self.logger.error(f"Failed to instantiate {rule_class.__name__} as orchestrator: {e}")
             return []
 
-    def _create_per_node_instances(self, rule_class: type, host_executors_dict: Dict[str, Any]) -> List[Rule]:
+    def _create_per_node_instances(self, rule_class: type[Rule], host_executors_dict: Dict[str, Any]) -> List[Rule]:
         """
         Create rule instances for each matching node.
 
@@ -210,7 +212,7 @@ def _create_per_node_instances(self, rule_class: type, host_executors_dict: Dict
 
         return instances
 
-    def _should_create_for_executor(self, rule_class: type, executor: Any) -> bool:
+    def _should_create_for_executor(self, rule_class: type[Rule], executor: Any) -> bool:
         """
         Check if rule should be created for the given executor.
 
@@ -231,7 +233,7 @@ def _should_create_for_executor(self, rule_class: type, executor: Any) -> bool:
 
         return True
 
-    def _matches_debug_filter(self, rule_class: type) -> bool:
+    def _matches_debug_filter(self, rule_class: type[Rule]) -> bool:
         """
         Check if rule matches debug filter (by unique_name or title).
 

src/in_cluster_checks/core/exceptions.py

@@ -44,7 +44,8 @@ def __init__(
 
         # Sanitize command before including in exception message
         safe_cmd = SecretFilter.sanitize(self.cmd)
-        super().__init__(f"{message} on {ip}: {safe_cmd}\nOutput: {output[:500]}")  # Limit output length
+        safe_output = SecretFilter.sanitize(output[:500])
+        super().__init__(f"{message} on {ip}: {safe_cmd}\nOutput: {safe_output}")
 
     def __str__(self):
         """Return formatted exception string (HC-style)."""

src/in_cluster_checks/core/executor_factory.py

@@ -91,7 +91,7 @@ def build_host_executors(self) -> Dict[str, NodeExecutor]:
 
         return self._host_executors_dict
 
-    def _get_internal_ip(self, node_dict: dict) -> str:
+    def _get_internal_ip(self, node_dict: dict) -> str | None:
         """
         Extract internal IP from node data.
 

src/in_cluster_checks/core/operations.py

@@ -35,6 +35,8 @@ class Operator:
     # e.g., [Objectives.ALL_NODES], [Objectives.ICE_CONTAINER], etc.
     objective_hosts = []
 
+    unique_name: str | None = None
+
     # Thread-safe debug output lock (prevents interleaved output in parallel execution)
     _debug_lock = threading.RLock()
 
@@ -76,7 +78,11 @@ def _debug_log(self, message: str):
             print(f"\n[DEBUG] [{self.get_host_name()}] {message}", flush=True)
 
     def run_cmd(
-        self, cmd: SafeCmdString, timeout: int = 120, hosts_cached_pool: dict = None, add_bash_timeout: bool = False
+        self,
+        cmd: SafeCmdString,
+        timeout: int = 120,
+        hosts_cached_pool: dict | None = None,
+        add_bash_timeout: bool = False,
     ) -> tuple:
         """
         Run command on host/container and log it.
@@ -141,7 +147,7 @@ def _run_cmd_use_cached(
         return res
 
     def get_output_from_run_cmd(
-        self, cmd: SafeCmdString, timeout: int = 30, message: str = None, hosts_cached_pool: dict = None
+        self, cmd: SafeCmdString, timeout: int = 30, message: str | None = None, hosts_cached_pool: dict | None = None
     ) -> str:
         """
         Run command, log it, and return stdout if successful.
@@ -315,14 +321,10 @@ def _enforce_have_document(self):
                 f"Add as class variable: unique_name = 'your_unique_name'"
             )
 
-    def get_unique_name(self) -> str:
+    def get_unique_name(self) -> str | None:
         """Get unique operation name (accessible as class or instance attribute)."""
         return self.unique_name
 
-    def get_severity(self) -> str:
-        """Get severity level (HC-style interface)."""
-        return self._severity if self._severity else "NA"
-
     def get_implication_tags(self) -> list:
         """Get implication tags (HC-style interface)."""
         return getattr(self, "_implication_tags", [])
@@ -505,7 +507,9 @@ class MyCollector(OrchestratorDataCollector):
             objective_hosts = [Objectives.ORCHESTRATOR]
 
             def collect_data(self):
-                network_obj = self.oc_api.select_resources("network.operator/cluster", single=True)
+                network_obj = self.oc_api.select_single_resource("network.operator/cluster")
+                if not network_obj:
+                    return None
                 return network_obj.model.spec.defaultNetwork.type
     """
 
@@ -531,7 +535,11 @@ def _validate_objective_hosts(self):
         pass
 
     def run_cmd(
-        self, cmd: SafeCmdString, timeout: int = 120, hosts_cached_pool: dict = None, add_bash_timeout: bool = False
+        self,
+        cmd: SafeCmdString,
+        timeout: int = 120,
+        hosts_cached_pool: dict | None = None,
+        add_bash_timeout: bool = False,
     ) -> tuple:
         """
         Not available for OrchestratorDataCollector - use oc_api methods instead.

src/in_cluster_checks/core/printer.py

@@ -311,12 +311,12 @@ def print_summary(self, domain_name: str) -> None:
             logger.info("")  # Empty line after host validations
 
     @staticmethod
-    def print_to_json(results: Dict[str, Any], output_file: str) -> None:
+    def print_to_json(results: List[Dict[str, Any]], output_file: str) -> None:
         """
         Write rule results to JSON file with restricted permissions (owner-only).
 
         Args:
-            results: Dictionary with rule results in Insights format
+            results: List of report dicts from format_results()
             output_file: Path to output JSON file
         """
         file_descriptor = os.open(output_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
@@ -359,7 +359,7 @@ def print_to_junit(results: List[Dict[str, Any]], output_file: str) -> None:
                     if run_time is not None:
                         tc_attrs["time"] = str(run_time)
 
-                    testcase = ET.SubElement(testsuite, "testcase", **tc_attrs)
+                    testcase = ET.SubElement(testsuite, "testcase", tc_attrs)
                     status = host_result.get("status")
                     message = _strip_xml_illegal_chars(host_result.get("message", ""))
 
@@ -380,7 +380,7 @@ def print_to_junit(results: List[Dict[str, Any]], output_file: str) -> None:
         tree.write(output_file, encoding="unicode", xml_declaration=True)
 
     @staticmethod
-    def format_results(flow_results: list, rule_component_map: Dict[str, str]) -> Dict[str, Any]:
+    def format_results(flow_results: list, rule_component_map: Dict[str, str]) -> List[Dict[str, Any]]:
         """
         Format multiple flow results into Insights-compatible structure.
 
@@ -393,21 +393,17 @@ def format_results(flow_results: list, rule_component_map: Dict[str, str]) -> Di
             rule_component_map: Map of {rule_name: full_component_path}
 
         Returns:
-            Formatted results dictionary:
+            List of report dicts, each containing:
             {
-                "in_cluster_rules": [
-                    {
-                        "rule_id": "domain|rule",
-                        "component": "...",
-                        "key": "rule",
-                        "status": "aggregated_status",  # Worst status across all hosts
-                        "description": "...",
-                        "domain": "...",
-                        "details": [  # Array of host results
-                            {"node_ip": "...", "node_name": "...", "status": "...", ...},
-                            {"node_ip": "...", "node_name": "...", "status": "...", ...}
-                        ]
-                    }
+                "rule_id": "domain|rule",
+                "component": "...",
+                "key": "rule",
+                "status": "aggregated_status",  # Worst status across all hosts
+                "description": "...",
+                "domain": "...",
+                "details": [  # Array of host results
+                    {"node_ip": "...", "node_name": "...", "status": "...", ...},
+                    {"node_ip": "...", "node_name": "...", "status": "...", ...}
                 ]
             }
         """

src/in_cluster_checks/core/rule.py

@@ -78,7 +78,7 @@ def get_roles_for_current_deployment(self) -> list:
         return self.objective_hosts
 
     @classmethod
-    def get_unique_name_classmethod(cls) -> str:
+    def get_unique_name_classmethod(cls) -> str | None:
         """
         Get unique operation name without instantiation.
 

src/in_cluster_checks/rules/hw_fw_details/hw_fw_base.py

@@ -182,7 +182,7 @@ def compare_within_groups(self, collected_data: Dict, node_groups: Dict) -> Rule
 
         # Process each node group
         for group_label, executors in node_groups.items():
-            group_result = OrderedDict()
+            group_result: OrderedDict[str, Any] = OrderedDict()
             group_result["node_count"] = len(executors)
             group_result["nodes"] = [e.node_name for e in executors]
             group_result[category_key] = OrderedDict()
@@ -223,9 +223,10 @@ def compare_within_groups(self, collected_data: Dict, node_groups: Dict) -> Rule
                     collector_result["value"] = self._get_list_of_id_host_name_data(group_data)
 
                 # Create nested structure: topic -> name -> result (HC Blueprint format)
-                if topic not in group_result[category_key]:
-                    group_result[category_key][topic] = OrderedDict()
-                group_result[category_key][topic][name] = collector_result
+                category_data: OrderedDict = group_result[category_key]
+                if topic not in category_data:
+                    category_data[topic] = OrderedDict()
+                category_data[topic][name] = collector_result
 
             result_data[group_label] = group_result