[ASIM] Fix ITP workflow regarding ASimDnsActivityLogs ASimDataTester with datetime/string columns

.script/tests/asimParsersTest/ingestTimeParser/README.md

@@ -63,6 +63,13 @@ 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.
 
+Before ASIM data testing, the `ASimDnsActivityLogs` table slice normalizes
+`ThreatFirstReportedTime` and `ThreatLastReportedTime` to `datetime` and
+`ThreatOriginalRiskLevel` to `string`. This accommodates the documented
+physical-table/logical-ASIM type discrepancy, preferring the corrected `_d`
+and `_s` columns and falling back to casts of the legacy columns. This
+normalization affects only the data-tester input, not ingestion or QTP parity.
+
 Static validation requires `ParserFilterQuery` to contain a `where` stage when
 `ParserQuery` contains one. This is a presence check only; it does not require
 all parser `where` stages to be copied into the filter query.

.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/queries.py

@@ -453,6 +453,23 @@ def build_asim_tester_query(
     )
     schema_literal = _quote_kql_string(schema_name)
     if tester_name == "ASimDataTester":
+        if target_table == "ASimDnsActivityLogs":
+            # ASimDnsActivityLogs has a type discrepancy with logical ASIM:
+            # these timestamps are string and the original risk level is int.
+            # Prefer the corrected _d/_s columns before casting legacy values.
+            # https://learn.microsoft.com/en-us/azure/azure-monitor/reference/tables/asimdnsactivitylogs
+            target_expression += (
+                "\n| extend "
+                "ThreatFirstReportedTime = coalesce("
+                'column_ifexists("ThreatFirstReportedTime_d", datetime(null)), '
+                'todatetime(column_ifexists("ThreatFirstReportedTime", ""))), '
+                "ThreatLastReportedTime = coalesce("
+                'column_ifexists("ThreatLastReportedTime_d", datetime(null)), '
+                'todatetime(column_ifexists("ThreatLastReportedTime", ""))), '
+                "ThreatOriginalRiskLevel = coalesce("
+                'column_ifexists("ThreatOriginalRiskLevel_s", ""), '
+                'tostring(column_ifexists("ThreatOriginalRiskLevel", int(null))))'
+            )
         signature = "T:(*), selected_schema:string=''"
         invocation = (
             "_AsimValidationTargetRows\n"

.script/tests/asimParsersTest/ingestTimeParser/tests/test_asim_ingestion_validation.py

@@ -53,6 +53,62 @@ def query_kql(self, query, *, stage="query"):
 
 
 class QueryConstructionTests(unittest.TestCase):
+    def test_dns_data_tester_normalizes_physical_threat_columns(self):
+        query = VALIDATION.build_asim_tester_query(
+            tester_name="ASimDataTester",
+            tester_query="T | project Result = 'test'",
+            target_table="ASimDnsActivityLogs",
+            timestamps=[TIMESTAMP_1],
+            schema_name="Dns",
+        )
+
+        target_query, tester_query = query.split(
+            "let _AsimValidationTester", 1
+        )
+        self.assertIn(f"datetime({TIMESTAMP_1})", target_query)
+        for field in (
+            "ThreatFirstReportedTime",
+            "ThreatLastReportedTime",
+        ):
+            self.assertIn(
+                f'{field} = coalesce(column_ifexists("{field}_d", '
+                f'datetime(null)), todatetime(column_ifexists('
+                f'"{field}", "")))',
+                target_query,
+            )
+        self.assertIn(
+            'ThreatOriginalRiskLevel = coalesce(column_ifexists('
+            '"ThreatOriginalRiskLevel_s", ""), tostring(column_ifexists('
+            '"ThreatOriginalRiskLevel", int(null))))',
+            target_query,
+        )
+        self.assertIn(
+            "| invoke _AsimValidationTester('Dns')", tester_query
+        )
+
+    def test_data_tester_leaves_other_target_tables_unchanged(self):
+        for table, schema in (
+            ("ASimAuthenticationEventLogs", "Authentication"),
+            ("CustomDns_CL", "Dns"),
+        ):
+            with self.subTest(table=table):
+                query = VALIDATION.build_asim_tester_query(
+                    tester_name="ASimDataTester",
+                    tester_query="T | project Result = 'test'",
+                    target_table=table,
+                    timestamps=[TIMESTAMP_1],
+                    schema_name=schema,
+                )
+
+                self.assertIn(
+                    f"{table}\n| where TimeGenerated in "
+                    f"(datetime({TIMESTAMP_1})));",
+                    query,
+                )
+                self.assertNotIn("ThreatFirstReportedTime", query)
+                self.assertNotIn("ThreatLastReportedTime", query)
+                self.assertNotIn("ThreatOriginalRiskLevel", query)
+
     def test_builds_typed_fixture_filter_query(self):
         query = VALIDATION.build_filter_count_query(
             "source",