Add destination assertions to rule tests

panther_analysis_tool/detection_schemas/analysis_config_schema.json

@@ -1674,6 +1674,12 @@
         "ExpectedResult": {
           "type": "boolean"
         },
+        "ExpectedDestinations": {
+          "items": {
+            "type": "string"
+          },
+          "type": "array"
+        },
         "LogType": {
           "type": "string"
         },

panther_analysis_tool/main.py

@@ -48,6 +48,7 @@
 from panther_core.policy import TYPE_POLICY, Policy
 from panther_core.rule import TYPE_SCHEDULED_RULE, Detection, Rule
 from panther_core.testing import (
+    FunctionTestResult,
     TestCaseEvaluator,
     TestExpectations,
     TestResult,
@@ -142,9 +143,7 @@
 from panther_analysis_tool.directory import setup_temp
 from panther_analysis_tool.enriched_event_generator import EnrichedEventGenerator
 from panther_analysis_tool.log_schemas import user_defined
-from panther_analysis_tool.log_type_validator import (
-    split_analysis_by_log_type_support,
-)
+from panther_analysis_tool.log_type_validator import split_analysis_by_log_type_support
 from panther_analysis_tool.schemas import LOOKUP_TABLE_SCHEMA, SQL_LOOKUP_TABLE_SCHEMA
 from panther_analysis_tool.util import (
     BackendNotFoundException,
@@ -1748,6 +1747,17 @@ def _run_tests(  # pylint: disable=too-many-arguments,too-many-positional-argume
                 continue
             found_debug_unit_test = True
 
+        expected_destinations = unit_test.get("ExpectedDestinations")
+        test_destinations_by_name = destinations_by_name.copy()
+        if expected_destinations is not None:
+            for destination_name in expected_destinations:
+                test_destinations_by_name.setdefault(
+                    destination_name,
+                    FakeDestination(
+                        destination_id=str(uuid4()), destination_display_name=destination_name
+                    ),
+                )
+
         test_output = ""
         try:
             entry: dict = unit_test["Resource"] if "Resource" in unit_test else unit_test["Log"]
@@ -1780,10 +1790,12 @@ def _run_tests(  # pylint: disable=too-many-arguments,too-many-positional-argume
                 if mock_methods:
                     with patch.multiple(detection.module, **mock_methods):
                         result = detection.run(
-                            test_case, {}, destinations_by_name, batch_mode=False
+                            test_case, {}, test_destinations_by_name, batch_mode=False
                         )
                 else:
-                    result = detection.run(test_case, {}, destinations_by_name, batch_mode=False)
+                    result = detection.run(
+                        test_case, {}, test_destinations_by_name, batch_mode=False
+                    )
             test_output = ""
             if not debug_args or not debug_args.get("debug_mode", False):
                 test_output = cast(io.StringIO, test_output_buf).getvalue()
@@ -1825,6 +1837,17 @@ def _run_tests(  # pylint: disable=too-many-arguments,too-many-positional-argume
                 print(test_output)
 
         # print results
+        expected_destination_ids = (
+            None
+            if expected_destinations is None
+            else (
+                ["SKIP"]
+                if not expected_destinations
+                else [
+                    test_destinations_by_name[name].destination_id for name in expected_destinations
+                ]
+            )
+        )
         spec = TestSpecification(
             id=unit_test["Name"],
             name=unit_test["Name"],
@@ -1847,6 +1870,17 @@ def _run_tests(  # pylint: disable=too-many-arguments,too-many-positional-argume
             test_result.functions.referenceFunction = None
             test_result.functions.uniqueFunction = None
 
+        if expected_destination_ids is not None and sorted(expected_destination_ids) != sorted(
+            result.destinations_output or []
+        ):
+            test_result.passed = False
+            if test_result.functions.destinationsFunction is None:
+                test_result.functions.destinationsFunction = FunctionTestResult(
+                    output="null", error=None, matched=False
+                )
+            else:
+                test_result.functions.destinationsFunction.matched = False
+
         if all_test_results:
             test_result_str = status_passed if test_result.passed else status_errored
             stored_test_results = getattr(all_test_results, test_result_str)

panther_analysis_tool/schemas.py

@@ -178,6 +178,7 @@ def validate(  # type: ignore
                     "LogType"
                 ): str,  # Not needed anymore, optional for backwards compatibility
                 "ExpectedResult": bool,
+                Optional("ExpectedDestinations"): [str],
                 "Log": object,
                 Optional("Mocks"): [MOCK_SCHEMA],
             }

tests/fixtures/detections/expected_destinations/example_expected_destinations.py

@@ -0,0 +1,10 @@
+def rule(_event):
+    return True
+
+
+def destinations(event):
+    if event.get("route_to_multiple_destinations", False):
+        return ["secondary-destination", "test-destination"]
+    if event.get("route_alert", False):
+        return ["test-destination"]
+    return []

tests/fixtures/detections/expected_destinations/example_expected_destinations.yml

@@ -0,0 +1,38 @@
+AnalysisType: rule
+Enabled: true
+Filename: example_expected_destinations.py
+RuleID: Example.Rule.Expected.Destinations
+LogTypes:
+  - AWS.CloudTrail
+Severity: Low
+Tests:
+  - Name: Routes alert
+    ExpectedResult: true
+    ExpectedDestinations:
+      - test-destination
+    Log:
+      {
+        "route_alert": true,
+      }
+  - Name: Routes alert regardless of destination order
+    ExpectedResult: true
+    ExpectedDestinations:
+      - test-destination
+      - secondary-destination
+    Log:
+      {
+        "route_to_multiple_destinations": true,
+      }
+  - Name: Routes signal
+    ExpectedResult: true
+    ExpectedDestinations: []
+    Log:
+      {
+        "route_alert": false,
+      }
+  - Name: Uses destination without expectation
+    ExpectedResult: true
+    Log:
+      {
+        "route_alert": true,
+      }

tests/fixtures/detections/expected_destinations_mismatch/example_expected_destinations_mismatch.py

@@ -0,0 +1,6 @@
+def rule(_event):
+    return True
+
+
+def destinations(_event):
+    return None

tests/fixtures/detections/expected_destinations_mismatch/example_expected_destinations_mismatch.yml

@@ -0,0 +1,14 @@
+AnalysisType: rule
+Enabled: true
+Filename: example_expected_destinations_mismatch.py
+RuleID: Example.Rule.Expected.Destinations.Mismatch
+CreateAlert: false
+LogTypes:
+  - AWS.CloudTrail
+Severity: Low
+Tests:
+  - Name: Routes signal unexpectedly
+    ExpectedResult: true
+    ExpectedDestinations:
+      - test-destination
+    Log: {}

tests/unit/panther_analysis_tool/test_main.py

@@ -973,6 +973,41 @@ def test_available_destination_names_valid_name_returned(self) -> None:
         )
         self.assertEqual(return_code, 0)
 
+    def test_expected_destinations(self) -> None:
+        return_code, _ = mock_test_analysis(
+            self,
+            [
+                "test",
+                "--path",
+                f"{DETECTIONS_FIXTURES_PATH}/expected_destinations",
+            ],
+        )
+        self.assertEqual(return_code, 0)
+
+    def test_expected_destinations_mismatch(self) -> None:
+        return_code, _ = mock_test_analysis(
+            self,
+            [
+                "test",
+                "--path",
+                f"{DETECTIONS_FIXTURES_PATH}/expected_destinations_mismatch",
+            ],
+        )
+        self.assertEqual(return_code, 1)
+
+    def test_expected_destinations_do_not_leak(self) -> None:
+        return_code, _ = mock_test_analysis(
+            self,
+            [
+                "test",
+                "--path",
+                f"{DETECTIONS_FIXTURES_PATH}/expected_destinations",
+                "--available-destination",
+                "allowed-destination",
+            ],
+        )
+        self.assertEqual(return_code, 1)
+
     def test_invalid_query(self) -> None:
         return_code, invalid_specs = mock_test_analysis(
             self, f"test --path {FIXTURES_PATH}/queries/invalid".split()