Expanding coverage and other metrics in summary.yml

contentctl/actions/detection_testing/infrastructures/DetectionTestingInfrastructure.py

@@ -10,7 +10,6 @@
 from tempfile import TemporaryDirectory, mktemp
 from ssl import SSLEOFError, SSLZeroReturnError
 from sys import stdout
-#from dataclasses import dataclass
 from shutil import copyfile
 from typing import Union, Optional
 
@@ -29,7 +28,7 @@
 from contentctl.objects.base_test import BaseTest
 from contentctl.objects.unit_test import UnitTest
 from contentctl.objects.integration_test import IntegrationTest
-from contentctl.objects.unit_test_attack_data import UnitTestAttackData
+from contentctl.objects.test_attack_data import TestAttackData
 from contentctl.objects.unit_test_result import UnitTestResult
 from contentctl.objects.integration_test_result import IntegrationTestResult
 from contentctl.objects.test_group import TestGroup
@@ -1179,7 +1178,7 @@ def retry_search_until_timeout(
 
         return
 
-    def delete_attack_data(self, attack_data_files: list[UnitTestAttackData]):
+    def delete_attack_data(self, attack_data_files: list[TestAttackData]):
         for attack_data_file in attack_data_files:
             index = attack_data_file.custom_index or self.sync_obj.replay_index
             host = attack_data_file.host or self.sync_obj.replay_host
@@ -1212,7 +1211,7 @@ def replay_attack_data_files(
 
     def replay_attack_data_file(
         self,
-        attack_data_file: UnitTestAttackData,
+        attack_data_file: TestAttackData,
         tmp_dir: str,
         test_group: TestGroup,
         test_group_start_time: float,
@@ -1280,7 +1279,7 @@ def replay_attack_data_file(
     def hec_raw_replay(
         self,
         tempfile: str,
-        attack_data_file: UnitTestAttackData,
+        attack_data_file: TestAttackData,
         verify_ssl: bool = False,
     ):
         if verify_ssl is False:

contentctl/actions/detection_testing/views/DetectionTestingView.py

@@ -1,5 +1,6 @@
 import abc
 import datetime
+from typing import Any
 
 from pydantic import BaseModel
 
@@ -10,6 +11,7 @@
 )
 from contentctl.helper.utils import Utils
 from contentctl.objects.enums import DetectionStatus
+from contentctl.objects.base_test_result import TestResultStatus
 
 
 class DetectionTestingView(BaseModel, abc.ABC):
@@ -74,18 +76,23 @@ def getSummaryObject(
         self,
         test_result_fields: list[str] = ["success", "message", "exception", "status", "duration", "wait_duration"],
         test_job_fields: list[str] = ["resultCount", "runDuration"],
-    ) -> dict:
+    ) -> dict[str, dict[str, Any] | list[dict[str, Any]] | str]:
         """
         Iterates over detections, consolidating results into a single dict and aggregating metrics
         :param test_result_fields: fields to pull from the test result
         :param test_job_fields: fields to pull from the job content of the test result
         :returns: summary dict
         """
-        # Init the list of tested detections, and some metrics aggregate counters
-        tested_detections = []
+        # Init the list of tested and skipped detections, and some metrics aggregate counters
+        tested_detections: list[dict[str, Any]] = []
+        skipped_detections: list[dict[str, Any]] = []
         total_pass = 0
         total_fail = 0
         total_skipped = 0
+        total_production = 0
+        total_experimental = 0
+        total_deprecated = 0
+        total_manual = 0
 
         # Iterate the detections tested (anything in the output queue was tested)
         for detection in self.sync_obj.outputQueue:
@@ -95,46 +102,57 @@ def getSummaryObject(
             )
 
             # Aggregate detection pass/fail metrics
-            if summary["success"] is False:
+            if detection.test_status == TestResultStatus.FAIL:
                 total_fail += 1
+            elif detection.test_status == TestResultStatus.PASS:
+                total_pass += 1
+            elif detection.test_status == TestResultStatus.SKIP:
+                total_skipped += 1
+
+            # Aggregate production status metrics
+            if detection.status == DetectionStatus.production.value:                                # type: ignore
+                total_production += 1
+            elif detection.status == DetectionStatus.experimental.value:                            # type: ignore
+                total_experimental += 1
+            elif detection.status == DetectionStatus.deprecated.value:                              # type: ignore
+                total_deprecated += 1
+
+            # Check if the detection is manual_test
+            if detection.tags.manual_test is not None:
+                total_manual += 1
+
+            # Append to our list (skipped or tested)
+            if detection.test_status == TestResultStatus.SKIP:
+                skipped_detections.append(summary)
             else:
-                #Test is marked as a success, but we need to determine if there were skipped unit tests
-                #SKIPPED tests still show a success in this field, but we want to count them differently
-                pass_increment = 1
-                for test in summary.get("tests"):
-                    if test.get("test_type") == "unit" and test.get("status") == "skip":
-                        total_skipped += 1
-                        #Test should not count as a pass, so do not increment the count
-                        pass_increment = 0
-                        break
-                total_pass += pass_increment
-
-
-            # Append to our list
-            tested_detections.append(summary)
-
-        # Sort s.t. all failures appear first (then by name)
-        #Second short condition is a hack to get detections with unit skipped tests to appear above pass tests
-        tested_detections.sort(key=lambda x: (x["success"], 0 if x.get("tests",[{}])[0].get("status","status_missing")=="skip" else 1, x["name"]))
+                tested_detections.append(summary)
+
+        # Sort tested detections s.t. all failures appear first, then by name
+        tested_detections.sort(
+            key=lambda x: (
+                x["success"],
+                x["name"]
+            )
+        )
+
+        # Sort skipped detections s.t. detections w/ tests appear before those w/o, then by name
+        skipped_detections.sort(
+            key=lambda x: (
+                0 if len(x["tests"]) > 0 else 1,
+                x["name"]
+            )
+        )
 
         # Aggregate summaries for the untested detections (anything still in the input queue was untested)
         total_untested = len(self.sync_obj.inputQueue)
-        untested_detections = []
+        untested_detections: list[dict[str, Any]] = []
         for detection in self.sync_obj.inputQueue:
             untested_detections.append(detection.get_summary())
 
         # Sort by detection name
         untested_detections.sort(key=lambda x: x["name"])
 
-        # Get lists of detections (name only) that were skipped due to their status (experimental or deprecated)
-        experimental_detections = sorted([
-            detection.name for detection in self.sync_obj.skippedQueue if detection.status == DetectionStatus.experimental.value
-        ])
-        deprecated_detections = sorted([
-            detection.name for detection in self.sync_obj.skippedQueue if detection.status == DetectionStatus.deprecated.value
-        ])
-
-        # If any detection failed, the overall success is False
+        # If any detection failed, or if there are untested detections, the overall success is False
         if (total_fail + len(untested_detections)) == 0:
             overall_success = True
         else:
@@ -143,33 +161,40 @@ def getSummaryObject(
         # Compute total detections
         total_detections = total_fail + total_pass + total_untested + total_skipped
 
+        # Compute total detections actually tested (at least one test not skipped)
+        total_tested_detections = total_fail + total_pass
 
         # Compute the percentage of completion for testing, as well as the success rate
         percent_complete = Utils.getPercent(
             len(tested_detections), len(untested_detections), 1
         )
         success_rate = Utils.getPercent(
-            total_pass, total_detections-total_skipped, 1
+            total_pass, total_tested_detections, 1
         )
 
-        # TODO (#230): expand testing metrics reported
+        # TODO (#230): expand testing metrics reported (and make nested)
         # Construct and return the larger results dict
         result_dict = {
             "summary": {
+                "mode": self.config.getModeName(),
+                "enable_integration_testing": self.config.enable_integration_testing,
                 "success": overall_success,
                 "total_detections": total_detections,
+                "total_tested_detections": total_tested_detections,
                 "total_pass": total_pass,
                 "total_fail": total_fail,
                 "total_skipped": total_skipped,
                 "total_untested": total_untested,
-                "total_experimental_or_deprecated": len(deprecated_detections+experimental_detections),
+                "total_production": total_production,
+                "total_experimental": total_experimental,
+                "total_deprecated": total_deprecated,
+                "total_manual": total_manual,
+                "total_other_skips": total_skipped - total_deprecated - total_experimental - total_manual,
                 "success_rate": success_rate,
             },
             "tested_detections": tested_detections,
+            "skipped_detections": skipped_detections,
             "untested_detections": untested_detections,
             "percent_complete": percent_complete,
-            "deprecated_detections": deprecated_detections,
-            "experimental_detections": experimental_detections
-
         }
         return result_dict

contentctl/actions/detection_testing/views/DetectionTestingViewFile.py

@@ -13,7 +13,6 @@ class DetectionTestingViewFile(DetectionTestingView):
     output_filename: str = OUTPUT_FILENAME
 
     def getOutputFilePath(self) -> pathlib.Path:
-
         folder_path = pathlib.Path('.') / self.output_folder
         output_file = folder_path / self.output_filename
 
@@ -27,13 +26,12 @@ def stop(self):
         output_file = self.getOutputFilePath()
 
         folder_path.mkdir(parents=True, exist_ok=True)
-
-
+
         result_dict = self.getSummaryObject()
-        
+
         # use the yaml writer class
         with open(output_file, "w") as res:
-            res.write(yaml.safe_dump(result_dict,sort_keys=False))
+            res.write(yaml.safe_dump(result_dict, sort_keys=False))
 
     def showStatus(self, interval: int = 60):
         pass

contentctl/actions/test.py

@@ -45,34 +45,17 @@ class TestInputDto:
 
 class Test:
 
-    def filter_detections(self, input_dto: TestInputDto)->TestInputDto:
-
+    def filter_tests(self, input_dto: TestInputDto) -> TestInputDto:
         if not input_dto.config.enable_integration_testing:
-            #Skip all integraiton tests if integration testing is not enabled:
+            # Skip all integraiton tests if integration testing is not enabled:
             for detection in input_dto.detections:
                 for test in detection.tests:
                     if isinstance(test, IntegrationTest):
                         test.skip("TEST SKIPPED: Skipping all integration tests")
-
-        list_after_filtering:List[Detection] = []
-        #extra filtering which may be removed/modified in the future
-        for detection in input_dto.detections:
-            if (detection.status != DetectionStatus.production.value):
-                #print(f"{detection.name} - Not testing because [STATUS: {detection.status}]")
-                pass
-            elif detection.type == AnalyticsType.Correlation:
-                #print(f"{detection.name} - Not testing because [  TYPE: {detection.type}]")
-                pass
-            else:
-                list_after_filtering.append(detection)
-
-        return TestInputDto(list_after_filtering, input_dto.config)
-
-
-    def execute(self, input_dto: TestInputDto) -> bool:
 
-
+        return input_dto
 
+    def execute(self, input_dto: TestInputDto) -> bool:
         output_dto = DetectionTestingManagerOutputDto()
 
         web = DetectionTestingViewWeb(config=input_dto.config, sync_obj=output_dto)
@@ -87,26 +70,33 @@ def execute(self, input_dto: TestInputDto) -> bool:
         manager = DetectionTestingManager(
             input_dto=manager_input_dto, output_dto=output_dto
         )
-
+
+        mode = input_dto.config.getModeName()
         if len(input_dto.detections) == 0:
-            print(f"With Detection Testing Mode '{input_dto.config.getModeName()}', there were [0] detections found to test.\nAs such, we will quit immediately.")
-            # Directly call stop so that the summary.yml will be generated. Of course it will not have any test results, but we still want it to contain
-            # a summary showing that now detections were tested.
+            print(
+                f"With Detection Testing Mode '{mode}', there were [0] detections found to test."
+                "\nAs such, we will quit immediately."
+            )
+            # Directly call stop so that the summary.yml will be generated. Of course it will not
+            # have any test results, but we still want it to contain a summary showing that now
+            # detections were tested.
             file.stop()
         else:
-            print(f"MODE: [{input_dto.config.getModeName()}] - Test [{len(input_dto.detections)}] detections")
-            if input_dto.config.mode in [DetectionTestingMode.changes, DetectionTestingMode.selected]:
-                files_string = '\n- '.join([str(pathlib.Path(detection.file_path).relative_to(input_dto.config.path)) for detection in input_dto.detections])
+            print(f"MODE: [{mode}] - Test [{len(input_dto.detections)}] detections")
+            if mode in [DetectionTestingMode.changes.value, DetectionTestingMode.selected.value]:
+                files_string = '\n- '.join(
+                    [str(pathlib.Path(detection.file_path).relative_to(input_dto.config.path)) for detection in input_dto.detections]
+                )
                 print(f"Detections:\n- {files_string}")
 
             manager.setup()
             manager.execute()
-        
+
         try:
             summary_results = file.getSummaryObject()
             summary = summary_results.get("summary", {})
 
-            print("Test Summary")
+            print(f"Test Summary (mode: {summary.get('mode','Error')})")
             print(f"\tSuccess                      : {summary.get('success',False)}")
             print(
                 f"\tSuccess Rate                 : {summary.get('success_rate','ERROR')}"
@@ -115,10 +105,28 @@ def execute(self, input_dto: TestInputDto) -> bool:
                 f"\tTotal Detections             : {summary.get('total_detections','ERROR')}"
             )
             print(
-                f"\tPassed Detections            : {summary.get('total_pass','ERROR')}"
+                f"\tTotal Tested Detections      : {summary.get('total_tested_detections','ERROR')}"
+            )
+            print(
+                f"\t  Passed Detections          : {summary.get('total_pass','ERROR')}"
+            )
+            print(
+                f"\t  Failed Detections          : {summary.get('total_fail','ERROR')}"
+            )
+            print(
+                f"\tSkipped Detections           : {summary.get('total_skipped','ERROR')}"
+            )
+            print(
+                f"\t  Experimental Detections    : {summary.get('total_experimental','ERROR')}"
+            )
+            print(
+                f"\t  Deprecated Detections      : {summary.get('total_deprecated','ERROR')}"
+            )
+            print(
+                f"\t  Manually Tested Detections : {summary.get('total_manual','ERROR')}"
             )
             print(
-                f"\tFailed Detections            : {summary.get('total_fail','ERROR')}"
+                f"\t  Other Skipped Detections   : {summary.get('total_other_skips','ERROR')}"
             )
             print(
                 f"\tUntested Detections          : {summary.get('total_untested','ERROR')}"

contentctl/contentctl.py

@@ -114,9 +114,9 @@ def test_common_func(config:test_common):
 
     t = Test()
 
-    # Remove detections that we do not want to test because they are
-    # not production, the correct type, or manual_test only
-    filted_test_input_dto = t.filter_detections(test_input_dto)
+    # Remove detections or disable tests that we do not want to test (e.g. integration testing is
+    # disabled)
+    filted_test_input_dto = t.filter_tests(test_input_dto)
 
     if config.plan_only:
         #Emit the test plan and quit. Do not actually run the test