adding ls capability with regex pattern expected values

Author: alexandraBara

nodescraper/interfaces/dataplugin.py

@@ -142,15 +142,15 @@ def collect(
             Union[SystemInteractionLevel, str]
         ] = SystemInteractionLevel.INTERACTIVE,
         preserve_connection: bool = False,
-        collection_args: Optional[Union[TCollectArg, dict]] = None,
+        collection_args: Optional[TCollectArg] = None,
     ) -> TaskResult:
         """Run data collector task
 
         Args:
             max_event_priority_level (Union[EventPriority, str], optional): priority limit for events. Defaults to EventPriority.CRITICAL.
             system_interaction_level (Union[SystemInteractionLevel, str], optional): system interaction level. Defaults to SystemInteractionLevel.INTERACTIVE.
             preserve_connection (bool, optional): whether we should close the connection after data collection. Defaults to False.
-            collection_args (Optional[Union[TCollectArg  , dict]], optional): args for data collection. Defaults to None.
+            collection_args (Optional[TCollectArg], optional): args for data collection (validated model). Defaults to None.
 
         Returns:
             TaskResult: task result for data collection
@@ -195,6 +195,13 @@ def collect(
                     message="Connection not available, data collection skipped",
                 )
             else:
+                if (
+                    collection_args is not None
+                    and isinstance(collection_args, dict)
+                    and hasattr(self, "COLLECTOR_ARGS")
+                    and self.COLLECTOR_ARGS is not None
+                ):
+                    collection_args = self.COLLECTOR_ARGS.model_validate(collection_args)
 
                 collection_task = self.COLLECTOR(
                     system_info=self.system_info,
@@ -264,6 +271,14 @@ def analyze(
             )
             return self.analysis_result
 
+        if (
+            analysis_args is not None
+            and isinstance(analysis_args, dict)
+            and hasattr(self, "ANALYZER_ARGS")
+            and self.ANALYZER_ARGS is not None
+        ):
+            analysis_args = self.ANALYZER_ARGS.model_validate(analysis_args)
+
         analyzer_task = self.ANALYZER(
             self.system_info,
             logger=self.logger,

nodescraper/plugins/inband/sys_settings/analyzer_args.py

@@ -25,20 +25,22 @@
 ###############################################################################
 from typing import Optional
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from nodescraper.models import AnalyzerArgs
 
 
 class SysfsCheck(BaseModel):
-    """One sysfs check: path to read, acceptable values, and display name.
+    """One sysfs check: path to read, acceptable values or pattern, and display name.
 
-    If expected is an empty list, the check is treated as passing (no constraint).
+    For file paths: use expected (list of acceptable values); if empty, check passes.
+    For directory paths: use pattern (regex); at least one directory entry must match (e.g. ^hsn[0-9]+).
     """
 
     path: str
-    expected: list[str]
+    expected: list[str] = Field(default_factory=list)
     name: str
+    pattern: Optional[str] = None
 
 
 class SysSettingsAnalyzerArgs(AnalyzerArgs):
@@ -51,16 +53,29 @@ class SysSettingsAnalyzerArgs(AnalyzerArgs):
     checks: Optional[list[SysfsCheck]] = None
 
     def paths_to_collect(self) -> list[str]:
-        """Return the unique sysfs paths from checks, for use by the collector.
+        """Return unique sysfs file paths from checks (those without pattern), for use by the collector."""
+        if not self.checks:
+            return []
+        seen = set()
+        out = []
+        for c in self.checks:
+            if c.pattern:
+                continue
+            p = c.path.rstrip("/")
+            if p not in seen:
+                seen.add(p)
+                out.append(c.path)
+        return out
 
-        Returns:
-            List of unique path strings from self.checks, preserving order of first occurrence.
-        """
+    def paths_to_list(self) -> list[str]:
+        """Return unique sysfs directory paths from checks (those with pattern), for listing (ls)."""
         if not self.checks:
             return []
         seen = set()
         out = []
         for c in self.checks:
+            if not c.pattern:
+                continue
             p = c.path.rstrip("/")
             if p not in seen:
                 seen.add(p)

nodescraper/plugins/inband/sys_settings/collector_args.py

@@ -27,6 +27,11 @@
 
 
 class SysSettingsCollectorArgs(BaseModel):
-    """Collection args for SysSettingsCollector: list of sysfs paths to read."""
+    """Collection args for SysSettingsCollector.
+
+    paths: sysfs paths to read (cat).
+    directory_paths: sysfs paths to list (ls -1); use for checks that match entry names by regex.
+    """
 
     paths: list[str] = []
+    directory_paths: list[str] = []

nodescraper/plugins/inband/sys_settings/sys_settings_analyzer.py

@@ -23,7 +23,8 @@
 # SOFTWARE.
 #
 ###############################################################################
-from typing import Optional, cast
+import re
+from typing import List, Optional, Union, cast
 
 from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus
 from nodescraper.interfaces import DataAnalyzer
@@ -64,14 +65,44 @@ def analyze_data(
         Returns:
             TaskResult with status OK if all checks pass, ERROR if any mismatch or missing path.
         """
-        mismatches = {}
+        mismatches: dict[str, dict[str, Union[Optional[str], List[str]]]] = {}
 
         if not args or not args.checks:
             self.result.status = ExecutionStatus.OK
             self.result.message = "No checks configured."
             return self.result
 
         for check in args.checks:
+            raw_reading = data.readings.get(check.path) or data.readings.get(check.path.rstrip("/"))
+
+            if check.pattern:
+                # Directory-listing check: at least one line must match the regex (e.g. ^hsn[0-9]+)
+                if raw_reading is None:
+                    mismatches[check.name] = {
+                        "path": check.path,
+                        "pattern": check.pattern,
+                        "actual": None,
+                        "reason": "path not collected by this plugin",
+                    }
+                    continue
+                try:
+                    pat = re.compile(check.pattern)
+                except re.error:
+                    mismatches[check.name] = {
+                        "path": check.path,
+                        "pattern": check.pattern,
+                        "reason": "invalid regex",
+                    }
+                    continue
+                lines = [ln.strip() for ln in raw_reading.splitlines() if ln.strip()]
+                if not any(pat.search(ln) for ln in lines):
+                    mismatches[check.name] = {
+                        "path": check.path,
+                        "pattern": check.pattern,
+                        "actual": lines,
+                    }
+                continue
+
             actual = _get_actual_for_path(data, check.path)
             if actual is None:
                 mismatches[check.name] = {
@@ -98,12 +129,15 @@ def analyze_data(
             parts = []
             for name, info in mismatches.items():
                 path = info.get("path", "")
-                expected = info.get("expected")
-                actual = cast(Optional[str], info.get("actual"))
                 reason = info.get("reason")
+                pattern = info.get("pattern")
                 if reason:
-                    part = f"{name} ({path})"
+                    part = f"{name} ({path}): {reason}"
+                elif pattern is not None:
+                    part = f"{name} ({path}): no entry matching pattern {pattern!r}"
                 else:
+                    expected = info.get("expected")
+                    actual = cast(Optional[str], info.get("actual"))
                     part = f"{name} ({path}): expected one of {expected}, actual {repr(actual)}"
                 parts.append(part)
             self.result.message = "Sysfs mismatch: " + "; ".join(parts)

nodescraper/plugins/inband/sys_settings/sys_settings_collector.py

@@ -38,7 +38,7 @@
 
 
 def _parse_bracketed_setting(content: str) -> Optional[str]:
-    """Extract the active setting from sysfs content (value in square brackets).
+    """Extract the active setting from sysfs content.
 
     Args:
         content: Raw sysfs file content (e.g. "[always] madvise never").
@@ -56,19 +56,30 @@ def _paths_from_args(args: Optional[SysSettingsCollectorArgs]) -> list[str]:
     """Extract list of sysfs paths from collection args.
 
     Args:
-        args: Collector args containing paths to read, or None. May be a dict.
+        args: Collector args containing paths to read, or None.
 
     Returns:
         List of sysfs paths; empty if args is None or args.paths is empty.
     """
     if args is None:
         return []
-    paths = args.get("paths") if isinstance(args, dict) else getattr(args, "paths", None)
-    return list(paths) if paths else []
+    return list(args.paths) if args.paths else []
+
+
+def _directory_paths_from_args(args: Optional[SysSettingsCollectorArgs]) -> list[str]:
+    """Extract list of sysfs directory paths to list from collection args."""
+    if args is None:
+        return []
+    return list(args.directory_paths) if args.directory_paths else []
 
 
 def _path_under_sys(path: str) -> Optional[str]:
-    """Normalize path to the suffix under /sys/ for use in 'cat /sys/{}'."""
+    """Normalize path to the suffix under /sys/ for use in 'cat /sys/{}'.
+
+    Accepts paths like '/sys/kernel/...' or 'kernel/...'. Returns the relative
+    part (e.g. 'kernel/mm/transparent_hugepage/enabled'). Returns None if path
+    contains '..' (e.g. /sys/../etc/passwd, /sys/something/../../etc) or is not under /sys.
+    """
     if ".." in path:
         return None
     p = path.strip().lstrip("/")
@@ -80,17 +91,18 @@ def _path_under_sys(path: str) -> Optional[str]:
 
 
 def _sysfs_full_path(suffix: str) -> str:
-    """Return full path /sys/{suffix}."""
+    """Return full path /sys/{suffix} for use as readings key."""
     return f"/sys/{suffix}"
 
 
 class SysSettingsCollector(InBandDataCollector[SysSettingsDataModel, SysSettingsCollectorArgs]):
-    """Collect sysfs settings from user-specified paths (paths come from config/args)."""
+    """Collect sysfs settings from user-specified paths."""
 
     DATA_MODEL = SysSettingsDataModel
     SUPPORTED_OS_FAMILY: set[OSFamily] = {OSFamily.LINUX}
 
     CMD = "cat /sys/{}"
+    CMD_LS = "ls -1 /sys/{}"
 
     def collect_data(
         self, args: Optional[SysSettingsCollectorArgs] = None
@@ -113,7 +125,16 @@ def collect_data(
             return self.result, None
 
         paths = _paths_from_args(args)
-        if not paths:
+        directory_paths = _directory_paths_from_args(args)
+        if directory_paths:
+            self._log_event(
+                category=EventCategory.OS,
+                description=f"Sysfs directory_paths to list: {directory_paths}",
+                data={"directory_paths": directory_paths},
+                priority=EventPriority.INFO,
+                console_log=True,
+            )
+        if not paths and not directory_paths:
             self.result.message = "No paths configured for sysfs collection"
             self.result.status = ExecutionStatus.NOT_RAN
             return self.result, None
@@ -144,6 +165,30 @@ def collect_data(
                     console_log=True,
                 )
 
+        for path in directory_paths:
+            suffix = _path_under_sys(path)
+            if suffix is None:
+                self._log_event(
+                    category=EventCategory.OS,
+                    description=f"Skipping directory path not under /sys or invalid: {path!r}",
+                    data={"path": path},
+                    priority=EventPriority.WARNING,
+                    console_log=True,
+                )
+                continue
+            full_path = _sysfs_full_path(suffix)
+            res = self._run_sut_cmd(self.CMD_LS.format(suffix), sudo=False)
+            if res.exit_code == 0:
+                readings[full_path] = res.stdout.strip() if res.stdout else ""
+            else:
+                self._log_event(
+                    category=EventCategory.OS,
+                    description=f"Failed to list sysfs path: {full_path}",
+                    data={"exit_code": res.exit_code},
+                    priority=EventPriority.WARNING,
+                    console_log=True,
+                )
+
         if not readings:
             self.result.message = "Sysfs settings not read"
             self.result.status = ExecutionStatus.ERROR

test/functional/test_sys_settings_plugin.py

@@ -30,6 +30,11 @@
 import pytest
 
 
+def _project_root() -> Path:
+    """Return project root (directory containing plugin_config.json)."""
+    return Path(__file__).resolve().parent.parent.parent
+
+
 @pytest.fixture
 def fixtures_dir():
     """Return path to fixtures directory."""
@@ -42,6 +47,12 @@ def sys_settings_config_file(fixtures_dir):
     return fixtures_dir / "sys_settings_plugin_config.json"
 
 
+@pytest.fixture
+def plugin_config_json():
+    """Return path to project root plugin_config.json (full config: paths + directory_paths + checks)."""
+    return _project_root() / "plugin_config.json"
+
+
 def test_sys_settings_plugin_with_config_file(run_cli_command, sys_settings_config_file, tmp_path):
     """Test SysSettingsPlugin using config file with collection_args and analysis_args."""
     assert sys_settings_config_file.exists(), f"Config file not found: {sys_settings_config_file}"
@@ -85,3 +96,20 @@ def test_sys_settings_plugin_output_contains_plugin_result(
     output = result.stdout + result.stderr
     # Table or status line should mention the plugin
     assert "SysSettingsPlugin" in output
+
+
+def test_sys_settings_plugin_with_plugin_config_json(run_cli_command, plugin_config_json, tmp_path):
+    """Functional test: run node-scraper with project plugin_config.json (paths + directory_paths + checks)."""
+    assert plugin_config_json.exists(), f"Config not found: {plugin_config_json}"
+
+    log_path = str(tmp_path / "logs_plugin_config")
+    result = run_cli_command(
+        ["--log-path", log_path, "--plugin-configs", str(plugin_config_json)], check=False
+    )
+
+    assert result.returncode in [0, 1, 2]
+    output = result.stdout + result.stderr
+    assert len(output) > 0
+    assert "SysSettingsPlugin" in output
+    # Config exercises paths + directory_paths (/sys/class/net) + pattern check; collector/analyzer run
+    assert "Sysfs" in output or "sys" in output.lower()