Improve quickstart deployment validation

.github/schemas/quickstart-metadata.schema.json

@@ -0,0 +1,43 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://github.com/Azure/terraform/blob/master/.github/schemas/quickstart-metadata.schema.json",
+  "title": "Azure Terraform quickstart deployment metadata",
+  "type": "object",
+  "additionalProperties": false,
+  "required": [
+    "$schema",
+    "testResult"
+  ],
+  "properties": {
+    "$schema": {
+      "type": "string",
+      "const": "../../.github/schemas/quickstart-metadata.schema.json"
+    },
+    "testResult": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": [
+        "correlationId",
+        "timestamp"
+      ],
+      "properties": {
+        "correlationId": {
+          "type": "string",
+          "format": "uuid",
+          "description": "Value supplied to ARM_CORRELATION_REQUEST_ID for the validated Terraform apply."
+        },
+        "timestamp": {
+          "type": "string",
+          "format": "date-time",
+          "pattern": "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}(?:\\.\\d+)?Z$",
+          "description": "RFC 3339 UTC time ending in Z, captured immediately before the validated Terraform apply started."
+        },
+        "terraformVersion": {
+          "type": "string",
+          "pattern": "^\\d+\\.\\d+\\.\\d+(?:[-+][0-9A-Za-z.-]+)?$",
+          "description": "Terraform CLI version used for the apply."
+        }
+      }
+    }
+  }
+}

.github/scripts/test_validate_arm_correlation.py

@@ -0,0 +1,108 @@
+import importlib.util
+import json
+import unittest.mock
+import pathlib
+import unittest
+
+
+SCRIPT = pathlib.Path(__file__).with_name("validate-arm-correlation.py")
+SPEC = importlib.util.spec_from_file_location("arm_validator", SCRIPT)
+VALIDATOR = importlib.util.module_from_spec(SPEC)
+assert SPEC.loader is not None
+SPEC.loader.exec_module(VALIDATOR)
+
+
+class ValidateArmCorrelationTests(unittest.TestCase):
+    def test_build_body_uses_string_encoded_properties(self) -> None:
+        body = json.loads(
+            VALIDATOR.build_body(
+                "12345678-1234-1234-1234-1234567890ab",
+                "2026-10-06T18:00:00Z",
+            )
+        )
+
+        self.assertEqual("Requests", body["db"])
+        self.assertIsInstance(body["properties"], str)
+        self.assertEqual(
+            "12345678-1234-1234-1234-1234567890ab",
+            json.loads(body["properties"])["Parameters"]["cid"],
+        )
+
+    def test_primary_row_maps_columns(self) -> None:
+        response = {
+            "Tables": [
+                {
+                    "Columns": [
+                        {"ColumnName": "totalWrites"},
+                        {"ColumnName": "failedWrites"},
+                    ],
+                    "Rows": [[3, 0]],
+                }
+            ]
+        }
+
+        self.assertEqual(
+            {"totalWrites": 3, "failedWrites": 0},
+            VALIDATOR.primary_row(response),
+        )
+
+    @unittest.mock.patch.object(VALIDATOR.shutil, "which")
+    def test_azure_cli_finds_command(self, which: unittest.mock.Mock) -> None:
+        which.return_value = "/usr/bin/az"
+
+        with unittest.mock.patch.object(VALIDATOR.os, "name", "posix"):
+            self.assertEqual(["/usr/bin/az"], VALIDATOR.azure_cli_command())
+
+    def test_success_requires_writes_and_no_final_failures(self) -> None:
+        self.assertEqual(
+            [],
+            VALIDATOR.validate_summary(
+                {
+                    "totalWrites": 3,
+                    "completedWrites": 2,
+                    "acceptedAsyncWrites": 1,
+                    "failedWrites": 0,
+                }
+            ),
+        )
+
+    def test_final_failed_write_is_rejected(self) -> None:
+        errors = VALIDATOR.validate_summary(
+            {
+                "totalWrites": 3,
+                "completedWrites": 1,
+                "acceptedAsyncWrites": 1,
+                "failedWrites": 1,
+            }
+        )
+
+        self.assertTrue(any("final failed ARM write" in error for error in errors))
+
+    def test_summary_comparison_detects_ingestion_changes(self) -> None:
+        first = {
+            "totalWrites": 1,
+            "completedWrites": 1,
+            "acceptedAsyncWrites": 0,
+            "failedWrites": 0,
+        }
+        second = {
+            "totalWrites": 2,
+            "completedWrites": 1,
+            "acceptedAsyncWrites": 0,
+            "failedWrites": 1,
+        }
+
+        self.assertNotEqual(first, second)
+
+    def test_snapshot_requires_all_cluster_queries(self) -> None:
+        self.assertTrue(VALIDATOR.snapshot_is_complete([{"totalWrites": 1}], []))
+        self.assertFalse(
+            VALIDATOR.snapshot_is_complete(
+                [{"totalWrites": 1}],
+                ["https://armprodweu.westeurope.kusto.windows.net"],
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/scripts/test_validate_quickstart_metadata.py

@@ -0,0 +1,153 @@
+import importlib.util
+import datetime
+import json
+import pathlib
+import re
+import tempfile
+import unittest
+
+
+SCRIPT = pathlib.Path(__file__).with_name("validate-quickstart-metadata.py")
+SCHEMA = SCRIPT.parent.parent / "schemas" / "quickstart-metadata.schema.json"
+SPEC = importlib.util.spec_from_file_location("metadata_validator", SCRIPT)
+VALIDATOR = importlib.util.module_from_spec(SPEC)
+assert SPEC.loader is not None
+SPEC.loader.exec_module(VALIDATOR)
+
+
+class ValidateMetadataTests(unittest.TestCase):
+    def current_timestamp(self) -> str:
+        return (
+            datetime.datetime.now(datetime.timezone.utc)
+            .replace(microsecond=0)
+            .isoformat()
+            .replace("+00:00", "Z")
+        )
+
+    def write_metadata(self, data: dict) -> pathlib.Path:
+        temporary_directory = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary_directory.cleanup)
+        path = pathlib.Path(temporary_directory.name) / "metadata.json"
+        path.write_text(json.dumps(data), encoding="utf-8")
+        return path
+
+    def test_accepts_valid_metadata(self) -> None:
+        path = self.write_metadata(
+            {
+                "$schema": "../../.github/schemas/quickstart-metadata.schema.json",
+                "testResult": {
+                    "correlationId": "12345678-1234-1234-1234-1234567890ab",
+                    "timestamp": self.current_timestamp(),
+                    "terraformVersion": "1.13.0",
+                },
+            }
+        )
+
+        self.assertEqual([], VALIDATOR.validate_metadata(path))
+
+    def test_rejects_invalid_correlation_id(self) -> None:
+        path = self.write_metadata(
+            {
+                "$schema": "../../.github/schemas/quickstart-metadata.schema.json",
+                "testResult": {
+                    "correlationId": "not-a-guid",
+                    "timestamp": self.current_timestamp(),
+                },
+            }
+        )
+
+        self.assertTrue(
+            any("dashed UUID" in error for error in VALIDATOR.validate_metadata(path))
+        )
+
+    def test_requires_utc_timestamp(self) -> None:
+        path = self.write_metadata(
+            {
+                "$schema": "../../.github/schemas/quickstart-metadata.schema.json",
+                "testResult": {
+                    "correlationId": "12345678-1234-1234-1234-1234567890ab",
+                    "timestamp": "2026-08-18T17:00:00-07:00",
+                },
+            }
+        )
+
+        self.assertTrue(
+            any(
+                "timestamp" in error
+                for error in VALIDATOR.validate_metadata(path)
+            )
+        )
+
+    def test_schema_matches_python_utc_timestamp_contract(self) -> None:
+        schema = json.loads(SCHEMA.read_text(encoding="utf-8"))
+        pattern = schema["properties"]["testResult"]["properties"]["timestamp"][
+            "pattern"
+        ]
+
+        self.assertEqual(VALIDATOR.TIMESTAMP_RE.pattern, pattern)
+        self.assertIsNotNone(re.fullmatch(pattern, "2026-08-18T17:00:00Z"))
+        self.assertIsNotNone(re.fullmatch(pattern, "2026-08-18T17:00:00.123Z"))
+        for invalid in (
+            "2026-08-18T17:00:00-07:00",
+            "2026-08-18 17:00:00Z",
+            "20260818T170000Z",
+            "2026-08-18T17:00Z",
+            "2026-08-18T17:00:00,123Z",
+        ):
+            self.assertIsNone(re.fullmatch(pattern, invalid))
+            self.assertIsNone(VALIDATOR.parse_timestamp(invalid))
+
+    def test_reads_test_result_value(self) -> None:
+        data = {"testResult": {"correlationId": "example"}}
+
+        self.assertEqual(
+            "example",
+            VALIDATOR.test_result_value(data, "correlationId"),
+        )
+        self.assertIsNone(VALIDATOR.test_result_value({}, "correlationId"))
+
+    def test_rejects_stale_timestamp(self) -> None:
+        path = self.write_metadata(
+            {
+                "$schema": "../../.github/schemas/quickstart-metadata.schema.json",
+                "testResult": {
+                    "correlationId": "12345678-1234-1234-1234-1234567890ab",
+                    "timestamp": "2000-01-01T00:00:00Z",
+                },
+            }
+        )
+
+        self.assertTrue(
+            any("days old" in error for error in VALIDATOR.validate_metadata(path))
+        )
+
+    def test_allows_stale_timestamp_when_freshness_is_not_required(self) -> None:
+        path = self.write_metadata(
+            {
+                "$schema": "../../.github/schemas/quickstart-metadata.schema.json",
+                "testResult": {
+                    "correlationId": "12345678-1234-1234-1234-1234567890ab",
+                    "timestamp": "2000-01-01T00:00:00Z",
+                },
+            }
+        )
+
+        self.assertEqual(
+            [],
+            VALIDATOR.validate_metadata(path, enforce_freshness=False),
+        )
+
+    def test_recognizes_terraform_json_configuration(self) -> None:
+        self.assertTrue("main.tf.json".endswith(VALIDATOR.CONFIGURATION_SUFFIXES))
+        self.assertTrue(
+            "testing.auto.tfvars.json".endswith(VALIDATOR.CONFIGURATION_SUFFIXES)
+        )
+
+    def test_metadata_change_requires_metadata(self) -> None:
+        self.assertTrue(
+            VALIDATOR.should_require_metadata(False, False, True)
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()