added utests

Author: alexandraBara

nodescraper/plugins/inband/sysctl/sysctl_collector.py

@@ -92,7 +92,6 @@ def collect_data(
                 category=EventCategory.OS,
                 description="Sysctl settings not read",
                 priority=EventPriority.CRITICAL,
-                console_log=True,
             )
             self.result.message = "Sysctl settings not read"
             self.result.status = ExecutionStatus.ERROR

test/unit/plugin/test_sysctl_analyzer.py

@@ -0,0 +1,41 @@
+import pytest
+
+from nodescraper.enums import ExecutionStatus
+from nodescraper.plugins.inband.sysctl.analyzer_args import SysctlAnalyzerArgs
+from nodescraper.plugins.inband.sysctl.sysctl_analyzer import SysctlAnalyzer
+from nodescraper.plugins.inband.sysctl.sysctldata import SysctlDataModel
+
+
+@pytest.fixture
+def analyzer(system_info):
+    return SysctlAnalyzer(system_info=system_info)
+
+
+@pytest.fixture
+def correct_data():
+    return SysctlDataModel(
+        vm_swappiness=1,
+        vm_numa_balancing=2,
+        vm_oom_kill_allocating_task=3,
+        vm_compaction_proactiveness=4,
+        vm_compact_unevictable_allowed=5,
+        vm_extfrag_threshold=6,
+        vm_zone_reclaim_mode=7,
+        vm_dirty_background_ratio=8,
+        vm_dirty_ratio=9,
+        vm_dirty_writeback_centisecs=10,
+        kernel_numa_balancing=11,
+    )
+
+
+def test_analyzer_all_match(analyzer, correct_data):
+    args = SysctlAnalyzerArgs.build_from_model(correct_data)
+    result = analyzer.analyze_data(correct_data, args)
+    assert result.status == ExecutionStatus.OK
+
+
+def test_analyzer_mismatch(analyzer, correct_data):
+    args = SysctlAnalyzerArgs(exp_vm_swappiness=3, exp_vm_numa_balancing=4)
+    result = analyzer.analyze_data(correct_data, args)
+    assert result.status == ExecutionStatus.ERROR
+    assert "Sysctl parameters mismatch detected" in result.message

test/unit/plugin/test_sysctl_collector.py

@@ -0,0 +1,53 @@
+from types import SimpleNamespace
+
+import pytest
+
+from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus, OSFamily
+from nodescraper.plugins.inband.sysctl.sysctl_collector import SysctlCollector
+from nodescraper.plugins.inband.sysctl.sysctldata import SysctlDataModel
+
+
+@pytest.fixture
+def linux_sysctl_collector(system_info, conn_mock):
+    system_info.os_family = OSFamily.LINUX
+    return SysctlCollector(system_info, conn_mock)
+
+
+def make_artifact(cmd, exit_code, stdout):
+    return SimpleNamespace(command=cmd, exit_code=exit_code, stdout=stdout, stderr="")
+
+
+def test_collect_data_all_fields_success(linux_sysctl_collector):
+    sysctl_fields = SysctlDataModel.model_fields.keys()
+    responses = [
+        make_artifact(f"sysctl -n {f.replace('_', '.', 1)}", 0, "111") for f in sysctl_fields
+    ]
+
+    linux_sysctl_collector._run_sut_cmd = lambda cmd, seq=responses: seq.pop(0)
+
+    result, data = linux_sysctl_collector.collect_data()
+
+    assert result.status == ExecutionStatus.OK
+    assert isinstance(data, SysctlDataModel)
+    for field in SysctlDataModel.model_fields:
+        assert getattr(data, field) == 111
+
+    event = result.events[-1]
+    assert event.category == "SYSCTL_READ"
+    assert event.priority == EventPriority.INFO.value
+    assert result.message == "SYSCTL data collected"
+
+
+def test_collect_data_all_fail(linux_sysctl_collector):
+    def always_fail(cmd):
+        return make_artifact(cmd, 1, "")
+
+    linux_sysctl_collector._run_sut_cmd = always_fail
+    result, data = linux_sysctl_collector.collect_data()
+
+    assert result.status == ExecutionStatus.ERROR
+    assert data is None
+
+    evt = result.events[0]
+    assert evt.category == EventCategory.OS.value
+    assert "Sysctl settings not read" in result.message