diff --git a/.script/tests/asimParsersTest/ASimFilteringTest.py b/.script/tests/asimParsersTest/ASimFilteringTest.py index 1d8ad5fcffb..697e6f94502 100644 --- a/.script/tests/asimParsersTest/ASimFilteringTest.py +++ b/.script/tests/asimParsersTest/ASimFilteringTest.py @@ -25,10 +25,6 @@ import re import subprocess from datetime import datetime, timedelta, timezone -from azure.monitor.query import LogsQueryClient, LogsQueryStatus -from azure.identity import DefaultAzureCredential -from azure.identity import InteractiveBrowserCredential -from azure.core.exceptions import HttpResponseError import csv DUMMY_VALUE = "\'!not_REAL_vAlUe\'" @@ -73,8 +69,10 @@ } def attempt_to_connect(): try: + from azure.monitor.query import LogsQueryClient, LogsQueryStatus + from azure.identity import DefaultAzureCredential + credential = DefaultAzureCredential() - #credential = InteractiveBrowserCredential() # Uncomment this line if you want to use the interactive browser credential for testing purposes client = LogsQueryClient(credential) empty_query = "" response = client.query_workspace( @@ -91,15 +89,7 @@ def attempt_to_connect(): sys.stdout.flush() # Explicitly flush stdout return None -# Authenticating the user -client = attempt_to_connect() -if client is None: - print("::error::Couldn't connect to workspace with DefaultAzureCredential.") - sys.stdout.flush() # Explicitly flush stdout - sys.exit() -else: - # Proceed with using 'client' for further operations - pass +client = None def get_parser(parser_path): try: @@ -273,6 +263,13 @@ def handle_test_failure(parser_file_path, error_message=None): sys.exit(1) # Uncomment this line to fail workflow when tests are not successful. def main(): + global client + client = attempt_to_connect() + if client is None: + print("::error::Couldn't connect to workspace with DefaultAzureCredential.") + sys.stdout.flush() + sys.exit(1) + # Get modified ASIM Parser files along with their status current_directory = os.path.dirname(os.path.abspath(__file__)) @@ -375,12 +372,16 @@ def tests_main_func(self): if schema_of_parser not in all_schemas_parameters: self.fail(f"Schema: {schema_of_parser} - Not an existing schema or not supported by the validations script") param_to_column_mapping = all_schemas_parameters[schema_of_parser] + if schema_of_parser == "EmailEvent": + self.email_event_default_test(parser_file["ParserParams"], query_definition) for param in parser_file['ParserParams']: param_name = param['Name'] with self.subTest(): if param_name not in param_to_column_mapping: self.fail(f"parameter: {param_name} - No such parameter in {schema_of_parser} schema") column_name_in_table = param_to_column_mapping[param_name] + if schema_of_parser == "EmailEvent" and param_name == "pack": + self.pack_test(query_definition) self.send_param_to_test(param, query_definition, columns_in_answer, column_name_in_table) self.legacy_eventtype_compatibility_test(parser_file, query_definition) @@ -401,15 +402,20 @@ def send_param_to_test(self, param, query_definition, columns_in_answer, column_ """ param_name = param['Name'] param_type = param['Type'] - # pack parameter is not checked by the validations if (param_name == "pack"): return if (param_name == "disabled"): self.disabled_test(query_definition) - elif column_name_in_table not in columns_in_answer: + elif isinstance(column_name_in_table, tuple): + self.email_recipient_test(param_name, query_definition, column_name_in_table, columns_in_answer) + elif column_name_in_table not in columns_in_answer: + print_and_flush( + f"Parameter: {param_name} - mapped column {column_name_in_table} is unavailable; " + "filtering validation is partial." + ) + self.unavailable_column_test(param, query_definition) return elif (param_type == "datetime"): - #pass #TODO add test for datetime self.datetime_test(param, query_definition, column_name_in_table) elif (param_type == "dynamic"): self.dynamic_test(param, query_definition, column_name_in_table) @@ -475,6 +481,120 @@ def datetime_test(self, param, query_definition, column_name_in_table): num_of_rows_with_filter_in_query = len(filtered_response.tables[0].rows) self.assertLess(num_of_rows_with_filter_in_query, num_of_rows_when_no_filters_in_query, f"Parameter: {param_name} - Expected to have less results after filtering. Filtered by value: {datetime_mid_point_value}") + def get_count_for_query(self, query): + response = self.send_query(query + " | count\n") + return response.tables[0].rows[0][0] + + def unavailable_column_test(self, param, query_definition): + param_name = param["Name"] + param_type = param["Type"] + default = param["Default"] + if param_type == "string": + default = f"'{default}'" + baseline_count = self.get_count_for_query(query_definition + "query()\n") + default_count = self.get_count_for_query( + query_definition + f"query({param_name}={default})\n" + ) + self.assertEqual(default_count, baseline_count) + if param_type == "dynamic": + absent_value = f"dynamic([{DUMMY_VALUE}])" + elif param_type == "datetime": + absent_value = ( + "datetime(9999-12-31T23:59:59Z)" + if param_name == "starttime" + else "datetime(0001-01-01T00:00:00Z)" + ) + else: + absent_value = INT_DUMMY_VALUE if param_type == "int" else DUMMY_VALUE + absent_count = self.get_count_for_query( + query_definition + f"query({param_name}={absent_value})\n" + ) + self.assertEqual( + absent_count, + 0, + f"Parameter: {param_name} - Returned results for an absent value." + ) + + def email_event_default_test(self, parser_params, query_definition): + defaults = [] + for param in parser_params: + default = param["Default"] + if param["Type"] == "string": + default = f"'{default}'" + defaults.append(f"{param['Name']}={default}") + baseline_count = self.get_count_for_query(query_definition + "query()\n") + explicit_default_count = self.get_count_for_query( + query_definition + f"query({', '.join(defaults)})\n" + ) + self.assertEqual( + explicit_default_count, + baseline_count, + "EmailEvent - Explicit default filters changed the unfiltered result set." + ) + + def pack_test(self, query_definition): + baseline_count = self.get_count_for_query(query_definition + "query(pack=false)\n") + packed_count = self.get_count_for_query(query_definition + "query(pack=true)\n") + self.assertEqual(packed_count, baseline_count, "Parameter: pack - Changed the number of results.") + + def email_recipient_test(self, parameter_name, query_definition, recipient_columns, columns_in_answer): + available_columns = [column for column in recipient_columns if column in columns_in_answer] + baseline_count = self.get_count_for_query(query_definition + "query()\n") + empty_count = self.get_count_for_query( + query_definition + f"query({parameter_name}=dynamic([]))\n" + ) + self.assertEqual(empty_count, baseline_count, f"Parameter: {parameter_name} - Empty array changed the unfiltered result set.") + if not available_columns: + print_and_flush( + f"Parameter: {parameter_name} - recipient columns are unavailable; " + "matching validation is partial." + ) + absent_count = self.get_count_for_query( + query_definition + + f"query({parameter_name}=dynamic([{DUMMY_VALUE}]))\n" + ) + self.assertEqual(absent_count, 0, f"Parameter: {parameter_name} - Returned results for a fictive recipient.") + return + + array_values = ", ".join( + f"coalesce({column}, dynamic([]))" for column in available_columns + ) + value_query = ( + query_definition + + "query() | project recipient_values=array_concat(" + + array_values + + ") | mv-expand recipient=recipient_values " + "| where isnotempty(tostring(recipient)) | distinct recipient=tostring(recipient) | take 2\n" + ) + values = [row[0] for row in self.send_query(value_query).tables[0].rows] + if not values: + print_and_flush( + f"Parameter: {parameter_name} - no recipient values are present; " + "matching validation is partial." + ) + else: + values_literal = create_values_string(values) + for selected_values in ([values[0]], values) if len(values) > 1 else ([values[0]],): + filtered_count = self.get_count_for_query( + query_definition + + f"query({parameter_name}=dynamic([{create_values_string(selected_values)}]))\n" + ) + self.assertGreater( + filtered_count, 0, + f"Parameter: {parameter_name} - No rows matched recipient values {values_literal}." + ) + self.assertLessEqual(filtered_count, baseline_count) + if filtered_count == baseline_count: + print_and_flush( + f"Parameter: {parameter_name} - recipient values matched all rows; " + "selectivity validation is partial." + ) + absent_count = self.get_count_for_query( + query_definition + + f"query({parameter_name}=dynamic([{DUMMY_VALUE}]))\n" + ) + self.assertEqual(absent_count, 0, f"Parameter: {parameter_name} - Returned results for a fictive recipient.") + def legacy_eventtype_compatibility_test(self, parser_file, query_definition): parser_name = parser_file.get('ParserName') compatible_values = LEGACY_EVENTTYPE_COMPATIBILITY.get(parser_name) @@ -513,15 +633,15 @@ def scalar_test(self, param, query_definition, column_name_in_table): no_filter_query = query_definition + create_execution_string_without_parameters(column_name_in_table) no_filter_response = self.send_query(no_filter_query) self.assertNotEqual(len(no_filter_response.tables[0].rows) , 0 , f"No data for parameter:{param_name}") - with self.subTest(): - self.assertNotEqual(len(no_filter_response.tables[0].rows), 1, f"Only one value exists for parameter: {param_name} - validations for this parameter are partial" ) + if len(no_filter_response.tables[0].rows) == 1: + print_and_flush( + f"Parameter: {param_name} - only one value is present; " + "selectivity validation is partial." + ) # Taking the first value returned in the response selected_value = no_filter_response.tables[0].rows[0][0] value_to_filter = f"\'{selected_value}\'" if param['Type']=="string" else selected_value query_with_filter = query_definition + create_execution_strings_with_one_parameter(param_name, value_to_filter, column_name_in_table) - if selected_value=="": - query_with_filter = query_definition + f"query() | where isempty({column_name_in_table}) | summarize count() by {column_name_in_table}\n" - # Performing a filtering by the first value returned in the first response self.scalar_test_check_filtering(param_name, query_with_filter, value_to_filter) @@ -595,7 +715,7 @@ def get_response_for_query_with_one_parameter(self, parameter_name, query_defini return self.get_response_for_query_with_parameters(parameter_name, query_definition, column_name_in_table, filtering_with_one_value_list) - def dynamic_tests_assertions(self, parameter_name, num_of_rows_with_filter_in_query, filter_parameters, num_of_rows_when_no_filters_in_query): + def dynamic_tests_assertions(self, parameter_name, num_of_rows_with_filter_in_query, filter_parameters, num_of_rows_when_no_filters_in_query, require_fewer_results): """ Performing assertions for dynamic tests with parameter filtering. @@ -608,17 +728,51 @@ def dynamic_tests_assertions(self, parameter_name, num_of_rows_with_filter_in_qu """ with self.subTest(): self.assertNotEqual(0, num_of_rows_with_filter_in_query, f"Parameter: {parameter_name} - Got no results at all after filtering. Filtered by value: {filter_parameters}") - with self.subTest(): - if (num_of_rows_when_no_filters_in_query == 1): - self.assertEqual(1, num_of_rows_when_no_filters_in_query, f"Parameter: {parameter_name} - Expected to have one result after filtering. Filtered by value: {filter_parameters}") + if require_fewer_results: + self.assertLess( + num_of_rows_with_filter_in_query, + num_of_rows_when_no_filters_in_query, + f"Parameter: {parameter_name} - Expected fewer distinct values after filtering." + ) + else: + self.assertLessEqual( + num_of_rows_with_filter_in_query, + num_of_rows_when_no_filters_in_query, + f"Parameter: {parameter_name} - Filtering returned more values than the unfiltered query." + ) + if not require_fewer_results and num_of_rows_with_filter_in_query == num_of_rows_when_no_filters_in_query: + print_and_flush( + f"Parameter: {parameter_name} - filter values matched all distinct output values; " + "multi-value selectivity validation is partial." + ) + + def dynamic_response_matches_filter(self, parameter_name, response, filter_values): + normalized_filters = [str(value).casefold() for value in filter_values] + for row in response.tables[0].rows: + output_value = str(row[COLUMN_INDEX_IN_ROW]).casefold() + if parameter_name.endswith("_has_any_prefix"): + matches = any(output_value.startswith(value) for value in normalized_filters) + elif parameter_name.endswith("_has_all"): + matches = all(value in output_value for value in normalized_filters) + elif parameter_name.endswith("_has_any"): + matches = any(value in output_value for value in normalized_filters) else: - self.assertLess(num_of_rows_with_filter_in_query, num_of_rows_when_no_filters_in_query, f"Parameter: {parameter_name} - Expected to have less results after filtering. Filtered by value: {filter_parameters}") + matches = output_value in normalized_filters + self.assertTrue( + matches, + f"Parameter: {parameter_name} - Filter returned an unrelated value: {output_value}" + ) - # Failing the test with corresponding message if values list is empty - def fail_if_values_list_is_empty(self, values_list,parameter_name, test_type): + # Report partial coverage when no usable values are available. + def values_list_is_available(self, values_list, parameter_name, test_type): if len(values_list) == 0: - self.fail(f"Parameter: {parameter_name} - Unable to find values to perform {test_type} tests") + print_and_flush( + f"Parameter: {parameter_name} - no values are available for {test_type} tests; " + "matching validation is partial." + ) + return False + return True def dynamic_tests_helper(self, parameter_name, query_definition, num_of_rows_when_no_filters_in_query, column_name_in_table, values_list, test_type): @@ -634,21 +788,58 @@ def dynamic_tests_helper(self, parameter_name, query_definition, num_of_rows_whe values_list : List of at most two values that will be used to perform filtering. test_type : Name of the specific tests performed. It can be default testing or specific testing for has_any/has_any_prefix parameters """ - self.fail_if_values_list_is_empty(values_list, parameter_name, test_type) + if not self.values_list_is_available(values_list, parameter_name, test_type): + self.dynamic_tests_check_fictive_value(parameter_name, query_definition, column_name_in_table) + return # Performing filtering with one value filtering_with_one_value_response, one_value_string = self.get_response_for_query_with_one_parameter(parameter_name, query_definition, column_name_in_table, values_list) num_of_rows_with_one_parameter_in_query = len(filtering_with_one_value_response.tables[0].rows) - self.dynamic_tests_assertions(parameter_name, num_of_rows_with_one_parameter_in_query, one_value_string, num_of_rows_when_no_filters_in_query) + self.dynamic_response_matches_filter(parameter_name, filtering_with_one_value_response, [values_list[0]]) + self.dynamic_tests_assertions( + parameter_name, + num_of_rows_with_one_parameter_in_query, + one_value_string, + num_of_rows_when_no_filters_in_query, + num_of_rows_when_no_filters_in_query > 1, + ) - # Performing filtering with two values if possible - if len(values_list) == 1 or num_of_rows_when_no_filters_in_query <= MAX_FILTERING_PARAMETERS: - # Skip self.fail if values_list contains both "Logon" and "Logoff" and parameter_name is "eventtype_in" - if not (set(values_list) == {"Logon", "Logoff"} and parameter_name == "eventtype_in"): - self.fail(f"Parameter: {parameter_name} - Not enough data to perform two values {test_type} tests") - filtering_response, values_string = self.get_response_for_query_with_parameters(parameter_name, query_definition, column_name_in_table, values_list) - num_of_rows_with_parameters_in_query = len(filtering_response.tables[0].rows) - self.dynamic_tests_assertions(parameter_name, num_of_rows_with_parameters_in_query, values_string, num_of_rows_when_no_filters_in_query) + if len(values_list) > 1: + second_value_response, _ = self.get_response_for_query_with_one_parameter( + parameter_name, query_definition, column_name_in_table, [values_list[1]] + ) + self.dynamic_response_matches_filter(parameter_name, second_value_response, [values_list[1]]) + filtering_response, values_string = self.get_response_for_query_with_parameters(parameter_name, query_definition, column_name_in_table, values_list) + num_of_rows_with_parameters_in_query = len(filtering_response.tables[0].rows) + self.dynamic_response_matches_filter(parameter_name, filtering_response, values_list) + self.dynamic_tests_assertions( + parameter_name, + num_of_rows_with_parameters_in_query, + values_string, + num_of_rows_when_no_filters_in_query, + num_of_rows_when_no_filters_in_query > len(values_list), + ) + expected_values = { + str(row[COLUMN_INDEX_IN_ROW]) + for row in filtering_with_one_value_response.tables[0].rows + } | { + str(row[COLUMN_INDEX_IN_ROW]) + for row in second_value_response.tables[0].rows + } + actual_values = { + str(row[COLUMN_INDEX_IN_ROW]) + for row in filtering_response.tables[0].rows + } + self.assertEqual( + actual_values, + expected_values, + f"Parameter: {parameter_name} - Multi-value filter did not return the union of individual matches." + ) + else: + print_and_flush( + f"Parameter: {parameter_name} - only one value is available for {test_type} tests; " + "multi-value validation is partial." + ) # Performing a query with a non-existing value, expecting to return no results self.dynamic_tests_check_fictive_value(parameter_name, query_definition, column_name_in_table) @@ -732,22 +923,43 @@ def has_all_test(self, parameter_name, query_definition, no_filter_rows, column_ column_name_in_table : The name of the column in the query response on which the parameter performs filtering """ values_list = self.get_values_for_dynamic_tests(no_filter_rows) - self.fail_if_values_list_is_empty(values_list, parameter_name, "has_all") + if not self.values_list_is_available(values_list, parameter_name, "has_all"): + self.dynamic_tests_check_fictive_value(parameter_name, query_definition, column_name_in_table) + return # Performing filtering with one value filtering_with_one_value_response, one_value_string = self.get_response_for_query_with_one_parameter(parameter_name, query_definition, column_name_in_table, values_list) num_of_rows_when_no_filters_in_query = len (no_filter_rows) num_of_rows_with_one_parameter_in_query = len(filtering_with_one_value_response.tables[0].rows) - self.dynamic_tests_assertions(parameter_name, num_of_rows_with_one_parameter_in_query, one_value_string, num_of_rows_when_no_filters_in_query) + self.dynamic_response_matches_filter(parameter_name, filtering_with_one_value_response, [values_list[0]]) + self.dynamic_tests_assertions( + parameter_name, + num_of_rows_with_one_parameter_in_query, + one_value_string, + num_of_rows_when_no_filters_in_query, + num_of_rows_when_no_filters_in_query > 1, + ) # Performing filtering with two values if possible first_part_of_a_value , second_part_of_a_value = get_splitted_parts_of_string(values_list) if first_part_of_a_value is None: - self.fail(f"Parameter: {parameter_name} - has_all tests performed for only one filtering value") + print_and_flush( + f"Parameter: {parameter_name} - no value can be split for a multi-value has_all check; " + "multi-value validation is partial." + ) + self.dynamic_tests_check_fictive_value(parameter_name, query_definition, column_name_in_table) + return splitted_parts_list = [first_part_of_a_value , second_part_of_a_value] filtering_response, values_string = self.get_response_for_query_with_parameters(parameter_name, query_definition, column_name_in_table, splitted_parts_list) num_of_rows_with_parameters_in_query = len(filtering_response.tables[0].rows) - self.dynamic_tests_assertions(parameter_name, num_of_rows_with_parameters_in_query, values_string, num_of_rows_when_no_filters_in_query) + self.dynamic_response_matches_filter(parameter_name, filtering_response, splitted_parts_list) + self.dynamic_tests_assertions( + parameter_name, + num_of_rows_with_parameters_in_query, + values_string, + num_of_rows_when_no_filters_in_query, + num_of_rows_when_no_filters_in_query > len(splitted_parts_list), + ) # Performing a query with a non-existing value, expecting to return no results self.dynamic_tests_check_fictive_value(parameter_name, query_definition, column_name_in_table) @@ -806,8 +1018,21 @@ def dynamic_test(self, parameter, query_definition, column_name_in_table): no_filter_response = self.send_query(no_filter_query) no_filter_rows = no_filter_response.tables[0].rows self.assertNotEqual(len(no_filter_rows) , 0 , f"No data for parameter:{parameter_name}") - with self.subTest(): - self.assertNotEqual(len(no_filter_rows), 1, f"Only one value exists for parameter: {parameter_name} - validations for this parameter are partial" ) + if len(no_filter_rows) == 1: + print_and_flush( + f"Parameter: {parameter_name} - only one value is present; " + "selectivity validation is partial." + ) + if parameter["Default"] == "dynamic([])": + baseline_count = self.get_count_for_query(query_definition + "query()\n") + empty_filter_count = self.get_count_for_query( + query_definition + f"query({parameter_name}=dynamic([]))\n" + ) + self.assertEqual( + empty_filter_count, + baseline_count, + f"Parameter: {parameter_name} - Empty array changed the unfiltered result set." + ) # Specific tests parameters with certain suffixes if parameter_name.endswith('has_any'): @@ -856,7 +1081,12 @@ def send_query(self, query_str): query_str : A string with the KQL query that will be sent to the workspace. """ failed_query_message = f"The following query failed:\n{query_str}" + from azure.monitor.query import LogsQueryStatus + from azure.core.exceptions import HttpResponseError + try: + if client is None: + self.fail("Azure Logs Query client is not initialized") response = client.query_workspace( workspace_id = ws_id, query = query_str, @@ -960,6 +1190,19 @@ def send_query(self, query_str): "srcipaddr" : "SrcIpAddr", "starttime" : "EventStartTime" }, + "EmailEvent" : + { + "starttime" : "EventStartTime", + "endtime" : "EventEndTime", + "ipaddr_has_any_prefix" : "IpAddr", + "emailfromaddress_has_any" : "EmailFromAddress", + "emailrecipient_has_any" : ("EmailToRecipients", "EmailCCRecipients", "EmailBCCRecipients"), + "emailsubject_has_any" : "EmailSubject", + "emaildirection_in" : "EmailDirection", + "emaildeliverylocation_in" : "EmailDeliveryLocation", + "eventresult" : "EventResult", + "pack" : "" + }, "FileEvent" : { "actorusername_has_any" : "ActorUsername", diff --git a/.script/tests/asimParsersTest/test_asim_filtering_validator.py b/.script/tests/asimParsersTest/test_asim_filtering_validator.py new file mode 100644 index 00000000000..c1252e9b618 --- /dev/null +++ b/.script/tests/asimParsersTest/test_asim_filtering_validator.py @@ -0,0 +1,291 @@ +import ast +import importlib.util +import os +import re +import unittest +from types import SimpleNamespace +from unittest.mock import Mock + + +SCRIPT_PATH = os.path.join(os.path.dirname(__file__), "ASimFilteringTest.py") +SPEC = importlib.util.spec_from_file_location("asim_filtering_validator", SCRIPT_PATH) +validator = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(validator) + +EXPECTED_EMAIL_EVENT_PARAMETERS = { + "starttime": "EventStartTime", + "endtime": "EventEndTime", + "ipaddr_has_any_prefix": "IpAddr", + "emailfromaddress_has_any": "EmailFromAddress", + "emailrecipient_has_any": ( + "EmailToRecipients", + "EmailCCRecipients", + "EmailBCCRecipients", + ), + "emailsubject_has_any": "EmailSubject", + "emaildirection_in": "EmailDirection", + "emaildeliverylocation_in": "EmailDeliveryLocation", + "eventresult": "EventResult", + "pack": "", +} + + +def response(rows): + return SimpleNamespace(tables=[SimpleNamespace(rows=rows)]) + + +def filtered_rows_for_query(query, parameter_name, rows): + parameter_match = re.search(rf"{parameter_name}=dynamic\(\[([^\]]*)\]\)", query) + if parameter_match is None: + return rows + values = re.findall(r"@'([^']*)'", parameter_match.group(1)) + if validator.DUMMY_VALUE.strip("'") in query: + return [] + matches = [] + for row in rows: + output_value = str(row[0]).casefold() + normalized_values = [value.casefold() for value in values] + if parameter_name.endswith("_has_any_prefix"): + matched = any(output_value.startswith(value) for value in normalized_values) + elif parameter_name.endswith("_has_any"): + matched = any(value in output_value for value in normalized_values) + else: + matched = output_value in normalized_values + if matched: + matches.append(row) + return matches + + +class EmailEventFilteringValidatorTests(unittest.TestCase): + def make_test(self, query_responses): + test = validator.FilteringTest() + test.send_query = Mock(side_effect=query_responses) + return test + + def test_email_event_catalog_maps_all_standard_parameters(self): + self.assertEqual( + validator.all_schemas_parameters["EmailEvent"], + EXPECTED_EMAIL_EVENT_PARAMETERS, + ) + + def test_legacy_asim_tester_catalog_matches_shared_validator(self): + legacy_path = os.path.join( + os.path.dirname(__file__), + "..", + "..", + "..", + "ASIM", + "dev", + "ASimTester", + "filteringTest", + "ASimFilteringTest.py", + ) + with open(legacy_path, encoding="utf-8") as legacy_file: + syntax_tree = ast.parse(legacy_file.read()) + mapping_assignment = next( + node + for node in syntax_tree.body + if isinstance(node, ast.Assign) + and any( + isinstance(target, ast.Name) and target.id == "all_schemas_parameters" + for target in node.targets + ) + ) + legacy_mapping = ast.literal_eval(mapping_assignment.value) + self.assertEqual(legacy_mapping["EmailEvent"], EXPECTED_EMAIL_EVENT_PARAMETERS) + + def test_email_event_catalog_covers_the_unifier_parser_parameters(self): + parser_path = os.path.join( + os.path.dirname(__file__), + "..", + "..", + "..", + "Parsers", + "ASimEmailEvent", + "Parsers", + "imEmailEvent.yaml", + ) + parser = validator.get_parser(parser_path) + self.assertEqual( + set(validator.all_schemas_parameters["EmailEvent"]), + {param["Name"] for param in parser["ParserParams"]}, + ) + + def test_email_event_defaults_and_pack_control_preserve_rows(self): + test = self.make_test( + [ + response([[7]]), + response([[7]]), + response([[7]]), + response([[7]]), + ] + ) + test.email_event_default_test( + [ + {"Name": "ipaddr_has_any_prefix", "Type": "dynamic", "Default": "dynamic([])"}, + {"Name": "eventresult", "Type": "string", "Default": "*"}, + ], + "let query=(ipaddr_has_any_prefix:dynamic=dynamic([]),eventresult:string='*') { Logs };\n", + ) + test.pack_test("let query(pack:bool=false) { Logs };\n") + self.assertIn("query(ipaddr_has_any_prefix=dynamic([]), eventresult='*')", test.send_query.call_args_list[1].args[0]) + self.assertIn("query(pack=true)", test.send_query.call_args_list[3].args[0]) + + def test_existing_schema_mappings_and_disabled_control_remain_supported(self): + self.assertEqual( + validator.all_schemas_parameters["Authentication"]["eventtype_in"], + "EventType", + ) + self.assertEqual( + validator.all_schemas_parameters["NetworkSession"]["srcipaddr_has_any_prefix"], + "SrcIpAddr", + ) + test = self.make_test([response([[0]]), response([[5]])]) + test.disabled_test("let query(disabled:bool=false) { Logs };\n") + self.assertIn("query(disabled=true)", test.send_query.call_args_list[0].args[0]) + self.assertIn("query(disabled=false)", test.send_query.call_args_list[1].args[0]) + + def test_datetime_filter_preserves_original_timegenerated_behavior(self): + test = self.make_test( + [ + response([["2026-10-21T00:00:00Z"], ["2026-10-22T00:00:00Z"]]), + response([["2026-10-21T12:00:00Z", 2, 86400000000]]), + response([["2026-10-22T00:00:00Z"]]), + ] + ) + test.datetime_test( + {"Name": "starttime", "Type": "datetime"}, + "let query(starttime:datetime=datetime(null)) { Logs };\n", + "EventStartTime", + ) + query_text = "\n".join(call.args[0] for call in test.send_query.call_args_list) + self.assertIn("max(TimeGenerated)", query_text) + self.assertIn("project TimeGenerated", query_text) + self.assertNotIn("EventStartTime", query_text) + self.assertIn("query(starttime=datetime(2026-10-21T12:00:00Z))", query_text) + self.assertEqual(test.send_query.call_count, 3) + + def test_sparse_maillog_filter_checks_match_and_absent_values(self): + test = self.make_test( + [ + response([["Delivered", 4]]), + response([["Delivered", 4]]), + response([]), + ] + ) + test.scalar_test( + {"Name": "eventresult", "Type": "string"}, + "let query(eventresult:string='*') { MailLog };\n", + "EventResult", + ) + self.assertIn("query(eventresult='Delivered')", test.send_query.call_args_list[1].args[0]) + self.assertIn("!not_REAL_vAlUe", test.send_query.call_args_list[2].args[0]) + + def test_sparse_dynamic_filter_checks_match_empty_and_absent_values(self): + test = self.make_test( + [ + response([["Inbound", 2]]), + response([[2]]), + response([[2]]), + response([["Inbound", 2]]), + response([]), + ] + ) + test.dynamic_test( + {"Name": "emaildirection_in", "Type": "dynamic", "Default": "dynamic([])"}, + "let query(emaildirection_in:dynamic=dynamic([])) { MailLog };\n", + "EmailDirection", + ) + self.assertIn("emaildirection_in=dynamic([])", test.send_query.call_args_list[2].args[0]) + self.assertIn("emaildirection_in=dynamic", test.send_query.call_args_list[4].args[0]) + + def test_message_sender_and_subject_has_any_filters(self): + cases = [ + ("emailfromaddress_has_any", "EmailFromAddress", "alice@alpha.test", "bob@beta.test"), + ("emailsubject_has_any", "EmailSubject", "Invoice 1042", "Reset access"), + ] + for parameter_name, column_name, first_value, second_value in cases: + with self.subTest(parameter=parameter_name): + baseline_rows = [[first_value, 2], [second_value, 1]] + test = self.make_test([]) + + def send_query(query): + if "| count" in query: + return response([[3]]) + if f"query() | summarize count() by {column_name}" in query: + return response(baseline_rows) + return response(filtered_rows_for_query(query, parameter_name, baseline_rows)) + + test.send_query.side_effect = send_query + test.dynamic_test( + {"Name": parameter_name, "Type": "dynamic", "Default": "dynamic([])"}, + f"let query({parameter_name}:dynamic=dynamic([])) {{ Message }};\n", + column_name, + ) + self.assertTrue( + any(f"{parameter_name}=dynamic([" in call.args[0] for call in test.send_query.call_args_list) + ) + + def test_ip_prefix_filter_uses_distinct_network_prefixes(self): + test = self.make_test([]) + + def send_query(query): + if "| count" in query: + return response([[4]]) + if "query() | summarize count() by IpAddr" in query: + rows = [["192.0.2.10", 2], ["198.51.100.3", 2]] + return response(rows) + return response( + filtered_rows_for_query( + query, + "ipaddr_has_any_prefix", + [["192.0.2.10", 2], ["198.51.100.3", 2]], + ) + ) + + test.send_query.side_effect = send_query + test.dynamic_test( + {"Name": "ipaddr_has_any_prefix", "Type": "dynamic", "Default": "dynamic([])"}, + "let query(ipaddr_has_any_prefix:dynamic=dynamic([])) { Message };\n", + "IpAddr", + ) + self.assertTrue( + any("192.0.2." in call.args[0] or "198.51.100." in call.args[0] + for call in test.send_query.call_args_list) + ) + + def test_unavailable_column_still_checks_empty_and_absent_filters(self): + test = self.make_test([response([[5]]), response([[5]]), response([[0]])]) + test.unavailable_column_test( + {"Name": "emaildirection_in", "Type": "dynamic", "Default": "dynamic([])"}, + "let query(emaildirection_in:dynamic=dynamic([])) { MailLog };\n", + ) + self.assertIn("query(emaildirection_in=dynamic([]))", test.send_query.call_args_list[1].args[0]) + self.assertIn("!not_REAL_vAlUe", test.send_query.call_args_list[2].args[0]) + + def test_recipient_filter_searches_all_recipient_arrays(self): + test = self.make_test( + [ + response([[3]]), + response([[3]]), + response([["alice@example.test"], ["bob@example.test"]]), + response([[1]]), + response([[2]]), + response([[0]]), + ] + ) + test.email_recipient_test( + "emailrecipient_has_any", + "let query(emailrecipient_has_any:dynamic=dynamic([])) { EmailLogs };\n", + ("EmailToRecipients", "EmailCCRecipients", "EmailBCCRecipients"), + {"EmailToRecipients", "EmailCCRecipients", "EmailBCCRecipients"}, + ) + value_query = test.send_query.call_args_list[2].args[0] + self.assertIn("array_concat(coalesce(EmailToRecipients, dynamic([])), coalesce(EmailCCRecipients, dynamic([])), coalesce(EmailBCCRecipients, dynamic([])))", value_query) + self.assertIn("mv-expand recipient=recipient_values", value_query) + self.assertIn("emailrecipient_has_any=dynamic", test.send_query.call_args_list[3].args[0]) + self.assertIn("emailrecipient_has_any=dynamic", test.send_query.call_args_list[5].args[0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/ASIM/dev/ASimTester/filteringTest/ASimFilteringTest.py b/ASIM/dev/ASimTester/filteringTest/ASimFilteringTest.py index 71b05abebd3..d16b6151eb8 100644 --- a/ASIM/dev/ASimTester/filteringTest/ASimFilteringTest.py +++ b/ASIM/dev/ASimTester/filteringTest/ASimFilteringTest.py @@ -223,12 +223,16 @@ def tests_main_func(self): if schema_of_parser not in all_schemas_parameters: self.fail(f"Schema: {schema_of_parser} - Not an existing schema or not supported by the validations script") param_to_column_mapping = all_schemas_parameters[schema_of_parser] + if schema_of_parser == "EmailEvent": + self.email_event_default_test(parser_file["ParserParams"], query_definition) for param in parser_file['ParserParams']: param_name = param['Name'] with self.subTest(): if param_name not in param_to_column_mapping: self.fail(f"parameter: {param_name} - No such parameter in {schema_of_parser} schema") column_name_in_table = param_to_column_mapping[param_name] + if schema_of_parser == "EmailEvent" and param_name == "pack": + self.pack_test(query_definition) self.send_param_to_test(param, query_definition, columns_in_answer, column_name_in_table) @@ -245,12 +249,18 @@ def send_param_to_test(self, param, query_definition, columns_in_answer, column_ """ param_name = param['Name'] param_type = param['Type'] - # pack parameter is not checked by the validations if (param_name == "pack"): return if (param_name == "disabled"): self.disabled_test(query_definition) - elif column_name_in_table not in columns_in_answer: + elif isinstance(column_name_in_table, tuple): + self.email_recipient_test(param_name, query_definition, column_name_in_table, columns_in_answer) + elif column_name_in_table not in columns_in_answer: + print( + f"Parameter: {param_name} - mapped column {column_name_in_table} is unavailable; " + "filtering validation is partial." + ) + self.unavailable_column_test(param, query_definition) return elif (param_type == "datetime"): pass #TODO add test for datetime @@ -281,6 +291,98 @@ def check_required_fields(self, parser_file): file = file[field_name] if len(missing_fields) != 0: self.fail(f"The following fields are missing in the file:\n{missing_fields}") + + def get_count_for_query(self, query): + return self.send_query(query + " | count\n").tables[0].rows[0][0] + + def unavailable_column_test(self, param, query_definition): + param_name = param["Name"] + param_type = param["Type"] + default = param["Default"] + if param_type == "string": + default = f"'{default}'" + baseline = self.get_count_for_query(query_definition + "query()\n") + self.assertEqual( + self.get_count_for_query(query_definition + f"query({param_name}={default})\n"), + baseline, + ) + if param_type == "dynamic": + absent_value = f"dynamic([{DUMMY_VALUE}])" + elif param_type == "datetime": + absent_value = ( + "datetime(9999-12-31T23:59:59Z)" + if param_name == "starttime" + else "datetime(0001-01-01T00:00:00Z)" + ) + else: + absent_value = INT_DUMMY_VALUE if param_type == "int" else DUMMY_VALUE + self.assertEqual( + self.get_count_for_query(query_definition + f"query({param_name}={absent_value})\n"), + 0, + f"Parameter: {param_name} - Returned results for an absent value.", + ) + + def email_event_default_test(self, parser_params, query_definition): + defaults = [] + for param in parser_params: + default = param["Default"] + if param["Type"] == "string": + default = f"'{default}'" + defaults.append(f"{param['Name']}={default}") + baseline = self.get_count_for_query(query_definition + "query()\n") + explicit_defaults = self.get_count_for_query( + query_definition + f"query({', '.join(defaults)})\n" + ) + self.assertEqual(explicit_defaults, baseline, "EmailEvent defaults changed the unfiltered result set.") + + def pack_test(self, query_definition): + baseline = self.get_count_for_query(query_definition + "query(pack=false)\n") + packed = self.get_count_for_query(query_definition + "query(pack=true)\n") + self.assertEqual(packed, baseline, "Parameter: pack - Changed the number of results.") + + def email_recipient_test(self, parameter_name, query_definition, recipient_columns, columns_in_answer): + baseline = self.get_count_for_query(query_definition + "query()\n") + empty_count = self.get_count_for_query( + query_definition + f"query({parameter_name}=dynamic([]))\n" + ) + self.assertEqual(empty_count, baseline, f"Parameter: {parameter_name} - Empty array changed results.") + available_columns = [column for column in recipient_columns if column in columns_in_answer] + if not available_columns: + print(f"Parameter: {parameter_name} - recipient columns are unavailable; matching validation is partial.") + values = [] + else: + array_values = ", ".join( + f"coalesce({column}, dynamic([]))" for column in available_columns + ) + values = [ + row[0] + for row in self.send_query( + query_definition + + "query() | project recipient_values=array_concat(" + + array_values + + ") | mv-expand recipient=recipient_values " + "| where isnotempty(tostring(recipient)) | distinct recipient=tostring(recipient) | take 2\n" + ).tables[0].rows + ] + if not values: + print(f"Parameter: {parameter_name} - no recipient values are present; matching validation is partial.") + if len(values) > 1: + candidate_value_lists = ([values[0]], values) + elif values: + candidate_value_lists = ([values[0]],) + else: + candidate_value_lists = () + for selected_values in candidate_value_lists: + matched_count = self.get_count_for_query( + query_definition + + f"query({parameter_name}=dynamic([{create_values_string(selected_values)}]))\n" + ) + self.assertGreater(matched_count, 0, f"Parameter: {parameter_name} - No recipient match.") + self.assertLessEqual(matched_count, baseline) + absent_count = self.get_count_for_query( + query_definition + f"query({parameter_name}=dynamic([{DUMMY_VALUE}]))\n" + ) + self.assertEqual(absent_count, 0, f"Parameter: {parameter_name} - Matched a fictive recipient.") def scalar_test(self, param, query_definition, column_name_in_table): @@ -297,15 +399,15 @@ def scalar_test(self, param, query_definition, column_name_in_table): no_filter_query = query_definition + create_execution_string_without_parameters(column_name_in_table) no_filter_response = self.send_query(no_filter_query) self.assertNotEqual(len(no_filter_response.tables[0].rows) , 0 , f"No data for parameter:{param_name}") - with self.subTest(): - self.assertNotEqual(len(no_filter_response.tables[0].rows), 1, f"Only one value exists for parameter: {param_name} - validations for this parameter are partial" ) + if len(no_filter_response.tables[0].rows) == 1: + print( + f"Parameter: {param_name} - only one value is present; " + "selectivity validation is partial." + ) # Taking the first value returned in the response selected_value = no_filter_response.tables[0].rows[0][0] value_to_filter = f"\'{selected_value}\'" if param['Type']=="string" else selected_value query_with_filter = query_definition + create_execution_strings_with_one_parameter(param_name, value_to_filter, column_name_in_table) - if selected_value=="": - query_with_filter = query_definition + f"query() | where isempty({column_name_in_table}) | summarize count() by {column_name_in_table}\n" - # Performing a filtering by the first value returned in the first response self.scalar_test_check_filtering(param_name, query_with_filter, value_to_filter) @@ -379,7 +481,7 @@ def get_response_for_query_with_one_parameter(self, parameter_name, query_defini return self.get_response_for_query_with_parameters(parameter_name, query_definition, column_name_in_table, filtering_with_one_value_list) - def dynamic_tests_assertions(self, parameter_name, num_of_rows_with_filter_in_query, filter_parameters, num_of_rows_when_no_filters_in_query): + def dynamic_tests_assertions(self, parameter_name, num_of_rows_with_filter_in_query, filter_parameters, num_of_rows_when_no_filters_in_query, require_fewer_results): """ Performing assertions for dynamic tests with parameter filtering. @@ -392,17 +494,49 @@ def dynamic_tests_assertions(self, parameter_name, num_of_rows_with_filter_in_qu """ with self.subTest(): self.assertNotEqual(0, num_of_rows_with_filter_in_query, f"Parameter: {parameter_name} - Got no results at all after filtering. Filtered by value: {filter_parameters}") - with self.subTest(): - if (num_of_rows_when_no_filters_in_query == 1): - self.assertEqual(1, num_of_rows_when_no_filters_in_query, f"Parameter: {parameter_name} - Expected to have one result after filtering. Filtered by value: {filter_parameters}") - else: - self.assertLess(num_of_rows_with_filter_in_query, num_of_rows_when_no_filters_in_query, f"Parameter: {parameter_name} - Expected to have less results after filtering. Filtered by value: {filter_parameters}") + if require_fewer_results: + self.assertLess( + num_of_rows_with_filter_in_query, + num_of_rows_when_no_filters_in_query, + f"Parameter: {parameter_name} - Expected fewer distinct values after filtering." + ) + else: + self.assertLessEqual( + num_of_rows_with_filter_in_query, + num_of_rows_when_no_filters_in_query, + f"Parameter: {parameter_name} - Filtering returned more values than the unfiltered query." + ) + if num_of_rows_with_filter_in_query == num_of_rows_when_no_filters_in_query: + print( + f"Parameter: {parameter_name} - filter values matched all distinct output values; " + "multi-value selectivity validation is partial." + ) + def dynamic_response_matches_filter(self, parameter_name, response, filter_values): + normalized_filters = [str(value).casefold() for value in filter_values] + for row in response.tables[0].rows: + output_value = str(row[COLUMN_INDEX_IN_ROW]).casefold() + if parameter_name.endswith("_has_any_prefix"): + matches = any(output_value.startswith(value) for value in normalized_filters) + elif parameter_name.endswith("_has_all"): + matches = all(value in output_value for value in normalized_filters) + elif parameter_name.endswith("_has_any"): + matches = any(value in output_value for value in normalized_filters) + else: + matches = output_value in normalized_filters + self.assertTrue( + matches, + f"Parameter: {parameter_name} - Filter returned an unrelated value: {output_value}", + ) - # Failing the test with corresponding message if values list is empty - def fail_if_values_list_is_empty(self, values_list,parameter_name, test_type): + def values_list_is_available(self, values_list, parameter_name, test_type): if len(values_list) == 0: - self.fail(f"Parameter: {parameter_name} - Unable to find values to perform {test_type} tests") + print( + f"Parameter: {parameter_name} - no values are available for {test_type} tests; " + "matching validation is partial." + ) + return False + return True def dynamic_tests_helper(self, parameter_name, query_definition, num_of_rows_when_no_filters_in_query, column_name_in_table, values_list, test_type): @@ -418,19 +552,60 @@ def dynamic_tests_helper(self, parameter_name, query_definition, num_of_rows_whe values_list : List of at most two values that will be used to perform filtering. test_type : Name of the specific tests performed. It can be default testing or specific testing for has_any/has_any_prefix parameters """ - self.fail_if_values_list_is_empty(values_list, parameter_name, test_type) + if not self.values_list_is_available(values_list, parameter_name, test_type): + self.dynamic_tests_check_fictive_value(parameter_name, query_definition, column_name_in_table) + return # Performing filtering with one value filtering_with_one_value_response, one_value_string = self.get_response_for_query_with_one_parameter(parameter_name, query_definition, column_name_in_table, values_list) num_of_rows_with_one_parameter_in_query = len(filtering_with_one_value_response.tables[0].rows) - self.dynamic_tests_assertions(parameter_name, num_of_rows_with_one_parameter_in_query, one_value_string, num_of_rows_when_no_filters_in_query) - - # Performing filtering with two values if possible - if len(values_list) == 1 or num_of_rows_when_no_filters_in_query <= MAX_FILTERING_PARAMETERS: - self.fail(f"Parameter: {parameter_name} - Not enough data to perform two values {test_type} tests") - filtering_response, values_string = self.get_response_for_query_with_parameters(parameter_name, query_definition, column_name_in_table, values_list) - num_of_rows_with_parameters_in_query = len(filtering_response.tables[0].rows) - self.dynamic_tests_assertions(parameter_name, num_of_rows_with_parameters_in_query, values_string, num_of_rows_when_no_filters_in_query) + self.dynamic_response_matches_filter(parameter_name, filtering_with_one_value_response, [values_list[0]]) + self.dynamic_tests_assertions( + parameter_name, + num_of_rows_with_one_parameter_in_query, + one_value_string, + num_of_rows_when_no_filters_in_query, + num_of_rows_when_no_filters_in_query > 1, + ) + + if len(values_list) > 1: + second_value_response, _ = self.get_response_for_query_with_one_parameter( + parameter_name, query_definition, column_name_in_table, [values_list[1]] + ) + self.dynamic_response_matches_filter(parameter_name, second_value_response, [values_list[1]]) + filtering_response, values_string = self.get_response_for_query_with_parameters( + parameter_name, query_definition, column_name_in_table, values_list + ) + num_of_rows_with_parameters_in_query = len(filtering_response.tables[0].rows) + self.dynamic_response_matches_filter(parameter_name, filtering_response, values_list) + self.dynamic_tests_assertions( + parameter_name, + num_of_rows_with_parameters_in_query, + values_string, + num_of_rows_when_no_filters_in_query, + num_of_rows_when_no_filters_in_query > len(values_list), + ) + expected_values = { + str(row[COLUMN_INDEX_IN_ROW]) + for row in filtering_with_one_value_response.tables[0].rows + } | { + str(row[COLUMN_INDEX_IN_ROW]) + for row in second_value_response.tables[0].rows + } + actual_values = { + str(row[COLUMN_INDEX_IN_ROW]) + for row in filtering_response.tables[0].rows + } + self.assertEqual( + actual_values, + expected_values, + f"Parameter: {parameter_name} - Multi-value filter did not return the union of individual matches." + ) + else: + print( + f"Parameter: {parameter_name} - only one value is available for {test_type} tests; " + "multi-value validation is partial." + ) # Performing a query with a non-existing value, expecting to return no results self.dynamic_tests_check_fictive_value(parameter_name, query_definition, column_name_in_table) @@ -508,22 +683,43 @@ def has_all_test(self, parameter_name, query_definition, no_filter_rows, column_ column_name_in_table : The name of the column in the query response on which the parameter performs filtering """ values_list = self.get_values_for_dynamic_tests(no_filter_rows) - self.fail_if_values_list_is_empty(values_list, parameter_name, "has_all") + if not self.values_list_is_available(values_list, parameter_name, "has_all"): + self.dynamic_tests_check_fictive_value(parameter_name, query_definition, column_name_in_table) + return # Performing filtering with one value filtering_with_one_value_response, one_value_string = self.get_response_for_query_with_one_parameter(parameter_name, query_definition, column_name_in_table, values_list) num_of_rows_when_no_filters_in_query = len (no_filter_rows) num_of_rows_with_one_parameter_in_query = len(filtering_with_one_value_response.tables[0].rows) - self.dynamic_tests_assertions(parameter_name, num_of_rows_with_one_parameter_in_query, one_value_string, num_of_rows_when_no_filters_in_query) + self.dynamic_response_matches_filter(parameter_name, filtering_with_one_value_response, [values_list[0]]) + self.dynamic_tests_assertions( + parameter_name, + num_of_rows_with_one_parameter_in_query, + one_value_string, + num_of_rows_when_no_filters_in_query, + num_of_rows_when_no_filters_in_query > 1, + ) # Performing filtering with two values if possible first_part_of_a_value , second_part_of_a_value = get_splitted_parts_of_string(values_list) if first_part_of_a_value is None: - self.fail(f"Parameter: {parameter_name} - has_all tests performed for only one filtering value") + print( + f"Parameter: {parameter_name} - no value can be split for a multi-value has_all check; " + "multi-value validation is partial." + ) + self.dynamic_tests_check_fictive_value(parameter_name, query_definition, column_name_in_table) + return splitted_parts_list = [first_part_of_a_value , second_part_of_a_value] filtering_response, values_string = self.get_response_for_query_with_parameters(parameter_name, query_definition, column_name_in_table, splitted_parts_list) num_of_rows_with_parameters_in_query = len(filtering_response.tables[0].rows) - self.dynamic_tests_assertions(parameter_name, num_of_rows_with_parameters_in_query, values_string, num_of_rows_when_no_filters_in_query) + self.dynamic_response_matches_filter(parameter_name, filtering_response, splitted_parts_list) + self.dynamic_tests_assertions( + parameter_name, + num_of_rows_with_parameters_in_query, + values_string, + num_of_rows_when_no_filters_in_query, + num_of_rows_when_no_filters_in_query > len(splitted_parts_list), + ) # Performing a query with a non-existing value, expecting to return no results self.dynamic_tests_check_fictive_value(parameter_name, query_definition, column_name_in_table) @@ -582,8 +778,21 @@ def dynamic_test(self, parameter, query_definition, column_name_in_table): no_filter_response = self.send_query(no_filter_query) no_filter_rows = no_filter_response.tables[0].rows self.assertNotEqual(len(no_filter_rows) , 0 , f"No data for parameter:{parameter_name}") - with self.subTest(): - self.assertNotEqual(len(no_filter_rows), 1, f"Only one value exists for parameter: {parameter_name} - validations for this parameter are partial" ) + if len(no_filter_rows) == 1: + print( + f"Parameter: {parameter_name} - only one value is present; " + "selectivity validation is partial." + ) + if parameter["Default"] == "dynamic([])": + baseline_count = self.get_count_for_query(query_definition + "query()\n") + empty_count = self.get_count_for_query( + query_definition + f"query({parameter_name}=dynamic([]))\n" + ) + self.assertEqual( + empty_count, + baseline_count, + f"Parameter: {parameter_name} - Empty array changed the unfiltered result set.", + ) # Specific tests parameters with certain suffixes if parameter_name.endswith('has_any'): @@ -689,6 +898,19 @@ def send_query(self, query_str): "srcipaddr" : "SrcIpAddr", "starttime" : "EventStartTime" }, + "EmailEvent" : + { + "starttime" : "EventStartTime", + "endtime" : "EventEndTime", + "ipaddr_has_any_prefix" : "IpAddr", + "emailfromaddress_has_any" : "EmailFromAddress", + "emailrecipient_has_any" : ("EmailToRecipients", "EmailCCRecipients", "EmailBCCRecipients"), + "emailsubject_has_any" : "EmailSubject", + "emaildirection_in" : "EmailDirection", + "emaildeliverylocation_in" : "EmailDeliveryLocation", + "eventresult" : "EventResult", + "pack" : "" + }, "NetworkSession" : { "disabled" : "", diff --git a/ASIM/dev/ASimTester/filteringTest/Readme.md b/ASIM/dev/ASimTester/filteringTest/Readme.md index 28fc1f43f46..c0a2a876340 100644 --- a/ASIM/dev/ASimTester/filteringTest/Readme.md +++ b/ASIM/dev/ASimTester/filteringTest/Readme.md @@ -60,6 +60,8 @@ python ASimFilteringTest.py 123-456 C:\parsers\vimDnsMicrosoftOMSfict.yaml 99 ## Results +The filtering tests support EmailEvent parsers, including recipient matching across the To, CC, and BCC arrays. If a workspace has only constant or unavailable field values, the validator reports partial coverage while still checking defaults, matching values when present, and absent values. Datetime validation retains its previous behavior: this legacy script does not exercise available datetime columns, while the shared `.script` validator checks datetime filters against TimeGenerated. + - If all the validations that the test script performed have passed you will receive an "ok" message: ![enter image description here](https://github.com/Azure/Azure-Sentinel/assets/126081432/e3e130e1-6f73-4479-9e1f-2cdfd8023c45)