From 165399c66f127903b95922acaa08012e209b67d1 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <> Date: Fri, 23 Oct 2026 11:09:51 -0700 Subject: [PATCH 1/2] Canonicalize timestamps and persist replay rows Normalize returned TimeGenerated values through Kusto ticks and retain observed optional raw-replay schemas, rows, and timestamp counts in validation artifacts. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: f180e6f0-55c0-4d78-8707-7ed0bb43c8de --- .../ingestTimeParser/README.md | 5 + .../asim_ingestion/azure/client.py | 11 +- .../asim_ingestion/azure/common.py | 84 +++++++++- .../asim_ingestion/azure/runner.py | 59 ++++++- .../asim_ingestion/runtime/queries.py | 16 +- .../tests/test_asim_ingestion_harness.py | 158 ++++++++++++++++++ .../tests/test_asim_ingestion_validation.py | 50 ++++++ 7 files changed, 362 insertions(+), 21 deletions(-) diff --git a/.script/tests/asimParsersTest/ingestTimeParser/README.md b/.script/tests/asimParsersTest/ingestTimeParser/README.md index 0f54cf5ef92..9d9fa538ab5 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/README.md +++ b/.script/tests/asimParsersTest/ingestTimeParser/README.md @@ -62,6 +62,11 @@ 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. +`qtpSourceReplay` persists the latest `observedSchema`, `observedRows`, and +`observedCountsByTimestamp` from raw-table visibility checks. These remain +available in `validation-result.json` when visibility detects unexpected, +duplicate, or incomplete rows. + QTP parity stores the complete `qtp_schema`, `qtp_rows`, `itp_schema`, and `itp_rows` under `outputValidation`. It derives `intersect_schema` from shared column names, excluding the platform-managed `Type` column, and records both diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/client.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/client.py index 304cf808ce8..176f785960d 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/client.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/client.py @@ -511,7 +511,10 @@ def wait_for_exact_rows( raise HarnessError( stage, "Exact timestamp query returned unexpected timestamps.", - details={"unexpected": unexpected}, + details={ + "unexpected": unexpected, + "observedQueryTable": query_table, + }, ) observed_counts = { timestamp: raw_observed_counts.get(timestamp, 0) @@ -530,7 +533,10 @@ def wait_for_exact_rows( stage, "Exact timestamp slice contains more rows than " "expected for one or more timestamps.", - details={"overages": overages}, + details={ + "overages": overages, + "observedQueryTable": query_table, + }, ) if observed_counts == expected_counts: if stable_once: @@ -546,6 +552,7 @@ def wait_for_exact_rows( details={ "expectedCounts": expected_counts, "observedCounts": observed_counts, + "observedQueryTable": query_table, }, ) delay = self.poll_delays[ diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/common.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/common.py index 0c634ca207b..916c8084300 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/common.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/common.py @@ -17,6 +17,11 @@ import requests import yaml +from asim_ingestion.fixtures.timestamps import ( + parse_kusto_datetime_ticks, + ticks_to_kusto_datetime, +) + HARNESS_RESULT_VERSION = 1 SNAPSHOT_VERSION = 1 @@ -41,10 +46,6 @@ IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]*$") RESOURCE_NAME_RE = re.compile(r"[^a-z0-9-]+") -KUSTO_TIMESTAMP_RE = re.compile( - r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{7}Z$" -) - REPLAY_EXCLUDED_COLUMNS = { "_BilledSize", "_IsBillable", @@ -178,11 +179,14 @@ def _require_identifier(value: str, field: str) -> str: def _validate_timestamp(value: str) -> str: - if not KUSTO_TIMESTAMP_RE.fullmatch(value): - raise ValueError( - f"Expected exact Kusto timestamp with seven digits: {value}" + try: + return ticks_to_kusto_datetime( + parse_kusto_datetime_ticks(value) ) - return value + except (TypeError, ValueError) as exc: + raise ValueError( + f"Expected a valid Kusto timestamp: {value}" + ) from exc def _compact_json_bytes(value: Any) -> bytes: @@ -410,6 +414,70 @@ def _query_timestamp_counts( return dict(counts) +def build_observed_query_rows( + query_table: Mapping[str, Any], + *, + stage: str, +) -> 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( + stage, "Log Analytics result must contain columns and rows." + ) + schema: list[dict[str, str]] = [] + column_names: list[str] = [] + for column in columns: + if not isinstance(column, Mapping): + raise HarnessError( + stage, "Invalid Log Analytics column metadata." + ) + name = column.get("name") + column_type = column.get("type") + if not isinstance(name, str) or not isinstance(column_type, str): + raise HarnessError( + stage, "Invalid Log Analytics column metadata." + ) + if name in column_names: + raise HarnessError( + stage, + f"Log Analytics returned duplicate column {name}.", + ) + column_names.append(name) + schema.append( + { + "name": name, + "type": _normalize_dcr_type(column_type), + } + ) + type_by_name = { + column["name"]: column["type"] for column in schema + } + rows: list[dict[str, Any]] = [] + for raw_row in raw_rows: + if ( + not isinstance(raw_row, list) + or len(raw_row) != len(column_names) + ): + raise HarnessError(stage, "Invalid Log Analytics row shape.") + row = { + name: _json_safe_value(value, type_by_name[name]) + for name, value in zip(column_names, raw_row) + } + timestamp = row.get("TimeGenerated") + if isinstance(timestamp, str): + row["TimeGenerated"] = _validate_timestamp(timestamp) + rows.append(row) + return { + "schema": schema, + "rows": rows, + "countsByTimestamp": _query_timestamp_counts( + query_table, + stage, + ), + } + + def build_upload_batches( rows: Sequence[Mapping[str, Any]], max_bytes: int = 1_000_000, diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py index 9f753a135a8..7901f3add62 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py @@ -14,21 +14,43 @@ _validate_timestamp, _write_json_no_clobber, build_fixture_snapshot, + build_observed_query_rows, build_resource_names, count_rows_by_timestamp, ) def _error_payload(exc: HarnessError) -> dict[str, Any]: + details = exc.details + if isinstance(details, Mapping) and "observedQueryTable" in details: + details = dict(details) + details.pop("observedQueryTable", None) return { "type": type(exc).__name__, "stage": exc.stage, "message": str(exc), "statusCode": exc.status_code, - "details": exc.details, + "details": details, } +def _record_qtp_observation( + qtp_replay: dict[str, Any], + query_table: Any, +) -> None: + if not isinstance(query_table, Mapping): + return + observed = build_observed_query_rows( + query_table, + stage="raw-visibility", + ) + qtp_replay["observedSchema"] = observed["schema"] + qtp_replay["observedRows"] = observed["rows"] + qtp_replay["observedCountsByTimestamp"] = observed[ + "countsByTimestamp" + ] + + def _output_failure( output_validation: Mapping[str, Any], ) -> HarnessError | None: @@ -143,6 +165,9 @@ def run( ), "inputStream": input_stream or None, "error": None, + "observedSchema": [], + "observedRows": [], + "observedCountsByTimestamp": {}, }, "cleanup": cleanup, "telemetry": { @@ -288,9 +313,7 @@ def run( output_stream=input_stream, ) result["rawDcr"] = raw_handle.to_dict() - raw_counts = count_rows_by_timestamp( - rows - ) + raw_counts = count_rows_by_timestamp(rows) with self._stage("raw-upload"): result["rawUpload"] = self.upload_rows( raw_handle, @@ -299,14 +322,32 @@ def run( expected_counts_by_timestamp=raw_counts, ) with self._stage("raw-visibility"): - self.wait_for_exact_rows( - source_table, - timestamps, - raw_counts, - stage="raw-visibility", + observed_query_table = ( + self.wait_for_exact_rows( + source_table, + timestamps, + raw_counts, + stage="raw-visibility", + ) ) + _record_qtp_observation( + qtp_replay, + observed_query_table, + ) qtp_replay["status"] = "passed" except HarnessError as exc: + details = exc.details + if isinstance(details, Mapping): + _record_qtp_observation( + qtp_replay, + details.get("observedQueryTable"), + ) + cleaned_details = dict(details) + cleaned_details.pop( + "observedQueryTable", + None, + ) + exc.details = cleaned_details qtp_replay["status"] = "warning" qtp_replay["error"] = _error_payload(exc) diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/queries.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/queries.py index 335347aa34b..a323b86ef93 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/queries.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/queries.py @@ -9,7 +9,10 @@ import yaml -from asim_ingestion.fixtures.timestamps import parse_kusto_datetime_ticks +from asim_ingestion.fixtures.timestamps import ( + parse_kusto_datetime_ticks, + ticks_to_kusto_datetime, +) from asim_ingestion.parser.kql import split_kql_statements, split_pipeline MAX_MISMATCH_EXAMPLES = 10 @@ -253,11 +256,20 @@ def _parse_filter_counts( timestamps: Sequence[str], ) -> dict[str, int]: _, rows = _normalize_query_table(query_table, "filter") - expected_timestamps = set(timestamps) + expected_timestamps = { + ticks_to_kusto_datetime( + parse_kusto_datetime_ticks(str(timestamp)) + ) + for timestamp in timestamps + } counts = {timestamp: 0 for timestamp in expected_timestamps} for row in rows: timestamp = row.get("TimeGenerated") count = row.get("_AsimSelectedCount") + if isinstance(timestamp, str): + timestamp = ticks_to_kusto_datetime( + parse_kusto_datetime_ticks(timestamp) + ) if not isinstance(timestamp, str) or timestamp not in counts: raise ValidationError( "filter", 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 83081fe738a..fd7ba82f55e 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py @@ -319,6 +319,36 @@ def test_wait_for_rows_polls_until_exact_count(self): self.assertEqual(1, len(table["rows"])) self.assertEqual([1, 5], clock.sleeps) + def test_wait_for_rows_canonicalizes_returned_timestamp(self): + harness, _, clock = self.harness( + [ + FakeResponse( + 200, + query_payload( + [{"name": "TimeGenerated", "type": "datetime"}], + [["2026-10-16T19:55:00.123456Z"]], + ), + ), + FakeResponse( + 200, + query_payload( + [{"name": "TimeGenerated", "type": "datetime"}], + [["2026-10-16T19:55:00.123456Z"]], + ), + ), + ] + ) + + table = harness.wait_for_exact_rows( + "Contoso_CL", + ["2026-10-16T19:55:00.1234560Z"], + {"2026-10-16T19:55:00.1234560Z": 1}, + stage="raw-visibility", + ) + + self.assertEqual(1, len(table["rows"])) + self.assertEqual([5], clock.sleeps) + def test_wait_for_rows_fails_on_contamination(self): harness, _, _ = self.harness( [ @@ -818,6 +848,20 @@ def test_qtp_source_replay_uses_itp_input_stream(self): "Microsoft-SentinelContoso", result["rawDcr"]["outputStream"], ) + self.assertEqual( + [{"name": "TimeGenerated", "type": "datetime"}], + result["qtpSourceReplay"]["observedSchema"], + ) + self.assertEqual( + [{"TimeGenerated": timestamp}], + result["qtpSourceReplay"]["observedRows"], + ) + self.assertEqual( + {timestamp: 1}, + result["qtpSourceReplay"][ + "observedCountsByTimestamp" + ], + ) self.assertEqual(1, validation_runner.output_calls) self.assertEqual(1, validation_runner.parity_calls) @@ -914,6 +958,120 @@ def test_qtp_raw_dcr_failure_warns_and_blocks_parity(self): self.assertEqual(1, validation_runner.output_calls) self.assertEqual(0, validation_runner.parity_calls) + def test_qtp_raw_visibility_warning_persists_rows(self): + timestamp = "2026-10-16T19:55:00.1234567Z" + target_result = query_payload( + [{"name": "TimeGenerated", "type": "datetime"}], + [[timestamp]], + ) + duplicate_source_result = query_payload( + [ + {"name": "TimeGenerated", "type": "datetime"}, + {"name": "Value", "type": "string"}, + ], + [ + [timestamp, "one"], + [timestamp, "one"], + ], + ) + 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, target_result), + FakeResponse(200, target_result), + FakeResponse(201), # raw DCR PUT + FakeResponse( + 200, + { + "id": "/subscriptions/s/raw", + "properties": { + "provisioningState": "Succeeded", + "immutableId": "raw-immutable", + }, + }, + ), + FakeResponse(204), # raw upload + FakeResponse(200, duplicate_source_result), + FakeResponse(204), # raw cleanup + FakeResponse(204), # parser cleanup + ] + ) + prepared = { + "targetId": "parser-a", + "sourceRawTable": "Contoso", + "rowCount": 1, + "schema": [ + {"name": "TimeGenerated", "type": "datetime"}, + {"name": "Value", "type": "string"}, + ], + "namespace": { + "pullRequest": "123", + "runId": "456", + "runAttempt": 1, + "timestampValues": [timestamp], + }, + "rows": [ + { + "TimeGenerated": timestamp, + "Value": "one", + } + ], + } + 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"}, + ) + + replay = result["qtpSourceReplay"] + self.assertEqual("completed", result["status"]) + self.assertEqual("warning", replay["status"]) + self.assertEqual("raw-visibility", replay["error"]["stage"]) + self.assertNotIn( + "observedQueryTable", + replay["error"]["details"], + ) + self.assertEqual(2, len(replay["observedRows"])) + self.assertEqual( + {timestamp: 2}, + replay["observedCountsByTimestamp"], + ) + 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 = { 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 9f1d74db086..90f6c41d089 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_validation.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_validation.py @@ -246,6 +246,56 @@ def test_selective_filter_reports_selected_and_rejected_rows(self): ) self.assertNotIn("materialize(Contoso_CL", filter_query) + def test_filter_counts_canonicalize_trailing_zero_timestamp(self): + client = FakeQueryClient( + [ + ( + "filter", + query_table( + [ + ("TimeGenerated", "datetime"), + ("_AsimSelectedCount", "long"), + ], + [ + { + "TimeGenerated": + "2026-10-19T20:00:00.123456Z", + "_AsimSelectedCount": 1, + } + ], + ), + ) + ] + ) + snapshot = { + "timestampValues": [ + "2026-10-19T20:00:00.1234560Z" + ], + "schema": [ + {"name": "TimeGenerated", "type": "datetime"} + ], + "rows": [ + { + "TimeGenerated": + "2026-10-19T20:00:00.1234560Z" + } + ], + } + + result = self.validator().evaluate_filter( + client, + snapshot, + { + "SourceRawTable": "Contoso_CL", + "ParserFilterQuery": "source", + }, + ) + + self.assertEqual( + {"2026-10-19T20:00:00.1234560Z": 1}, + result["selectedCountsByTimestamp"], + ) + def test_selective_filter_requires_rejected_fixture_row(self): client = FakeQueryClient( [ From 212682f204ac0ca4110cfd2c03be47a0ad1d7db4 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <> Date: Fri, 23 Oct 2026 11:14:28 -0700 Subject: [PATCH 2/2] Fail ASIM validation after raw DCR creation Keep raw table and DCR establishment warning-only, but fail the target when optional raw ingestion, QTP execution, or parity fails after the raw DCR is available. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: f180e6f0-55c0-4d78-8707-7ed0bb43c8de --- .../ingestTimeParser/README.md | 7 +- .../asim_ingestion/azure/runner.py | 120 +++++++++++------- .../asim_ingestion/workflow/reporting.py | 35 +---- .../tests/test_asim_ingestion_harness.py | 29 +++-- .../tests/test_asim_ingestion_workflow.py | 72 ++++++++++- 5 files changed, 165 insertions(+), 98 deletions(-) diff --git a/.script/tests/asimParsersTest/ingestTimeParser/README.md b/.script/tests/asimParsersTest/ingestTimeParser/README.md index 9d9fa538ab5..673177f83aa 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/README.md +++ b/.script/tests/asimParsersTest/ingestTimeParser/README.md @@ -58,9 +58,10 @@ discover target ``` 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. +optional. If its table or DCR cannot be established, the report records a +warning and blocks QTP/parity without failing the required parser validation +stages. Once the raw DCR exists, upload or visibility failures fail the target; +QTP execution and parity also fail the target when they run and fail. `qtpSourceReplay` persists the latest `observedSchema`, `observedRows`, and `observedCountsByTimestamp` from raw-table visibility checks. These remain diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py index 7901f3add62..fdeefb63177 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py @@ -51,6 +51,25 @@ def _record_qtp_observation( ] +def _block_optional_parity( + output_validation: Mapping[str, Any], + *, + detail: str, + qtp_replay: Mapping[str, Any], +) -> dict[str, Any]: + blocked = dict(output_validation) + blocked["queryTimeExecution"] = { + "status": "blocked", + "detail": detail, + } + blocked["parity"] = { + "status": "blocked", + "detail": detail, + } + blocked["qtpSourceReplay"] = qtp_replay + return blocked + + def _output_failure( output_validation: Mapping[str, Any], ) -> HarnessError | None: @@ -312,44 +331,60 @@ def run( transform_kql="source", output_stream=input_stream, ) + except HarnessError as exc: + qtp_replay["status"] = "warning" + qtp_replay["error"] = _error_payload(exc) + else: 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"): - observed_query_table = ( - self.wait_for_exact_rows( - source_table, - timestamps, - raw_counts, - stage="raw-visibility", + try: + 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"): + observed_query_table = ( + self.wait_for_exact_rows( + source_table, + timestamps, + raw_counts, + stage="raw-visibility", + ) ) - ) - _record_qtp_observation( - qtp_replay, - observed_query_table, - ) - qtp_replay["status"] = "passed" - except HarnessError as exc: - details = exc.details - if isinstance(details, Mapping): _record_qtp_observation( qtp_replay, - details.get("observedQueryTable"), + observed_query_table, ) - cleaned_details = dict(details) - cleaned_details.pop( - "observedQueryTable", - None, + qtp_replay["status"] = "passed" + except HarnessError as exc: + details = exc.details + if isinstance(details, Mapping): + _record_qtp_observation( + qtp_replay, + details.get("observedQueryTable"), + ) + cleaned_details = dict(details) + cleaned_details.pop( + "observedQueryTable", + None, + ) + exc.details = cleaned_details + qtp_replay["status"] = "failed" + qtp_replay["error"] = _error_payload(exc) + result["outputValidation"] = ( + _block_optional_parity( + result["outputValidation"], + detail=( + "Blocked by QTP raw ingestion " + "failure." + ), + qtp_replay=qtp_replay, + ) ) - exc.details = cleaned_details - qtp_replay["status"] = "warning" - qtp_replay["error"] = _error_payload(exc) + raise if ( validation_runner is not None @@ -358,22 +393,13 @@ def run( 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 + output_validation = _block_optional_parity( + result["outputValidation"], + detail=( + "Skipped because QTP raw DCR " + "creation failed." + ), + qtp_replay=qtp_replay, ) else: parity_validation = dict( diff --git a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/workflow/reporting.py b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/workflow/reporting.py index d1c71eea558..feab509a38e 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/workflow/reporting.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/workflow/reporting.py @@ -326,11 +326,6 @@ def build_report( 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 = ( not parser_dcr_value and error_stage in {"dcr-create", "dcr-provisioning"} @@ -359,32 +354,12 @@ def build_report( 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( ( - "warning" - if qtp_raw_ingestion_warning + "failed" + if qtp_replay_status == "failed" else "blocked" if qtp_replay_status == "warning" - else "failed" - if raw_ingestion_failed else "passed" if validation.get("rawUpload") else "not-applicable" @@ -393,11 +368,9 @@ def build_report( ), ( qtp_replay_detail - if qtp_raw_ingestion_warning - else "Blocked by QTP raw DCR warning." + if qtp_replay_status == "failed" + else "Blocked by QTP raw DCR creation warning." if qtp_replay_status == "warning" - else error_detail - if raw_ingestion_failed else "" ), { 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 fd7ba82f55e..af2c134a1f3 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_harness.py @@ -958,7 +958,7 @@ def test_qtp_raw_dcr_failure_warns_and_blocks_parity(self): self.assertEqual(1, validation_runner.output_calls) self.assertEqual(0, validation_runner.parity_calls) - def test_qtp_raw_visibility_warning_persists_rows(self): + def test_qtp_raw_visibility_failure_fails_and_persists_rows(self): timestamp = "2026-10-16T19:55:00.1234567Z" target_result = query_payload( [{"name": "TimeGenerated", "type": "datetime"}], @@ -1038,17 +1038,20 @@ def test_qtp_raw_visibility_warning_persists_rows(self): 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"}, - ) + with self.assertRaisesRegex( + HARNESS.HarnessError, + "more rows than expected", + ): + harness.run( + prepared, + parser, + snapshot_output=Path(directory) / "snapshot.json", + validation_runner=validation_runner, + query_time_document={"ParserQuery": "Contoso"}, + ) - replay = result["qtpSourceReplay"] - self.assertEqual("completed", result["status"]) - self.assertEqual("warning", replay["status"]) + replay = harness.last_result["qtpSourceReplay"] + self.assertEqual("failed", replay["status"]) self.assertEqual("raw-visibility", replay["error"]["stage"]) self.assertNotIn( "observedQueryTable", @@ -1061,13 +1064,13 @@ def test_qtp_raw_visibility_warning_persists_rows(self): ) self.assertEqual( "blocked", - result["outputValidation"][ + harness.last_result["outputValidation"][ "queryTimeExecution" ]["status"], ) self.assertEqual( "blocked", - result["outputValidation"]["parity"]["status"], + harness.last_result["outputValidation"]["parity"]["status"], ) self.assertEqual(1, validation_runner.output_calls) self.assertEqual(0, validation_runner.parity_calls) 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 db507edff90..3072c4beb4b 100644 --- a/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_workflow.py +++ b/.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_workflow.py @@ -817,11 +817,15 @@ def test_qtp_raw_dcr_failure_warns_without_failing_parser(self): } validation["outputValidation"]["queryTimeExecution"] = { "status": "blocked", - "detail": "Skipped because QTP source replay failed.", + "detail": ( + "Skipped because QTP raw DCR creation failed." + ), } validation["outputValidation"]["parity"] = { "status": "blocked", - "detail": "Skipped because QTP source replay failed.", + "detail": ( + "Skipped because QTP raw DCR creation failed." + ), } report = self.build(validation) @@ -849,14 +853,14 @@ def test_qtp_raw_dcr_failure_warns_without_failing_parser(self): report["stages"]["qtpExecution"]["status"], ) self.assertEqual( - "Skipped because QTP source replay failed.", + "Skipped because QTP raw DCR creation failed.", report["stages"]["qtpExecution"]["detail"], ) self.assertEqual( "blocked", report["stages"]["parity"]["status"] ) self.assertEqual( - "Skipped because QTP source replay failed.", + "Skipped because QTP raw DCR creation failed.", report["stages"]["parity"]["detail"], ) output = io.StringIO() @@ -868,6 +872,66 @@ def test_qtp_raw_dcr_failure_warns_without_failing_parser(self): ) self.assertIn("InvalidStream", output.getvalue()) + def test_qtp_raw_ingestion_failure_fails_overall(self): + validation = self.base_validation() + validation["status"] = "failed" + validation["rawDcr"] = {"resourceId": "/raw"} + validation["rawUpload"] = [{"status": "accepted"}] + validation["qtpSourceReplay"] = { + "status": "failed", + "inputStream": "Custom-Contoso_CL", + "error": { + "stage": "raw-visibility", + "message": "Exact timestamp rows were duplicated.", + }, + "observedSchema": [ + {"name": "TimeGenerated", "type": "datetime"} + ], + "observedRows": [ + {"TimeGenerated": "timestamp"}, + {"TimeGenerated": "timestamp"}, + ], + "observedCountsByTimestamp": {"timestamp": 2}, + } + validation["outputValidation"]["queryTimeExecution"] = { + "status": "blocked", + "detail": "Blocked by QTP raw ingestion failure.", + } + validation["outputValidation"]["parity"] = { + "status": "blocked", + "detail": "Blocked by QTP raw ingestion failure.", + } + validation["error"] = { + "stage": "raw-visibility", + "message": "Exact timestamp rows were duplicated.", + } + + report = self.build(validation) + + self.assertEqual("failed", report["status"]) + self.assertTrue(report["requiredFailure"]) + self.assertEqual( + "passed", report["stages"]["rawDcr"]["status"] + ) + self.assertEqual( + "failed", report["stages"]["rawIngestion"]["status"] + ) + self.assertEqual( + "blocked", + report["stages"]["qtpExecution"]["status"], + ) + self.assertEqual( + "blocked", report["stages"]["parity"]["status"] + ) + self.assertEqual( + 2, + len( + report["stages"]["rawIngestion"]["evidence"][ + "qtpSourceReplay" + ]["observedRows"] + ), + ) + def test_target_visibility_failure_marks_parser_ingestion_failed(self): validation = self.base_validation() validation["status"] = "failed"