From 54d3b20a687f039b16aece4a1f8dec68e4fe60a5 Mon Sep 17 00:00:00 2001 From: Aditi Bhoge Date: Fri, 23 Oct 2026 12:27:57 -0700 Subject: [PATCH 1/2] Add EmailEvent filtering validator support Register standard EmailEvent filters in both CI validators, validate recipient arrays and event-time boundaries, and report sparse data as partial while still checking matches and absent values. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9d6cfd8c-4385-4bd6-8f53-a9dc74446ece --- .../asimParsersTest/ASimFilteringTest.py | 396 +++++++++++++++--- .../test_asim_filtering_validator.py | 291 +++++++++++++ .../filteringTest/ASimFilteringTest.py | 330 +++++++++++++-- ASIM/dev/ASimTester/filteringTest/Readme.md | 2 + 4 files changed, 921 insertions(+), 98 deletions(-) create mode 100644 .script/tests/asimParsersTest/test_asim_filtering_validator.py diff --git a/.script/tests/asimParsersTest/ASimFilteringTest.py b/.script/tests/asimParsersTest/ASimFilteringTest.py index 1d8ad5fcffb..42edefe857f 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) @@ -441,39 +447,174 @@ def check_required_fields(self, parser_file): def datetime_test(self, param, query_definition, column_name_in_table): param_name = param['Name'] - # Get count of rows without filtering - no_filter_query = query_definition + f"query() | project TimeGenerated \n" + no_filter_query = query_definition + f"query() | project {column_name_in_table} \n" no_filter_response = self.send_query(no_filter_query) num_of_rows_when_no_filters_in_query = len(no_filter_response.tables[0].rows) - self.assertNotEqual(len(no_filter_response.tables[0].rows) , 0 , f"No data for parameter:{param_name}") + self.assertNotEqual(num_of_rows_when_no_filters_in_query, 0, f"No data for parameter:{param_name}") datetime_mid_point_query = query_definition + ( - "query() " - "| summarize max_TimeGenerated = max(TimeGenerated), " - "min_TimeGenerated = min(TimeGenerated), " - "distinct_TimeGenerated = count_distinct(TimeGenerated) \n" - "| extend timeSpanMicroseconds = datetime_diff('microsecond', max_TimeGenerated, min_TimeGenerated) \n" - "| project mid_point = min_TimeGenerated + ((max_TimeGenerated - min_TimeGenerated) / 2), " - "distinct_TimeGenerated, timeSpanMicroseconds" + f"query() | summarize max_filter_time = max({column_name_in_table}), " + f"min_filter_time = min({column_name_in_table}), " + f"distinct_filter_time = count_distinct({column_name_in_table}) \n" + "| extend timeSpanMicroseconds = datetime_diff('microsecond', max_filter_time, min_filter_time) \n" + "| project mid_point = min_filter_time + ((max_filter_time - min_filter_time) / 2), " + "min_filter_time, max_filter_time, distinct_filter_time, timeSpanMicroseconds" ) datetime_mid_point_response = self.send_query(datetime_mid_point_query) midpoint_row = datetime_mid_point_response.tables[0].rows[0] - distinct_time_generated = midpoint_row[1] - time_span_microseconds = midpoint_row[2] - if distinct_time_generated < 2 or time_span_microseconds < 2: + min_filter_time = midpoint_row[1] + max_filter_time = midpoint_row[2] + distinct_time_generated = midpoint_row[3] + time_span_microseconds = midpoint_row[4] + if distinct_time_generated == 0 or min_filter_time is None or max_filter_time is None: print_and_flush( - f"Parameter: {param_name} - TimeGenerated values do not span a testable range; " + f"Parameter: {param_name} - {column_name_in_table} has no non-null values; " "datetime filtering validation is partial." ) return mid_point_datetime_value = midpoint_row[0] datetime_mid_point_value = f"datetime({mid_point_datetime_value})" - # Get count of rows after applying filtering - query_with_filter = query_definition + f"query({param_name}={datetime_mid_point_value})\n" - filtered_response = self.send_query(query_with_filter) - self.assertNotEqual(len(filtered_response.tables[0].rows) , 0 , f"No data for parameter:{param_name}") - 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}") + filtered_count = self.get_count_for_query( + query_definition + f"query({param_name}={datetime_mid_point_value})\n" + ) + self.assertGreater(filtered_count, 0, f"Parameter: {param_name} - Expected results for an in-range filter.") + if distinct_time_generated >= 2 and time_span_microseconds >= 2: + self.assertLess( + filtered_count, + num_of_rows_when_no_filters_in_query, + f"Parameter: {param_name} - Expected fewer results for an in-range filter." + ) + else: + print_and_flush( + f"Parameter: {param_name} - {column_name_in_table} values do not span a testable range; " + "datetime filtering validation is partial." + ) + + boundary_aggregation = "max" if param_name == "starttime" else "min" + boundary_offset = 1 if param_name == "starttime" else -1 + absent_count = self.get_count_for_query( + query_definition + + f"let filter_boundary = toscalar(query() | summarize {boundary_aggregation}({column_name_in_table}));\n" + + f"query({param_name}=datetime_add('microsecond', {boundary_offset}, filter_boundary))\n" + ) + self.assertEqual( + absent_count, 0, + f"Parameter: {param_name} - Returned results beyond the {column_name_in_table} data boundary." + ) + + 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') @@ -513,15 +654,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 +736,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 +749,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 +809,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 +944,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 +1039,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 +1102,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 +1211,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..c3f00fd33c7 --- /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_uses_schema_event_time_and_checks_out_of_range(self): + test = self.make_test( + [ + response([["2026-10-21T00:00:00Z"], ["2026-10-22T00:00:00Z"]]), + response([["2026-10-21T12:00:00Z", "2026-10-21T00:00:00Z", "2026-10-22T00:00:00Z", 2, 86400000000]]), + response([[1]]), + response([[0]]), + ] + ) + 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(EventStartTime)", query_text) + self.assertIn("project EventStartTime", query_text) + self.assertNotIn("TimeGenerated", query_text) + self.assertIn("datetime_add('microsecond', 1, filter_boundary)", query_text) + + 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..089f65364f8 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,15 +249,21 @@ 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 + self.datetime_test(param, query_definition, column_name_in_table) elif (param_type == "dynamic"): self.dynamic_test(param, query_definition, column_name_in_table) else: @@ -281,6 +291,142 @@ 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 datetime_test(self, param, query_definition, column_name_in_table): + param_name = param["Name"] + no_filter_rows = self.send_query( + query_definition + f"query() | project {column_name_in_table}\n" + ).tables[0].rows + self.assertNotEqual(len(no_filter_rows), 0, f"No data for parameter:{param_name}") + summary = self.send_query( + query_definition + + f"query() | summarize max_filter_time=max({column_name_in_table}), " + + f"min_filter_time=min({column_name_in_table}), " + + f"distinct_filter_time=count_distinct({column_name_in_table}) " + + "| extend timeSpanMicroseconds=datetime_diff('microsecond', max_filter_time, min_filter_time) " + + "| project mid_point=min_filter_time + ((max_filter_time - min_filter_time) / 2), " + + "min_filter_time, max_filter_time, distinct_filter_time, timeSpanMicroseconds\n" + ).tables[0].rows[0] + min_time, max_time = summary[1], summary[2] + if summary[3] == 0 or min_time is None or max_time is None: + print( + f"Parameter: {param_name} - {column_name_in_table} has no non-null values; " + "datetime filtering validation is partial." + ) + return + + in_range_count = self.get_count_for_query( + query_definition + f"query({param_name}=datetime({summary[0]}))\n" + ) + self.assertGreater(in_range_count, 0, f"Parameter: {param_name} - Expected an in-range match.") + if summary[3] > 1 and summary[4] >= 2: + self.assertLess(in_range_count, len(no_filter_rows)) + else: + print( + f"Parameter: {param_name} - {column_name_in_table} values do not span a testable range; " + "datetime filtering validation is partial." + ) + + operation = "max" if param_name == "starttime" else "min" + offset = 1 if param_name == "starttime" else -1 + out_of_range_count = self.get_count_for_query( + query_definition + + f"let filter_boundary=toscalar(query() | summarize {operation}({column_name_in_table}));\n" + + f"query({param_name}=datetime_add('microsecond', {offset}, filter_boundary))\n" + ) + self.assertEqual(out_of_range_count, 0, f"Parameter: {param_name} - Matched beyond the data boundary.") + + 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 +443,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 +525,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 +538,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 +596,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 +727,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 +822,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 +942,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..a98b736c3da 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 filters are checked against the mapped EventStartTime and EventEndTime columns. + - 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) From 503da959e26b3565cc0a1c6a68d7b8fbe2bb7af3 Mon Sep 17 00:00:00 2001 From: Aditi Bhoge Date: Fri, 23 Oct 2026 22:22:42 -0600 Subject: [PATCH 2/2] Restore previous datetime filtering validation behavior Revert mapped event-time validation and boundary changes while retaining EmailEvent catalog and non-time filter support. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9d6cfd8c-4385-4bd6-8f53-a9dc74446ece --- .../asimParsersTest/ASimFilteringTest.py | 61 ++++++------------- .../test_asim_filtering_validator.py | 16 ++--- .../filteringTest/ASimFilteringTest.py | 46 +------------- ASIM/dev/ASimTester/filteringTest/Readme.md | 2 +- 4 files changed, 30 insertions(+), 95 deletions(-) diff --git a/.script/tests/asimParsersTest/ASimFilteringTest.py b/.script/tests/asimParsersTest/ASimFilteringTest.py index 42edefe857f..697e6f94502 100644 --- a/.script/tests/asimParsersTest/ASimFilteringTest.py +++ b/.script/tests/asimParsersTest/ASimFilteringTest.py @@ -447,60 +447,39 @@ def check_required_fields(self, parser_file): def datetime_test(self, param, query_definition, column_name_in_table): param_name = param['Name'] - no_filter_query = query_definition + f"query() | project {column_name_in_table} \n" + # Get count of rows without filtering + no_filter_query = query_definition + f"query() | project TimeGenerated \n" no_filter_response = self.send_query(no_filter_query) num_of_rows_when_no_filters_in_query = len(no_filter_response.tables[0].rows) - self.assertNotEqual(num_of_rows_when_no_filters_in_query, 0, f"No data for parameter:{param_name}") + self.assertNotEqual(len(no_filter_response.tables[0].rows) , 0 , f"No data for parameter:{param_name}") datetime_mid_point_query = query_definition + ( - f"query() | summarize max_filter_time = max({column_name_in_table}), " - f"min_filter_time = min({column_name_in_table}), " - f"distinct_filter_time = count_distinct({column_name_in_table}) \n" - "| extend timeSpanMicroseconds = datetime_diff('microsecond', max_filter_time, min_filter_time) \n" - "| project mid_point = min_filter_time + ((max_filter_time - min_filter_time) / 2), " - "min_filter_time, max_filter_time, distinct_filter_time, timeSpanMicroseconds" + "query() " + "| summarize max_TimeGenerated = max(TimeGenerated), " + "min_TimeGenerated = min(TimeGenerated), " + "distinct_TimeGenerated = count_distinct(TimeGenerated) \n" + "| extend timeSpanMicroseconds = datetime_diff('microsecond', max_TimeGenerated, min_TimeGenerated) \n" + "| project mid_point = min_TimeGenerated + ((max_TimeGenerated - min_TimeGenerated) / 2), " + "distinct_TimeGenerated, timeSpanMicroseconds" ) datetime_mid_point_response = self.send_query(datetime_mid_point_query) midpoint_row = datetime_mid_point_response.tables[0].rows[0] - min_filter_time = midpoint_row[1] - max_filter_time = midpoint_row[2] - distinct_time_generated = midpoint_row[3] - time_span_microseconds = midpoint_row[4] - if distinct_time_generated == 0 or min_filter_time is None or max_filter_time is None: + distinct_time_generated = midpoint_row[1] + time_span_microseconds = midpoint_row[2] + if distinct_time_generated < 2 or time_span_microseconds < 2: print_and_flush( - f"Parameter: {param_name} - {column_name_in_table} has no non-null values; " + f"Parameter: {param_name} - TimeGenerated values do not span a testable range; " "datetime filtering validation is partial." ) return mid_point_datetime_value = midpoint_row[0] datetime_mid_point_value = f"datetime({mid_point_datetime_value})" - filtered_count = self.get_count_for_query( - query_definition + f"query({param_name}={datetime_mid_point_value})\n" - ) - self.assertGreater(filtered_count, 0, f"Parameter: {param_name} - Expected results for an in-range filter.") - if distinct_time_generated >= 2 and time_span_microseconds >= 2: - self.assertLess( - filtered_count, - num_of_rows_when_no_filters_in_query, - f"Parameter: {param_name} - Expected fewer results for an in-range filter." - ) - else: - print_and_flush( - f"Parameter: {param_name} - {column_name_in_table} values do not span a testable range; " - "datetime filtering validation is partial." - ) - - boundary_aggregation = "max" if param_name == "starttime" else "min" - boundary_offset = 1 if param_name == "starttime" else -1 - absent_count = self.get_count_for_query( - query_definition - + f"let filter_boundary = toscalar(query() | summarize {boundary_aggregation}({column_name_in_table}));\n" - + f"query({param_name}=datetime_add('microsecond', {boundary_offset}, filter_boundary))\n" - ) - self.assertEqual( - absent_count, 0, - f"Parameter: {param_name} - Returned results beyond the {column_name_in_table} data boundary." - ) + # Get count of rows after applying filtering + query_with_filter = query_definition + f"query({param_name}={datetime_mid_point_value})\n" + filtered_response = self.send_query(query_with_filter) + self.assertNotEqual(len(filtered_response.tables[0].rows) , 0 , f"No data for parameter:{param_name}") + 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") diff --git a/.script/tests/asimParsersTest/test_asim_filtering_validator.py b/.script/tests/asimParsersTest/test_asim_filtering_validator.py index c3f00fd33c7..c1252e9b618 100644 --- a/.script/tests/asimParsersTest/test_asim_filtering_validator.py +++ b/.script/tests/asimParsersTest/test_asim_filtering_validator.py @@ -145,13 +145,12 @@ def test_existing_schema_mappings_and_disabled_control_remain_supported(self): 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_uses_schema_event_time_and_checks_out_of_range(self): + 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", "2026-10-21T00:00:00Z", "2026-10-22T00:00:00Z", 2, 86400000000]]), - response([[1]]), - response([[0]]), + response([["2026-10-21T12:00:00Z", 2, 86400000000]]), + response([["2026-10-22T00:00:00Z"]]), ] ) test.datetime_test( @@ -160,10 +159,11 @@ def test_datetime_filter_uses_schema_event_time_and_checks_out_of_range(self): "EventStartTime", ) query_text = "\n".join(call.args[0] for call in test.send_query.call_args_list) - self.assertIn("max(EventStartTime)", query_text) - self.assertIn("project EventStartTime", query_text) - self.assertNotIn("TimeGenerated", query_text) - self.assertIn("datetime_add('microsecond', 1, filter_boundary)", query_text) + 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( diff --git a/ASIM/dev/ASimTester/filteringTest/ASimFilteringTest.py b/ASIM/dev/ASimTester/filteringTest/ASimFilteringTest.py index 089f65364f8..d16b6151eb8 100644 --- a/ASIM/dev/ASimTester/filteringTest/ASimFilteringTest.py +++ b/ASIM/dev/ASimTester/filteringTest/ASimFilteringTest.py @@ -263,7 +263,7 @@ def send_param_to_test(self, param, query_definition, columns_in_answer, column_ self.unavailable_column_test(param, query_definition) return elif (param_type == "datetime"): - self.datetime_test(param, query_definition, column_name_in_table) + pass #TODO add test for datetime elif (param_type == "dynamic"): self.dynamic_test(param, query_definition, column_name_in_table) else: @@ -292,50 +292,6 @@ def check_required_fields(self, parser_file): if len(missing_fields) != 0: self.fail(f"The following fields are missing in the file:\n{missing_fields}") - def datetime_test(self, param, query_definition, column_name_in_table): - param_name = param["Name"] - no_filter_rows = self.send_query( - query_definition + f"query() | project {column_name_in_table}\n" - ).tables[0].rows - self.assertNotEqual(len(no_filter_rows), 0, f"No data for parameter:{param_name}") - summary = self.send_query( - query_definition - + f"query() | summarize max_filter_time=max({column_name_in_table}), " - + f"min_filter_time=min({column_name_in_table}), " - + f"distinct_filter_time=count_distinct({column_name_in_table}) " - + "| extend timeSpanMicroseconds=datetime_diff('microsecond', max_filter_time, min_filter_time) " - + "| project mid_point=min_filter_time + ((max_filter_time - min_filter_time) / 2), " - + "min_filter_time, max_filter_time, distinct_filter_time, timeSpanMicroseconds\n" - ).tables[0].rows[0] - min_time, max_time = summary[1], summary[2] - if summary[3] == 0 or min_time is None or max_time is None: - print( - f"Parameter: {param_name} - {column_name_in_table} has no non-null values; " - "datetime filtering validation is partial." - ) - return - - in_range_count = self.get_count_for_query( - query_definition + f"query({param_name}=datetime({summary[0]}))\n" - ) - self.assertGreater(in_range_count, 0, f"Parameter: {param_name} - Expected an in-range match.") - if summary[3] > 1 and summary[4] >= 2: - self.assertLess(in_range_count, len(no_filter_rows)) - else: - print( - f"Parameter: {param_name} - {column_name_in_table} values do not span a testable range; " - "datetime filtering validation is partial." - ) - - operation = "max" if param_name == "starttime" else "min" - offset = 1 if param_name == "starttime" else -1 - out_of_range_count = self.get_count_for_query( - query_definition - + f"let filter_boundary=toscalar(query() | summarize {operation}({column_name_in_table}));\n" - + f"query({param_name}=datetime_add('microsecond', {offset}, filter_boundary))\n" - ) - self.assertEqual(out_of_range_count, 0, f"Parameter: {param_name} - Matched beyond the data boundary.") - def get_count_for_query(self, query): return self.send_query(query + " | count\n").tables[0].rows[0][0] diff --git a/ASIM/dev/ASimTester/filteringTest/Readme.md b/ASIM/dev/ASimTester/filteringTest/Readme.md index a98b736c3da..c0a2a876340 100644 --- a/ASIM/dev/ASimTester/filteringTest/Readme.md +++ b/ASIM/dev/ASimTester/filteringTest/Readme.md @@ -60,7 +60,7 @@ 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 filters are checked against the mapped EventStartTime and EventEndTime columns. +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: