added utest

Author: alexandraBara

nodescraper/plugins/inband/nvme/nvme_collector.py

@@ -55,7 +55,7 @@ def collect_data(
                 priority=EventPriority.WARNING,
             )
             self.result.message = "NVMe data collection skipped on Windows"
-            self.result.status = ExecutionStatus.SKIPPED
+            self.result.status = ExecutionStatus.NOT_RAN
             return self.result, None
 
         commands = [

test/unit/plugin/test_nvme_collector.py

@@ -0,0 +1,104 @@
+###############################################################################
+#
+# MIT License
+#
+# Copyright (c) 2025 Advanced Micro Devices, Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+###############################################################################
+from unittest.mock import MagicMock
+
+import pytest
+
+from nodescraper.enums import EventPriority, ExecutionStatus, OSFamily
+from nodescraper.enums.systeminteraction import SystemInteractionLevel
+from nodescraper.models import TaskResult
+from nodescraper.plugins.inband.nvme.nvme_collector import NvmeCollector
+from nodescraper.plugins.inband.nvme.nvmedata import NvmeDataModel
+
+
+@pytest.fixture
+def collector(system_info, conn_mock):
+    c = NvmeCollector(
+        system_info=system_info,
+        system_interaction_level=SystemInteractionLevel.PASSIVE,
+        connection=conn_mock,
+    )
+    c._log_event = MagicMock()
+    c._run_sut_cmd = MagicMock()
+    c.result = TaskResult()
+    return c
+
+
+def test_skips_on_windows(collector):
+    collector.system_info = MagicMock(os_family=OSFamily.WINDOWS)
+    result, data = collector.collect_data()
+
+    assert result.status == ExecutionStatus.NOT_RAN
+    assert data is None
+    collector._log_event.assert_called_once()
+    assert "Windows" in collector._log_event.call_args.kwargs["description"]
+
+
+def test_successful_collection(collector):
+    collector.system_info = MagicMock(os_family=OSFamily.LINUX)
+
+    collector._run_sut_cmd.return_value = MagicMock(exit_code=0, stdout="output")
+
+    result, data = collector.collect_data()
+
+    assert result.status == ExecutionStatus.OK
+    assert result.message == "NVMe data successfully collected"
+    assert isinstance(data, NvmeDataModel)
+    assert collector._run_sut_cmd.call_count == 7
+    assert any(
+        "Collected NVMe data" in call.kwargs["description"]
+        for call in collector._log_event.call_args_list
+    )
+
+
+def test_partial_failures(collector):
+    collector.system_info = MagicMock(os_family=OSFamily.LINUX)
+
+    def fake_cmd(cmd, sudo):
+        return MagicMock(exit_code=0 if "smart-log" in cmd else 1, stdout="out")
+
+    collector._run_sut_cmd.side_effect = fake_cmd
+
+    result, data = collector.collect_data()
+
+    assert result.status in {ExecutionStatus.OK, ExecutionStatus.ERROR}
+    assert collector._log_event.call_count >= 1
+
+
+def test_no_data_collected(collector):
+    collector.system_info = MagicMock(os_family=OSFamily.LINUX)
+
+    collector._run_sut_cmd.return_value = MagicMock(exit_code=1, stdout="")
+
+    result, data = collector.collect_data()
+
+    assert result.status == ExecutionStatus.ERROR
+    assert data is None
+    assert "No NVMe data collected" in result.message
+    assert any(
+        call.kwargs["priority"] == EventPriority.CRITICAL
+        for call in collector._log_event.call_args_list
+    )