moved the CMDs into class vars for doc extraction

Author: alexandraBara

nodescraper/plugins/inband/bios/bios_collector.py

@@ -35,6 +35,8 @@ class BiosCollector(InBandDataCollector[BiosDataModel, None]):
     """Collect BIOS details"""
 
     DATA_MODEL = BiosDataModel
+    CMD_WINDOWS = "wmic bios get SMBIOSBIOSVersion /Value"
+    CMD = "sh -c 'cat /sys/devices/virtual/dmi/id/bios_version'"
 
     def collect_data(
         self,
@@ -49,13 +51,13 @@ def collect_data(
         bios = None
 
         if self.system_info.os_family == OSFamily.WINDOWS:
-            res = self._run_sut_cmd("wmic bios get SMBIOSBIOSVersion /Value")
+            res = self._run_sut_cmd(self.CMD_WINDOWS)
             if res.exit_code == 0:
                 bios = [line for line in res.stdout.splitlines() if "SMBIOSBIOSVersion=" in line][
                     0
                 ].split("=")[1]
         else:
-            res = self._run_sut_cmd("sh -c 'cat /sys/devices/virtual/dmi/id/bios_version'")
+            res = self._run_sut_cmd(self.CMD)
             if res.exit_code == 0:
                 bios = res.stdout
 

nodescraper/plugins/inband/cmdline/cmdline_collector.py

@@ -37,6 +37,8 @@ class CmdlineCollector(InBandDataCollector[CmdlineDataModel, None]):
 
     DATA_MODEL = CmdlineDataModel
 
+    CMD = "cat /proc/cmdline"
+
     def collect_data(
         self,
         args=None,
@@ -47,7 +49,7 @@ def collect_data(
         Returns:
             tuple[TaskResult, CmdlineDataModel | None]: tuple containing the task result and the cmdline data model if successful, otherwise None.
         """
-        res = self._run_sut_cmd("cat /proc/cmdline")
+        res = self._run_sut_cmd(self.CMD)
         cmdline_data = None
         if res.exit_code == 0:
             cmdline_data = CmdlineDataModel(cmdline=res.stdout)

nodescraper/plugins/inband/dimm/dimm_collector.py

@@ -38,6 +38,9 @@ class DimmCollector(InBandDataCollector[DimmDataModel, DimmCollectorArgs]):
 
     DATA_MODEL = DimmDataModel
 
+    CMD_WINDOWS = "wmic memorychip get Capacity"
+    CMD = """sh -c 'dmidecode -t 17 | tr -s " " | grep -v "Volatile\\|None\\|Module" | grep Size' 2>/dev/null"""
+
     def collect_data(
         self,
         args: Optional[DimmCollectorArgs] = None,
@@ -48,7 +51,7 @@ def collect_data(
 
         dimm_str = None
         if self.system_info.os_family == OSFamily.WINDOWS:
-            res = self._run_sut_cmd("wmic memorychip get Capacity")
+            res = self._run_sut_cmd(self.CMD_WINDOWS)
             if res.exit_code == 0:
                 capacities = {}
                 total = 0
@@ -69,10 +72,7 @@ def collect_data(
                 self.result.message = "Skipping sudo plugin"
                 self.result.status = ExecutionStatus.NOT_RAN
                 return self.result, None
-            res = self._run_sut_cmd(
-                """sh -c 'dmidecode -t 17 | tr -s " " | grep -v "Volatile\\|None\\|Module" | grep Size' 2>/dev/null""",
-                sudo=True,
-            )
+            res = self._run_sut_cmd(self.CMD, sudo=True)
             if res.exit_code == 0:
                 total = 0
                 topology = {}

nodescraper/plugins/inband/dkms/dkms_collector.py

@@ -37,6 +37,9 @@ class DkmsCollector(InBandDataCollector[DkmsDataModel, None]):
 
     DATA_MODEL = DkmsDataModel
 
+    CMD_STATUS = "dkms status"
+    CMD_VERSION = "dkms --version"
+
     def collect_data(
         self,
         args=None,
@@ -49,8 +52,8 @@ def collect_data(
         """
 
         dkms_data = DkmsDataModel()
-        dkms_status = self._run_sut_cmd("dkms status")
-        dkms_version = self._run_sut_cmd("dkms --version")
+        dkms_status = self._run_sut_cmd(self.CMD_STATUS)
+        dkms_version = self._run_sut_cmd(self.CMD_VERSION)
 
         if dkms_status.exit_code == 0:
             dkms_data.status = dkms_status.stdout

nodescraper/plugins/inband/dmesg/dmesg_collector.py

@@ -42,15 +42,15 @@ class DmesgCollector(InBandDataCollector[DmesgData, DmesgCollectorArgs]):
 
     DATA_MODEL = DmesgData
 
-    DMESG_CMD = "dmesg --time-format iso -x"
+    CMD = "dmesg --time-format iso -x"
 
-    DMESG_LOGS_CMD = (
+    CMD_LOGS = (
         r"ls -1 /var/log/dmesg* 2>/dev/null | grep -E '^/var/log/dmesg(\.[0-9]+(\.gz)?)?$' || true"
     )
 
     def _collect_dmesg_rotations(self):
         """Collect dmesg logs"""
-        list_res = self._run_sut_cmd(self.DMESG_LOGS_CMD, sudo=True)
+        list_res = self._run_sut_cmd(self.CMD_LOGS, sudo=True)
         paths = [p.strip() for p in (list_res.stdout or "").splitlines() if p.strip()]
         if not paths:
             self._log_event(
@@ -122,7 +122,7 @@ def _get_dmesg_content(self) -> str:
         """
 
         self.logger.info("Running dmesg command on system")
-        res = self._run_sut_cmd(self.DMESG_CMD, sudo=True, log_artifact=False)
+        res = self._run_sut_cmd(self.CMD, sudo=True, log_artifact=False)
         if res.exit_code != 0:
             self._log_event(
                 category=EventCategory.OS,

nodescraper/plugins/inband/journal/journal_collector.py

@@ -36,15 +36,15 @@ class JournalCollector(InBandDataCollector[JournalData, None]):
 
     SUPPORTED_OS_FAMILY = {OSFamily.LINUX}
     DATA_MODEL = JournalData
+    CMD = "journalctl --no-pager --system --output=short-iso"
 
     def _read_with_journalctl(self):
         """Read journal logs using journalctl
 
         Returns:
             str|None: system journal read
         """
-        cmd = "journalctl --no-pager --system --output=short-iso"
-        res = self._run_sut_cmd(cmd, sudo=True, log_artifact=False, strip=False)
+        res = self._run_sut_cmd(self.CMD, sudo=True, log_artifact=False, strip=False)
 
         if res.exit_code != 0:
             self._log_event(

nodescraper/plugins/inband/kernel/kernel_collector.py

@@ -34,6 +34,8 @@ class KernelCollector(InBandDataCollector[KernelDataModel, None]):
     """Read kernel version"""
 
     DATA_MODEL = KernelDataModel
+    CMD_WINDOWS = "wmic os get Version /Value"
+    CMD = "sh -c 'uname -r'"
 
     def collect_data(
         self,
@@ -45,15 +47,16 @@ def collect_data(
         Returns:
             tuple[TaskResult, KernelDataModel | None]: tuple containing the task result and kernel data model or None if not found.
         """
+
         kernel = None
         if self.system_info.os_family == OSFamily.WINDOWS:
-            res = self._run_sut_cmd("wmic os get Version /Value")
+            res = self._run_sut_cmd(self.CMD_WINDOWS)
             if res.exit_code == 0:
                 kernel = [line for line in res.stdout.splitlines() if "Version=" in line][0].split(
                     "="
                 )[1]
         else:
-            res = self._run_sut_cmd("sh -c 'uname -r'")
+            res = self._run_sut_cmd(self.CMD)
             if res.exit_code == 0:
                 kernel = res.stdout
 

nodescraper/plugins/inband/kernel_module/kernel_module_collector.py

@@ -35,6 +35,8 @@ class KernelModuleCollector(InBandDataCollector[KernelModuleDataModel, None]):
     """Read kernel modules and associated parameters"""
 
     DATA_MODEL = KernelModuleDataModel
+    CMD_WINDOWS = "wmic os get Version /Value"
+    CMD = "cat /proc/modules"
 
     def parse_proc_modules(self, output: dict) -> dict:
         """Parse command output and return dict of modules
@@ -91,7 +93,7 @@ def collect_all_module_info(self) -> tuple[dict, CommandArtifact]:
             tuple[dict, CommandArtifact]: modules found and exit code
         """
         modules = {}
-        res = self._run_sut_cmd("cat /proc/modules")
+        res = self._run_sut_cmd(self.CMD)
         if res.exit_code != 0:
             self._log_event(
                 category=EventCategory.OS,
@@ -128,7 +130,7 @@ def collect_data(self, args=None) -> tuple[TaskResult, KernelModuleDataModel | N
         kernel_modules = {}
         km_data: KernelModuleDataModel | None = None
         if self.system_info.os_family == OSFamily.WINDOWS:
-            res = self._run_sut_cmd("wmic os get Version /Value")
+            res = self._run_sut_cmd(self.CMD_WINDOWS)
             if res.exit_code == 0:
                 for line in res.stdout.splitlines():
                     if line.startswith("Version="):

nodescraper/plugins/inband/memory/memory_collector.py

@@ -37,6 +37,11 @@ class MemoryCollector(InBandDataCollector[MemoryDataModel, None]):
 
     DATA_MODEL = MemoryDataModel
 
+    CMD_WINDOWS = (
+        "wmic OS get FreePhysicalMemory /Value; wmic ComputerSystem get TotalPhysicalMemory /Value"
+    )
+    CMD = "free -b"
+
     def collect_data(self, args=None) -> tuple[TaskResult, MemoryDataModel | None]:
         """
         Collects memory usage details from the system.
@@ -46,16 +51,14 @@ def collect_data(self, args=None) -> tuple[TaskResult, MemoryDataModel | None]:
         """
         mem_free, mem_total = None, None
         if self.system_info.os_family == OSFamily.WINDOWS:
-            os_memory_cmd = self._run_sut_cmd(
-                "wmic OS get FreePhysicalMemory /Value; wmic ComputerSystem get TotalPhysicalMemory /Value"
-            )
+            os_memory_cmd = self._run_sut_cmd(self.CMD_WINDOWS)
             if os_memory_cmd.exit_code == 0:
                 mem_free = re.search(r"FreePhysicalMemory=(\d+)", os_memory_cmd.stdout).group(
                     1
                 )  # bytes
                 mem_total = re.search(r"TotalPhysicalMemory=(\d+)", os_memory_cmd.stdout).group(1)
         else:
-            os_memory_cmd = self._run_sut_cmd("free -b")
+            os_memory_cmd = self._run_sut_cmd(self.CMD)
             if os_memory_cmd.exit_code == 0:
                 pattern = r"Mem:\s+(\d\.?\d*\w+)\s+\d\.?\d*\w+\s+(\d\.?\d*\w+)"
                 mem_free = re.search(pattern, os_memory_cmd.stdout).group(2)

nodescraper/plugins/inband/nvme/nvme_collector.py

@@ -75,7 +75,7 @@ def collect_data(
 
         for dev in nvme_devices:
             device_data = {}
-            commands = {
+            CMD = {
                 "smart_log": f"nvme smart-log {dev}",
                 "error_log": f"nvme error-log {dev} --log-entries=256",
                 "id_ctrl": f"nvme id-ctrl {dev}",
@@ -86,7 +86,7 @@ def collect_data(
                 "telemetry_log": f"nvme telemetry-log {dev} --output-file={dev}_{f_name}",
             }
 
-            for key, cmd in commands.items():
+            for key, cmd in CMD.items():
                 res = self._run_sut_cmd(cmd, sudo=True)
                 if "--output-file" in cmd:
                     _ = self._read_sut_file(filename=f"{dev}_{f_name}", encoding=None)