diff --git a/.script/tests/asimParsersTest/ingestTimeParser/README.md b/.script/tests/asimParsersTest/ingestTimeParser/README.md index b718f6f38a4..7caf91ee414 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/README.md +++ b/.script/tests/asimParsersTest/ingestTimeParser/README.md @@ -48,15 +48,27 @@ asim_ingestion/ discover target → static parser validation → prepare immutable fixture - → raw pass-through ingestion - → freeze raw-table source snapshot - → evaluate ParserFilterQuery - → execute ParserQuery through a real DCR + → freeze a typed parser-input snapshot from the fixture + → evaluate ParserFilterQuery against an inline fixture datatable + → upload fixture rows directly through the parser DCR → run ASIM data checks - → optionally execute the mapped QTP and compare parity + → optionally replay the fixture to SourceRawTable using InputStream + → execute the mapped QTP and compare parity when replay succeeds → clean up DCRs and publish the report ``` +Parser validation does not depend on raw-table replay. QTP source replay is +optional: if its table, DCR, upload, or visibility stage fails, the report +records a warning and blocks QTP/parity without failing the required parser +validation stages. + +QTP parity stores the complete `qtp_schema`, `qtp_rows`, `itp_schema`, and +`itp_rows` under `outputValidation`. It derives `intersect_schema` from shared +column names and records both QTP and ITP types. A shared-column type mismatch +fails schema parity. When shared types are consistent, it filters both row +sets into `qtp_rows_filtered` and `itp_rows_filtered` and runs the typed +duplicate-aware row comparison across that shared schema. + ## Tests Run the complete regression suite from the repository root: diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/common.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/common.py index 913c04ecf29..0c634ca207b 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/common.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/common.py @@ -97,6 +97,15 @@ def validate_outputs( ) -> Mapping[str, Any]: raise NotImplementedError + def validate_parity( + self, + query_client: Any, + source_snapshot: Mapping[str, Any], + parser_document: Mapping[str, Any], + query_time_document: Mapping[str, Any], + ) -> Mapping[str, Any]: + raise NotImplementedError + class DefaultAzureTokenProvider: def __init__(self): @@ -253,97 +262,105 @@ def build_exact_timestamp_query( ) -def _normalize_query_columns( - columns: Sequence[Mapping[str, Any]], -) -> list[dict[str, str]]: - schema: list[dict[str, str]] = [] - seen: set[str] = set() - for column in columns: - name = column.get("name") - column_type = column.get("type") - if not isinstance(name, str) or not isinstance(column_type, str): - raise HarnessError( - "query", - "Log Analytics returned an invalid column definition.", - details=column, - ) - if name in seen: - raise HarnessError( - "query", f"Log Analytics returned duplicate column {name}." - ) - seen.add(name) - schema.append( - {"name": name, "type": _normalize_dcr_type(column_type)} - ) - return schema - - -def build_source_snapshot( +def build_fixture_snapshot( *, target_id: str, table_name: str, timestamps: Sequence[str], - query_table: Mapping[str, Any], + schema: Sequence[Mapping[str, str]], + fixture_rows: Sequence[Mapping[str, Any]], ) -> dict[str, Any]: - columns = query_table.get("columns") - raw_rows = query_table.get("rows") - if not isinstance(columns, list) or not isinstance(raw_rows, list): - raise HarnessError( - "source-snapshot", - "Log Analytics query response must contain columns and rows.", + normalized_schema: list[dict[str, str]] = [ + { + "name": str(column["name"]), + "type": _normalize_dcr_type(str(column["type"])), + } + for column in schema + ] + known_names = { + column["name"] for column in normalized_schema + } + extra_names = [ + str(name) + for row in fixture_rows + for name in row + if str(name) not in known_names + and str(name) not in REPLAY_EXCLUDED_COLUMNS + ] + for name in dict.fromkeys(extra_names): + values = [ + row.get(name) + for row in fixture_rows + if row.get(name) is not None + ] + if values and all(isinstance(value, bool) for value in values): + column_type = "boolean" + elif values and all( + isinstance(value, int) and not isinstance(value, bool) + for value in values + ): + column_type = "long" + elif values and all( + isinstance(value, (int, float)) + and not isinstance(value, bool) + for value in values + ): + column_type = "real" + elif values and all( + isinstance(value, (dict, list)) + for value in values + ): + column_type = "dynamic" + else: + column_type = "string" + normalized_schema.append( + {"name": name, "type": column_type} ) - query_schema = _normalize_query_columns(columns) - column_names = [column["name"] for column in query_schema] - if any(not isinstance(row, list) for row in raw_rows): + known_names.add(name) + column_names = [column["name"] for column in normalized_schema] + if len(column_names) != len(set(column_names)): raise HarnessError( - "source-snapshot", "Query rows must be arrays." + "source-snapshot", + "Fixture schema contains duplicate columns.", ) - - included_schema = [ - column - for column in query_schema - if column["name"] not in REPLAY_EXCLUDED_COLUMNS - ] - included_names = [column["name"] for column in included_schema] type_by_name = { - column["name"]: column["type"] for column in included_schema + column["name"]: column["type"] + for column in normalized_schema } timestamp_set = {_validate_timestamp(value) for value in timestamps} - rows: list[dict[str, Any]] = [] - for row in raw_rows: - if len(row) != len(column_names): - raise HarnessError( - "source-snapshot", - "Query row length does not match the returned schema.", - ) - source = dict(zip(column_names, row)) - prepared = { - name: _json_safe_value(source.get(name), type_by_name[name]) - for name in included_names + rows = [ + { + name: _json_safe_value(row.get(name), type_by_name[name]) + for name in column_names } - timestamp = prepared.get("TimeGenerated") + for row in fixture_rows + ] + for row in rows: + timestamp = row.get("TimeGenerated") if not isinstance(timestamp, str) or timestamp not in timestamp_set: raise HarnessError( "source-snapshot", - "Query returned a row outside the exact timestamp set.", + "Fixture row is outside the exact timestamp set.", details={"TimeGenerated": timestamp}, ) - rows.append(prepared) - timestamp_hash = hashlib.sha256( "\n".join(sorted(timestamp_set)).encode("utf-8") ).hexdigest() + fixture_columns = { + str(name) + for row in fixture_rows + for name in row + } return { "version": SNAPSHOT_VERSION, "targetId": target_id, "sourceRawTable": table_name, + "authority": "prepared-fixture", "rowCount": len(rows), "timestampValues": sorted(timestamp_set), "timestampSetHash": f"sha256:{timestamp_hash}", - "excludedColumns": sorted( - set(column_names) & REPLAY_EXCLUDED_COLUMNS - ), - "schema": included_schema, + "excludedColumns": sorted(fixture_columns - set(column_names)), + "schema": normalized_schema, "rows": rows, } diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py index 3c11414eaaa..9f753a135a8 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py @@ -9,15 +9,48 @@ from .client import AzureMonitorClient from .common import ( HARNESS_RESULT_VERSION, - DcrHandle, + HarnessError, ValidationRunner, _validate_timestamp, _write_json_no_clobber, + build_fixture_snapshot, build_resource_names, - build_source_snapshot, count_rows_by_timestamp, ) + +def _error_payload(exc: HarnessError) -> dict[str, Any]: + return { + "type": type(exc).__name__, + "stage": exc.stage, + "message": str(exc), + "statusCode": exc.status_code, + "details": exc.details, + } + + +def _output_failure( + output_validation: Mapping[str, Any], +) -> HarnessError | None: + if output_validation.get("status") == "passed": + return None + failure = output_validation.get("failure") + if isinstance(failure, Mapping): + return HarnessError( + str(failure.get("stage") or "output-validation"), + str( + failure.get("message") + or "Output validation failed." + ), + details=failure.get("details"), + ) + return HarnessError( + "output-validation", + "Output validation failed.", + details=output_validation, + ) + + class AsimIngestionHarness(AzureMonitorClient): def run( self, @@ -35,6 +68,9 @@ def run( namespace = prepared_fixture["namespace"] source_table = str(prepared_fixture["sourceRawTable"]) target_table = str(parser_document["TargetASIMTable"]) + input_stream = str( + parser_document.get("InputStream") or "" + ).strip() parser_query = compact_kql( str(parser_document["ParserQuery"]) ) @@ -84,8 +120,6 @@ def run( str(namespace["runId"]), int(namespace["runAttempt"]), ) - raw_handle: DcrHandle | None = None - parser_handle: DcrHandle | None = None attempted_dcr_ids: list[str] = [] cleanup: list[dict[str, Any]] = [] result: dict[str, Any] = { @@ -101,6 +135,15 @@ def run( "snapshotPath": str(snapshot_output), "filterValidation": None, "outputValidation": None, + "qtpSourceReplay": { + "status": ( + "pending" + if query_time_document is not None + else "not-applicable" + ), + "inputStream": input_stream or None, + "error": None, + }, "cleanup": cleanup, "telemetry": { "stageTimingsSeconds": self.stage_timings, @@ -110,48 +153,17 @@ def run( self.last_result = result run_started = self.clock.monotonic() try: - with self._stage("raw-table"): - self.ensure_raw_table(source_table, schema) - raw_resource_id = self.dcr_resource_id(names["rawDcr"]) - attempted_dcr_ids.append(raw_resource_id) - with self._stage("raw-dcr"): - raw_handle = self.create_dcr( - name=names["rawDcr"], - stream_name=names["rawStream"], - destination_name=names["destination"], - schema=schema, - transform_kql="source", - output_stream=( - f"Custom-{source_table}" - if source_table.lower().endswith("_cl") - else f"Microsoft-{source_table}" - ), - ) - result["rawDcr"] = raw_handle.to_dict() - raw_counts = count_rows_by_timestamp(rows) - with self._stage("raw-upload"): - result["rawUpload"] = self.upload_rows( - raw_handle, - rows, - reconcile_table=source_table, - expected_counts_by_timestamp=raw_counts, - ) - with self._stage("raw-visibility"): - raw_query_table = self.wait_for_exact_rows( - source_table, - timestamps, - raw_counts, - stage="raw-visibility", - ) with self._stage("source-snapshot"): - snapshot = build_source_snapshot( + snapshot = build_fixture_snapshot( target_id=target_id, table_name=source_table, timestamps=timestamps, - query_table=raw_query_table, + schema=schema, + fixture_rows=rows, ) _write_json_no_clobber(snapshot, snapshot_output) result["snapshot"] = { + "authority": snapshot["authority"], "rowCount": snapshot["rowCount"], "timestampSetHash": snapshot["timestampSetHash"], "excludedColumns": snapshot["excludedColumns"], @@ -224,16 +236,125 @@ def run( normalized_target_counts, stage="target-visibility", ) + if validation_runner is not None: with self._stage("output-validation"): - result["outputValidation"] = dict( + output_validation = dict( validation_runner.validate_outputs( self, snapshot, parser_document, - query_time_document, + None, + ) + ) + result["outputValidation"] = output_validation + output_failure = _output_failure( + output_validation + ) + if output_failure is not None: + raise output_failure + + if ( + query_time_document is not None + and validation_runner is not None + ): + qtp_replay = result["qtpSourceReplay"] + try: + if ( + not input_stream + or input_stream.casefold() == "none" + ): + raise HarnessError( + "raw-dcr", + "QTP source replay requires a non-none " + "InputStream.", + ) + with self._stage("raw-table"): + self.ensure_raw_table( + source_table, + schema, + ) + raw_resource_id = self.dcr_resource_id( + names["rawDcr"] + ) + attempted_dcr_ids.append(raw_resource_id) + with self._stage("raw-dcr"): + raw_handle = self.create_dcr( + name=names["rawDcr"], + stream_name=names["rawStream"], + destination_name=names["destination"], + schema=schema, + transform_kql="source", + output_stream=input_stream, + ) + result["rawDcr"] = raw_handle.to_dict() + raw_counts = count_rows_by_timestamp( + rows + ) + with self._stage("raw-upload"): + result["rawUpload"] = self.upload_rows( + raw_handle, + rows, + reconcile_table=source_table, + expected_counts_by_timestamp=raw_counts, + ) + with self._stage("raw-visibility"): + self.wait_for_exact_rows( + source_table, + timestamps, + raw_counts, + stage="raw-visibility", + ) + qtp_replay["status"] = "passed" + except HarnessError as exc: + qtp_replay["status"] = "warning" + qtp_replay["error"] = _error_payload(exc) + + if ( + validation_runner is not None + and query_time_document is not None + ): + with self._stage("output-validation"): + qtp_replay = result["qtpSourceReplay"] + if qtp_replay.get("status") == "warning": + output_validation = dict( + result["outputValidation"] + ) + blocked_detail = ( + "Skipped because QTP source replay failed." + ) + output_validation["queryTimeExecution"] = { + "status": "blocked", + "detail": blocked_detail, + } + output_validation["parity"] = { + "status": "blocked", + "detail": blocked_detail, + } + output_validation["qtpSourceReplay"] = ( + qtp_replay + ) + else: + parity_validation = dict( + validation_runner.validate_parity( + self, + snapshot, + parser_document, + query_time_document, + ) + ) + output_validation = dict( + result["outputValidation"] + ) + output_validation.update( + parity_validation ) + result["outputValidation"] = output_validation + output_failure = _output_failure( + output_validation ) + if output_failure is not None: + raise output_failure result["status"] = "completed" return result finally: diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/queries.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/queries.py index eec335193b5..ebb67440383 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/queries.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/queries.py @@ -49,6 +49,12 @@ def _quote_kql_string(value: str) -> str: return "'" + value.replace("'", "''") + "'" +def _quote_column_name(value: str) -> str: + if IDENTIFIER_RE.fullmatch(value): + return value + return "['" + value.replace("'", "''") + "']" + + def _timestamp_literals(timestamps: Sequence[str]) -> str: unique = sorted(set(timestamps)) if not unique: @@ -60,27 +66,102 @@ def _timestamp_literals(timestamps: Sequence[str]) -> str: return ", ".join(f"datetime({value})" for value in unique) +def _datatable_type(value: str) -> str: + normalized = _normalize_column_type(value) + return "bool" if normalized == "boolean" else normalized + + +def _datatable_value(value: Any, column_type: str) -> str: + normalized = _normalize_column_type(column_type) + if value is None: + return f"{_datatable_type(normalized)}(null)" + if normalized == "string": + return "@'" + str(value).replace("'", "''") + "'" + if normalized == "datetime": + parse_kusto_datetime_ticks(str(value)) + return f"datetime({value})" + if normalized == "boolean": + if not isinstance(value, bool): + raise ValidationError( + "query-build", + f"Boolean fixture value is invalid: {value!r}", + ) + return "true" if value else "false" + if normalized in {"int", "long"}: + if isinstance(value, bool) or not isinstance(value, int): + raise ValidationError( + "query-build", + f"Integer fixture value is invalid: {value!r}", + ) + return str(value) + if normalized == "real": + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValidationError( + "query-build", + f"Real fixture value is invalid: {value!r}", + ) + return repr(float(value)) + if normalized == "dynamic": + return ( + "dynamic(" + + json.dumps( + value, + ensure_ascii=True, + separators=(",", ":"), + sort_keys=True, + ) + + ")" + ) + if normalized == "guid": + return f"guid({value})" + raise ValidationError( + "query-build", + f"Unsupported fixture schema type: {column_type}", + ) + + +def build_fixture_datatable( + schema: Sequence[Mapping[str, str]], + rows: Sequence[Mapping[str, Any]], +) -> str: + if not schema: + raise ValidationError( + "query-build", "Fixture schema must not be empty." + ) + columns = [ + ( + str(column["name"]), + str(column["type"]), + ) + for column in schema + ] + declaration = ", ".join( + f"{_quote_column_name(name)}:{_datatable_type(column_type)}" + for name, column_type in columns + ) + values = [ + ", ".join( + _datatable_value(row.get(name), column_type) + for name, column_type in columns + ) + for row in rows + ] + return ( + f"datatable({declaration})\n[\n" + + ",\n".join(values) + + "\n]" + ) + + def build_exact_slice_expression( table_name: str, timestamps: Sequence[str], - projected_columns: Sequence[str] | None = None, ) -> str: table_name = _require_identifier(table_name, "table_name") - query = ( + return ( f"{table_name}\n" f"| where TimeGenerated in ({_timestamp_literals(timestamps)})" ) - if projected_columns is not None: - if not projected_columns: - raise ValidationError( - "query-build", "At least one projected column is required." - ) - columns = [ - _require_identifier(column, "projected column") - for column in projected_columns - ] - query += "\n| project " + ", ".join(columns) - return query def _is_source_passthrough(filter_query: str) -> bool: @@ -92,12 +173,14 @@ def _is_source_passthrough(filter_query: str) -> bool: def build_filter_count_query( - source_table: str, - timestamps: Sequence[str], filter_query: str, + *, + source_schema: Sequence[Mapping[str, str]], + source_rows: Sequence[Mapping[str, Any]], ) -> str: - source_expression = build_exact_slice_expression( - source_table, timestamps + source_expression = build_fixture_datatable( + source_schema, + source_rows, ) return ( f"let source = materialize({source_expression});\n" @@ -349,12 +432,10 @@ def build_asim_tester_query( target_table: str, timestamps: Sequence[str], schema_name: str, - projected_columns: Sequence[str] | None = None, ) -> str: target_expression = build_exact_slice_expression( target_table, timestamps, - projected_columns, ) schema_literal = _quote_kql_string(schema_name) if tester_name == "ASimDataTester": diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/validator.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/validator.py index 5662e592767..e2470dedd7d 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/validator.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/validator.py @@ -3,7 +3,7 @@ from __future__ import annotations from pathlib import Path -from typing import Any, Mapping +from typing import Any, Mapping, Sequence from .parity import compare_parser_outputs from .queries import ( @@ -21,6 +21,63 @@ build_query_time_parser_query, ) + +def _intersect_schemas( + qtp_schema: Sequence[Mapping[str, str]], + itp_schema: Sequence[Mapping[str, str]], +) -> tuple[ + list[dict[str, Any]], + list[dict[str, str]], + list[dict[str, str]], +]: + itp_types = { + str(column["name"]): str(column["type"]) + for column in itp_schema + } + intersect_schema: list[dict[str, Any]] = [] + qtp_intersect_schema: list[dict[str, str]] = [] + itp_intersect_schema: list[dict[str, str]] = [] + for column in qtp_schema: + name = str(column["name"]) + if name not in itp_types: + continue + qtp_type = str(column["type"]) + itp_type = itp_types[name] + intersect_schema.append( + { + "name": name, + "qtpType": qtp_type, + "itpType": itp_type, + "typeConsistent": qtp_type == itp_type, + } + ) + qtp_intersect_schema.append( + {"name": name, "type": qtp_type} + ) + itp_intersect_schema.append( + {"name": name, "type": itp_type} + ) + return ( + intersect_schema, + qtp_intersect_schema, + itp_intersect_schema, + ) + + +def _filter_rows( + rows: Sequence[Mapping[str, Any]], + schema: Sequence[Mapping[str, Any]], +) -> list[dict[str, Any]]: + columns = [str(column["name"]) for column in schema] + return [ + { + column: row.get(column) + for column in columns + } + for row in rows + ] + + class AsimIngestionValidator: def __init__(self, repo_root: Path): self.repo_root = repo_root.resolve() @@ -34,13 +91,15 @@ def evaluate_filter( source_snapshot: Mapping[str, Any], parser_document: Mapping[str, Any], ) -> Mapping[str, Any]: - source_table = parser_document.get("SourceRawTable") filter_query = parser_document.get("ParserFilterQuery") timestamps = source_snapshot.get("timestampValues") + source_schema = source_snapshot.get("schema") + source_rows = source_snapshot.get("rows") if ( - not isinstance(source_table, str) - or not isinstance(filter_query, str) + not isinstance(filter_query, str) or not isinstance(timestamps, list) + or not isinstance(source_schema, list) + or not isinstance(source_rows, list) ): raise ValidationError( "filter", @@ -48,9 +107,9 @@ def evaluate_filter( ) raw_counts = _count_snapshot_rows(source_snapshot) query = build_filter_count_query( - source_table, - [str(value) for value in timestamps], filter_query, + source_schema=source_schema, + source_rows=source_rows, ) selected_counts = _parse_filter_counts( query_client.query_kql(query, stage="filter"), @@ -104,7 +163,6 @@ def validate_outputs( query_time_document: Mapping[str, Any] | None, ) -> Mapping[str, Any]: target_table = parser_document.get("TargetASIMTable") - source_table = parser_document.get("SourceRawTable") schema_name = ( parser_document.get("Normalization", {}).get("Schema") if isinstance( @@ -115,7 +173,6 @@ def validate_outputs( timestamps = source_snapshot.get("timestampValues") if ( not isinstance(target_table, str) - or not isinstance(source_table, str) or not isinstance(schema_name, str) or not isinstance(timestamps, list) ): @@ -160,46 +217,101 @@ def validate_outputs( if query_time_document is None: return result + parity_result = dict( + self.validate_parity( + query_client, + source_snapshot, + parser_document, + query_time_document, + ) + ) + result.update(parity_result) + return result + + def validate_parity( + self, + query_client: QueryClient, + source_snapshot: Mapping[str, Any], + parser_document: Mapping[str, Any], + query_time_document: Mapping[str, Any], + ) -> Mapping[str, Any]: + target_table = parser_document.get("TargetASIMTable") + source_table = parser_document.get("SourceRawTable") + timestamps = source_snapshot.get("timestampValues") + if ( + not isinstance(target_table, str) + or not isinstance(source_table, str) + or not isinstance(timestamps, list) + ): + raise ValidationError( + "parity", + "Parser or source snapshot is missing parity metadata.", + ) + timestamp_values = [str(value) for value in timestamps] qtp_query = build_query_time_parser_query( query_time_document, source_table, timestamp_values, ) - expected_schema, expected_rows = _normalize_query_table( + qtp_schema, qtp_rows = _normalize_query_table( query_client.query_kql( qtp_query, stage="qtp-execution" ), "qtp-execution", ) - comparison_columns = [ - column["name"] for column in expected_schema - ] - actual_schema, actual_rows = _normalize_query_table( + itp_schema, itp_rows = _normalize_query_table( query_client.query_kql( build_exact_slice_expression( target_table, timestamp_values, - comparison_columns, ), stage="target-output", ), "target-output", ) - result["queryTimeExecution"] = { - "status": "passed", - "rowCount": len(expected_rows), - } + ( + intersect_schema, + qtp_intersect_schema, + itp_intersect_schema, + ) = _intersect_schemas( + qtp_schema, + itp_schema, + ) + qtp_rows_filtered = _filter_rows( + qtp_rows, + intersect_schema, + ) + itp_rows_filtered = _filter_rows( + itp_rows, + intersect_schema, + ) parity = compare_parser_outputs( - expected_schema=expected_schema, - expected_rows=expected_rows, - actual_schema=actual_schema, - actual_rows=actual_rows, + expected_schema=qtp_intersect_schema, + expected_rows=qtp_rows_filtered, + actual_schema=itp_intersect_schema, + actual_rows=itp_rows_filtered, ) - result["parity"] = parity + result: dict[str, Any] = { + "status": parity["status"], + "queryTimeExecution": { + "status": "passed", + "rowCount": len(qtp_rows), + }, + "parity": parity, + "qtp_schema": qtp_schema, + "qtp_rows": qtp_rows, + "itp_schema": itp_schema, + "itp_rows": itp_rows, + "intersect_schema": intersect_schema, + "qtp_rows_filtered": qtp_rows_filtered, + "itp_rows_filtered": itp_rows_filtered, + } if parity["status"] != "passed": - raise ValidationError( - "parity", - "Query-time and ingestion-time outputs differ.", - details=parity, - ) + result["failure"] = { + "stage": "parity", + "message": ( + "Query-time and ingestion-time outputs differ." + ), + "details": parity, + } return result diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/workflow/reporting.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/workflow/reporting.py index d891f040c5d..4cfc996ee3e 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/workflow/reporting.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/workflow/reporting.py @@ -31,6 +31,50 @@ def _stage( return result +def _nested_error_messages(value: Any) -> list[str]: + if isinstance(value, list): + return [ + message + for item in value + for message in _nested_error_messages(item) + ] + if not isinstance(value, Mapping): + return [] + + messages: list[str] = [] + for key in ("details", "innererror", "error"): + messages.extend(_nested_error_messages(value.get(key))) + + message = value.get("message") + if isinstance(message, str) and message.strip(): + code = value.get("code") + target = value.get("target") + formatted = ( + f"{code}: {message.strip()}" + if isinstance(code, str) and code.strip() + else message.strip() + ) + if isinstance(target, str) and target.strip(): + formatted += f" [{target.strip()}]" + messages.append(formatted) + return messages + + +def _error_detail(error: Any) -> str: + if not isinstance(error, Mapping): + return "" + fragments: list[str] = [] + message = error.get("message") + if isinstance(message, str) and message.strip(): + fragments.append(message.strip()) + fragments.extend(_nested_error_messages(error.get("details"))) + unique: list[str] = [] + for fragment in fragments: + if fragment not in unique: + unique.append(fragment) + return "; ".join(unique) + + def build_report( *, target: Mapping[str, Any], @@ -125,11 +169,7 @@ def build_report( if isinstance(error, Mapping) else None ) - error_message = ( - error.get("message") - if isinstance(error, Mapping) - else "" - ) + error_detail = _error_detail(error) if isinstance(prepared_fixture, Mapping): fixture = _stage( @@ -158,7 +198,7 @@ def build_report( fixture = _stage( "failed", ( - error_message + error_detail if error_stage == "fixture-preparation" else "Fixture preparation did not complete." ), @@ -166,42 +206,95 @@ def build_report( raw_dcr_value = validation.get("rawDcr") parser_dcr_value = validation.get("parserDcr") - raw_dcr_failed = ( - not raw_dcr_value - and error_stage - in {"raw-table", "dcr-create", "dcr-provisioning"} + qtp_source_replay = validation.get("qtpSourceReplay") + qtp_replay_status = ( + qtp_source_replay.get("status") + if isinstance(qtp_source_replay, Mapping) + else None + ) + qtp_replay_error = ( + qtp_source_replay.get("error") + if isinstance(qtp_source_replay, Mapping) + else None + ) + qtp_replay_detail = _error_detail(qtp_replay_error) + qtp_replay_error_stage = ( + qtp_replay_error.get("stage") + if isinstance(qtp_replay_error, Mapping) + else None ) parser_dcr_failed = ( - bool(raw_dcr_value) - and not parser_dcr_value + not parser_dcr_value and error_stage in {"dcr-create", "dcr-provisioning"} ) raw_dcr = _stage( ( "passed" if raw_dcr_value - else "failed" - if raw_dcr_failed + else "warning" + if qtp_replay_status == "warning" + else "not-applicable" + if qtp_replay_status == "not-applicable" else "not-run" ), - error_message if raw_dcr_failed else "", - raw_dcr_value, + ( + qtp_replay_detail + if qtp_replay_status == "warning" + else "" + ), + ( + { + "dcr": raw_dcr_value, + "qtpSourceReplay": qtp_source_replay, + } + if isinstance(qtp_source_replay, Mapping) + else raw_dcr_value + ), ) raw_ingestion_failed = error_stage in { "upload", "upload-reconciliation", "raw-visibility", + "source-snapshot", } and not validation.get("parserDcr") + qtp_raw_ingestion_warning = ( + qtp_replay_status == "warning" + and ( + bool(raw_dcr_value) + or qtp_replay_error_stage + in { + "upload", + "upload-reconciliation", + "raw-visibility", + } + ) + ) raw_ingestion = _stage( - "failed" - if raw_ingestion_failed - else "passed" - if validation.get("rawUpload") - else "not-run", - error_message if raw_ingestion_failed else "", + ( + "warning" + if qtp_raw_ingestion_warning + else "blocked" + if qtp_replay_status == "warning" + else "failed" + if raw_ingestion_failed + else "passed" + if validation.get("rawUpload") + else "not-applicable" + if qtp_replay_status == "not-applicable" + else "not-run" + ), + ( + qtp_replay_detail + if qtp_raw_ingestion_warning + else "Blocked by QTP raw DCR warning." + if qtp_replay_status == "warning" + else error_detail + if raw_ingestion_failed + else "" + ), { "upload": validation.get("rawUpload", []), - "snapshot": validation.get("snapshot"), + "qtpSourceReplay": qtp_source_replay, }, ) filter_validation = validation.get("filterValidation") @@ -214,7 +307,7 @@ def build_report( if error_stage == "filter" else "not-run" ), - error_message if error_stage == "filter" else "", + error_detail if error_stage == "filter" else "", filter_validation, ) parser_dcr = _stage( @@ -225,7 +318,7 @@ def build_report( if parser_dcr_failed else "not-run" ), - error_message if parser_dcr_failed else "", + error_detail if parser_dcr_failed else "", parser_dcr_value, ) parser_ingestion_failed = error_stage in { @@ -239,7 +332,7 @@ def build_report( else "passed" if validation.get("parserUpload") else "not-run", - error_message if parser_ingestion_failed else "", + error_detail if parser_ingestion_failed else "", validation.get("parserUpload", []), ) @@ -268,18 +361,21 @@ def build_report( "passed" if validation.get("parserDcr") else ( "failed" if error_stage == "output-validation" else "not-run" ), - error_message if error_stage == "output-validation" else "", + error_detail if error_stage == "output-validation" else "", schema_validation, ) data_stage = _stage( ( str(data_validation.get("status")) if isinstance(data_validation, Mapping) + else "passed" + if error_stage + in {"qtp-execution", "target-output", "parity"} else "failed" if error_stage == "asim-data" else "not-run" ), - error_message if error_stage == "asim-data" else "", + error_detail if error_stage == "asim-data" else "", data_validation, ) if counterpart["status"] in {"not-applicable", "blocked"}: @@ -293,12 +389,18 @@ def build_report( str(qtp_execution.get("status")) if isinstance(qtp_execution, Mapping) else "passed" - if error_stage == "parity" + if error_stage in {"target-output", "parity"} else "failed" if error_stage == "qtp-execution" else "not-run" ), - error_message if error_stage == "qtp-execution" else "", + ( + error_detail + if error_stage == "qtp-execution" + else str(qtp_execution.get("detail") or "") + if isinstance(qtp_execution, Mapping) + else "" + ), qtp_execution, ) parity_stage = _stage( @@ -306,14 +408,20 @@ def build_report( str(parity_validation.get("status")) if isinstance(parity_validation, Mapping) else "failed" - if error_stage == "parity" + if error_stage in {"target-output", "parity"} else "not-run" ), - error_message if error_stage == "parity" else "", + ( + error_detail + if error_stage in {"target-output", "parity"} + else str(parity_validation.get("detail") or "") + if isinstance(parity_validation, Mapping) + else "" + ), parity_validation or ( error.get("details") if isinstance(error, Mapping) - and error_stage == "parity" + and error_stage in {"target-output", "parity"} else None ), ) @@ -332,13 +440,28 @@ def build_report( and item.get("status") == "warning" for item in cleanup_items ) + cleanup_details = [ + _error_detail(item) + for item in cleanup_items + if isinstance(item, Mapping) + and item.get("status") == "warning" + ] + cleanup_details = [ + detail for detail in cleanup_details if detail + ] cleanup_stage = _stage( "warning" if cleanup_warning else "passed" if cleanup_items else "not-run", - "Cleanup reported a warning." if cleanup_warning else "", + ( + "; ".join(dict.fromkeys(cleanup_details)) + if cleanup_details + else "Cleanup reported a warning." + if cleanup_warning + else "" + ), cleanup_items, ) @@ -427,17 +550,32 @@ def emit_annotations(report: Mapping[str, Any]) -> None: stages = report.get("stages", {}) if isinstance(stages, Mapping): for stage_name, stage in stages.items(): - if ( - not isinstance(stage, Mapping) - or stage.get("status") != "failed" - ): + if not isinstance(stage, Mapping): + continue + stage_status = stage.get("status") + if stage_status not in {"failed", "warning"}: continue detail = _short_message( - stage.get("detail") or "Stage failed." + stage.get("detail") + or ( + "Stage failed." + if stage_status == "failed" + else "Stage reported a warning." + ) + ) + command = ( + annotation_kind + if stage_status == "failed" + else "warning" + ) + title_status = ( + "failure" + if stage_status == "failed" + else "warning" ) print( - f"::{annotation_kind} title=ASIM {stage_name} " - f"failure::{target_id}: " + f"::{command} title=ASIM {stage_name} " + f"{title_status}::{target_id}: " f"{_github_command_message(detail)}" ) if status == "failed": @@ -450,9 +588,3 @@ def emit_annotations(report: Mapping[str, Any]) -> None: "::error title=ASIM ingestion validation failed::" f"{target_id} failed validation in {mode} mode." ) - cleanup = report.get("stages", {}).get("cleanup", {}) - if isinstance(cleanup, Mapping) and cleanup.get("status") == "warning": - print( - "::warning title=ASIM DCR cleanup warning::" - f"{target_id} cleanup was incomplete." - ) diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion_harness.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion_harness.py index cde19d7f936..a8e7429f46d 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion_harness.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion_harness.py @@ -40,7 +40,6 @@ _write_json_no_clobber, build_exact_timestamp_query, build_resource_names, - build_source_snapshot, build_upload_batches, count_rows_by_timestamp, ) @@ -60,7 +59,6 @@ "ValidationRunner", "build_exact_timestamp_query", "build_resource_names", - "build_source_snapshot", "build_upload_batches", "count_rows_by_timestamp", "main", diff --git a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py index d959af3b139..83081fe738a 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py @@ -3,6 +3,7 @@ import sys import tempfile import unittest +import unittest.mock from pathlib import Path import requests @@ -63,10 +64,18 @@ def get_token(self, scope): class FakeValidationRunner: - def __init__(self, timestamp): + def __init__( + self, + timestamp, + output_result=None, + parity_result=None, + ): self.timestamp = timestamp + self.output_result = output_result + self.parity_result = parity_result self.filter_calls = 0 self.output_calls = 0 + self.parity_calls = 0 def evaluate_filter( self, query_client, source_snapshot, parser_document @@ -87,11 +96,28 @@ def validate_outputs( query_time_document, ): self.output_calls += 1 - return { + return self.output_result or { "status": "passed", "parity": {"status": "not-applicable"}, } + def validate_parity( + self, + query_client, + source_snapshot, + parser_document, + query_time_document, + ): + self.parity_calls += 1 + return self.parity_result or { + "status": "passed", + "queryTimeExecution": { + "status": "passed", + "rowCount": 1, + }, + "parity": {"status": "passed"}, + } + class FailingFilterRunner: def evaluate_filter( @@ -108,6 +134,15 @@ def validate_outputs( ): raise AssertionError("Output validation must not run.") + def validate_parity( + self, + query_client, + source_snapshot, + parser_document, + query_time_document, + ): + raise AssertionError("Parity validation must not run.") + class FakeClock: def __init__(self): @@ -168,63 +203,6 @@ def test_builds_exact_timestamp_query(self): ) self.assertIn("join kind=inner", query) - def test_snapshot_excludes_platform_columns_and_normalizes_types(self): - snapshot = HARNESS.build_source_snapshot( - target_id="parser-a", - table_name="Contoso_CL", - timestamps=["2026-10-16T19:55:00.1234567Z"], - query_table={ - "columns": [ - {"name": "TimeGenerated", "type": "datetime"}, - {"name": "Type", "type": "string"}, - {"name": "TenantId", "type": "guid"}, - {"name": "Payload", "type": "dynamic"}, - {"name": "_ResourceId", "type": "string"}, - ], - "rows": [ - [ - "2026-10-16T19:55:00.1234567Z", - "Contoso_CL", - "tenant", - '{"key":"value"}', - "/resource", - ] - ], - }, - ) - schema = { - column["name"]: column["type"] - for column in snapshot["schema"] - } - self.assertEqual("string", schema["TenantId"]) - self.assertEqual( - {"key": "value"}, snapshot["rows"][0]["Payload"] - ) - self.assertNotIn("_ResourceId", schema) - self.assertIn("Type", schema) - self.assertEqual(["_ResourceId"], snapshot["excludedColumns"]) - - def test_snapshot_rejects_rows_outside_timestamp_set(self): - with self.assertRaisesRegex( - HARNESS.HarnessError, "outside the exact timestamp" - ): - HARNESS.build_source_snapshot( - target_id="parser-a", - table_name="Contoso_CL", - timestamps=["2026-10-16T19:55:00.1234567Z"], - query_table={ - "columns": [ - { - "name": "TimeGenerated", - "type": "datetime", - } - ], - "rows": [ - ["2026-10-16T19:55:01.1234567Z"] - ], - }, - ) - def test_batches_keep_noncontiguous_equal_timestamps_together(self): rows = [ { @@ -468,36 +446,17 @@ def test_cleanup_uses_independent_deadline(self): self.assertEqual("deleted", result["status"]) self.assertEqual("DELETE", session.calls[0]["method"]) - def test_full_run_builds_raw_and_parser_dcrs_and_snapshot(self): + def test_full_run_uploads_fixture_directly_to_parser_dcr(self): timestamp = "2026-10-16T19:55:00.1234567Z" query_result = query_payload( [ {"name": "TimeGenerated", "type": "datetime"}, - {"name": "Type", "type": "string"}, - {"name": "TenantId", "type": "guid"}, {"name": "Value", "type": "string"}, - {"name": "_ResourceId", "type": "string"}, ], - [[timestamp, "Contoso_CL", "tenant", "one", "/resource"]], + [[timestamp, "one"]], ) harness, session, _ = self.harness( [ - FakeResponse(202), # custom table PUT - FakeResponse(200, {"properties": {"provisioningState": "Succeeded"}}), - FakeResponse(201), # raw DCR PUT - FakeResponse( - 200, - { - "id": "/subscriptions/s/raw", - "properties": { - "provisioningState": "Succeeded", - "immutableId": "raw-immutable", - }, - }, - ), - FakeResponse(204), # raw upload - FakeResponse(200, query_result), # raw snapshot - FakeResponse(200, query_result), # raw stable FakeResponse(201), # parser DCR PUT FakeResponse( 200, @@ -513,7 +472,6 @@ def test_full_run_builds_raw_and_parser_dcrs_and_snapshot(self): FakeResponse(200, query_result), # target visibility FakeResponse(200, query_result), # target stable FakeResponse(204), # parser cleanup - FakeResponse(404), # raw cleanup ] ) prepared = { @@ -545,7 +503,7 @@ def test_full_run_builds_raw_and_parser_dcrs_and_snapshot(self): "TargetASIMTable": "ASimAuthenticationEventLogs", "ParserQuery": ( " source // remove this comment\n" - " | project TimeGenerated, Type, TenantId, Value" + " | project TimeGenerated, Type, Value" ), } with tempfile.TemporaryDirectory() as directory: @@ -562,16 +520,21 @@ def test_full_run_builds_raw_and_parser_dcrs_and_snapshot(self): self.assertEqual("completed", result["status"]) self.assertEqual( - ["deleted", "not-found"], + ["deleted"], [entry["status"] for entry in result["cleanup"]], ) - self.assertNotIn("_ResourceId", { - column["name"] for column in snapshot["schema"] - }) - self.assertEqual("string", { - column["name"]: column["type"] - for column in snapshot["schema"] - }["TenantId"]) + self.assertEqual("prepared-fixture", snapshot["authority"]) + self.assertEqual([], snapshot["excludedColumns"]) + self.assertEqual( + ["TimeGenerated", "Value", "Type"], + [column["name"] for column in snapshot["schema"]], + ) + self.assertEqual("Contoso_CL", snapshot["rows"][0]["Type"]) + self.assertIsNone(result["rawDcr"]) + self.assertEqual( + "not-applicable", + result["qtpSourceReplay"]["status"], + ) dcr_puts = [ call @@ -579,17 +542,12 @@ def test_full_run_builds_raw_and_parser_dcrs_and_snapshot(self): if call["method"] == "PUT" and "dataCollectionRules" in call["url"] ] - self.assertEqual(2, len(dcr_puts)) - raw_flow = dcr_puts[0]["json"]["properties"]["dataFlows"][0] - parser_flow = dcr_puts[1]["json"]["properties"]["dataFlows"][0] - self.assertEqual("source", raw_flow["transformKql"]) - self.assertEqual( - "Custom-Contoso_CL", raw_flow["outputStream"] - ) + self.assertEqual(1, len(dcr_puts)) + parser_flow = dcr_puts[0]["json"]["properties"]["dataFlows"][0] self.assertEqual( ( "source\n" - "| project TimeGenerated, Type, TenantId, Value" + "| project TimeGenerated, Type, Value" ), parser_flow["transformKql"], ) @@ -602,17 +560,12 @@ def test_run_cleans_up_dcr_when_provisioning_fails(self): timestamp = "2026-10-16T19:55:00.1234567Z" harness, session, _ = self.harness( [ - FakeResponse(202), # custom table PUT - FakeResponse( - 200, - {"properties": {"provisioningState": "Succeeded"}}, - ), - FakeResponse(201), # raw DCR PUT + FakeResponse(201), # parser DCR PUT FakeResponse( 200, {"properties": {"provisioningState": "Failed"}}, ), - FakeResponse(204), # cleanup attempted resource + FakeResponse(204), # parser cleanup ] ) prepared = { @@ -628,7 +581,12 @@ def test_run_cleans_up_dcr_when_provisioning_fails(self): "runAttempt": 1, "timestampValues": [timestamp], }, - "rows": [{"TimeGenerated": timestamp}], + "rows": [ + { + "TimeGenerated": timestamp, + "Type": "Contoso", + } + ], } parser = { "ParserName": "ASimAuthenticationContoso", @@ -669,20 +627,6 @@ def test_full_run_uses_validation_runner_counts_and_outputs(self): ) harness, _, _ = self.harness( [ - FakeResponse(201), # raw DCR PUT - FakeResponse( - 200, - { - "id": "/subscriptions/s/raw", - "properties": { - "provisioningState": "Succeeded", - "immutableId": "raw-immutable", - }, - }, - ), - FakeResponse(204), # raw upload - FakeResponse(200, query_result), # raw snapshot - FakeResponse(200, query_result), # raw stable FakeResponse(201), # parser DCR PUT FakeResponse( 200, @@ -698,7 +642,6 @@ def test_full_run_uses_validation_runner_counts_and_outputs(self): FakeResponse(200, query_result), # target visibility FakeResponse(200, query_result), # target stable FakeResponse(204), # parser cleanup - FakeResponse(204), # raw cleanup ] ) prepared = { @@ -761,18 +704,29 @@ def test_full_run_uses_validation_runner_counts_and_outputs(self): ) self.assertTrue(result["telemetry"]["requestAttempts"]) - def test_filter_validation_failure_still_cleans_up_raw_dcr(self): + def test_qtp_source_replay_uses_itp_input_stream(self): timestamp = "2026-10-16T19:55:00.1234567Z" query_result = query_payload( - [ - {"name": "TimeGenerated", "type": "datetime"}, - {"name": "Type", "type": "string"}, - ], - [[timestamp, "Contoso"]], + [{"name": "TimeGenerated", "type": "datetime"}], + [[timestamp]], ) harness, session, _ = self.harness( [ - FakeResponse(201), # raw DCR PUT + FakeResponse(201), # parser DCR PUT + FakeResponse( + 200, + { + "id": "/subscriptions/s/parser", + "properties": { + "provisioningState": "Succeeded", + "immutableId": "parser-immutable", + }, + }, + ), + FakeResponse(204), # parser upload + FakeResponse(200, query_result), # target visibility + FakeResponse(200, query_result), # target stable + FakeResponse(201), # QTP raw DCR PUT FakeResponse( 200, { @@ -784,9 +738,127 @@ def test_filter_validation_failure_still_cleans_up_raw_dcr(self): }, ), FakeResponse(204), # raw upload - FakeResponse(200, query_result), # raw snapshot + FakeResponse(200, query_result), # raw visibility FakeResponse(200, query_result), # raw stable FakeResponse(204), # raw cleanup + FakeResponse(204), # parser cleanup + ] + ) + prepared = { + "targetId": "parser-a", + "sourceRawTable": "Contoso", + "rowCount": 1, + "schema": [ + {"name": "TimeGenerated", "type": "datetime"} + ], + "namespace": { + "pullRequest": "123", + "runId": "456", + "runAttempt": 1, + "timestampValues": [timestamp], + }, + "rows": [ + { + "TimeGenerated": timestamp, + "Type": "Contoso", + } + ], + } + parser = { + "ParserName": "ASimAuthenticationContoso", + "InputStream": "Microsoft-SentinelContoso", + "SourceRawTable": "Contoso", + "TargetASIMTable": "ASimAuthenticationEventLogs", + "ParserQuery": "source", + } + validation_runner = FakeValidationRunner(timestamp) + + with tempfile.TemporaryDirectory() as directory: + result = harness.run( + prepared, + parser, + snapshot_output=Path(directory) / "snapshot.json", + validation_runner=validation_runner, + query_time_document={"ParserQuery": "Contoso"}, + ) + + dcr_puts = [ + call["json"]["properties"] + for call in session.calls + if call["method"] == "PUT" + and "dataCollectionRules" in call["url"] + ] + self.assertEqual(2, len(dcr_puts)) + self.assertEqual( + "Microsoft-ASimAuthenticationEventLogs", + dcr_puts[0]["dataFlows"][0]["outputStream"], + ) + self.assertEqual( + "Microsoft-SentinelContoso", + dcr_puts[1]["dataFlows"][0]["outputStream"], + ) + parser_columns = next( + iter(dcr_puts[0]["streamDeclarations"].values()) + )["columns"] + raw_columns = next( + iter(dcr_puts[1]["streamDeclarations"].values()) + )["columns"] + self.assertIn( + "Type", + {column["name"] for column in parser_columns}, + ) + self.assertNotIn( + "Type", + {column["name"] for column in raw_columns}, + ) + self.assertEqual( + "passed", result["qtpSourceReplay"]["status"] + ) + self.assertEqual( + "Microsoft-SentinelContoso", + result["rawDcr"]["outputStream"], + ) + self.assertEqual(1, validation_runner.output_calls) + self.assertEqual(1, validation_runner.parity_calls) + + def test_qtp_raw_dcr_failure_warns_and_blocks_parity(self): + timestamp = "2026-10-16T19:55:00.1234567Z" + query_result = query_payload( + [{"name": "TimeGenerated", "type": "datetime"}], + [[timestamp]], + ) + harness, _, _ = self.harness( + [ + FakeResponse(201), # parser DCR PUT + FakeResponse( + 200, + { + "id": "/subscriptions/s/parser", + "properties": { + "provisioningState": "Succeeded", + "immutableId": "parser-immutable", + }, + }, + ), + FakeResponse(204), # parser upload + FakeResponse(200, query_result), # target visibility + FakeResponse(200, query_result), # target stable + FakeResponse( + 400, + { + "error": { + "code": "InvalidPayload", + "details": [ + { + "code": "InvalidStream", + "message": "Stream is unsupported.", + } + ], + } + }, + ), + FakeResponse(404), # raw cleanup + FakeResponse(204), # parser cleanup ] ) prepared = { @@ -804,6 +876,175 @@ def test_filter_validation_failure_still_cleans_up_raw_dcr(self): }, "rows": [{"TimeGenerated": timestamp}], } + parser = { + "ParserName": "ASimAuthenticationContoso", + "InputStream": "Microsoft-SentinelContoso", + "SourceRawTable": "Contoso", + "TargetASIMTable": "ASimAuthenticationEventLogs", + "ParserQuery": "source", + } + validation_runner = FakeValidationRunner(timestamp) + + with tempfile.TemporaryDirectory() as directory: + result = harness.run( + prepared, + parser, + snapshot_output=Path(directory) / "snapshot.json", + validation_runner=validation_runner, + query_time_document={"ParserQuery": "Contoso"}, + ) + + self.assertEqual("completed", result["status"]) + self.assertIsNone(result["rawDcr"]) + self.assertEqual( + "warning", result["qtpSourceReplay"]["status"] + ) + self.assertEqual( + "dcr-create", + result["qtpSourceReplay"]["error"]["stage"], + ) + self.assertEqual( + "blocked", + result["outputValidation"]["queryTimeExecution"]["status"], + ) + self.assertEqual( + "blocked", + result["outputValidation"]["parity"]["status"], + ) + self.assertEqual(1, validation_runner.output_calls) + self.assertEqual(0, validation_runner.parity_calls) + + def test_failed_output_validation_is_preserved_before_raise(self): + timestamp = "2026-10-16T19:55:00.1234567Z" + query_table = { + "columns": [ + {"name": "TimeGenerated", "type": "datetime"}, + {"name": "Type", "type": "string"}, + {"name": "Value", "type": "string"}, + ], + "rows": [[timestamp, "Contoso_CL", "one"]], + } + harness, _, _ = self.harness([]) + prepared = { + "version": 1, + "targetId": "parser-a", + "sourceRawTable": "Contoso_CL", + "rowCount": 1, + "schema": [ + {"name": "TimeGenerated", "type": "datetime"}, + {"name": "Value", "type": "string"}, + ], + "namespace": { + "pullRequest": "123", + "runId": "456", + "runAttempt": 1, + "timestampValues": [timestamp], + }, + "rows": [ + { + "TimeGenerated": timestamp, + "Type": "Contoso_CL", + "Value": "one", + } + ], + } + parser = { + "ParserName": "ASimAuthenticationContoso", + "SourceRawTable": "Contoso_CL", + "TargetASIMTable": "ASimAuthenticationEventLogs", + "ParserQuery": "source", + } + output_result = { + "status": "failed", + "failure": { + "stage": "parity", + "message": "rows differ", + "details": {"missingRowCount": 1}, + }, + "qtp_schema": [{"name": "Value", "type": "string"}], + "qtp_rows": [{"Value": "expected"}], + "itp_schema": [{"name": "Value", "type": "string"}], + "itp_rows": [{"Value": "actual"}], + "intersect_schema": [ + {"name": "Value", "type": "string"} + ], + "qtp_rows_filtered": [{"Value": "expected"}], + "itp_rows_filtered": [{"Value": "actual"}], + "parity": {"status": "failed"}, + } + validation_runner = FakeValidationRunner( + timestamp, + output_result=output_result, + ) + handles = [ + HARNESS.DcrHandle( + "parser", + "/dcr/parser", + "parser-immutable", + "Custom-parser", + "Microsoft-ASimAuthenticationEventLogs", + ), + ] + + with tempfile.TemporaryDirectory() as directory: + with ( + unittest.mock.patch.object( + harness, + "create_dcr", + side_effect=handles, + ), + unittest.mock.patch.object( + harness, + "upload_rows", + return_value=[{"status": "accepted"}], + ), + unittest.mock.patch.object( + harness, + "wait_for_exact_rows", + return_value=query_table, + ), + unittest.mock.patch.object( + harness, + "delete_dcr", + return_value={"status": "deleted"}, + ), + ): + with self.assertRaisesRegex( + HARNESS.HarnessError, + "rows differ", + ): + harness.run( + prepared, + parser, + snapshot_output=( + Path(directory) / "snapshot.json" + ), + validation_runner=validation_runner, + ) + + self.assertEqual( + output_result, + harness.last_result["outputValidation"], + ) + + def test_filter_validation_failure_creates_no_dcr(self): + timestamp = "2026-10-16T19:55:00.1234567Z" + harness, session, _ = self.harness([]) + prepared = { + "targetId": "parser-a", + "sourceRawTable": "Contoso", + "rowCount": 1, + "schema": [ + {"name": "TimeGenerated", "type": "datetime"} + ], + "namespace": { + "pullRequest": "123", + "runId": "456", + "runAttempt": 1, + "timestampValues": [timestamp], + }, + "rows": [{"TimeGenerated": timestamp}], + } parser = { "ParserName": "ASimAuthenticationContoso", "SourceRawTable": "Contoso", @@ -822,7 +1063,8 @@ def test_filter_validation_failure_still_cleans_up_raw_dcr(self): validation_runner=FailingFilterRunner(), ) - self.assertEqual("DELETE", session.calls[-1]["method"]) + self.assertEqual([], session.calls) + self.assertEqual([], harness.last_result["cleanup"]) def test_full_run_rejects_incomplete_expected_count_map(self): harness, _, _ = self.harness([]) diff --git a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_validation.py b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_validation.py index 571e1f98631..6fb785c56f5 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_validation.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_validation.py @@ -53,16 +53,52 @@ def query_kql(self, query, *, stage="query"): class QueryConstructionTests(unittest.TestCase): - def test_builds_exact_filter_query(self): + def test_builds_typed_fixture_filter_query(self): query = VALIDATION.build_filter_count_query( - "Contoso_CL", - [TIMESTAMP_1, TIMESTAMP_2], - "source\n| where EventType == 'Login'", + "source", + source_schema=[ + {"name": "TimeGenerated", "type": "datetime"}, + {"name": "Message", "type": "string"}, + {"name": "Enabled", "type": "boolean"}, + {"name": "Count", "type": "long"}, + {"name": "Score", "type": "real"}, + {"name": "Payload", "type": "dynamic"}, + {"name": "CorrelationId", "type": "guid"}, + {"name": "Column Name", "type": "string"}, + ], + source_rows=[ + { + "TimeGenerated": TIMESTAMP_1, + "Message": r"C:\logs\//keep 'quoted'", + "Enabled": True, + "Count": None, + "Score": 1.5, + "Payload": {"a": [1, True]}, + "CorrelationId": ( + "00000000-0000-0000-0000-000000000001" + ), + "Column Name": "quoted", + } + ], ) - self.assertIn("let source = materialize(Contoso_CL", query) - self.assertIn(f"datetime({TIMESTAMP_1})", query) - self.assertIn("where EventType == 'Login'", query) + self.assertIn( + "datatable(TimeGenerated:datetime, Message:string, " + "Enabled:bool, Count:long, Score:real, " + "Payload:dynamic, CorrelationId:guid, " + "['Column Name']:string)", + query, + ) + self.assertIn( + r"@'C:\logs\//keep ''quoted'''", + query, + ) + self.assertIn("true, long(null), 1.5", query) + self.assertIn('dynamic({"a":[1,true]})', query) + self.assertIn( + "guid(00000000-0000-0000-0000-000000000001)", + query, + ) self.assertIn("_AsimSelectedCount=count()", query) def test_qtp_query_overrides_disabled_and_pack(self): @@ -144,6 +180,10 @@ class FilterValidationTests(unittest.TestCase): def snapshot(self): return { "timestampValues": [TIMESTAMP_1, TIMESTAMP_2], + "schema": [ + {"name": "TimeGenerated", "type": "datetime"}, + {"name": "EventType", "type": "string"}, + ], "rows": [ {"TimeGenerated": TIMESTAMP_1, "EventType": "Login"}, {"TimeGenerated": TIMESTAMP_1, "EventType": "Logout"}, @@ -195,6 +235,12 @@ def test_selective_filter_reports_selected_and_rejected_rows(self): {TIMESTAMP_1: 1, TIMESTAMP_2: 0}, result["selectedCountsByTimestamp"], ) + filter_query = client.queries[0]["query"] + self.assertIn( + "datatable(TimeGenerated:datetime, EventType:string)", + filter_query, + ) + self.assertNotIn("materialize(Contoso_CL", filter_query) def test_selective_filter_requires_rejected_fixture_row(self): client = FakeQueryClient( @@ -581,18 +627,50 @@ def source_snapshot(self): "rows": [{"TimeGenerated": TIMESTAMP_1}], } - def output_table(self): + def qtp_output_table(self, *, event_vendor="Contoso"): return query_table( [ ("TimeGenerated", "datetime"), ("EventUid", "string"), ("EventVendor", "string"), + ("EventSchema", "string"), + ("QtpOnly", "string"), + ("DifferentType", "string"), ], [ { "TimeGenerated": TIMESTAMP_1, "EventUid": "event-1", - "EventVendor": "Contoso", + "EventVendor": event_vendor, + "EventSchema": "Authentication", + "QtpOnly": "qtp", + "DifferentType": "1", + } + ], + ) + + def itp_output_table( + self, + *, + event_vendor="Contoso", + different_type="string", + ): + different_value = "1" if different_type == "string" else 1 + return query_table( + [ + ("TimeGenerated", "datetime"), + ("EventUid", "string"), + ("EventVendor", "string"), + ("ItpOnly", "string"), + ("DifferentType", different_type), + ], + [ + { + "TimeGenerated": TIMESTAMP_1, + "EventUid": "event-1", + "EventVendor": event_vendor, + "ItpOnly": "itp", + "DifferentType": different_value, } ], ) @@ -610,8 +688,8 @@ def test_runs_asim_tests_and_optional_parity(self): [{"Result": "(1) Warning: example"}], ), ), - ("qtp-execution", self.output_table()), - ("target-output", self.output_table()), + ("qtp-execution", self.qtp_output_table()), + ("target-output", self.itp_output_table()), ] ) qtp = { @@ -647,6 +725,49 @@ def test_runs_asim_tests_and_optional_parity(self): ) self.assertEqual("passed", result["dataTest"]["status"]) self.assertEqual("passed", result["parity"]["status"]) + self.assertEqual( + 1, + sum( + item["stage"] == "asim-data" + for item in client.queries + ), + ) + self.assertEqual( + [ + "TimeGenerated", + "EventUid", + "EventVendor", + "DifferentType", + ], + [ + column["name"] + for column in result["intersect_schema"] + ], + ) + self.assertEqual( + [ + { + "TimeGenerated": TIMESTAMP_1, + "EventUid": "event-1", + "EventVendor": "Contoso", + "DifferentType": "1", + } + ], + result["qtp_rows_filtered"], + ) + self.assertEqual( + result["qtp_rows_filtered"], + result["itp_rows_filtered"], + ) + self.assertTrue( + all( + column["typeConsistent"] + for column in result["intersect_schema"] + ) + ) + self.assertIn("EventSchema", result["qtp_rows"][0]) + self.assertNotIn("EventSchema", result["itp_rows"][0]) + self.assertIn("ItpOnly", result["itp_rows"][0]) data_query = next( item["query"] for item in client.queries @@ -666,10 +787,92 @@ def test_runs_asim_tests_and_optional_parity(self): for item in client.queries if item["stage"] == "target-output" ) - self.assertIn( - "| project TimeGenerated, EventUid, EventVendor", - target_query, + self.assertNotIn("| project", target_query) + + def test_parity_failure_returns_full_and_filtered_artifacts(self): + client = FakeQueryClient( + [ + ("asim-data", self.empty_tester_result()), + ("qtp-execution", self.qtp_output_table()), + ( + "target-output", + self.itp_output_table(event_vendor="Different"), + ), + ] + ) + qtp = { + "ParserParams": [], + "ParserQuery": ( + "Contoso_CL\n" + "| project TimeGenerated, EventUid, EventVendor" + ), + } + + result = self.validator().validate_outputs( + client, + self.source_snapshot(), + self.parser_document(), + qtp, + ) + + self.assertEqual("failed", result["status"]) + self.assertEqual("parity", result["failure"]["stage"]) + self.assertEqual("failed", result["parity"]["status"]) + self.assertTrue(result["qtp_schema"]) + self.assertTrue(result["qtp_rows"]) + self.assertTrue(result["itp_schema"]) + self.assertTrue(result["itp_rows"]) + self.assertTrue(result["intersect_schema"]) + self.assertTrue(result["qtp_rows_filtered"]) + self.assertTrue(result["itp_rows_filtered"]) + + def test_shared_column_type_mismatch_fails_schema_parity(self): + client = FakeQueryClient( + [ + ("asim-data", self.empty_tester_result()), + ("qtp-execution", self.qtp_output_table()), + ( + "target-output", + self.itp_output_table(different_type="long"), + ), + ] + ) + qtp = { + "ParserParams": [], + "ParserQuery": ( + "Contoso_CL\n" + "| project TimeGenerated, EventUid, EventVendor" + ), + } + + result = self.validator().validate_outputs( + client, + self.source_snapshot(), + self.parser_document(), + qtp, + ) + + self.assertEqual("failed", result["status"]) + mismatch = next( + column + for column in result["intersect_schema"] + if column["name"] == "DifferentType" + ) + self.assertEqual("string", mismatch["qtpType"]) + self.assertEqual("long", mismatch["itpType"]) + self.assertFalse(mismatch["typeConsistent"]) + self.assertEqual("failed", result["parity"]["status"]) + self.assertEqual( + [ + { + "column": "DifferentType", + "queryTimeType": "string", + "ingestTimeType": "long", + } + ], + result["parity"]["schema"]["typeMismatches"], ) + self.assertIsNone(result["parity"]["rows"]) def test_without_qtp_reports_parity_not_applicable(self): client = FakeQueryClient( diff --git a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_workflow.py b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_workflow.py index bdd446ea7fe..0084a305556 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_workflow.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_workflow.py @@ -499,7 +499,7 @@ def test_parser_dcr_failure_does_not_add_detail_to_raw_dcr(self): report["stages"]["parserDcr"]["detail"], ) - def test_raw_dcr_failure_adds_detail_only_to_raw_dcr(self): + def test_direct_parser_dcr_failure_adds_detail_to_parser_dcr(self): validation = self.base_validation() validation["status"] = "failed" validation["rawDcr"] = None @@ -516,15 +516,193 @@ def test_raw_dcr_failure_adds_detail_only_to_raw_dcr(self): report = self.build(validation) - self.assertEqual("failed", report["stages"]["rawDcr"]["status"]) + self.assertEqual( + "not-run", report["stages"]["rawDcr"]["status"] + ) + self.assertEqual("", report["stages"]["rawDcr"]["detail"]) + self.assertEqual( + "failed", report["stages"]["parserDcr"]["status"] + ) self.assertEqual( "Provisioning failed.", + report["stages"]["parserDcr"]["detail"], + ) + + def test_stage_summary_includes_nested_provider_error(self): + validation = self.base_validation() + validation["status"] = "failed" + validation["rawDcr"] = None + validation["parserDcr"] = None + validation["rawUpload"] = [] + validation["parserUpload"] = [] + validation["snapshot"] = None + validation["filterValidation"] = None + validation["outputValidation"] = None + validation["error"] = { + "stage": "dcr-create", + "message": "PUT failed with HTTP 400.", + "details": { + "error": { + "code": "InvalidPayload", + "message": "Data collection rule is invalid", + "details": [ + { + "code": "InvalidStream", + "message": ( + "'OutputStream' stream " + "'Microsoft-SalesforceAuditTrail' must " + "be a custom stream or one of the " + "allowed streams." + ), + "target": ( + "Properties.DataFlows[0].OutputStream" + ), + } + ], + } + }, + } + + report = self.build(validation) + detail = report["stages"]["parserDcr"]["detail"] + summary = WORKFLOW.render_summary(report) + + self.assertIn("PUT failed with HTTP 400.", detail) + self.assertIn("InvalidStream:", detail) + self.assertIn("Microsoft-SalesforceAuditTrail", detail) + self.assertIn( + "Properties.DataFlows[0].OutputStream", + detail, + ) + self.assertIn("Microsoft-SalesforceAuditTrail", summary) + + def test_target_output_error_is_exposed_as_parity_failure(self): + validation = self.base_validation() + validation["status"] = "failed" + validation["outputValidation"] = None + validation["error"] = { + "stage": "target-output", + "message": "POST failed with HTTP 400.", + "details": { + "error": { + "code": "BadArgumentError", + "innererror": { + "code": "SEM0100", + "message": ( + "Failed to resolve scalar expression " + "named 'EventSchema'" + ), + }, + "message": ( + "The request had some invalid properties" + ), + } + }, + } + + report = self.build(validation) + + self.assertEqual("passed", report["stages"]["data"]["status"]) + self.assertEqual( + "passed", report["stages"]["qtpExecution"]["status"] + ) + self.assertEqual( + "failed", report["stages"]["parity"]["status"] + ) + self.assertIn( + "EventSchema", + report["stages"]["parity"]["detail"], + ) + + def test_cleanup_warning_exposes_cleanup_error(self): + report = self.build( + self.base_validation(), + cleanup={ + "resources": [ + { + "resourceId": "/raw", + "status": "warning", + "message": "DELETE failed with HTTP 503.", + } + ] + }, + ) + + self.assertEqual( + "DELETE failed with HTTP 503.", + report["stages"]["cleanup"]["detail"], + ) + + def test_qtp_raw_dcr_failure_warns_without_failing_parser(self): + validation = self.base_validation() + validation["rawDcr"] = None + validation["rawUpload"] = [] + validation["qtpSourceReplay"] = { + "status": "warning", + "inputStream": "Microsoft-SentinelContoso", + "error": { + "stage": "dcr-create", + "message": "PUT failed with HTTP 400.", + "details": { + "error": { + "code": "InvalidStream", + "message": "Stream is unsupported.", + } + }, + }, + } + validation["outputValidation"]["queryTimeExecution"] = { + "status": "blocked", + "detail": "Skipped because QTP source replay failed.", + } + validation["outputValidation"]["parity"] = { + "status": "blocked", + "detail": "Skipped because QTP source replay failed.", + } + + report = self.build(validation) + + self.assertEqual("passed", report["status"]) + self.assertFalse(report["requiredFailure"]) + self.assertEqual( + "warning", report["stages"]["rawDcr"]["status"] + ) + self.assertIn( + "InvalidStream", report["stages"]["rawDcr"]["detail"], ) self.assertEqual( - "not-run", report["stages"]["parserDcr"]["status"] + "blocked", report["stages"]["rawIngestion"]["status"] + ) + self.assertEqual( + "passed", report["stages"]["parserDcr"]["status"] + ) + self.assertEqual( + "passed", report["stages"]["parserIngestion"]["status"] + ) + self.assertEqual( + "blocked", + report["stages"]["qtpExecution"]["status"], + ) + self.assertEqual( + "Skipped because QTP source replay failed.", + report["stages"]["qtpExecution"]["detail"], + ) + self.assertEqual( + "blocked", report["stages"]["parity"]["status"] + ) + self.assertEqual( + "Skipped because QTP source replay failed.", + report["stages"]["parity"]["detail"], + ) + output = io.StringIO() + with redirect_stdout(output): + WORKFLOW.emit_annotations(report) + self.assertIn( + "::warning title=ASIM rawDcr warning::", + output.getvalue(), ) - self.assertEqual("", report["stages"]["parserDcr"]["detail"]) + self.assertIn("InvalidStream", output.getvalue()) def test_target_visibility_failure_marks_parser_ingestion_failed(self): validation = self.base_validation()