Merge pull request #395 from Limmen/read_emulation

unit test read_emulation_util

simulation-system/libs/csle-common/tests/test_read_emulation_util.py

@@ -0,0 +1,136 @@
+from csle_common.util.read_emulation_statistics_util import ReadEmulationStatisticsUtil
+from csle_collector.host_manager.dao.host_metrics import HostMetrics
+from csle_collector.ossec_ids_manager.dao.ossec_ids_alert_counters import OSSECIdsAlertCounters
+from csle_collector.snort_ids_manager.dao.snort_ids_alert_counters import SnortIdsAlertCounters
+from csle_collector.snort_ids_manager.dao.snort_ids_rule_counters import SnortIdsRuleCounters
+import csle_collector.constants.constants as constants
+import numpy as np
+
+
+class TestReadEmulationUtilSuite:
+    """
+    Test suite for read_emulation_util
+    """
+
+    def test_average_host_metrics(self) -> None:
+        """
+        Test the method that computes the average metrics from a list of host metrics
+
+        :return: None
+        """
+        metrics1 = HostMetrics(
+            num_logged_in_users=5,
+            num_failed_login_attempts=2,
+            num_open_connections=10,
+            num_login_events=3,
+            num_processes=50,
+            num_users=5,
+        )
+        metrics2 = HostMetrics(
+            num_logged_in_users=7,
+            num_failed_login_attempts=4,
+            num_open_connections=14,
+            num_login_events=5,
+            num_processes=60,
+            num_users=6,
+        )
+
+        metrics3 = HostMetrics(
+            num_logged_in_users=6,
+            num_failed_login_attempts=3,
+            num_open_connections=12,
+            num_login_events=4,
+            num_processes=55,
+            num_users=7,
+        )
+
+        average_metrics = ReadEmulationStatisticsUtil.average_host_metrics([metrics1, metrics2, metrics3])
+
+        assert average_metrics.num_logged_in_users == 18
+        assert average_metrics.num_failed_login_attempts == 9
+        assert average_metrics.num_open_connections == 36
+        assert average_metrics.num_login_events == 12
+        assert average_metrics.num_processes == 165
+        assert average_metrics.num_users == 18
+
+    def test_average_ossec_metrics(self) -> None:
+        """
+        Test the method that computes the average metrics from a list of OSSEC metrics
+
+        :return: None
+        """
+        metrics1 = OSSECIdsAlertCounters()
+        metrics1.level_alerts = list(np.zeros(16))
+        metrics1.group_alerts = list(np.zeros(len(set(constants.OSSEC.OSSEC_IDS_ALERT_GROUP_ID.values()))))
+        metrics1.severe_alerts = 5
+        metrics1.warning_alerts = 3
+        metrics1.total_alerts = 10
+        metrics1.alerts_weighted_by_level = 20
+
+        metrics2 = OSSECIdsAlertCounters()
+        metrics2.level_alerts = list(np.zeros(16))
+        metrics2.group_alerts = list(np.zeros(len(set(constants.OSSEC.OSSEC_IDS_ALERT_GROUP_ID.values()))))
+        metrics2.severe_alerts = 10
+        metrics2.warning_alerts = 6
+        metrics2.total_alerts = 20
+        metrics2.alerts_weighted_by_level = 40
+
+        average_metrics = ReadEmulationStatisticsUtil.average_ossec_metrics([metrics1, metrics2])
+
+        assert average_metrics.level_alerts == [0] * 16
+        assert average_metrics.group_alerts == [0] * len(set(constants.OSSEC.OSSEC_IDS_ALERT_GROUP_ID.values()))
+        assert average_metrics.severe_alerts == 15
+        assert average_metrics.warning_alerts == 9
+        assert average_metrics.total_alerts == 30
+        assert average_metrics.alerts_weighted_by_level == 60
+
+    def test_average_snort_metrics(self) -> None:
+        """
+        Test the method that computes the average metrics from a list of Snort metrics
+
+        :return: None
+        """
+        metrics1 = SnortIdsAlertCounters()
+        metrics1.priority_alerts = [0] * 4
+        metrics1.class_alerts = []
+        for i in range(len(set(constants.SNORT_IDS_ROUTER.SNORT_ALERT_IDS_ID.values()))):
+            metrics1.class_alerts.append(0)
+        metrics1.severe_alerts = 5
+        metrics1.warning_alerts = 3
+        metrics1.total_alerts = 10
+        metrics1.alerts_weighted_by_priority = 20
+
+        metrics2 = SnortIdsAlertCounters()
+        metrics2.priority_alerts = [0] * 4
+        metrics2.class_alerts = []
+        for i in range(len(set(constants.SNORT_IDS_ROUTER.SNORT_ALERT_IDS_ID.values()))):
+            metrics2.class_alerts.append(0)
+        metrics2.severe_alerts = 10
+        metrics2.warning_alerts = 6
+        metrics2.total_alerts = 20
+        metrics2.alerts_weighted_by_priority = 40
+
+        average_metrics = ReadEmulationStatisticsUtil.average_snort_metrics([metrics1, metrics2])
+
+        assert average_metrics.priority_alerts == [0] * 4
+        assert average_metrics.class_alerts == [0] * len(set(constants.SNORT_IDS_ROUTER.SNORT_ALERT_IDS_ID.values()))
+        assert average_metrics.severe_alerts == 15
+        assert average_metrics.warning_alerts == 9
+        assert average_metrics.total_alerts == 30
+        assert average_metrics.alerts_weighted_by_priority == 60
+
+    def test_average_snort_rule_metrics(self) -> None:
+        """
+        Test the method that computes the average metrics from a list of Snort rule metrics
+
+        :return: None
+        """
+        metrics1 = SnortIdsRuleCounters()
+        metrics1.rule_alerts = {}
+
+        metrics2 = SnortIdsAlertCounters()
+        metrics2.rule_alerts = {}
+
+        average_metrics = ReadEmulationStatisticsUtil.average_snort_rule_metrics([metrics1, metrics2])
+
+        assert average_metrics.rule_alerts == {}