diff --git a/src/aks-preview/HISTORY.rst b/src/aks-preview/HISTORY.rst
index e20aa860346..a80895e4413 100644
--- a/src/aks-preview/HISTORY.rst
+++ b/src/aks-preview/HISTORY.rst
@@ -12,6 +12,7 @@ To release a new version, please select a new version number (usually plus 1 to
 Pending
 +++++++
 * `az aks alert-config add`: Reject an empty `--name` before looking up existing configurations instead of reporting that it already exists.
+* `az aks enable-addons`, `az aks disable-addons` and `az aks update`: Keep Container Insights monitoring and container network log settings consistent with the legacy monitoring addon when updating clusters.
 
 22.0.0b7
 +++++++++
diff --git a/src/aks-preview/azext_aks_preview/custom.py b/src/aks-preview/azext_aks_preview/custom.py
index f925723cec4..78986c60b06 100644
--- a/src/aks-preview/azext_aks_preview/custom.py
+++ b/src/aks-preview/azext_aks_preview/custom.py
@@ -4085,6 +4085,19 @@ def _update_addons(cmd,  # pylint: disable=too-many-branches,too-many-statements
                     raise CLIError(f"The addon {addon} is not installed.")
             addon_profiles[addon].config = None
         addon_profiles[addon].enabled = enable
+        if addon == CONST_MONITORING_ADDON_NAME:
+            monitor_profile = getattr(instance, "azure_monitor_profile", None)
+            if getattr(monitor_profile, "container_insights", None) is not None:
+                # Reset canonical monitoring values along with the legacy addon config.
+                ContainerInsights = cmd.get_models(
+                    "ManagedClusterAzureMonitorProfileContainerInsights",
+                    resource_type=CUSTOM_MGMT_AKS_PREVIEW,
+                    operation_group="managed_clusters",
+                )
+                monitor_profile.container_insights = ContainerInsights(
+                    enabled=enable,
+                    log_analytics_workspace_resource_id=workspace_resource_id if enable else None,
+                )
 
     instance.addon_profiles = addon_profiles
 
diff --git a/src/aks-preview/azext_aks_preview/managed_cluster_decorator.py b/src/aks-preview/azext_aks_preview/managed_cluster_decorator.py
index a9a88b3c8e8..7fef93edcb7 100644
--- a/src/aks-preview/azext_aks_preview/managed_cluster_decorator.py
+++ b/src/aks-preview/azext_aks_preview/managed_cluster_decorator.py
@@ -6662,6 +6662,11 @@ def update_monitoring_profile_flow_logs(self, mc: ManagedCluster) -> ManagedClus
                     config = monitoring_addon_profile.config or {}
                     config["enableRetinaNetworkFlags"] = str(container_network_logs_enabled)
                     mc.addon_profiles[monitoring_addon_key].config = config
+            container_insights = getattr(mc.azure_monitor_profile, "container_insights", None)
+            if container_insights is not None:
+                container_insights.container_network_logs = (
+                    "Enabled" if container_network_logs_enabled else "Disabled"
+                )
 
         # When enabling CNL, the DCR must be updated to add the high-scale stream.
         # Set the postprocessing intermediate so that the update path calls ensure_container_insights.
diff --git a/src/aks-preview/azext_aks_preview/tests/latest/data/setup_proxy.sh b/src/aks-preview/azext_aks_preview/tests/latest/data/setup_proxy.sh
index 879a0988688..9d0765970ae 100644
--- a/src/aks-preview/azext_aks_preview/tests/latest/data/setup_proxy.sh
+++ b/src/aks-preview/azext_aks_preview/tests/latest/data/setup_proxy.sh
@@ -1,5 +1,5 @@
 #!/usr/bin/env bash
-set -x
+set -euxo pipefail
 
 echo "setting up"
 WORKDIR="${1:-$(mktemp -d)}"
@@ -7,22 +7,14 @@ echo "setting up ${WORKDIR}"
 
 pushd "$WORKDIR"
 
-apt update -y && apt install -y apt-transport-https curl gnupg make gcc < /dev/null
-
-# add diladele apt key
-wget -qO - https://packages.diladele.com/diladele_pub.asc | apt-key add -
-
-# add new repo
-tee /etc/apt/sources.list.d/squid413-ubuntu20.diladele.com.list <<EOF
-deb https://squid413-ubuntu20.diladele.com/ubuntu/ focal main
-EOF
-
-# and install
-apt-get update && apt-get install -y squid-common squid-openssl squidclient libecap3 libecap3-dev < /dev/null
+apt-get -o Acquire::Retries=3 update
+DEBIAN_FRONTEND=noninteractive apt-get install -y ca-certificates curl squid-openssl < /dev/null
 
 mkdir -p /var/lib/squid
 
-/usr/lib/squid/security_file_certgen -c -s /var/lib/squid/ssl_db -M 4MB || true
+if [ ! -d /var/lib/squid/ssl_db ]; then
+    /usr/lib/squid/security_file_certgen -c -s /var/lib/squid/ssl_db -M 4MB
+fi
 
 chown -R proxy:proxy /var/lib/squid
 
@@ -129,11 +121,14 @@ update-ca-certificates
 sed -i 's~http_access deny all~http_access allow all~' /etc/squid/squid.conf
 sed -i "s~http_port 3128~http_port $HOST:3128\nhttps_port $HOST:3129 tls-cert=/etc/squid/squidc.pem tls-key=/etc/squid/squidk.pem~" /etc/squid/squid.conf
 
+squid -k parse
 systemctl restart squid
-systemctl status squid
-
-# validation, fails VM creation if commands fail
-curl -fsSl -o /dev/null -w '%{http_code}\n' -x http://${HOST}:3128/ -I http://www.google.com
-curl -fsSl -o /dev/null -w '%{http_code}\n' -x http://${HOST}:3128/ -I https://www.google.com
-curl -fsSl -o /dev/null -w '%{http_code}\n' -x https://${HOST}:3129/ -I http://www.google.com
-curl -fsSl -o /dev/null -w '%{http_code}\n' -x https://${HOST}:3129/ -I https://www.google.com
+systemctl is-active --quiet squid
+
+# Fail cloud-init unless both proxy listeners can reach the required endpoints.
+for proxy in "http://${HOST}:3128/" "https://${HOST}:3129/"; do
+    for endpoint in "http://packages.microsoft.com/" "https://mcr.microsoft.com/v2/"; do
+        curl -fsSL --connect-timeout 10 --max-time 30 --retry 2 --retry-connrefused \
+            -o /dev/null -w '%{http_code}\n' -x "$proxy" "$endpoint"
+    done
+done
diff --git a/src/aks-preview/azext_aks_preview/tests/latest/test_aks_commands.py b/src/aks-preview/azext_aks_preview/tests/latest/test_aks_commands.py
index 1a0c323a217..e2d5332f6e5 100644
--- a/src/aks-preview/azext_aks_preview/tests/latest/test_aks_commands.py
+++ b/src/aks-preview/azext_aks_preview/tests/latest/test_aks_commands.py
@@ -119,6 +119,7 @@ def _is_transient_operation_conflict(ex):
         return (
             "Another operation is in progress" in message or
             "Operation is not allowed because there's an in-progress" in message or
+            "AKSOperationPreempted" in message or
             "in-progress PutExtensionAddonHandler.PUT operation" in message or
             "is in Updating state, please wait for it to succeed" in message or
             "ProvisioningState of extension: Updating" in message or
@@ -255,6 +256,10 @@ def _execute_with_transient_conflict_retry(self, command, expect_failure):
                 ):
                     show_command = self._build_show_command_for_already_existing_resource(command)
                     if show_command is not None:
+                        recovery_command = getattr(self, "_retried_create_recovery_command", None)
+                        if recovery_command and re.match(r"^aks\s+create\b", command.strip()):
+                            logging.warning("Resuming metrics configuration after a retried cluster create.")
+                            return self._execute_with_transient_conflict_retry(recovery_command, False)
                         logging.warning(
                             "Resource already exists after a retried create/add; the earlier "
                             "attempt's async operation likely already succeeded server-side. "
@@ -291,7 +296,10 @@ def _execute_with_transient_conflict_retry(self, command, expect_failure):
                             return show_result
                 if (
                     expect_failure or
-                    not self._is_transient_operation_conflict(ex) or
+                    not (
+                        self._is_transient_operation_conflict(ex) or
+                        self._is_private_dns_role_assignment_pending(command, ex)
+                    ) or
                     attempt == max_retries - 1
                 ):
                     raise
@@ -308,6 +316,48 @@ def _execute_with_transient_conflict_retry(self, command, expect_failure):
 
         raise AssertionError("unreachable")
 
+    def _cmd_with_retried_create_recovery(self, command, recovery_command, checks=None):
+        previous_command = getattr(self, "_retried_create_recovery_command", None)
+        self._retried_create_recovery_command = self._apply_kwargs(recovery_command)
+        try:
+            return self.cmd(command, checks=checks)
+        finally:
+            self._retried_create_recovery_command = previous_command
+
+    def _is_private_dns_role_assignment_pending(self, command, ex):
+        zone_id = getattr(self, "_private_dns_role_assignment_scope", None)
+        if not zone_id or not re.match(r"^aks\s+create\b", command.strip()):
+            return False
+        requested_zone = self._extract_cli_option(command, "--private-dns-zone")
+        message = str(ex).casefold()
+        return (
+            requested_zone is not None and requested_zone.casefold() == zone_id.casefold() and
+            "(resourcemissingpermissionerror)" in message and
+            f"resource {zone_id.casefold()}." in message and
+            "not allowed for action microsoft.network/privatednszones/read" in message
+        )
+
+    def _cmd_with_private_dns_role_assignment_retry(self, command, zone_id, checks=None):
+        # Only the caller that just granted this zone's role may retry propagation.
+        previous_scope = getattr(self, "_private_dns_role_assignment_scope", None)
+        self._private_dns_role_assignment_scope = zone_id
+        try:
+            return self.cmd(command, checks=checks)
+        finally:
+            self._private_dns_role_assignment_scope = previous_scope
+
+    def _reencrypt_kms_secrets(self):
+        if not self.is_live:
+            return
+        # Private-vault rotation requires every secret to use the current key first.
+        self.cmd(
+            "aks command invoke --resource-group={resource_group} --name={name} "
+            "--command \"bash -o pipefail -c "
+            "'kubectl get secrets --all-namespaces -o json | kubectl replace -f -'\" "
+            "--output json",
+            checks=[self.check("provisioningState", "Succeeded"), self.check("exitCode", 0)],
+        )
+
     def _refetch_settled_aks_result(self, resource_id, fallback_result):
         from azure.cli.testsdk.base import execute
 
@@ -399,6 +449,19 @@ def _cmd_with_retry(self, command, checks, expect_failure):
 
         return result
 
+    def _reimage_nodepool(self, resource_group, cluster_name, nodepool_name):
+        self.cmd(
+            f"aks nodepool upgrade --resource-group {resource_group} --cluster-name {cluster_name} "
+            f"--name {nodepool_name} --node-image-only --yes",
+            checks=[self.is_empty()],
+        )
+        # The upgradeNodeImageVersion action returns no resource body.
+        self.cmd(
+            f"aks nodepool show --resource-group {resource_group} --cluster-name {cluster_name} "
+            f"--name {nodepool_name}",
+            checks=[self.check("provisioningState", "Succeeded")],
+        )
+
     def _create_log_analytics_workspace(self, resource_group_location):
         workspace_name = self.create_random_name("clilaw", 16)
         workspace_location = (
@@ -494,8 +557,9 @@ def _cmd_or_skip_if_managed_system_unavailable(self, cmd, checks=None):
             return self.cmd(cmd, checks=checks)
         except Exception as ex:  # pylint: disable=broad-except
             message = str(ex)
-            if "ManagedSystem" in message and re.search(
-                r"not (?:whitelisted|allowed|enabled|supported|available|registered)",
+            if "managedsystem" in message.lower() and re.search(
+                r"not (?:whitelisted|allowed|enabled|supported|available|registered)"
+                r"|only supports? whitelisted subscriptions",
                 message,
                 re.IGNORECASE,
             ):
@@ -560,12 +624,12 @@ def _cmd_or_skip_if_os_sku_retired(self, cmd, os_sku, checks=None):
                 )
             raise
 
-    def _cmd_or_skip_if_feature_unavailable(self, cmd, feature_name, checks=None):
-        """Run a command that depends on a preview feature unlocked via `--aks-custom-headers`.
+    def _cmd_or_skip_if_feature_unavailable(self, cmd, feature_name, checks=None, feature_aliases=()):
+        """Run a command that requires a preview feature in the test environment.
 
         Some `AKSHTTPCustomFeatures` preview flags additionally require the test
         subscription to be on a service-side allowlist; the header alone is not
-        sufficient to unlock them (mirrors the verified behavior seen for the
+        sufficient to unlock them or make them available in every region (mirrors the behavior seen for the
         `ManagedSystem` agent pool mode preview). If the service rejects the request
         because this subscription isn't enrolled, skip with a precise reason instead
         of failing the test; any other failure (e.g. a real CLI/service regression)
@@ -575,17 +639,35 @@ def _cmd_or_skip_if_feature_unavailable(self, cmd, feature_name, checks=None):
             return self.cmd(cmd, checks=checks)
         except Exception as ex:  # pylint: disable=broad-except
             message = str(ex)
-            if feature_name.lower() in message.lower() and re.search(
-                r"not (?:whitelisted|allowed|enabled|supported|available|registered)",
+            feature_matches = any(
+                name.lower() in message.lower() for name in (feature_name, *feature_aliases)
+            )
+            if feature_matches and re.search(
+                r"not (?:yet )?(?:whitelisted|allowed|enabled|supported|available|registered)",
                 message,
                 re.IGNORECASE,
             ):
                 self.skipTest(
-                    f"This subscription is not enrolled for the {feature_name} preview "
-                    f"despite the aks-custom-headers override: {message}"
+                    f"The {feature_name} preview is unavailable in this test environment: {message}"
                 )
             raise
 
+    def _cmd_or_skip_if_region_unavailable(self, command, location, checks=None, vm_size=None):
+        try:
+            return self.cmd(command, checks=checks)
+        except (CLIError, HttpResponseError) as ex:
+            message = str(ex).lower()
+            expected_location = location.replace(" ", "").lower()
+            if f"'{expected_location}'" not in message:
+                raise
+            if vm_size is None:
+                unavailable = "(availabilityzonenotsupported)" in message
+            else:
+                unavailable = "(vmsizenotsupported)" in message and f"'{vm_size.lower()}'" in message
+            if unavailable:
+                self.skipTest(f"Required test capability is unavailable in {location}: {ex}")
+            raise
+
     def _get_lts_version(self, location):
         """Return the latest LTS version in the given location."""
         data = self.cmd(
@@ -928,8 +1010,9 @@ def test_aks_create_and_update_with_managed_nat_gateway_standardv2(
             "--nat-gateway-managed-outbound-ip-count 2 "
             "--ssh-key-value={ssh_key_value}"
         )
-        self.cmd(
+        self._cmd_or_skip_if_feature_unavailable(
             create_cmd,
+            "managedNATGatewayV2",
             checks=[
                 self.check("provisioningState", "Succeeded"),
                 self.check("networkProfile.outboundType", "managedNATGateway"),
@@ -9860,6 +9943,21 @@ def test_aks_create_update_cluster_node_hardening(
             checks=[self.is_empty()],
         )
 
+    def _wait_for_http_proxy(self):
+        script = (
+            "timeout 600 cloud-init status --wait && systemctl is-active --quiet squid && "
+            "curl -fsSL --connect-timeout 10 --max-time 30 --retry 2 --retry-connrefused "
+            "-x http://cli-proxy-vm:3128/ https://mcr.microsoft.com/v2/ -o /dev/null && "
+            "curl -fsSL --connect-timeout 10 --max-time 30 --retry 2 --retry-connrefused "
+            "-x https://cli-proxy-vm:3129/ https://mcr.microsoft.com/v2/ -o /dev/null && "
+            "echo AKS_PROXY_READY"
+        )
+        self.cmd(
+            "vm run-command invoke --resource-group={resource_group} --name cli-proxy-vm "
+            f'--command-id RunShellScript --scripts "{script}" -o json',
+            checks=[self.check("contains(join('', value[].message), `\"\\nAKS_PROXY_READY\\n\"`)", True)],
+        )
+
     def _setup_http_proxy_cluster(self, resource_group, _resource_group_location, aks_name):
         """Shared setup for the http-proxy tests below: create a VNet/subnet,
         a proxy VM (with cloud-init that stands up an actual HTTP(S) proxy),
@@ -9873,8 +9971,8 @@ def _setup_http_proxy_cluster(self, resource_group, _resource_group_location, ak
         sequentially in a single live test run. Factoring the common
         VNet/VM/initial-create steps into this helper lets each of those
         follow-on scenarios (update / disable / re-enable) run as its own
-        independent live test, each well under the 1-hour live-test budget,
-        without dropping any of the original coverage.
+        independent live test. Verify the proxy before starting the cluster
+        so broken bootstrap/egress does not consume the full test timeout.
         """
         self.kwargs.update(
             {
@@ -9910,7 +10008,7 @@ def _setup_http_proxy_cluster(self, resource_group, _resource_group_location, ak
         create_vm_cmd = 'vm create \
             --resource-group={resource_group} \
             --name=cli-proxy-vm \
-            --image Canonical:0001-com-ubuntu-server-focal:20_04-lts:latest \
+            --image Canonical:ubuntu-24_04-lts:server:latest \
             --ssh-key-values @{ssh_key_value} \
             --public-ip-address "" \
             --custom-data {custom_data_path} \
@@ -9931,6 +10029,7 @@ def _setup_http_proxy_cluster(self, resource_group, _resource_group_location, ak
         assert subnet_id is not None
 
         self.cmd(create_vm_cmd)
+        self._wait_for_http_proxy()
 
         self.kwargs.update(
             {
@@ -9966,8 +10065,7 @@ def _setup_http_proxy_cluster(self, resource_group, _resource_group_location, ak
     # scenario is split below into three focused live tests that share the
     # ``_setup_http_proxy_cluster`` helper above. This preserves the exact
     # same assertions/coverage (initial create, config update, disable,
-    # re-enable) while keeping each individual test comfortably within the
-    # live-test time budget and letting them run in parallel (separate
+    # re-enable) while letting them run in parallel (separate
     # per-test resource groups).
     #
     # this case relatively frequently requires updating the corresponding recording file after network/virtualnetwork
@@ -10236,8 +10334,9 @@ def test_aks_create_fqdn_subdomain(self, resource_group, resource_group_location
             "--enable-private-cluster --private-dns-zone={zone_id} --enable-managed-identity --assign-identity {identity_resource_id} "
             "--ssh-key-value={ssh_key_value}"
         )
-        self.cmd(
+        self._cmd_with_private_dns_role_assignment_retry(
             create_cmd,
+            zone_id,
             checks=[
                 self.exists("privateFqdn"),
                 self.exists("fqdnSubdomain"),
@@ -10923,7 +11022,9 @@ def test_aks_enable_utlra_ssd(self, resource_group, resource_group_location):
             "--node-vm-size Standard_D2s_v3 --zones 1 2 3 --enable-ultra-ssd "
             "--ssh-key-value={ssh_key_value}"
         )
-        self.cmd(create_cmd, checks=[self.check("provisioningState", "Succeeded")])
+        self._cmd_or_skip_if_region_unavailable(
+            create_cmd, resource_group_location, checks=[self.check("provisioningState", "Succeeded")]
+        )
 
         # delete
         self.cmd(
@@ -13770,6 +13871,7 @@ def test_aks_create_with_azurekeyvaultkms_private_key_vault(
             "--azure-keyvault-kms-key-vault-network-access=Private --azure-keyvault-kms-key-vault-resource-id {kv_resource_id} "
             "-o json"
         )
+        self._reencrypt_kms_secrets()
         self.cmd(
             update_cmd,
             checks=[
@@ -14093,6 +14195,7 @@ def test_aks_create_with_azurekeyvaultkms_private_cluster_v1_private_key_vault(
             "--azure-keyvault-kms-key-vault-network-access=Private --azure-keyvault-kms-key-vault-resource-id {kv_resource_id} "
             "-o json"
         )
+        self._reencrypt_kms_secrets()
         self.cmd(
             update_cmd,
             checks=[
@@ -15960,8 +16063,9 @@ def test_aks_availability_zones(self, resource_group, resource_group_location):
             "aks create --resource-group={resource_group} --name={name} "
             "--node-count=1 --ssh-key-value={ssh_key_value} --zones {zones}"
         )
-        self.cmd(
+        self._cmd_or_skip_if_region_unavailable(
             create_cmd,
+            resource_group_location,
             checks=[
                 self.check("provisioningState", "Succeeded"),
                 self.check("agentPoolProfiles[0].availabilityZones[0]", "1"),
@@ -16378,12 +16482,17 @@ def _cleanup_backup(self, vault_name):
 
         # 1. Disable immutability + soft-delete on the vault so that BIs
         #    with active recovery points can be force-deleted.
+        # Use the API that supports reversible soft delete, as AKS vault creation does.
         try:
             self.cmd(
-                "dataprotection backup-vault update "
-                "-g {backup_rg} --vault-name {vault_name} "
+                "resource update --resource-group {backup_rg} --name {vault_name} "
+                "--resource-type Microsoft.DataProtection/backupVaults --api-version 2025-07-01 "
                 "--set properties.securitySettings.immutabilitySettings.state=Disabled "
-                "properties.securitySettings.softDeleteSettings.state=Off"
+                "properties.securitySettings.softDeleteSettings.state=Off",
+                checks=[
+                    self.check("properties.securitySettings.immutabilitySettings.state", "Disabled"),
+                    self.check("properties.securitySettings.softDeleteSettings.state", "Off"),
+                ],
             )
         except Exception as ex:  # pylint: disable=broad-except
             logging.warning(
@@ -17347,19 +17456,26 @@ def test_aks_create_with_control_plane_metrics(
                 "node_vm_size": node_vm_size,
             }
         )
+        self.kwargs["amw_id"] = self._create_azure_monitor_workspace(resource_group_location)
 
         # create: --enable-azure-monitor-metrics + --enable-control-plane-metrics
         create_cmd = (
             "aks create --resource-group={resource_group} --name={name} --location={location} "
             "--ssh-key-value={ssh_key_value} --node-vm-size={node_vm_size} --enable-managed-identity "
-            "--enable-azure-monitor-metrics --enable-control-plane-metrics --output=json"
+            "--enable-azure-monitor-metrics --enable-control-plane-metrics "
+            "--azure-monitor-workspace-resource-id={amw_id} --output=json"
         )
         # NOTE: ``--enable-control-plane-metrics`` on create is intentionally deferred to a
         # postprocessing PUT (after DCRA creation) to avoid scheduling the CCP pod before its
         # DCRA exists. The create response may therefore reflect the pre-flip state; assert
         # the final state via ``aks show`` after the cluster settles.
-        self.cmd(
+        self._cmd_with_retried_create_recovery(
             create_cmd,
+            recovery_command=(
+                "aks update --resource-group={resource_group} --name={name} "
+                "--enable-azure-monitor-metrics --enable-control-plane-metrics "
+                "--azure-monitor-workspace-resource-id={amw_id}"
+            ),
             checks=[
                 self.check("provisioningState", "Succeeded"),
                 self.check("azureMonitorProfile.metrics.enabled", True),
@@ -21735,7 +21851,9 @@ def test_aks_create_acns_with_flow_logs(
                 self.check("networkProfile.advancedNetworking.observability.enabled", True),
                 self.check("networkProfile.advancedNetworking.security.enabled", True),
                 self.check("addonProfiles.omsagent.enabled", True),
-                self.check("addonProfiles.omsagent.config.enableRetinaNetworkFlags", "True"),
+                self.check(
+                    "contains(['True', 'true'], addonProfiles.omsagent.config.enableRetinaNetworkFlags)", True
+                ),
             ],
         ).get_output_in_json()
 
@@ -21760,7 +21878,9 @@ def test_aks_create_acns_with_flow_logs(
             disable_cmd,
             checks=[
                 self.check("provisioningState", "Succeeded"),
-                self.check("addonProfiles.omsagent.config.enableRetinaNetworkFlags", "False"),
+                self.check(
+                    "contains(['False', 'false'], addonProfiles.omsagent.config.enableRetinaNetworkFlags)", True
+                ),
             ],
         )
 
@@ -21770,7 +21890,9 @@ def test_aks_create_acns_with_flow_logs(
             enable_cmd_update,
             checks=[
                 self.check("provisioningState", "Succeeded"),
-                self.check("addonProfiles.omsagent.config.enableRetinaNetworkFlags", "True"),
+                self.check(
+                    "contains(['True', 'true'], addonProfiles.omsagent.config.enableRetinaNetworkFlags)", True
+                ),
             ],
         )
 
@@ -22802,8 +22924,7 @@ def test_aks_applicationloadbalancer_update(self, resource_group, resource_group
 
         # update (currently makes a PUT no-op)
         update_cmd = (
-            "aks applicationloadbalancer update --resource-group={resource_group} --name={aks_name} "
-            "--aks-custom-headers AKSHTTPCustomFeatures=Microsoft.ContainerService/ApplicationLoadBalancerPreview"
+            "aks applicationloadbalancer update --resource-group={resource_group} --name={aks_name}"
         )
 
         self.cmd(
@@ -24333,11 +24454,7 @@ def test_aks_network_isolated_cluster(self, resource_group, resource_group_locat
 
         # reimage the existing nodepool so nodes pick up the Cache-based bootstrap artifacts
         # while they still have outbound network access
-        reimage_nodepool_cmd = (
-            "aks nodepool upgrade --resource-group {resource_group} --cluster-name {aks_name_2} "
-            "--name nodepool1 --node-image-only --yes"
-        )
-        self.cmd(reimage_nodepool_cmd, checks=[self.check("provisioningState", "Succeeded")])
+        self._reimage_nodepool(resource_group, aks_name_2, "nodepool1")
 
         # wait for the cluster to settle after the reimage before removing outbound access
         wait_after_reimage_cmd = (
@@ -25003,6 +25120,7 @@ def test_aks_bastion_enable_update_disable(
         self._cmd_or_skip_if_feature_unavailable(
             enable_cmd,
             "ManagedBastion",
+            feature_aliases=("bastionProfile",),
             checks=[
                 self.check("provisioningState", "Succeeded"),
                 self.check("networkProfile.bastionProfile.enabled", True),
@@ -25097,6 +25215,7 @@ def test_aks_bastion_enable_disable_byo_public_ip(
         self._cmd_or_skip_if_feature_unavailable(
             enable_cmd,
             "ManagedBastion",
+            feature_aliases=("bastionProfile",),
             checks=[
                 self.check("provisioningState", "Succeeded"),
                 self.check("networkProfile.bastionProfile.enabled", True),
@@ -25923,9 +26042,6 @@ def test_aks_run_blue_green_upgrade(self, resource_group, resource_group_locatio
 
     @AllowLargeResponse()
     @AKSCustomResourceGroupPreparer(
-        # eastus is capacity-constrained for standard_dc16ads_cc_v5 (verified live:
-        # "SkuNotAvailable ... Following SKUs have failed for Capacity Restrictions");
-        # eastus2 has capacity for this confidential-compute SKU.
         random_name_length=17, name_prefix="clitest", location="eastus2"
     )
     def test_aks_jwtauthenticator_cmds(self, resource_group, resource_group_location):
@@ -25935,13 +26051,8 @@ def test_aks_jwtauthenticator_cmds(self, resource_group, resource_group_location
         aks_name = self.create_random_name('cliakstest', 16)
         jwt_auth_name = self.create_random_name('jwt', 10)
 
-        # The confidential-compute SKU below is only verified to have capacity in eastus2.
-        # A compliance-mandated AZURE_CLI_TEST_FORCE_RESOURCE_GROUP_LOCATION override always
-        # wins for the resource group's own location (by design, regardless of
-        # preserve_default_location), but an AKS cluster can be created in a different region
-        # than its resource group. Pin the cluster's --location explicitly to the
-        # capacity-verified region so this test keeps passing under a forced resource-group
-        # location instead of failing with SkuNotAvailable.
+        # Keep the confidential-compute requirement; subscription availability is checked
+        # explicitly rather than replacing this SKU with an ordinary VM.
         cluster_location = "eastus2"
 
         self.kwargs.update({
@@ -25959,9 +26070,10 @@ def test_aks_jwtauthenticator_cmds(self, resource_group, resource_group_location
             "--vm-set-type=VirtualMachines --node-count 1 --vm-size standard_dc16ads_cc_v5 "
             "--enable-managed-identity --ssh-key-value={ssh_key_value} "
         )
-        self.cmd(create_cmd, checks=[
-            self.check('provisioningState', 'Succeeded'),
-        ])
+        self._cmd_or_skip_if_region_unavailable(
+            create_cmd, cluster_location, vm_size="standard_dc16ads_cc_v5",
+            checks=[self.check('provisioningState', 'Succeeded')],
+        )
 
         add_jwt_cmd = (
             "aks jwtauthenticator add --resource-group={resource_group} --cluster-name={name} "
diff --git a/src/aks-preview/azext_aks_preview/tests/latest/test_aks_provisioning_retry.py b/src/aks-preview/azext_aks_preview/tests/latest/test_aks_provisioning_retry.py
index 550166a4c7c..be9fe9b6351 100644
--- a/src/aks-preview/azext_aks_preview/tests/latest/test_aks_provisioning_retry.py
+++ b/src/aks-preview/azext_aks_preview/tests/latest/test_aks_provisioning_retry.py
@@ -4,6 +4,7 @@
 # --------------------------------------------------------------------------------------------
 
 import json
+import inspect
 import os
 import tempfile
 import unittest
@@ -196,6 +197,7 @@ def test_retries_only_known_transient_conflicts(
         messages = [
             "Operation is not allowed: Another operation is in progress.",
             "Operation is not allowed because there's an in-progress update managed cluster operation",
+            "(AKSOperationPreempted) This operation has been preempted by another operation. Please retry later.",
             "Operation is not allowed: in-progress PutExtensionAddonHandler.PUT operation",
             "The managed cluster test is in Updating state, please wait for it to succeed.",
             "ProvisioningState of extension: Updating",
@@ -328,6 +330,46 @@ def test_keyvault_validation_error_reraises_when_retries_exhausted(
 
 
 class TestAlreadyExistsConflictHandling(AKSRetryTestCase):
+    @patch.dict(os.environ, {"AZURE_CLI_TEST_OPERATION_MAX_RETRIES": "3"})
+    @patch("time.sleep")
+    @patch("azure.cli.testsdk.base.execute")
+    def test_metrics_create_recovery_completes_configuration_not_just_show(self, execute, _sleep):
+        instance = self._make_instance()
+        instance.cmd = lambda command, checks: instance._cmd_with_retry(command, checks, False)
+        recovery = "aks update -g rg -n cluster --enable-azure-monitor-metrics --enable-control-plane-metrics"
+        result = self._result({"azureMonitorProfile": {"metrics": {"controlPlane": {"enabled": True}}}})
+        execute.side_effect = [
+            CLIError("Another operation is in progress."),
+            CLIError("Cluster 'cluster' already exists."),
+            result,
+        ]
+        self.assertIs(instance._cmd_with_retried_create_recovery(
+            "aks create -g rg -n cluster", recovery,
+            checks=[JMESPathCheck("azureMonitorProfile.metrics.controlPlane.enabled", True)],
+        ), result)
+        execute.assert_called_with(instance.cli_ctx, recovery, expect_failure=False)
+        self.assertIsNone(instance._retried_create_recovery_command)
+
+    @patch.dict(os.environ, {"AZURE_CLI_TEST_OPERATION_MAX_RETRIES": "3"})
+    @patch("time.sleep")
+    @patch("azure.cli.testsdk.base.execute")
+    def test_metrics_recovery_error_is_not_hidden(self, execute, _sleep):
+        instance = self._make_instance()
+        instance.cmd = lambda command, checks: instance._cmd_with_retry(command, checks or [], False)
+        error = CLIError("Permission denied")
+        execute.side_effect = [
+            CLIError("Another operation is in progress."),
+            CLIError("Cluster 'cluster' already exists."),
+            error,
+        ]
+        with self.assertRaises(CLIError) as raised:
+            instance._cmd_with_retried_create_recovery(
+                "aks create -g rg -n cluster", "aks update -g rg -n cluster --enable-control-plane-metrics",
+            )
+        self.assertIs(raised.exception, error)
+        self.assertEqual(execute.call_count, 3)
+        self.assertIsNone(instance._retried_create_recovery_command)
+
     def test_is_resource_already_exists_conflict_detects_message(self):
         instance = self._make_instance()
 
@@ -688,6 +730,254 @@ def test_ambiguous_status_error_raises_with_expect_failure(
         mock_sleep.assert_not_called()
 
 
+class TestLiveScenarioRegressions(AKSRetryTestCase):
+    def test_backup_cleanup_keeps_scope_and_reversible_soft_delete_api(self):
+        instance = self._make_instance()
+        instance.kwargs.update({"backup_rg": "owned-rg", "vault_name": "owned-vault"})
+        instance.cmd = MagicMock(return_value=self._result([]))
+        instance._cleanup_backup("owned-vault")
+        first = instance.cmd.call_args_list[0]
+        self.assertIn("resource update --resource-group {backup_rg} --name {vault_name}", first.args[0])
+        self.assertIn("--resource-type Microsoft.DataProtection/backupVaults --api-version 2025-07-01", first.args[0])
+        self.assertIn("properties.securitySettings.softDeleteSettings.state=Off", first.args[0])
+        self.assertEqual(len(first.kwargs["checks"]), 2)
+
+    def test_control_plane_metrics_create_uses_workspace_and_explicit_recovery(self):
+        instance = self._make_instance()
+        instance.create_random_name = MagicMock(return_value="cluster")
+        instance.generate_ssh_keys = MagicMock(return_value="key")
+        instance._create_azure_monitor_workspace = MagicMock(return_value="/workspaces/dedicated")
+        instance._cmd_with_retried_create_recovery = MagicMock(side_effect=RuntimeError("stop at create"))
+        scenario = inspect.unwrap(instance.test_aks_create_with_control_plane_metrics)
+        with self.assertRaisesRegex(RuntimeError, "stop at create"):
+            scenario(instance, "rg", "westcentralus")
+        self.assertEqual(instance.kwargs["amw_id"], "/workspaces/dedicated")
+        call = instance._cmd_with_retried_create_recovery.call_args
+        for command in (call.args[0], call.kwargs["recovery_command"]):
+            self.assertIn("--azure-monitor-workspace-resource-id={amw_id}", command)
+            self.assertIn("--enable-azure-monitor-metrics", command)
+            self.assertIn("--enable-control-plane-metrics", command)
+
+    def test_kms_secret_reencryption_checks_remote_exit_status(self):
+        instance = self._make_instance()
+        instance.is_live = True
+        for exit_code in (0, 1, None):
+            with self.subTest(exit_code=exit_code):
+                result = self._result({"provisioningState": "Succeeded", "exitCode": exit_code})
+                instance.cmd = MagicMock(
+                    side_effect=lambda command, checks: result.assert_with_checks(checks)
+                )
+                if exit_code == 0:
+                    instance._reencrypt_kms_secrets()
+                else:
+                    with self.assertRaises(AssertionError):
+                        instance._reencrypt_kms_secrets()
+                command = instance.cmd.call_args.args[0]
+                self.assertIn("bash -o pipefail", command)
+                self.assertIn("kubectl get secrets --all-namespaces -o json | kubectl replace -f -", command)
+        instance.is_live = False
+        instance.cmd.reset_mock()
+        instance._reencrypt_kms_secrets()
+        instance.cmd.assert_not_called()
+
+    def test_both_private_kms_rotation_cases_reencrypt_before_updating_key(self):
+        for name in (
+            "test_aks_create_with_azurekeyvaultkms_private_key_vault",
+            "test_aks_create_with_azurekeyvaultkms_private_cluster_v1_private_key_vault",
+        ):
+            with self.subTest(name=name):
+                instance = self._make_instance()
+                instance.create_random_name = MagicMock(return_value="name")
+                instance.generate_ssh_keys = MagicMock(return_value="key")
+                instance._get_user_assigned_identity = MagicMock(return_value="/identity")
+                instance._get_principal_id_of_user_assigned_identity = MagicMock(return_value="principal")
+                instance._get_test_identity_object_id = MagicMock(return_value="test-principal")
+                instance.cmd = MagicMock(return_value=self._result({
+                    "id": "/vault", "key": {"kid": "key-version"},
+                }))
+                instance._reencrypt_kms_secrets = MagicMock(side_effect=RuntimeError("before rotation"))
+                scenario = inspect.unwrap(getattr(instance, name))
+                with self.assertRaisesRegex(RuntimeError, "before rotation"):
+                    scenario(instance, "rg", "westcentralus")
+                instance._reencrypt_kms_secrets.assert_called_once()
+                self.assertTrue(any(call.args[0].startswith("aks create ") for call in instance.cmd.call_args_list))
+                self.assertFalse(any(call.args[0].startswith("aks update ") for call in instance.cmd.call_args_list))
+
+    def test_proxy_readiness_requires_success_marker_not_just_run_command_status(self):
+        instance = self._make_instance()
+        for message, ready in [
+            ("[stdout]\nAKS_PROXY_READY\n\n[stderr]\n", True),
+            ("[stdout]\n\n[stderr]\nFailed to start squid\n", False),
+            ("Failed executing echo AKS_PROXY_READY", False),
+        ]:
+            with self.subTest(message=message):
+                result = self._result({"value": [{"code": "ProvisioningState/succeeded", "message": message}]})
+                instance.cmd = lambda command, checks: result.assert_with_checks(checks)
+                if ready:
+                    instance._wait_for_http_proxy()
+                else:
+                    with self.assertRaises(AssertionError):
+                        instance._wait_for_http_proxy()
+
+    def test_proxy_readiness_failure_prevents_cluster_creation(self):
+        instance = self._make_instance()
+        instance.generate_ssh_keys = MagicMock(return_value="key")
+        instance.cmd = MagicMock(return_value=self._result({"id": "/subnets/aks"}))
+        instance._wait_for_http_proxy = MagicMock(side_effect=AssertionError("Proxy not ready"))
+
+        with self.assertRaisesRegex(AssertionError, "Proxy not ready"):
+            instance._setup_http_proxy_cluster("rg", "westcentralus", "cluster")
+
+        commands = [call.args[0] for call in instance.cmd.call_args_list]
+        self.assertTrue(any(command.startswith("vm create") for command in commands))
+        self.assertFalse(any(command.startswith("aks create") for command in commands))
+
+    @patch("azure.cli.testsdk.base.execute")
+    def test_node_image_upgrade_accepts_real_empty_sdk_result(self, mock_execute):
+        from azure.cli.testsdk.base import ExecutionResult
+
+        cli_ctx = MagicMock()
+        cli_ctx.invoke.return_value = 0
+        action_result = ExecutionResult(cli_ctx, "aks nodepool upgrade --node-image-only")
+        pool_result = self._result({"provisioningState": "Succeeded"})
+        mock_execute.side_effect = [action_result, pool_result]
+        instance = self._make_instance()
+        instance.cmd = lambda command, checks=None: instance._cmd_with_retry(command, checks or [], False)
+
+        instance._reimage_nodepool("rg", "cluster", "pool")
+
+        self.assertEqual(mock_execute.call_count, 2)
+
+    def test_node_image_upgrade_checks_persisted_pool_not_empty_action_result(self):
+        instance = self._make_instance()
+        instance.cmd = MagicMock()
+
+        instance._reimage_nodepool("rg", "cluster", "pool")
+
+        self.assertEqual(instance.cmd.call_count, 2)
+        upgrade_call, show_call = instance.cmd.call_args_list
+        self.assertIn("--node-image-only", upgrade_call.args[0])
+        self.assertFalse(any(
+            instance._is_provisioning_state_check(check)
+            for check in upgrade_call.kwargs.get("checks", [])
+        ))
+        self.assertEqual(
+            show_call.args[0],
+            "aks nodepool show --resource-group rg --cluster-name cluster --name pool",
+        )
+        self.assertTrue(instance._is_provisioning_state_check(show_call.kwargs["checks"][0]))
+
+    def test_node_image_upgrade_failure_is_not_recovered_by_show(self):
+        instance = self._make_instance()
+        instance.cmd = MagicMock(side_effect=CLIError("Reimage failed"))
+
+        with self.assertRaisesRegex(CLIError, "Reimage failed"):
+            instance._reimage_nodepool("rg", "cluster", "pool")
+
+        instance.cmd.assert_called_once()
+
+    def test_alb_update_uses_supported_command_arguments(self):
+        from azext_aks_preview import ContainerServiceCommandsLoader
+        from azext_aks_preview.tests.latest.mocks import MockCLI
+
+        instance = self._make_instance()
+        instance.cmd = MagicMock()
+        instance._get_versions = MagicMock(return_value=("1.33.1", "1.34.1"))
+        instance.create_random_name = MagicMock(return_value="cluster")
+        instance.generate_ssh_keys = MagicMock(return_value="ssh-key")
+        scenario = inspect.unwrap(instance.test_aks_applicationloadbalancer_update)
+        scenario(instance, "rg", "eastus")
+
+        loader = ContainerServiceCommandsLoader(MockCLI())
+        loader.load_command_table(["aks", "applicationloadbalancer", "update"])
+        command = loader.command_table["aks applicationloadbalancer update"]
+        command.load_arguments()
+        self.assertNotIn("aks_custom_headers", command.arguments)
+        update_command = instance.cmd.call_args_list[1].args[0]
+        self.assertIn("aks applicationloadbalancer update", update_command)
+        self.assertNotIn("--aks-custom-headers", update_command)
+
+
+class TestFeatureAvailabilitySkip(AKSRetryTestCase):
+    def test_managed_system_allowlist_wording_skips(self):
+        instance = self._make_instance()
+        instance.cmd = MagicMock(side_effect=CLIError(
+            "(InvalidParameter) 'ManagedSystem' is in preview and only support whitelisted subscriptions"
+        ))
+        with self.assertRaises(unittest.SkipTest):
+            instance._cmd_or_skip_if_managed_system_unavailable("aks create")
+
+    def test_managed_system_validation_error_still_fails(self):
+        instance = self._make_instance()
+        instance.cmd = MagicMock(side_effect=CLIError(
+            "(InvalidParameter) Only one ManagedSystem pool is allowed per cluster"
+        ))
+        with self.assertRaises(CLIError):
+            instance._cmd_or_skip_if_managed_system_unavailable("aks nodepool add")
+
+    def test_feature_alias_and_region_errors_skip_precisely(self):
+        cases = [
+            ("ManagedBastion", ("bastionProfile",),
+             "(InvalidParameter) Field .properties.bastionProfile is not yet supported for managed cluster."),
+            ("managedNATGatewayV2", (),
+             "(UnsupportedOutboundType) Outbound type managedNATGatewayV2 is not supported in this region."),
+        ]
+        for feature, aliases, message in cases:
+            with self.subTest(feature=feature):
+                instance = self._make_instance()
+                instance.cmd = MagicMock(side_effect=CLIError(message))
+                with self.assertRaises(unittest.SkipTest):
+                    instance._cmd_or_skip_if_feature_unavailable(
+                        "aks update", feature, feature_aliases=aliases
+                    )
+
+    def test_unrelated_feature_and_validation_errors_still_fail(self):
+        messages = [
+            "(InvalidParameter) Field .properties.otherProfile is not yet supported for managed cluster.",
+            "(InvalidParameter) Field .properties.bastionProfile has an invalid value.",
+        ]
+        for message in messages:
+            with self.subTest(message=message):
+                instance = self._make_instance()
+                instance.cmd = MagicMock(side_effect=CLIError(message))
+                with self.assertRaises(CLIError):
+                    instance._cmd_or_skip_if_feature_unavailable(
+                        "aks update", "ManagedBastion", feature_aliases=("bastionProfile",)
+                    )
+
+    def test_region_or_required_vm_size_unavailable_skips(self):
+        cases = [
+            ("westcentralus", None, "(AvailabilityZoneNotSupported) The supported zones for location 'westcentralus' are ''"),
+            ("eastus2", "Standard_DC16ads_cc_v5",
+             "(VMSizeNotSupported) Virtual Machine size: 'standard_dc16ads_cc_v5' is not supported in location 'eastus2'."),
+        ]
+        for location, vm_size, message in cases:
+            with self.subTest(location=location):
+                instance = self._make_instance()
+                instance.cmd = MagicMock(side_effect=CLIError(message))
+                with self.assertRaises(unittest.SkipTest):
+                    instance._cmd_or_skip_if_region_unavailable(
+                        "aks create", location, vm_size=vm_size
+                    )
+
+    def test_region_skip_preserves_unrelated_and_capacity_failures(self):
+        cases = [
+            ("eastus", None, "(AvailabilityZoneNotSupported) The supported zones for location 'eastus2' are ''"),
+            ("eastus2", "Standard_DC16ads_cc_v5",
+             "(VMSizeNotSupported) Virtual Machine size: 'Standard_D2s_v3' is not supported in location 'eastus2'."),
+            ("eastus2", "Standard_DC16ads_cc_v5",
+             "(ErrCode_InsufficientVCPUQuota) Insufficient quota for Standard_DC16ads_cc_v5 in location 'eastus2'."),
+        ]
+        for location, vm_size, message in cases:
+            with self.subTest(message=message):
+                instance = self._make_instance()
+                instance.cmd = MagicMock(side_effect=CLIError(message))
+                with self.assertRaises(CLIError):
+                    instance._cmd_or_skip_if_region_unavailable(
+                        "aks create", location, vm_size=vm_size
+                    )
+
+
 class TestOsSkuRetirementSkip(AKSRetryTestCase):
     """
     Unit coverage for `_cmd_or_skip_if_os_sku_retired`, which was broadened this session
@@ -785,5 +1075,67 @@ def test_refetches_agentpool_with_native_show(self, mock_execute):
         )
 
 
+class TestPrivateDnsRoleAssignmentRetry(AKSRetryTestCase):
+    ZONE = "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Network/privateDnsZones/example"
+    COMMAND = f"aks create -g rg -n cluster --private-dns-zone={ZONE}"
+    ERROR = (
+        f"(ResourceMissingPermissionError) Permission to resource {ZONE}. "
+        "Check access result not allowed for action Microsoft.Network/privateDnsZones/read."
+    )
+
+    @patch.dict(os.environ, {"AZURE_CLI_TEST_OPERATION_MAX_RETRIES": "2"})
+    @patch("time.sleep")
+    @patch("azure.cli.testsdk.base.execute")
+    def test_retries_just_assigned_zone_and_preserves_bounded_failure(self, execute, sleep):
+        instance = self._make_instance()
+        instance._private_dns_role_assignment_scope = self.ZONE
+        error = CLIError(self.ERROR)
+        expected = self._result({"provisioningState": "Succeeded"})
+        execute.side_effect = [error, expected]
+        self.assertIs(instance._execute_with_transient_conflict_retry(self.COMMAND, False), expected)
+        sleep.assert_called_once()
+        execute.reset_mock(side_effect=True)
+        execute.side_effect = error
+        with self.assertRaises(CLIError) as raised:
+            instance._execute_with_transient_conflict_retry(self.COMMAND, False)
+        self.assertIs(raised.exception, error)
+        self.assertEqual(execute.call_count, 2)
+
+    def test_unrelated_permissions_commands_and_scopes_are_not_retryable(self):
+        for scope, command, message in (
+            (None, self.COMMAND, self.ERROR),
+            (self.ZONE + "-other", self.COMMAND, self.ERROR),
+            (self.ZONE, self.COMMAND + "-other", self.ERROR),
+            (self.ZONE, self.COMMAND.replace("aks create", "aks update"), self.ERROR),
+            (self.ZONE, self.COMMAND, self.ERROR.replace("example.", "example-other.")),
+            (self.ZONE, self.COMMAND, self.ERROR.replace("privateDnsZones/read", "privateDnsZones/write")),
+            (self.ZONE, self.COMMAND, self.ERROR.replace("ResourceMissingPermissionError", "AuthorizationFailed")),
+        ):
+            with self.subTest(scope=scope, command=command, message=message):
+                instance = self._make_instance()
+                instance._private_dns_role_assignment_scope = scope
+                self.assertFalse(instance._is_private_dns_role_assignment_pending(command, CLIError(message)))
+
+    def test_scope_is_case_insensitive_and_restored_after_failure(self):
+        instance = self._make_instance()
+        instance._private_dns_role_assignment_scope = self.ZONE.upper()
+        self.assertTrue(instance._is_private_dns_role_assignment_pending(self.COMMAND, CLIError(self.ERROR)))
+        instance.cmd = MagicMock(side_effect=CLIError("failed"))
+        with self.assertRaises(CLIError):
+            instance._cmd_with_private_dns_role_assignment_retry(self.COMMAND, self.ZONE)
+        self.assertEqual(instance._private_dns_role_assignment_scope, self.ZONE.upper())
+
+    @patch("azure.cli.testsdk.base.execute")
+    @patch("time.sleep")
+    def test_expected_failure_is_not_retried(self, sleep, execute):
+        instance = self._make_instance()
+        instance._private_dns_role_assignment_scope = self.ZONE
+        execute.side_effect = CLIError(self.ERROR)
+        with self.assertRaises(CLIError):
+            instance._execute_with_transient_conflict_retry(self.COMMAND, True)
+        execute.assert_called_once()
+        sleep.assert_not_called()
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/src/aks-preview/azext_aks_preview/tests/latest/test_custom.py b/src/aks-preview/azext_aks_preview/tests/latest/test_custom.py
index 8d2e76a7a08..4fb6eba9903 100644
--- a/src/aks-preview/azext_aks_preview/tests/latest/test_custom.py
+++ b/src/aks-preview/azext_aks_preview/tests/latest/test_custom.py
@@ -2,6 +2,7 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
+import json
 import unittest
 from unittest.mock import Mock, patch
 
@@ -231,6 +232,76 @@ def test_aks_enable_addons_virtual_node_with_none_agent_pool_profiles(self):
             self.assertNotIn("NoneType", str(type(e)))
 
 
+class TestMonitoringProfileConsistency(unittest.TestCase):
+    @staticmethod
+    def _cluster():
+        from azext_aks_preview.vendored_sdks.azure_mgmt_preview_aks import models
+
+        return models.ManagedCluster(
+            location="westus2",
+            addon_profiles={"omsagent": models.ManagedClusterAddonProfile(
+                enabled=True,
+                config={"enableRetinaNetworkFlags": "true", "useAADAuth": "true"},
+            )},
+            azure_monitor_profile=models.ManagedClusterAzureMonitorProfile(
+                container_insights=models.ManagedClusterAzureMonitorProfileContainerInsights(
+                    enabled=True, container_network_logs="Enabled",
+                ),
+                metrics=models.ManagedClusterAzureMonitorProfileMetrics(enabled=True),
+            ),
+        )
+
+    @staticmethod
+    def _wire_properties(cluster):
+        from azext_aks_preview.vendored_sdks.azure_mgmt_preview_aks import ContainerServiceClient
+
+        class RequestCaptured(Exception):
+            pass
+
+        with ContainerServiceClient(Mock(), "sub") as client:
+            with patch.object(client._client._pipeline, "run", side_effect=RequestCaptured) as send:
+                try:
+                    client.managed_clusters.begin_create_or_update("rg", "cluster", cluster)
+                except RequestCaptured:
+                    return json.loads(send.call_args.args[0].body)["properties"]
+        raise AssertionError("The preview SDK did not construct the managed cluster PUT")
+
+    def test_disable_and_reenable_monitoring_updates_both_wire_profiles(self):
+        from azext_aks_preview.custom import _update_addons
+
+        register_aks_preview_resource_type()
+        cmd = MockCmd(MockCLI())
+        cluster = self._cluster()
+        workspace = "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.OperationalInsights/workspaces/workspace"
+        for enable in (False, True):
+            with self.subTest(enable=enable):
+                _update_addons(cmd, cluster, "sub", "rg", "cluster", "monitoring", enable,
+                               workspace_resource_id=workspace)
+                payload = self._wire_properties(cluster)
+                insights = payload["azureMonitorProfile"]["containerInsights"]
+                self.assertEqual(insights["enabled"], enable)
+                self.assertEqual(payload["addonProfiles"]["omsagent"]["enabled"], enable)
+                self.assertNotEqual(insights.get("containerNetworkLogs"), "Enabled")
+                if enable:
+                    self.assertEqual(insights["logAnalyticsWorkspaceResourceId"], workspace)
+                self.assertTrue(payload["azureMonitorProfile"]["metrics"]["enabled"])
+
+    def test_disable_flow_logs_updates_both_wire_profiles(self):
+        from azext_aks_preview.managed_cluster_decorator import AKSPreviewManagedClusterUpdateDecorator
+
+        register_aks_preview_resource_type()
+        cluster = self._cluster()
+        decorator = AKSPreviewManagedClusterUpdateDecorator(
+            MockCmd(MockCLI()), Mock(), {"disable_container_network_logs": True}, CUSTOM_MGMT_AKS_PREVIEW,
+        )
+        decorator.context.attach_mc(cluster)
+        decorator.update_monitoring_profile_flow_logs(cluster)
+        payload = self._wire_properties(cluster)
+        self.assertEqual(payload["azureMonitorProfile"]["containerInsights"]["containerNetworkLogs"], "Disabled")
+        self.assertEqual(payload["addonProfiles"]["omsagent"]["config"]["enableRetinaNetworkFlags"].lower(), "false")
+        self.assertTrue(payload["azureMonitorProfile"]["metrics"]["enabled"])
+
+
 class TestAksAgentPoolGetBootstrapData(unittest.TestCase):
     def setUp(self):
         register_aks_preview_resource_type()