Bugfix/full storage

nodescraper/plugins/inband/storage/analyzer_args.py

@@ -51,3 +51,7 @@ class StorageAnalyzerArgs(AnalyzerArgs):
         default=False,
         description="If True, match device/mount names with regex; otherwise exact match.",
     )
+    skip_loop_devices: bool = Field(
+        default=True,
+        description="If True, skip /dev/loopN devices (read-only image mounts that always report 100 percent used) unless listed in check_devices.",
+    )

nodescraper/plugins/inband/storage/storage_analyzer.py

@@ -40,6 +40,19 @@ class StorageAnalyzer(DataAnalyzer[StorageDataModel, StorageAnalyzerArgs]):
 
     DATA_MODEL = StorageDataModel
 
+    LOOP_DEVICE_REGEX = re.compile(r"^/dev/loop\d+$")
+
+    def _is_loop_device(self, device_name: str) -> bool:
+        """Check if the device name is a loop device (e.g. /dev/loop0)
+
+        Args:
+            device_name (str): device name to check
+
+        Returns:
+            bool: True if the device name is a /dev/loopN device, False otherwise
+        """
+        return bool(self.LOOP_DEVICE_REGEX.match(device_name))
+
     def _matches_device_filter(
         self, device_name: str, exp_devices: list[str], regex_match: bool
     ) -> bool:
@@ -104,8 +117,12 @@ def analyze_data(
                     device_name, args.check_devices, args.regex_match
                 ):
                     continue
-            elif args.ignore_devices:
-                if self._matches_device_filter(device_name, args.ignore_devices, args.regex_match):
+            else:
+                if args.skip_loop_devices and self._is_loop_device(device_name):
+                    continue
+                if args.ignore_devices and self._matches_device_filter(
+                    device_name, args.ignore_devices, args.regex_match
+                ):
                     continue
 
             condition = False

test/functional/conftest.py

@@ -51,14 +51,12 @@ def redfish_conn_mock():
 def run_cli_command():
     """Fixture that returns a function to run CLI commands."""
 
-    def _run_command(args: List[str], check: bool = False, cwd=None):
+    def _run_command(args: List[str], check: bool = False):
         """Run a node-scraper CLI command.
 
         Args:
             args: List of command-line arguments
             check: If True, raise CalledProcessError on non-zero exit
-            cwd: Working directory for the command (default: the caller's). Use a tmp dir for
-                plugins whose probes write files to the cwd, e.g. wget saving index.html
 
         Returns:
             subprocess.CompletedProcess instance
@@ -70,7 +68,6 @@ def _run_command(args: List[str], check: bool = False, cwd=None):
             capture_output=True,
             text=True,
             check=check,
-            cwd=cwd,
         )
 
     return _run_command

test/functional/test_cli_no_console_log.py

@@ -28,7 +28,7 @@
 import sys
 
 
-def test_no_console_log_with_log_path_none_still_parses(tmp_path):
+def test_no_console_log_with_log_path_none_still_parses():
     """--log-path None + --no-console-log defaults to ./scraper_logs_* (no argparse error)."""
     result = subprocess.run(
         [
@@ -44,7 +44,6 @@ def test_no_console_log_with_log_path_none_still_parses(tmp_path):
         ],
         capture_output=True,
         text=True,
-        cwd=tmp_path,
     )
     assert result.returncode == 0, result.stderr
     combined = (result.stdout or "") + (result.stderr or "")

test/functional/test_network_plugin.py

@@ -158,7 +158,6 @@ def test_network_plugin_with_netprobe_wget(run_cli_command, tmp_path):
             "wget",
         ],
         check=False,
-        cwd=tmp_path,  # wget saves the fetched page (index.html, index.html.1, ...) into the cwd
     )
 
     assert result.returncode in [0, 1, 2]

test/functional/test_plugin_configs.py

@@ -181,34 +181,25 @@ def test_plugin_config_with_multiple_configs(run_cli_command, plugin_config_file
 def test_plugin_config_with_nonexistent_file(run_cli_command, tmp_path):
     """Test that a nonexistent config file path fails gracefully."""
     nonexistent_path = str(tmp_path / "nonexistent_config.json")
-    result = run_cli_command(
-        ["--log-path", str(tmp_path / "logs"), f"--plugin-configs={nonexistent_path}"],
-        check=False,
-    )
+    result = run_cli_command([f"--plugin-configs={nonexistent_path}"], check=False)
 
     assert result.returncode != 0
     output = (result.stdout + result.stderr).lower()
     assert "error" in output or "no plugin config found" in output
 
 
-def test_plugin_config_with_invalid_builtin_name(run_cli_command, tmp_path):
+def test_plugin_config_with_invalid_builtin_name(run_cli_command):
     """Test that an invalid built-in config name fails gracefully."""
-    result = run_cli_command(
-        ["--log-path", str(tmp_path / "logs"), "--plugin-configs=NonExistentConfig"],
-        check=False,
-    )
+    result = run_cli_command(["--plugin-configs=NonExistentConfig"], check=False)
 
     assert result.returncode != 0
     output = (result.stdout + result.stderr).lower()
     assert "error" in output or "no plugin config found" in output
 
 
-def test_plugin_config_with_invalid_json(run_cli_command, invalid_plugin_config, tmp_path):
+def test_plugin_config_with_invalid_json(run_cli_command, invalid_plugin_config):
     """Test that an invalid JSON file fails gracefully."""
-    result = run_cli_command(
-        ["--log-path", str(tmp_path / "logs"), f"--plugin-configs={invalid_plugin_config}"],
-        check=False,
-    )
+    result = run_cli_command([f"--plugin-configs={invalid_plugin_config}"], check=False)
 
     assert result.returncode != 0
     output = (result.stdout + result.stderr).lower()

test/functional/test_run_plugins.py

@@ -93,12 +93,9 @@ def test_run_all_plugins_together(run_cli_command, all_plugins, tmp_path):
     assert len(output) > 0
 
 
-def test_run_plugin_with_invalid_name(run_cli_command, tmp_path):
+def test_run_plugin_with_invalid_name(run_cli_command):
     """Test that running a non-existent plugin logs a warning and falls back to default config."""
-    result = run_cli_command(
-        ["--log-path", str(tmp_path / "logs"), "run-plugins", "NonExistentPlugin"],
-        check=False,
-    )
+    result = run_cli_command(["run-plugins", "NonExistentPlugin"], check=False)
 
     # Invalid plugin is ignored and default config runs instead
     # Exit code depends on whether default config plugins succeed

test/unit/framework/test_compare_runs.py

@@ -126,38 +126,32 @@ def test_run_compare_runs_no_data(caplog, tmp_path):
     (tmp_path / "run2").mkdir()
     logger = logging.getLogger()
     plugin_reg = PluginRegistry()
-    run_compare_runs(
-        str(tmp_path / "run1"),
-        str(tmp_path / "run2"),
-        plugin_reg,
-        logger,
-        artifact_dir=str(tmp_path),
-    )
+    run_compare_runs(str(tmp_path / "run1"), str(tmp_path / "run2"), plugin_reg, logger)
     assert "No plugin data found" in caplog.text
 
 
-def test_run_compare_runs_with_fixture_dirs(caplog, framework_fixtures_path, tmp_path):
+def test_run_compare_runs_with_fixture_dirs(caplog, framework_fixtures_path):
     """With two log dirs using fixture (same content), table is produced and no diffs."""
     caplog.set_level(logging.INFO)
     logger = logging.getLogger()
     plugin_reg = PluginRegistry()
     base = framework_fixtures_path / "log_dir"
-    run_compare_runs(str(base), str(base), plugin_reg, logger, artifact_dir=str(tmp_path))
+    run_compare_runs(str(base), str(base), plugin_reg, logger)
     assert "Loading run 1" in caplog.text
     assert "Loading run 2" in caplog.text
     assert "Plugin" in caplog.text
     assert "No differences" in caplog.text or "difference" in caplog.text.lower()
 
 
-def test_run_compare_runs_sysctl_fixture(caplog, framework_fixtures_path, tmp_path):
+def test_run_compare_runs_sysctl_fixture(caplog, framework_fixtures_path):
     """Compare runs using compare_runs_sysctl fixtures; run1 and run2 differ (SysctlPlugin)."""
     caplog.set_level(logging.INFO)
     logger = logging.getLogger()
     base = framework_fixtures_path / "compare_runs_sysctl"
     run1 = base / "run1"
     run2 = base / "run2"
     plugin_reg = PluginRegistry()
-    run_compare_runs(str(run1), str(run2), plugin_reg, logger, artifact_dir=str(tmp_path))
+    run_compare_runs(str(run1), str(run2), plugin_reg, logger)
     assert "Loading run 1" in caplog.text
     assert "Loading run 2" in caplog.text
     assert "Sysctl" in caplog.text or "sysctl" in caplog.text
@@ -173,7 +167,7 @@ def test_run_compare_runs_one_run_missing_plugin(caplog, framework_fixtures_path
     run2 = tmp_path / "run2"
     run2.mkdir()
     # run2 has no collector dirs
-    run_compare_runs(str(run1), str(run2), plugin_reg, logger, artifact_dir=str(tmp_path))
+    run_compare_runs(str(run1), str(run2), plugin_reg, logger)
     assert "Loading run 1" in caplog.text
     assert "Loading run 2" in caplog.text
     assert "not found in run 2" in caplog.text or "NOT_RAN" in caplog.text

test/unit/plugin/test_storage_analyzer.py

@@ -61,11 +61,45 @@ def multiple_dev():
     )
 
 
+@pytest.fixture
+def loop_dev_model():
+    return StorageDataModel(
+        storage_data={
+            "/dev/nvme0n1p2": DeviceStorageData(
+                total=943441641472,
+                free=869796294656,
+                used=25645850624,
+                percent=3,
+            ),
+            "/dev/loop1": DeviceStorageData(
+                total=1966080,
+                free=0,
+                used=1966080,
+                percent=100.0,
+            ),
+            "/dev/loop2": DeviceStorageData(
+                total=52559872,
+                free=0,
+                used=52559872,
+                percent=100.0,
+            ),
+        }
+    )
+
+
 @pytest.fixture
 def analyzer(system_info):
     return StorageAnalyzer(system_info=system_info)
 
 
+def _critical_devices(result):
+    return sorted(
+        e.data["offending_device"]["device"]
+        for e in result.events
+        if e.priority == EventPriority.CRITICAL
+    )
+
+
 def test_filter_exact_match(analyzer):
     assert analyzer._matches_device_filter("foo", ["foo", "bar"], regex_match=False)
     assert not analyzer._matches_device_filter("baz", ["foo", "bar"], regex_match=False)
@@ -166,3 +200,123 @@ def test_device_filter(analyzer, model_obj):
     )
     result2 = analyzer.analyze_data(model_obj, args2)
     assert result2.status == ExecutionStatus.OK
+
+
+def test_skip_loop_devices_default_true():
+    assert StorageAnalyzerArgs().skip_loop_devices is True
+
+
+@pytest.mark.parametrize(
+    "args",
+    [StorageAnalyzerArgs(min_required_free_space_prct=10), None],
+    ids=["explicit_args", "args_none"],
+)
+def test_loop_devices_skipped_by_default(analyzer, loop_dev_model, args):
+    result = analyzer.analyze_data(loop_dev_model, args)
+    assert result.status == ExecutionStatus.OK
+    assert result.events == []
+    assert not any(e.priority == EventPriority.CRITICAL for e in result.events)
+    assert "loop" not in result.message
+    assert result.message == "Sufficient disk space available on [/dev/nvme0n1p2]"
+
+
+def test_loop_devices_checked_when_skip_disabled(analyzer, loop_dev_model):
+    args = StorageAnalyzerArgs(min_required_free_space_prct=10, skip_loop_devices=False)
+    result = analyzer.analyze_data(loop_dev_model, args)
+    assert result.status == ExecutionStatus.ERROR
+    assert len(result.events) == 2
+    assert all(e.category == EventCategory.STORAGE.value for e in result.events)
+    assert all(e.priority == EventPriority.CRITICAL for e in result.events)
+    assert all("Insufficient disk space" in e.description for e in result.events)
+    assert _critical_devices(result) == ["/dev/loop1", "/dev/loop2"]
+    assert result.message.startswith("Insufficient disk space on [/dev/loop1, /dev/loop2]")
+
+
+def test_check_devices_overrides_loop_skip(analyzer, loop_dev_model):
+    args = StorageAnalyzerArgs(min_required_free_space_prct=10, check_devices=["/dev/loop1"])
+    assert args.skip_loop_devices is True
+    result = analyzer.analyze_data(loop_dev_model, args)
+    assert result.status == ExecutionStatus.ERROR
+    assert len(result.events) == 1
+    event = result.events[0]
+    assert event.category == EventCategory.STORAGE.value
+    assert event.priority == EventPriority.CRITICAL
+    assert "Insufficient disk space" in event.description
+    assert "/dev/loop1" in event.description
+    assert event.data["offending_device"]["device"] == "/dev/loop1"
+    assert result.message.startswith("Insufficient disk space on [/dev/loop1]")
+
+
+def test_loop_skip_does_not_hide_real_full_disk(analyzer, loop_dev_model):
+    loop_dev_model.storage_data["/dev/sda1"] = DeviceStorageData(
+        total=1000, free=50, used=950, percent=95
+    )
+    args = StorageAnalyzerArgs(min_required_free_space_prct=10)
+    result = analyzer.analyze_data(loop_dev_model, args)
+    assert result.status == ExecutionStatus.ERROR
+    assert len(result.events) == 1
+    event = result.events[0]
+    assert event.category == EventCategory.STORAGE.value
+    assert event.priority == EventPriority.CRITICAL
+    assert "Insufficient disk space" in event.description
+    assert event.data["offending_device"]["device"] == "/dev/sda1"
+    assert result.message.startswith("Insufficient disk space on [/dev/sda1]")
+    assert "loop" not in result.message
+
+
+def test_loop_skip_composes_with_ignore_devices(analyzer, loop_dev_model):
+    loop_dev_model.storage_data["/dev/sda1"] = DeviceStorageData(
+        total=1000, free=50, used=950, percent=95
+    )
+    args = StorageAnalyzerArgs(min_required_free_space_prct=10, ignore_devices=["/dev/sda1"])
+    result = analyzer.analyze_data(loop_dev_model, args)
+    assert result.status == ExecutionStatus.OK
+    assert result.events == []
+    assert result.message == "Sufficient disk space available on [/dev/nvme0n1p2]"
+
+
+@pytest.mark.parametrize(
+    "device_name, is_loop",
+    [
+        ("/dev/loop0", True),
+        ("/dev/loop12", True),
+        ("/dev/loopback-foo", False),
+        ("loop1", False),
+        ("/dev/loop", False),
+        ("/dev/sda1", False),
+        ("C:", False),
+    ],
+)
+def test_is_loop_device(analyzer, device_name, is_loop):
+    assert analyzer._is_loop_device(device_name) is is_loop
+
+
+@pytest.mark.parametrize(
+    "device_name, skipped",
+    [
+        ("/dev/loop0", True),
+        ("/dev/loop12", True),
+        ("/dev/loopback-foo", False),
+        ("loop1", False),
+    ],
+)
+def test_loop_skip_name_matching(analyzer, device_name, skipped):
+    model = StorageDataModel(
+        storage_data={
+            device_name: DeviceStorageData(total=1000, free=0, used=1000, percent=100.0),
+        }
+    )
+    args = StorageAnalyzerArgs(min_required_free_space_prct=10)
+    result = analyzer.analyze_data(model, args)
+    if skipped:
+        assert result.status == ExecutionStatus.OK
+        assert result.events == []
+        assert result.message == "Sufficient disk space available on []"
+    else:
+        assert result.status == ExecutionStatus.ERROR
+        assert len(result.events) == 1
+        event = result.events[0]
+        assert event.category == EventCategory.STORAGE.value
+        assert event.priority == EventPriority.CRITICAL
+        assert event.data["offending_device"]["device"] == device_name
+        assert result.message.startswith(f"Insufficient disk space on [{device_name}]")