custom.py

# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------

import os
import shutil
import subprocess
import tempfile
import yaml

from knack.log import get_logger
from knack.util import CLIError

from azure.cli.core.commands.client_factory import get_subscription_id
from azure.cli.core.util import sdk_no_wait, get_file_json, shell_safe_json_parse
from azure.cli.core import get_default_cli
from azure.mgmt.core.tools import parse_resource_id
from azure.cli.command_modules.acs._graph import resolve_object_id

from azext_fleet._client_factory import CUSTOM_MGMT_FLEET, cf_fleet_members, cf_fleets
from azext_fleet._helpers import is_rp_registered, print_or_merge_credentials
from azext_fleet._helpers import assign_network_contributor_role_to_subnet
from azext_fleet._helpers import get_msi_object_id
from azext_fleet._helpers import is_stdout_path
from azext_fleet.constants import UPGRADE_TYPE_CONTROLPLANEONLY
from azext_fleet.constants import UPGRADE_TYPE_FULL
from azext_fleet.constants import UPGRADE_TYPE_NODEIMAGEONLY
from azext_fleet.constants import UPGRADE_TYPE_ERROR_MESSAGES
from azext_fleet.constants import SUPPORTED_GATE_STATES_FILTERS
from azext_fleet.constants import SUPPORTED_GATE_STATES_PATCH
from azext_fleet.constants import FLEET_1P_APP_ID
from azext_fleet.vendored_sdks.v2026_03_02_preview.models import (
    PropagationPolicy,
    PlacementProfile,
    PlacementV1ClusterResourcePlacementSpec,
    PlacementV1PlacementPolicy,
    PropagationType,
    PlacementType,
)

logger = get_logger(__name__)


# pylint: disable=too-many-locals
def create_fleet(cmd,
                 client,
                 resource_group_name,
                 name,
                 location=None,
                 tags=None,
                 enable_hub=False,
                 vm_size=None,
                 dns_name_prefix=None,
                 enable_private_cluster=False,
                 enable_vnet_integration=False,
                 apiserver_subnet_id=None,
                 agent_subnet_id=None,
                 enable_managed_identity=False,
                 assign_identity=None,
                 no_wait=False):
    fleet_model = cmd.get_models(
        "Fleet",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleets"
    )

    poll_interval = 5
    if enable_hub:
        poll_interval = 30
        fleet_hub_profile_model = cmd.get_models(
            "FleetHubProfile",
            resource_type=CUSTOM_MGMT_FLEET,
            operation_group="fleets"
        )
        api_server_access_profile_model = cmd.get_models(
            "APIServerAccessProfile",
            resource_type=CUSTOM_MGMT_FLEET,
            operation_group="fleets"
        )
        agent_profile_model = cmd.get_models(
            "AgentProfile",
            resource_type=CUSTOM_MGMT_FLEET,
            operation_group="fleets",
        )
        api_server_access_profile = api_server_access_profile_model(
            enable_private_cluster=enable_private_cluster,
            enable_vnet_integration=enable_vnet_integration,
            subnet_id=apiserver_subnet_id
        )
        agent_profile = agent_profile_model(
            subnet_id=agent_subnet_id,
            vm_size=vm_size
        )
        fleet_hub_profile = fleet_hub_profile_model(
            dns_prefix=dns_name_prefix,
            api_server_access_profile=api_server_access_profile,
            agent_profile=agent_profile)
    else:
        if dns_name_prefix is not None or \
           enable_private_cluster or \
           enable_vnet_integration or \
           apiserver_subnet_id is not None or \
           agent_subnet_id is not None:
            raise CLIError(
                "The parameters for private cluster, vnet & subnet integration are only valid if hub is enabled.")
        fleet_hub_profile = None

    fleet_managed_service_identity_model = cmd.get_models(
        "ManagedServiceIdentity",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleets"
    )
    managed_service_identity = fleet_managed_service_identity_model(type="None")
    if enable_managed_identity:
        managed_service_identity.type = "SystemAssigned"
        if assign_identity is not None:
            user_assigned_identity_model = cmd.get_models(
                "UserAssignedIdentity",
                resource_type=CUSTOM_MGMT_FLEET,
                operation_group="fleets"
            )
            managed_service_identity.type = "UserAssigned"
            managed_service_identity.user_assigned_identities = {assign_identity: user_assigned_identity_model()}
    elif assign_identity is not None:
        raise CLIError("Cannot assign identity without enabling managed identity.")

    fleet = fleet_model(
        location=location,
        tags=tags,
        hub_profile=fleet_hub_profile,
        identity=managed_service_identity
    )

    if enable_private_cluster:
        # provider registration state being is checked to ensure that the Fleet service principal is available
        # to create the role assignment on the subnet
        if not is_rp_registered(cmd):
            raise CLIError("The Microsoft.ContainerService resource provider is not registered."
                           "Run `az provider register -n Microsoft.ContainerService --wait`.")
        object_id = resolve_object_id(cmd.cli_ctx, FLEET_1P_APP_ID)
        assign_network_contributor_role_to_subnet(cmd, object_id, agent_subnet_id)

    if enable_vnet_integration and assign_identity is not None:
        object_id = get_msi_object_id(cmd, assign_identity)
        assign_network_contributor_role_to_subnet(cmd, object_id, apiserver_subnet_id)
        assign_network_contributor_role_to_subnet(cmd, object_id, agent_subnet_id)

    return sdk_no_wait(no_wait,
                       client.begin_create_or_update,
                       resource_group_name,
                       name,
                       fleet,
                       polling_interval=poll_interval)


def update_fleet(cmd,
                 client,
                 resource_group_name,
                 name,
                 enable_managed_identity=None,
                 assign_identity=None,
                 tags=None,
                 no_wait=False):
    fleet_patch_model = cmd.get_models(
        "FleetPatch",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleets"
    )
    fleet_managed_service_identity_model = cmd.get_models(
        "ManagedServiceIdentity",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleets"
    )

    if enable_managed_identity is None:
        managed_service_identity = None
        if assign_identity is not None:
            raise CLIError("Cannot assign identity without enabling managed identity.")
    elif enable_managed_identity is False:
        managed_service_identity = fleet_managed_service_identity_model(type="None")
    else:
        managed_service_identity = fleet_managed_service_identity_model(type="SystemAssigned")
        if assign_identity is not None:
            user_assigned_identity_model = cmd.get_models(
                "UserAssignedIdentity",
                resource_type=CUSTOM_MGMT_FLEET,
                operation_group="fleets"
            )
            managed_service_identity.type = "UserAssigned"
            managed_service_identity.user_assigned_identities = {assign_identity: user_assigned_identity_model()}

    fleet_patch = fleet_patch_model(
        tags=tags,
        identity=managed_service_identity
    )

    return sdk_no_wait(no_wait, client.begin_update, resource_group_name, name, fleet_patch, polling_interval=5)


def show_fleet(cmd,  # pylint: disable=unused-argument
               client,
               resource_group_name,
               name):
    return client.get(resource_group_name, name)


def list_fleet(cmd,  # pylint: disable=unused-argument
               client,
               resource_group_name=None):
    if resource_group_name:
        return client.list_by_resource_group(resource_group_name)
    return client.list_by_subscription()


def delete_fleet(cmd,  # pylint: disable=unused-argument
                 client,
                 resource_group_name,
                 name,
                 no_wait=False):
    return sdk_no_wait(no_wait, client.begin_delete, resource_group_name, name, polling_interval=5)


def _convert_kubeconfig_to_azurecli(path):
    """
    Convert kubeconfig to use Azure CLI authentication if it uses devicecode.

    Args:
        path: Path to the kubeconfig file to convert
    """
    # Skip conversion if path is stdout
    if is_stdout_path(path):
        return

    if shutil.which("kubelogin"):
        try:
            subprocess.run(
                ["kubelogin", "convert-kubeconfig", "-l", "azurecli", "--kubeconfig", path],
                check=True,
                capture_output=True,
                text=True,
                timeout=60,
            )
            logger.warning("Converted kubeconfig to use Azure CLI authentication.")
        except subprocess.CalledProcessError as e:
            logger.warning("Failed to convert kubeconfig with kubelogin: %s", str(e))
        except subprocess.TimeoutExpired as e:
            logger.warning("kubelogin command timed out: %s", str(e))
        except Exception as e:  # pylint: disable=broad-except
            logger.warning("Error running kubelogin: %s", str(e))
    else:
        logger.warning(
            "The fleet hub cluster kubeconfig requires kubelogin. "
            "Please install kubelogin from https://github.com/Azure/kubelogin or run "
            "'az aks install-cli' to install both kubectl and kubelogin. "
            "After installing kubelogin, rerun 'az fleet get-credentials' and the "
            "kubeconfig will be converted automatically."
        )


def get_credentials(cmd,
                    client,
                    resource_group_name,
                    name,
                    path=os.path.join(os.path.expanduser('~'), '.kube', 'config'),
                    overwrite_existing=False,
                    context_name=None,
                    member_name=None,
                    skip_kubelogin_conversion=False):

    # If a member name is given, we use the cluster resource ID from the fleet member
    # to get that member cluster's credentials
    # Otherwise, we get the credentials for the fleet hub
    if member_name:
        fleet_members_client = cf_fleet_members(cmd.cli_ctx)

        try:
            fleet_member = fleet_members_client.get(resource_group_name, name, member_name)

            parsed_id = parse_resource_id(fleet_member.cluster_resource_id)
            if (parsed_id.get('resource_type').lower() != 'managedclusters' or
                    parsed_id.get('namespace').lower() != 'microsoft.containerservice'):
                raise CLIError(f"Fleet member '{member_name}' is not associated with an AKS managed cluster. "
                               f"Currently, only AKS managed clusters are supported for this command.")

            args = [
                'aks', 'get-credentials',
                '--subscription', parsed_id['subscription'],
                '--resource-group', parsed_id['resource_group'],
                '--name', parsed_id['resource_name'],
                '--context', context_name or member_name,
                '--file', path,
            ]

            if overwrite_existing:
                args.append("--overwrite-existing")

            exit_code = get_default_cli().invoke(args)
            if exit_code != 0:
                error_msg = (f"Failed to get credentials from managed cluster '{parsed_id['resource_name']}' "
                             f"for fleet member '{member_name}'")
                raise CLIError(error_msg)

        except Exception as exc:
            if isinstance(exc, CLIError):
                raise
            raise CLIError(f"Error getting credentials for fleet member '{member_name}': {str(exc)}") from exc

    else:
        credential_results = client.list_credentials(resource_group_name, name)
        if not credential_results:
            raise CLIError("No Kubernetes credentials found.")

        try:
            kubeconfig = credential_results.kubeconfigs[0].value.decode(encoding='UTF-8')
            print_or_merge_credentials(path, kubeconfig, overwrite_existing, context_name)
            # Fleet hub is always RBAC-enabled and should convert it with kubelogin so that
            # user doesn't have to manually run kubelogin convert-kubeconfig -l azurecli
            # every time after az fleet get-credentials
            if not skip_kubelogin_conversion:
                _convert_kubeconfig_to_azurecli(path)
        except (IndexError, ValueError) as exc:
            raise CLIError("Fail to find kubeconfig file.") from exc


def reconcile_fleet(cmd,  # pylint: disable=unused-argument
                    client,
                    resource_group_name,
                    name,
                    no_wait=False):

    poll_interval = 5
    fleet = client.get(resource_group_name, name)
    if fleet.hub_profile is not None:
        poll_interval = 30

    return sdk_no_wait(no_wait,
                       client.begin_create_or_update,
                       resource_group_name,
                       name,
                       fleet,
                       if_match=fleet.e_tag,
                       polling_interval=poll_interval)


def create_fleet_member(cmd,
                        client,
                        resource_group_name,
                        name,
                        fleet_name,
                        member_cluster_id,
                        update_group=None,
                        member_labels=None,
                        no_wait=False):
    fleet_member_model = cmd.get_models(
        "FleetMember",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleet_members"
    )
    fleet_member = fleet_member_model(cluster_resource_id=member_cluster_id, group=update_group, labels=member_labels)
    return sdk_no_wait(no_wait, client.begin_create, resource_group_name, fleet_name, name, fleet_member)


def update_fleet_member(cmd,
                        client,
                        resource_group_name,
                        name,
                        fleet_name,
                        update_group=None,
                        member_labels=None,
                        no_wait=False):
    fleet_member_update_model = cmd.get_models(
        "FleetMemberUpdate",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleet_members"
    )
    properties = fleet_member_update_model(group=update_group, labels=member_labels)
    return sdk_no_wait(no_wait, client.begin_update, resource_group_name, fleet_name, name, properties)


def list_fleet_member(cmd,  # pylint: disable=unused-argument
                      client,
                      resource_group_name,
                      fleet_name,
                      cluster_mesh_profile=None):
    filter_expr = None
    if cluster_mesh_profile:
        filter_expr = f"clusterMeshProfile eq {cluster_mesh_profile}"
    return client.list_by_fleet(resource_group_name, fleet_name, filter=filter_expr)


def show_fleet_member(cmd,  # pylint: disable=unused-argument
                      client,
                      resource_group_name,
                      fleet_name,
                      name):
    return client.get(resource_group_name, fleet_name, name)


def delete_fleet_member(cmd,  # pylint: disable=unused-argument
                        client,
                        resource_group_name,
                        fleet_name,
                        name,
                        no_wait=False):
    return sdk_no_wait(no_wait, client.begin_delete, resource_group_name, fleet_name, name)


def reconcile_fleet_member(cmd,  # pylint: disable=unused-argument
                           client,
                           resource_group_name,
                           name,
                           fleet_name,
                           no_wait=False):

    member = client.get(resource_group_name, fleet_name, name)
    return sdk_no_wait(no_wait,
                       client.begin_create,
                       resource_group_name,
                       fleet_name,
                       name,
                       member,
                       if_match=member.e_tag)


def create_update_run(cmd,
                      client,
                      resource_group_name,
                      fleet_name,
                      name,
                      upgrade_type,
                      node_image_selection=None,
                      kubernetes_version=None,
                      stages=None,
                      update_strategy_name=None,
                      no_wait=False):

    if upgrade_type in UPGRADE_TYPE_ERROR_MESSAGES:
        if (
            ((upgrade_type in (UPGRADE_TYPE_FULL, UPGRADE_TYPE_CONTROLPLANEONLY)) and kubernetes_version is None) or  # pylint: disable=line-too-long
            (upgrade_type == UPGRADE_TYPE_NODEIMAGEONLY and kubernetes_version is not None)
        ):
            raise CLIError(UPGRADE_TYPE_ERROR_MESSAGES[upgrade_type])
    else:
        raise CLIError((f"The upgrade type parameter '{upgrade_type}' is not valid."
                        f"Valid options are: '{UPGRADE_TYPE_FULL}', '{UPGRADE_TYPE_CONTROLPLANEONLY}', or '{UPGRADE_TYPE_NODEIMAGEONLY}'"))  # pylint: disable=line-too-long

    if upgrade_type == UPGRADE_TYPE_CONTROLPLANEONLY and node_image_selection is not None:
        raise CLIError("Node image selection must not be set when upgrade type is 'ControlPlaneOnly'.")

    if stages is not None and update_strategy_name is not None:
        raise CLIError("Cannot set stages when update strategy name is set.")

    update_run_strategy = get_update_run_strategy(cmd, "update_runs", stages)

    managed_cluster_upgrade_spec_model = cmd.get_models(
        "ManagedClusterUpgradeSpec",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="update_runs"
    )
    managed_cluster_update_model = cmd.get_models(
        "ManagedClusterUpdate",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="update_runs"
    )
    node_image_selection_model = cmd.get_models(
        "NodeImageSelection",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="update_runs"
    )
    update_run_model = cmd.get_models(
        "UpdateRun",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="update_runs"
    )

    managed_cluster_upgrade_spec = managed_cluster_upgrade_spec_model(
        type=upgrade_type, kubernetes_version=kubernetes_version)

    node_image_selection_type = None
    if upgrade_type != UPGRADE_TYPE_CONTROLPLANEONLY:
        if node_image_selection is None:
            node_image_selection = "Latest"
        node_image_selection_type = node_image_selection_model(type=node_image_selection)

    managed_cluster_update = managed_cluster_update_model(
        upgrade=managed_cluster_upgrade_spec,
        node_image_selection=node_image_selection_type)

    updateStrategyId = None
    if update_strategy_name is not None:
        subId = get_subscription_id(cmd.cli_ctx)
        updateStrategyId = (
            f"/subscriptions/{subId}/resourceGroups/{resource_group_name}"
            f"/providers/Microsoft.ContainerService/fleets/{fleet_name}/updateStrategies/{update_strategy_name}"
        )

    update_run = update_run_model(
        update_strategy_id=updateStrategyId,
        strategy=update_run_strategy,
        managed_cluster_update=managed_cluster_update)

    result = sdk_no_wait(no_wait, client.begin_create_or_update, resource_group_name, fleet_name, name, update_run)
    print("After successfully creating the run, you need to use the following command to start the run:"
          f"az fleet updaterun start --resource-group={resource_group_name} --fleet={fleet_name} --name={name}")
    return result


def show_update_run(cmd,  # pylint: disable=unused-argument
                    client,
                    resource_group_name,
                    fleet_name,
                    name):
    return client.get(resource_group_name, fleet_name, name)


def list_update_run(cmd,  # pylint: disable=unused-argument
                    client,
                    resource_group_name,
                    fleet_name):
    return client.list_by_fleet(resource_group_name, fleet_name)


def delete_update_run(cmd,  # pylint: disable=unused-argument
                      client,
                      resource_group_name,
                      fleet_name,
                      name,
                      no_wait=False):
    return sdk_no_wait(no_wait, client.begin_delete, resource_group_name, fleet_name, name)


def start_update_run(cmd,  # pylint: disable=unused-argument
                     client,
                     resource_group_name,
                     fleet_name,
                     name,
                     no_wait=False):
    return sdk_no_wait(no_wait, client.begin_start, resource_group_name, fleet_name, name)


def stop_update_run(cmd,  # pylint: disable=unused-argument
                    client,
                    resource_group_name,
                    fleet_name,
                    name,
                    no_wait=False):
    return sdk_no_wait(no_wait, client.begin_stop, resource_group_name, fleet_name, name)


def skip_update_run(cmd,  # pylint: disable=unused-argument
                    client,
                    resource_group_name,
                    fleet_name,
                    name,
                    targets=None,
                    no_wait=False):

    update_run_skip_properties_model = cmd.get_models(
        "SkipProperties",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="update_runs"
    )

    update_run_skip_target_model = cmd.get_models(
        "SkipTarget",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="update_runs"
    )

    skipTargets = []
    for target in targets:
        key, value = target.split(':')
        skipTargets.append(update_run_skip_target_model(
            type=key,
            name=value
        ))

    skip_properties = update_run_skip_properties_model(targets=skipTargets)
    return sdk_no_wait(no_wait, client.begin_skip, resource_group_name, fleet_name, name, skip_properties)


def get_update_run_strategy(cmd, operation_group, stages):
    if stages is None:
        return None

    # Check if the input is a file path or inline JSON
    if os.path.exists(stages):
        data = get_file_json(stages)
    else:
        data = shell_safe_json_parse(stages)

    update_group_model = cmd.get_models(
        "UpdateGroup",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group=operation_group
    )
    update_stage_model = cmd.get_models(
        "UpdateStage",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group=operation_group
    )
    update_run_strategy_model = cmd.get_models(
        "UpdateRunStrategy",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group=operation_group
    )

    update_stages = []

    for stage in data["stages"]:
        update_groups = []
        for group in stage["groups"]:
            update_groups.append(update_group_model(
                name=group["name"],
                max_concurrency=group.get("maxConcurrency"),
                before_gates=group.get("beforeGates", []),
                after_gates=group.get("afterGates", []),
            ))

        after_wait = stage.get("afterStageWaitInSeconds") or 0

        update_stages.append(update_stage_model(
            name=stage["name"],
            groups=update_groups,
            max_concurrency=stage.get("maxConcurrency"),
            before_gates=stage.get("beforeGates", []),
            after_gates=stage.get("afterGates", []),
            after_stage_wait_in_seconds=after_wait
        ))

    return update_run_strategy_model(stages=update_stages)


def create_fleet_update_strategy(cmd,
                                 client,
                                 resource_group_name,
                                 fleet_name,
                                 name,
                                 stages,
                                 no_wait=False):
    update_run_strategy_model = get_update_run_strategy(cmd, "fleet_update_strategies", stages)

    fleet_update_strategy_model = cmd.get_models(
        "FleetUpdateStrategy",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleet_update_strategies"
    )
    fleet_update_strategy = fleet_update_strategy_model(strategy=update_run_strategy_model)
    return sdk_no_wait(
        no_wait, client.begin_create_or_update, resource_group_name, fleet_name, name, fleet_update_strategy)


def show_fleet_update_strategy(cmd,  # pylint: disable=unused-argument
                               client,
                               resource_group_name,
                               fleet_name,
                               name):
    return client.get(resource_group_name, fleet_name, name)


def list_fleet_update_strategies(cmd,  # pylint: disable=unused-argument
                                 client,
                                 resource_group_name,
                                 fleet_name):
    return client.list_by_fleet(resource_group_name, fleet_name)


def delete_fleet_update_strategy(cmd,  # pylint: disable=unused-argument
                                 client,
                                 resource_group_name,
                                 fleet_name,
                                 name,
                                 no_wait=False):
    return sdk_no_wait(no_wait, client.begin_delete, resource_group_name, fleet_name, name)


def create_auto_upgrade_profile(cmd,  # pylint: disable=unused-argument
                                client,
                                resource_group_name,
                                fleet_name,
                                name,
                                channel,
                                update_strategy_id=None,
                                node_image_selection=None,
                                target_kubernetes_version=None,
                                long_term_support=False,
                                disabled=False,
                                no_wait=False):

    if channel == "NodeImage" and node_image_selection is not None:
        raise CLIError("node_image_selection must NOT be populated when channel type `NodeImage` is selected")

    upgrade_channel_model = cmd.get_models(
        "UpgradeChannel",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="auto_upgrade_profiles",
    )
    upgrade_channel = upgrade_channel_model(channel)

    auto_upgrade_node_image_selection = None
    if node_image_selection:
        auto_upgrade_node_image_selection_model = cmd.get_models(
            "AutoUpgradeNodeImageSelection",
            resource_type=CUSTOM_MGMT_FLEET,
            operation_group="auto_upgrade_profiles",
        )
        auto_upgrade_node_image_selection = auto_upgrade_node_image_selection_model(type=node_image_selection)

    auto_upgrade_profile_model = cmd.get_models(
        "AutoUpgradeProfile",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="auto_upgrade_profiles",
    )
    auto_upgrade_profile = auto_upgrade_profile_model(
        update_strategy_id=update_strategy_id,
        channel=upgrade_channel,
        node_image_selection=auto_upgrade_node_image_selection,
        target_kubernetes_version=target_kubernetes_version,
        long_term_support=long_term_support,
        disabled=disabled
    )

    return sdk_no_wait(no_wait,
                       client.begin_create_or_update,
                       resource_group_name,
                       fleet_name,
                       name,
                       auto_upgrade_profile)


def show_auto_upgrade_profile(cmd,  # pylint: disable=unused-argument
                              client,
                              resource_group_name,
                              fleet_name,
                              name):
    return client.get(resource_group_name, fleet_name, name)


def list_auto_upgrade_profiles(cmd,  # pylint: disable=unused-argument
                               client,
                               resource_group_name,
                               fleet_name):
    return client.list_by_fleet(resource_group_name, fleet_name)


def delete_auto_upgrade_profile(cmd,  # pylint: disable=unused-argument
                                client,
                                resource_group_name,
                                fleet_name,
                                name,
                                no_wait=False):
    return sdk_no_wait(no_wait, client.begin_delete, resource_group_name, fleet_name, name)


def generate_update_run(cmd,  # pylint: disable=unused-argument
                        client,
                        resource_group_name,
                        fleet_name,
                        auto_upgrade_profile_name,
                        no_wait=False):
    return sdk_no_wait(
        no_wait,
        client.begin_generate_update_run,
        resource_group_name,
        fleet_name,
        auto_upgrade_profile_name
    )


def list_gates_by_fleet(cmd,  # pylint: disable=unused-argument
                        client,
                        resource_group_name,
                        fleet_name,
                        state_filter=None):
    params = {}

    if state_filter:
        if state_filter not in SUPPORTED_GATE_STATES_FILTERS:
            raise CLIError(
                f"Unsupported gate state filter value: '{state_filter}'. "
                f"Allowed values are {SUPPORTED_GATE_STATES_FILTERS}"
            )

        params["$filter"] = f"state eq {state_filter}"

    return client.list_by_fleet(resource_group_name, fleet_name, params=params)


def show_gate(cmd,  # pylint: disable=unused-argument
              client,
              resource_group_name,
              fleet_name,
              gate_name):
    return client.get(resource_group_name, fleet_name, gate_name)


def _patch_gate(cmd,  # pylint: disable=unused-argument
                client,
                resource_group_name,
                fleet_name,
                gate_name,
                gate_state,
                no_wait=False):
    if gate_state not in SUPPORTED_GATE_STATES_PATCH:
        raise CLIError(
            f"Unsupported gate state value: '{gate_state}'. "
            f"Allowed values are {SUPPORTED_GATE_STATES_PATCH}"
        )

    gate_model = cmd.get_models(
        "GatePatch",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="gates"
    )
    gate_properties_model = cmd.get_models(
        "GatePatchProperties",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="gates"
    )

    properties = gate_properties_model(state=gate_state)
    patch_request = gate_model(properties=properties)

    return sdk_no_wait(no_wait, client.begin_update, resource_group_name, fleet_name, gate_name, patch_request)


def update_gate(cmd,  # pylint: disable=unused-argument
                client,
                resource_group_name,
                fleet_name,
                gate_name,
                gate_state=None,
                no_wait=False):
    return _patch_gate(cmd, client, resource_group_name, fleet_name, gate_name, gate_state, no_wait)


def approve_gate(cmd,  # pylint: disable=unused-argument
                 client,
                 resource_group_name,
                 fleet_name,
                 gate_name,
                 no_wait=False):
    return _patch_gate(cmd, client, resource_group_name, fleet_name, gate_name, "Completed", no_wait)


def create_managed_namespace(cmd,
                             client,
                             resource_group_name,
                             fleet_name,
                             managed_namespace_name,
                             tags=None,
                             labels=None,
                             annotations=None,
                             cpu_requests=None,
                             cpu_limits=None,
                             memory_requests=None,
                             memory_limits=None,
                             ingress_policy=None,
                             egress_policy=None,
                             delete_policy=None,
                             adoption_policy=None,
                             member_cluster_names=None,
                             no_wait=False):

    managed_namespace_model = cmd.get_models(
        "FleetManagedNamespace",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleet_managed_namespaces"
    )

    managed_namespace_properties_model = cmd.get_models(
        "ManagedNamespaceProperties",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleet_managed_namespaces"
    )

    fleet_managed_namespace_properties_model = cmd.get_models(
        "FleetManagedNamespaceProperties",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleet_managed_namespaces"
    )

    resource_quota_model = cmd.get_models(
        "ResourceQuota",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleet_managed_namespaces"
    )

    network_policy_model = cmd.get_models(
        "NetworkPolicy",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleet_managed_namespaces"
    )

    fleet_client = cf_fleets(cmd.cli_ctx)
    fleet = fleet_client.get(resource_group_name, fleet_name)

    default_resource_quota = None
    if cpu_requests or cpu_limits or memory_requests or memory_limits:
        resource_limits = {}
        if cpu_requests:
            resource_limits['cpu_request'] = cpu_requests
        if memory_requests:
            resource_limits['memory_request'] = memory_requests
        if cpu_limits:
            resource_limits['cpu_limit'] = cpu_limits
        if memory_limits:
            resource_limits['memory_limit'] = memory_limits
        default_resource_quota = resource_quota_model(**resource_limits)

    default_network_policy = None
    if ingress_policy or egress_policy:
        network_policies = {}
        if ingress_policy:
            network_policies['ingress'] = ingress_policy
        if egress_policy:
            network_policies['egress'] = egress_policy
        default_network_policy = network_policy_model(**network_policies)

    managed_namespace_props = managed_namespace_properties_model(
        labels=labels,
        annotations=annotations,
        default_resource_quota=default_resource_quota,
        default_network_policy=default_network_policy
    )

    propagation_policy = None
    if member_cluster_names:
        placement_policy = PlacementV1PlacementPolicy(
            placement_type=PlacementType.pick_fixed,
            cluster_names=member_cluster_names
        )
        placement_spec = PlacementV1ClusterResourcePlacementSpec(
            policy=placement_policy
        )
        placement_profile = PlacementProfile(
            default_cluster_resource_placement=placement_spec
        )
        propagation_policy = PropagationPolicy(
            type=PropagationType.placement,
            placement_profile=placement_profile
        )
    else:
        logger.warning("--member-cluster-names was empty; namespace will not be placed on any member clusters")

    fleet_managed_namespace_props = fleet_managed_namespace_properties_model(
        managed_namespace_properties=managed_namespace_props,
        adoption_policy=adoption_policy,
        delete_policy=delete_policy,
        propagation_policy=propagation_policy
    )

    managed_namespace = managed_namespace_model(
        location=fleet.location,
        tags=tags,
        properties=fleet_managed_namespace_props
    )

    return sdk_no_wait(
        no_wait,
        client.begin_create_or_update,
        resource_group_name,
        fleet_name,
        managed_namespace_name,
        managed_namespace,
        polling_interval=5
    )


def update_managed_namespace(cmd,
                             client,
                             resource_group_name,
                             fleet_name,
                             managed_namespace_name,
                             tags=None):
    """
    Update a fleet managed namespace. Currently only supports updating tags.
    """
    fleet_managed_namespace_patch_model = cmd.get_models(
        "FleetManagedNamespacePatch",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="fleet_managed_namespaces"
    )

    patch = fleet_managed_namespace_patch_model(tags=tags)

    return client.begin_update(
        resource_group_name=resource_group_name,
        fleet_name=fleet_name,
        managed_namespace_name=managed_namespace_name,
        properties=patch
    )


def delete_managed_namespace(cmd,  # pylint: disable=unused-argument
                             client,
                             resource_group_name,
                             fleet_name,
                             managed_namespace_name,
                             no_wait=False):
    return sdk_no_wait(
        no_wait,
        client.begin_delete,
        resource_group_name,
        fleet_name,
        managed_namespace_name,
        polling_interval=5
    )


def show_managed_namespace(cmd,  # pylint: disable=unused-argument
                           client,
                           resource_group_name,
                           fleet_name,
                           managed_namespace_name):
    return client.get(
        resource_group_name=resource_group_name,
        fleet_name=fleet_name,
        managed_namespace_name=managed_namespace_name
    )


def list_managed_namespaces(cmd,  # pylint: disable=unused-argument
                            client,
                            resource_group_name,
                            fleet_name):
    return client.list_by_fleet(
        resource_group_name=resource_group_name,
        fleet_name=fleet_name
    )


def get_namespace_credentials(cmd,
                              client,  # pylint: disable=unused-argument
                              resource_group_name,
                              fleet_name,
                              managed_namespace_name,
                              path=os.path.join(os.path.expanduser('~'), '.kube', 'config'),
                              overwrite_existing=True,
                              context_name=None,
                              member_name=None):
    """
    Get credentials for a fleet hub or managed cluster and modifies the kubeconfig to set the default namespace.
    """
    fleet_client = cf_fleets(cmd.cli_ctx)
    with tempfile.NamedTemporaryFile(mode='w+', suffix='.kubeconfig') as temp_file:
        get_credentials(
            cmd=cmd,
            client=fleet_client,
            resource_group_name=resource_group_name,
            name=fleet_name,
            path=temp_file.name,
            overwrite_existing=overwrite_existing,
            context_name=context_name,
            member_name=member_name,
            skip_kubelogin_conversion=True  # Skip here, we'll convert after namespace modification
        )

        with open(temp_file.name, 'r', encoding='utf-8') as f:
            kubeconfig = yaml.safe_load(f.read())

        ctx = next(ctx for ctx in kubeconfig['contexts']
                   if ctx['name'] == kubeconfig['current-context'])
        ctx['context']['namespace'] = managed_namespace_name

        modified_kubeconfig = yaml.dump(kubeconfig, default_flow_style=False)
        print_or_merge_credentials(path, modified_kubeconfig, overwrite_existing, context_name)
        print(f"Default namespace set to '{managed_namespace_name}' for context '{kubeconfig.get('current-context')}'")

        # Apply kubelogin conversion to the final file after namespace modification
        _convert_kubeconfig_to_azurecli(path)


def create_cluster_mesh_profile(cmd,
                                client,
                                resource_group_name,
                                fleet_name,
                                name,
                                member_selector=None,
                                no_wait=False):
    cluster_mesh_profile_model = cmd.get_models(
        "ClusterMeshProfile",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="cluster_mesh_profiles"
    )
    cluster_mesh_profile_properties_model = cmd.get_models(
        "ClusterMeshProfileProperties",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="cluster_mesh_profiles"
    )
    member_selector_model = cmd.get_models(
        "MemberSelector",
        resource_type=CUSTOM_MGMT_FLEET,
        operation_group="cluster_mesh_profiles"
    )

    selector = None
    if member_selector is not None:
        selector = member_selector_model(by_label=member_selector)

    properties = cluster_mesh_profile_properties_model(member_selector=selector)
    profile = cluster_mesh_profile_model(properties=properties)

    return sdk_no_wait(
        no_wait,
        client.begin_create_or_update,
        resource_group_name,
        fleet_name,
        name,
        profile
    )


def show_cluster_mesh_profile(cmd,  # pylint: disable=unused-argument
                              client,
                              resource_group_name,
                              fleet_name,
                              name):
    return client.get(resource_group_name, fleet_name, name)


def list_cluster_mesh_profiles(cmd,  # pylint: disable=unused-argument
                               client,
                               resource_group_name,
                               fleet_name):
    return client.list_by_fleet(resource_group_name, fleet_name)


def delete_cluster_mesh_profile(cmd,  # pylint: disable=unused-argument
                                client,
                                resource_group_name,
                                fleet_name,
                                name,
                                no_wait=False):
    return sdk_no_wait(no_wait, client.begin_delete, resource_group_name, fleet_name, name)


def apply_cluster_mesh_profile(cmd,
                               client,
                               resource_group_name,
                               fleet_name,
                               name,
                               what_if=False,
                               no_wait=False):
    if what_if:
        return _apply_cluster_mesh_what_if(cmd, resource_group_name, fleet_name, name)

    return sdk_no_wait(no_wait, client.begin_apply, resource_group_name, fleet_name, name)


def list_cluster_mesh_profile_members(cmd,
                                      client,  # pylint: disable=unused-argument
                                      resource_group_name,
                                      fleet_name,
                                      name,
                                      selector=False):
    """List fleet members for a cluster mesh profile.

    Modes:
      --name cmp-1              members currently applied to the mesh
                                  (server-side: $filter=clusterMeshProfile eq cmp-1)
      --name cmp-1 --selector   members matching the profile's label selector
                                  (server-side: $filter=clusterMeshProfile.Selector eq cmp-1)
    """
    members_client = cf_fleet_members(cmd.cli_ctx)
    if selector:
        filter_expr = f"clusterMeshProfile.Selector eq {name}"
    else:
        filter_expr = f"clusterMeshProfile eq {name}"
    return members_client.list_by_fleet(resource_group_name, fleet_name, filter=filter_expr)


def _apply_cluster_mesh_what_if(cmd, resource_group_name, fleet_name, name):
    """Simulate apply by comparing currently-applied members vs selector-matched members."""
    members_client = cf_fleet_members(cmd.cli_ctx)

    # Members currently in the mesh (already applied)
    current_filter = f"clusterMeshProfile eq {name}"
    current_members = {
        m.name: m for m in members_client.list_by_fleet(
            resource_group_name, fleet_name, filter=current_filter
        )
    }

    # Members that match the selector (would be in the mesh after apply)
    selector_filter = f"clusterMeshProfile.Selector eq {name}"
    desired_members = {
        m.name: m for m in members_client.list_by_fleet(
            resource_group_name, fleet_name, filter=selector_filter
        )
    }

    results = []
    all_names = set(current_members.keys()) | set(desired_members.keys())

    for member_name in sorted(all_names):
        in_current = member_name in current_members
        in_desired = member_name in desired_members

        member = desired_members.get(member_name) or current_members.get(member_name)

        mesh_state = None
        if member.mesh_properties and member.mesh_properties.status:
            mesh_state = member.mesh_properties.status.state

        if in_desired and not in_current:
            action = "Add"
        elif in_current and not in_desired:
            action = "Remove"
        else:
            action = "-"

        results.append({
            "ClusterResourceId": member.cluster_resource_id,
            "ETag": member.e_tag,
            "Name": member.name,
            "Action": action,
            "MeshMembershipState": mesh_state or "-"
        })

    return results