added collector args to dimm, dmesg and storage to support skip_sudo

Author: alexandraBara

nodescraper/cli/helper.py

@@ -331,7 +331,7 @@ def generate_reference_config(
     for obj in results:
         if obj.result_data.collection_result.status != ExecutionStatus.OK:
             logger.warning(
-                "Plugin: %s result status is %, skipping",
+                "Plugin: %s result status is %s, skipping",
                 obj.source,
                 obj.result_data.collection_result.status,
             )

nodescraper/pluginexecutor.py

@@ -243,15 +243,23 @@ def apply_global_args_to_plugin(
             else:
                 run_args[key] = global_args[key]
 
-        if "collection_args" in global_args and hasattr(plugin_class, "COLLECTOR_ARGS"):
+        if (
+            "collection_args" in global_args
+            and getattr(plugin_class, "COLLECTOR_ARGS", None) is not None
+            and "COLLECTOR_ARGS" in plugin_class.__dict__
+        ):
             plugin_fields = set(plugin_class.COLLECTOR_ARGS.__fields__.keys())
             filtered = {
                 k: v for k, v in global_args["collection_args"].items() if k in plugin_fields
             }
             if filtered:
                 run_args["collection_args"] = filtered
 
-        if "analysis_args" in global_args and hasattr(plugin_class, "ANALYZER_ARGS"):
+        if (
+            "analysis_args" in global_args
+            and getattr(plugin_class, "ANALYZER_ARGS", None) is not None
+            and "ANALYZER_ARGS" in plugin_class.__dict__
+        ):
             plugin_fields = set(plugin_class.ANALYZER_ARGS.__fields__.keys())
             filtered = {k: v for k, v in global_args["analysis_args"].items() if k in plugin_fields}
             if filtered:

nodescraper/plugins/inband/dimm/dimm_collector.py

@@ -50,6 +50,7 @@ def collect_data(
             self.result.message = "Skipping sudo plugin"
             self.result.status = ExecutionStatus.NOT_RAN
             return self.result, None
+
         dimm_str = None
         if self.system_info.os_family == OSFamily.WINDOWS:
             res = self._run_sut_cmd("wmic memorychip get Capacity")

nodescraper/plugins/inband/dmesg/dmesg_collector.py

@@ -23,14 +23,17 @@
 # SOFTWARE.
 #
 ###############################################################################
+from typing import Optional
+
 from nodescraper.base import InBandDataCollector
 from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus, OSFamily
 from nodescraper.models import TaskResult
 
+from .collector_args import DmesgCollectorArgs
 from .dmesgdata import DmesgData
 
 
-class DmesgCollector(InBandDataCollector[DmesgData, None]):
+class DmesgCollector(InBandDataCollector[DmesgData, DmesgCollectorArgs]):
     """Read dmesg log"""
 
     SUPPORTED_OS_FAMILY = {OSFamily.LINUX}
@@ -60,17 +63,21 @@ def _get_dmesg_content(self) -> str:
 
     def collect_data(
         self,
-        args=None,
+        args: Optional[DmesgCollectorArgs] = None,
     ) -> tuple[TaskResult, DmesgData | None]:
         """Collect dmesg data from the system
 
         Returns:
             tuple[TaskResult, DmesgData | None]: tuple containing the result of the task and the dmesg data if available
         """
+        if args is None:
+            args = DmesgCollectorArgs()
+
         if args.skip_sudo:
             self.result.message = "Skipping sudo plugin"
             self.result.status = ExecutionStatus.NOT_RAN
             return self.result, None
+
         dmesg_content = self._get_dmesg_content()
 
         if dmesg_content:

nodescraper/plugins/inband/storage/storage_collector.py

@@ -24,11 +24,13 @@
 #
 ###############################################################################
 import re
+from typing import Optional
 
 from nodescraper.base import InBandDataCollector
 from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus, OSFamily
 from nodescraper.models import TaskResult
 
+from .collector_args import StorageCollectorArgs
 from .storagedata import DeviceStorageData, StorageDataModel
 
 
@@ -37,8 +39,13 @@ class StorageCollector(InBandDataCollector[StorageDataModel, None]):
 
     DATA_MODEL = StorageDataModel
 
-    def collect_data(self, args: None = None) -> tuple[TaskResult, StorageDataModel | None]:
+    def collect_data(
+        self, args: Optional[StorageCollectorArgs] = None
+    ) -> tuple[TaskResult, StorageDataModel | None]:
         """read storage usage data"""
+        if args is None:
+            args = StorageCollectorArgs()
+
         if args.skip_sudo:
             self.result.message = "Skipping sudo plugin"
             self.result.status = ExecutionStatus.NOT_RAN