azext_edge/edge/params.py

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

"""
CLI parameter definitions.
"""

from azure.cli.core.commands.parameters import (
    get_enum_type,
    get_three_state_flag,
    tags_type,
)
from knack.arguments import CaseInsensitiveList

from azext_edge.edge.providers.edge_api.dataflow import DataflowResourceKinds

from ._validators import (
    validate_namespace,
    validate_resource_name,
)
from .common import OpsServiceType
from .providers.check.common import ResourceOutputDetailLevel
from .providers.edge_api import (
    DeviceRegistryResourceKinds,
    MqResourceKinds,
)
from .providers.orchestration.common import (
    EXTENSION_MONIKER_TO_ALIAS_MAP,
    TRUST_SETTING_KEYS,
    X509_ISSUER_REF_KEYS,
    CloneSummaryMode,
    CloneTemplateMode,
    ConfigSyncModeType,
    IdentityUsageType,
    KubernetesDistroType,
    ListenerProtocol,
    MqMemoryProfile,
    MqServiceType,
    SchemaFormat,
    SchemaType,
    TlsKeyAlgo,
    TlsKeyRotation,
)


def load_iotops_arguments(self, _):
    """
    Load CLI Args for Knack parser
    """

    with self.argument_context("iot ops") as context:
        context.argument(
            "context_name",
            options_list=["--context"],
            help="Kubeconfig context name to use for k8s cluster communication. "
            "If no context is provided current_context is used.",
            arg_group="K8s Cluster",
        )
        context.argument(
            "namespace",
            options_list=["--namespace", "-n"],
            help="K8s cluster namespace the command should operate against. "
            "If no namespace is provided the kubeconfig current_context namespace will be used. "
            "If not defined, the fallback value `azure-iot-operations` will be used. ",
            validator=validate_namespace,
        )
        context.argument(
            "confirm_yes",
            options_list=["--yes", "-y"],
            arg_type=get_three_state_flag(),
            help="Confirm [y]es without a prompt. Useful for CI and automation scenarios.",
        )
        context.argument(
            "no_progress",
            options_list=["--no-progress"],
            arg_type=get_three_state_flag(),
            help="Disable visual representation of work.",
        )
        context.argument(
            "force",
            options_list=["--force"],
            arg_type=get_three_state_flag(),
            help="Force the operation to execute.",
        )
        context.argument(
            "tags",
            options_list=["--tags"],
            arg_type=tags_type,
        )
        context.argument(
            "instance_name",
            options_list=["--name", "-n"],
            help="IoT Operations instance name.",
        )
        context.argument(
            "instance_description",
            options_list=["--description"],
            help="Description of the IoT Operations instance.",
        )
        context.argument(
            "broker_name",
            options_list=["--broker", "-b"],
            help="Mqtt broker name.",
        )
        context.argument(
            "mi_user_assigned",
            options_list=["--mi-user-assigned"],
            help="The resource Id for the desired user-assigned managed identity to use with the instance.",
        )
        context.argument(
            "federated_credential_name",
            options_list=["--fc"],
            help="The federated credential name.",
        )
        context.argument(
            "use_self_hosted_issuer",
            options_list=["--self-hosted-issuer"],
            arg_type=get_three_state_flag(),
            help="Use the self-hosted oidc issuer for federation.",
        )
        context.argument(
            "config_file",
            options_list=["--config-file"],
            help="Path to a config file containing resource properties in json format. The config file "
            "should contain an object with properties compatible with the ARM representation of the resource. "
            "The object correlates directly with 'properties:{}' of the ARM resource.",
            arg_group="Config",
        )
        context.argument(
            "show_config",
            options_list=["--show-config"],
            arg_type=get_three_state_flag(),
            help="Show the generated resource config instead of invoking the API with it.",
        )

    with self.argument_context("iot ops identity") as context:
        context.argument(
            "usage_type",
            options_list=["--usage"],
            arg_type=get_enum_type(IdentityUsageType),
            help="Indicates the usage type of the associated identity.",
        )

    with self.argument_context("iot ops show") as context:
        context.argument(
            "show_tree",
            options_list=["--tree"],
            arg_type=get_three_state_flag(),
            help="Use to visualize the IoT Operations deployment against the backing cluster.",
        )

    with self.argument_context("iot ops support") as context:
        context.argument(
            "ops_services",
            nargs="+",
            action="extend",
            options_list=["--ops-service", "--svc"],
            choices=CaseInsensitiveList(OpsServiceType.list()),
            help="The IoT Operations service the support bundle creation should apply to. "
            "If no service is provided, the operation will default to capture all services. "
            "--ops-service can be used one or more times.",
        )
        context.argument(
            "log_age_seconds",
            options_list=["--log-age"],
            help="Container log age in seconds.",
            type=int,
        )
        context.argument(
            "bundle_dir",
            options_list=["--bundle-dir"],
            help="The local directory the produced bundle will be saved to. "
            "If no directory is provided the current directory is used.",
        )
        context.argument(
            "include_mq_traces",
            options_list=["--broker-traces"],
            arg_type=get_three_state_flag(),
            help="Include mqtt broker traces in the support bundle. "
            "Usage may add considerable size to the produced bundle.",
        )

    with self.argument_context("iot ops check") as context:
        context.argument(
            "pre_deployment_checks",
            options_list=["--pre"],
            help="Run pre-requisite checks to determine if the minimum "
            "requirements of a service deployment are fulfilled.",
            arg_type=get_three_state_flag(),
        )
        context.argument(
            "post_deployment_checks",
            options_list=["--post"],
            help="Run post-deployment checks.",
            arg_type=get_three_state_flag(),
        )
        context.argument(
            "as_object",
            options_list=["--as-object"],
            help="Output check content and validations in a CI friendly data structure.",
            arg_type=get_three_state_flag(),
            arg_group="Format",
        )
        context.argument(
            "ops_service",
            options_list=["--ops-service", "--svc"],
            choices=CaseInsensitiveList(OpsServiceType.list_check_services()),
            help="The IoT Operations service deployment that will be evaluated.",
        )
        context.argument(
            "resource_kinds",
            nargs="*",
            options_list=["--resources"],
            choices=CaseInsensitiveList(
                set(
                    [
                        DeviceRegistryResourceKinds.ASSET.value,
                        DeviceRegistryResourceKinds.ASSETENDPOINTPROFILE.value,
                        MqResourceKinds.BROKER.value,
                        MqResourceKinds.BROKER_LISTENER.value,
                        MqResourceKinds.BROKER_AUTHENTICATION.value,
                        MqResourceKinds.BROKER_AUTHORIZATION.value,
                        DataflowResourceKinds.DATAFLOW.value,
                        DataflowResourceKinds.DATAFLOWENDPOINT.value,
                        DataflowResourceKinds.DATAFLOWPROFILE.value,
                    ]
                )
            ),
            help="Only run checks on specific resource kinds. Use space-separated values.",
        ),
        context.argument(
            "detail_level",
            options_list=["--detail-level"],
            default=ResourceOutputDetailLevel.summary.value,
            choices=ResourceOutputDetailLevel.list(),
            arg_type=get_enum_type(ResourceOutputDetailLevel),
            help="Controls the level of detail displayed in the check output. "
            "Choose 0 for a summary view (minimal output), "
            "1 for a detailed view (more comprehensive information), "
            "or 2 for a verbose view (all available information).",
        ),
        context.argument(
            "resource_name",
            options_list=["--resource-name", "--rn"],
            help="Only run checks for the specific resource name. "
            "The name is case insensitive. "
            "Glob patterns '*' and '?' are supported. "
            "Note: Only alphanumeric characters, hyphens, '?' and '*' are allowed.",
            validator=validate_resource_name,
        ),

    with self.argument_context("iot ops dataflow") as context:
        context.argument(
            "instance_name",
            options_list=["--instance", "-i"],
            help="IoT Operations instance name.",
        )
        context.argument(
            "dataflow_name",
            options_list=["--name", "-n"],
            help="Dataflow name.",
        )
        context.argument(
            "profile_name",
            options_list=["--profile", "-p"],
            help="Dataflow profile name.",
        )

    with self.argument_context("iot ops dataflow profile") as context:
        context.argument(
            "profile_name",
            options_list=["--name", "-n"],
            help="Dataflow profile name.",
        )

    with self.argument_context("iot ops dataflow endpoint") as context:
        context.argument(
            "endpoint_name",
            options_list=["--name", "-n"],
            help="Dataflow endpoint name.",
        )

    with self.argument_context("iot ops broker") as context:
        context.argument(
            "instance_name",
            options_list=["--instance", "-i"],
            help="IoT Operations instance name.",
        )
        context.argument(
            "broker_name",
            options_list=["--name", "-n"],
            help="Broker name.",
        )

    with self.argument_context("iot ops broker listener") as context:
        context.argument(
            "listener_name",
            options_list=["--name", "-n"],
            help="Listener name.",
        )
        context.argument(
            "broker_name",
            options_list=["--broker", "-b"],
            help="Broker name.",
        )

    with self.argument_context("iot ops broker listener port") as context:
        context.argument(
            "listener_name",
            options_list=["--listener", "-n"],
            help="Listener name.",
        )
        context.argument(
            "port",
            type=int,
            options_list=["--port"],
            help="Listener service port.",
        )
        context.argument(
            "nodeport",
            type=int,
            options_list=["--nodeport"],
            help="The listener service will exposes a static port on each Node's IP address. "
            "Only relevant when this port is associated with a NodePort listener.",
            arg_group="Node Port",
        )
        context.argument(
            "service_name",
            options_list=["--service-name"],
            help="Kubernetes service name of the listener. Used when a target listener does not exist.",
        )
        context.argument(
            "service_type",
            options_list=["--service-type"],
            arg_type=get_enum_type(MqServiceType, default=None),
            help="Kubernetes service type of the listener. Used when a target listener does not exist.",
        )
        context.argument(
            "protocol",
            options_list=["--protocol"],
            arg_type=get_enum_type(ListenerProtocol, default=None),
            help="Protocol to use for client connections.",
        )
        context.argument(
            "authn_ref", options_list=["--authn-ref"], help="Authentication reference (name).", arg_group="Auth"
        )
        context.argument(
            "authz_ref", options_list=["--authz-ref"], help="Authorization reference (name).", arg_group="Auth"
        )
        context.argument(
            "tls_auto_issuer_ref",
            options_list=["--tls-issuer-ref"],
            nargs="+",
            help="Cert-manager issuer reference. Format is space-separated "
            f"key=value pairs. The following keys are supported: `{'`, `'.join(X509_ISSUER_REF_KEYS)}`. "
            "`kind` and `name` are required, while `group` has a default value of 'cert-manager.io'.",
            arg_group="TLS Auto",
        )
        context.argument(
            "tls_auto_duration",
            options_list=["--tls-duration"],
            help="Lifetime of certificate. Must be specified using a time.Duration format (h|m|s). "
            "E.g. 240h for 240 hours and 45m for 45 minutes.",
            arg_group="TLS Auto",
        )
        context.argument(
            "tls_auto_key_algo",
            options_list=["--tls-key-algo"],
            arg_type=get_enum_type(TlsKeyAlgo, default=None),
            help="Algorithm for private key. ",
            arg_group="TLS Auto",
        )
        context.argument(
            "tls_auto_key_rotation_policy",
            options_list=["--tls-key-rotation"],
            arg_type=get_enum_type(TlsKeyRotation, default=None),
            help="Cert-manager private key rotation policy.",
            arg_group="TLS Auto",
        )
        context.argument(
            "tls_auto_renew_before",
            options_list=["--tls-renew-before"],
            help="When to begin renewing certificate. Must be specified using a Go time.Duration format (h|m|s). "
            "E.g. 240h for 240 hours and 45m for 45 minutes.",
            arg_group="TLS Auto",
        )
        context.argument(
            "tls_auto_san_dns",
            options_list=["--tls-san-dns"],
            nargs="+",
            help="DNS subject alternative names for the certificate. Use space-separated values.",
            arg_group="TLS Auto",
        )
        context.argument(
            "tls_auto_san_ip",
            options_list=["--tls-san-ip"],
            nargs="+",
            help="IP subject alternative names for the certificate. Use space-separated values.",
            arg_group="TLS Auto",
        )
        context.argument(
            "tls_auto_secret_name",
            options_list=["--tls-secret-name"],
            help="Secret for storing server certificate. Any existing data will be overwritten. This is a reference to "
            "the secret through an identifying name, not the secret itself.",
            arg_group="TLS Auto",
        )
        context.argument(
            "tls_manual_secret_ref",
            options_list=["--tls-man-secret-ref"],
            help="Secret containing an X.509 client certificate. This is a "
            "reference to the secret through an identifying name, not the secret itself.",
            arg_group="TLS Manual",
        )

    with self.argument_context("iot ops broker authn") as context:
        context.argument(
            "authn_name",
            options_list=["--name", "-n"],
            help="Mqtt broker authentication resource name.",
        )
        context.argument(
            "broker_name",
            options_list=["--broker", "-b"],
            help="Mqtt broker name.",
        )

    with self.argument_context("iot ops broker authn method add") as context:
        context.argument(
            "authn_name",
            options_list=["--authn", "-n"],
            help="Mqtt broker authentication resource name.",
        )
        context.argument(
            "sat_audiences",
            options_list=["--sat-aud"],
            nargs="+",
            help="Space-separated list of allowed audiences.",
            arg_group="SAT",
        )
        context.argument(
            "x509_client_ca_cm",
            options_list=["--x509-client-ca-ref"],
            help="Name of the configmap containing the trusted client ca cert resource. Default value is 'client-ca'.",
            arg_group="x509",
        )
        context.argument(
            "x509_attrs",
            options_list=["--x509-attr"],
            nargs="+",
            action="extend",
            help="Specify attributes in the authentication resource for authorizing clients based on their "
            "certificate properties. You can apply authorization rules to clients by using x509 certificates "
            "with these attributes. Format is space-separated key=value pairs where the key uses object dot notation "
            "such as 'a.b.c=value'. Can be used one or more times.",
            arg_group="x509",
        )
        context.argument(
            "custom_endpoint",
            options_list=["--custom-ep"],
            help="Endpoint to use for the custom auth service. Format is 'https://.*'.",
            arg_group="Custom",
        )
        context.argument(
            "custom_ca_cm",
            options_list=["--custom-ca-ref"],
            help="Name of the configmap containing the CA certificate for validating the "
            "custom authentication server's certificate.",
            arg_group="Custom",
        )
        context.argument(
            "custom_x509_secret_ref",
            options_list=["--custom-x509-secret-ref"],
            help="Reference to Kubernetes secret containing a client certificate.",
            arg_group="Custom",
        )
        context.argument(
            "custom_http_headers",
            options_list=["--custom-header"],
            nargs="+",
            action="extend",
            help="http headers to pass to the custom authentication server. Format is space-separated key=value pairs. "
            "Can be used one or more times.",
            arg_group="Custom",
        )

    with self.argument_context("iot ops broker authz") as context:
        context.argument(
            "authz_name",
            options_list=["--name", "-n"],
            help="Mqtt broker authorization resource name.",
        )
        context.argument(
            "broker_name",
            options_list=["--broker", "-b"],
            help="Mqtt broker name.",
        )

    for cmd_space in ["iot ops init", "iot ops create"]:
        with self.argument_context(cmd_space) as context:
            context.argument(
                "cluster_name",
                options_list=["--cluster"],
                help="Target cluster name for IoT Operations deployment.",
            )
            context.argument(
                "cluster_namespace",
                options_list=["--cluster-namespace"],
                help="The cluster namespace IoT Operations infra will be deployed to. Must be lowercase.",
            )
            context.argument(
                "custom_location_name",
                options_list=["--custom-location"],
                help="The custom location name corresponding to the IoT Operations deployment. "
                "The default is in the form 'location-{hash(5)}'.",
            )
            context.argument(
                "location",
                options_list=["--location"],
                help="The region that will be used for provisioned resource collateral. "
                "If not provided the connected cluster location will be used.",
            )
            context.argument(
                "enable_rsync_rules",
                options_list=["--enable-rsync"],
                arg_type=get_three_state_flag(),
                help="Resource sync rules will be included in the IoT Operations deployment.",
            )
            context.argument(
                "ensure_latest",
                options_list=["--ensure-latest"],
                arg_type=get_three_state_flag(),
                help="Ensure the latest IoT Ops CLI is being used, raising an error if an upgrade is available.",
            )
            # Schema Registry
            context.argument(
                "schema_registry_resource_id",
                options_list=["--sr-resource-id"],
                help="The schema registry resource Id to use with IoT Operations.",
            )
            # Akri
            context.argument(
                "container_runtime_socket",
                options_list=["--runtime-socket"],
                help="The default node path of the container runtime socket. If not provided (default), the "
                "socket path is determined by --kubernetes-distro.",
                arg_group="Akri",
            )
            context.argument(
                "kubernetes_distro",
                arg_type=get_enum_type(KubernetesDistroType),
                options_list=["--kubernetes-distro"],
                help="The Kubernetes distro to use for Akri configuration. The selected distro implies the "
                "default container runtime socket path when no --runtime-socket value is provided.",
                arg_group="Akri",
            )
            # Broker
            context.argument(
                "custom_broker_config_file",
                options_list=["--broker-config-file"],
                help="Path to a json file with custom broker config properties. "
                "File config content is used over individual broker config parameters. "
                "Useful for advanced scenarios. "
                "The expected format is described at https://aka.ms/aziotops-broker-config.",
                arg_group="Broker",
            )
            context.argument(
                "add_insecure_listener",
                options_list=[
                    "--add-insecure-listener",
                    context.deprecate(
                        target="--mq-insecure",
                        redirect="--add-insecure-listener",
                        hide=True,
                    ),
                ],
                arg_type=get_three_state_flag(),
                help="When enabled the mqtt broker deployment will include a listener "
                f"of service type {MqServiceType.LOADBALANCER.value}, bound to port 1883 with no authN or authZ. "
                "For non-production workloads only.",
                arg_group="Broker",
            )
            # Broker Config
            context.argument(
                "broker_frontend_replicas",
                type=int,
                options_list=["--broker-frontend-replicas", "--fr"],
                help="Mqtt broker frontend replicas. Min value: 1, max value: 16.",
                arg_group="Broker",
            )
            context.argument(
                "broker_frontend_workers",
                type=int,
                options_list=["--broker-frontend-workers", "--fw"],
                help="Mqtt broker frontend workers. Min value: 1, max value: 16.",
                arg_group="Broker",
            )
            context.argument(
                "broker_backend_redundancy_factor",
                type=int,
                options_list=["--broker-backend-rf", "--br"],
                help="Mqtt broker backend redundancy factor. Min value: 1, max value: 5.",
                arg_group="Broker",
            )
            context.argument(
                "broker_backend_workers",
                type=int,
                options_list=["--broker-backend-workers", "--bw"],
                help="Mqtt broker backend workers. Min value: 1, max value: 16.",
                arg_group="Broker",
            )
            context.argument(
                "broker_backend_partitions",
                type=int,
                options_list=["--broker-backend-part", "--bp"],
                help="Mqtt broker backend partitions. Min value: 1, max value: 16.",
                arg_group="Broker",
            )
            context.argument(
                "broker_memory_profile",
                arg_type=get_enum_type(MqMemoryProfile),
                options_list=["--broker-mem-profile", "--mp"],
                help="Mqtt broker memory profile.",
                arg_group="Broker",
            )
            context.argument(
                "broker_service_type",
                arg_type=get_enum_type(MqServiceType),
                options_list=["--broker-listener-type", "--lt"],
                help="Service type associated with the default mqtt broker listener.",
                arg_group="Broker",
                deprecate_info=context.deprecate(hide=True),
            )
            context.argument(
                "enable_fault_tolerance",
                arg_type=get_three_state_flag(),
                options_list=["--enable-fault-tolerance"],
                help="Enable fault tolerance for Azure Arc Container Storage. At least 3 cluster nodes are required.",
                arg_group="Container Storage",
            )
            context.argument(
                "dataflow_profile_instances",
                type=int,
                options_list=["--df-profile-instances"],
                help="The instance count associated with the default dataflow profile.",
                arg_group="Dataflow",
            )
            context.argument(
                "trust_settings",
                options_list=["--trust-settings"],
                nargs="+",
                action="store",
                help="Settings for user provided trust bundle. Used for component TLS. Format is space-separated "
                f"key=value pairs. The following keys are required: `{'`, `'.join(TRUST_SETTING_KEYS)}`. If not "
                "used, a system provided self-signed trust bundle is configured.",
                arg_group="Trust",
            )
            context.argument(
                "user_trust",
                options_list=["--user-trust", "--ut"],
                arg_type=get_three_state_flag(),
                help="Skip the deployment of the system cert-manager and trust-manager "
                "in favor of a user-provided configuration.",
                arg_group="Trust",
            )

            for moniker in EXTENSION_MONIKER_TO_ALIAS_MAP:
                alias = EXTENSION_MONIKER_TO_ALIAS_MAP[moniker]
                if alias in ["acs", "ssc", "ops"]:
                    context.argument(
                        f"{alias}_config",
                        options_list=[f"--{alias}-config"],
                        nargs="+",
                        action="extend",
                        help=f"{moniker} arc extension custom config. Format is space-separated key=value pairs "
                        f"or just the key. This option can be used one or more times.",
                        arg_group="Extension Config",
                    )
                    context.argument(
                        f"{alias}_version",
                        options_list=[f"--{alias}-version"],
                        help=f"Use to override the built-in {moniker} arc extension version.",
                        arg_group="Extension Config",
                        deprecate_info=context.deprecate(hide=True),
                    )
                    context.argument(
                        f"{alias}_train",
                        options_list=[f"--{alias}-train"],
                        help=f"Use to override the built-in {moniker} arc extension release train.",
                        arg_group="Extension Config",
                        deprecate_info=context.deprecate(hide=True),
                    )

    for cmd_space in ["iot ops create", "iot ops update"]:
        with self.argument_context(cmd_space) as context:
            context.argument(
                "instance_features",
                options_list=["--feature"],
                nargs="+",
                action="extend",
                help="Instance feature config. The settings of a component and/or it's mode can be configured. "
                "Component mode syntax is `{component}.mode={mode}` where known mode values are: "
                "`Stable`, `Preview` and `Disabled`. Component setting syntax is "
                "`{component}.settings.{setting}={value}` where known setting values are `Enabled` or `Disabled`. "
                "This option can be used one or more times.",
            )

    with self.argument_context("iot ops upgrade") as context:
        for moniker in EXTENSION_MONIKER_TO_ALIAS_MAP:
            alias = EXTENSION_MONIKER_TO_ALIAS_MAP[moniker]
            context.argument(
                f"{alias}_config",
                options_list=[f"--{alias}-config"],
                nargs="+",
                action="extend",
                help=f"{moniker} arc extension custom config. Format is space-separated key=value pairs "
                f"or just the key. This option can be used one or more times.",
                arg_group="Extension Config",
            )
            context.argument(
                f"{alias}_config_sync_mode",
                options_list=[f"--{alias}-config-sync"],
                help=f"{moniker} arc extension config sync mode. This option is applicable if an upgrade is "
                "requested to a known version. Mode 'full' will alter current config to the target, "
                "'add' will apply additive changes only, 'none' is a no-op.",
                arg_type=get_enum_type(ConfigSyncModeType, default=ConfigSyncModeType.FULL.value),
                arg_group="Extension Config",
                deprecate_info=context.deprecate(hide=True),
            )
            context.argument(
                f"{alias}_version",
                options_list=[f"--{alias}-version"],
                help=f"Use to override the built-in {moniker} arc extension version.",
                arg_group="Extension Config",
                deprecate_info=context.deprecate(hide=True),
            )
            context.argument(
                f"{alias}_train",
                options_list=[f"--{alias}-train"],
                help=f"Use to override the built-in {moniker} arc extension release train.",
                arg_group="Extension Config",
                deprecate_info=context.deprecate(hide=True),
            )
        context.argument(
            "force",
            options_list=["--force"],
            arg_type=get_three_state_flag(),
            help="Force the operation to continue. Use to get around guards, such as those preventing downgrade.",
            arg_group="Extension Config",
            deprecate_info=context.deprecate(hide=True),
        )

    with self.argument_context("iot ops delete") as context:
        context.argument(
            "include_dependencies",
            options_list=["--include-deps"],
            arg_type=get_three_state_flag(),
            help="Indicates the command should remove IoT Operations dependencies. "
            "This option is intended to reverse the application of init.",
        )
        context.argument(
            "cluster_name",
            options_list=["--cluster"],
            help="Target cluster name for IoT Operations deletion.",
        )

    with self.argument_context("iot ops secretsync") as context:
        context.argument(
            "keyvault_resource_id",
            options_list=["--kv-resource-id"],
            help="Key Vault ARM resource Id.",
        )
        context.argument(
            "spc_name",
            options_list=["--spc"],
            help="The default secret provider class name for secret sync enablement. "
            "The default pattern is 'spc-ops-{hash}'.",
        )
        context.argument(
            "skip_role_assignments",
            options_list=["--skip-ra"],
            arg_type=get_three_state_flag(),
            help="When used the role assignment step of the operation will be skipped.",
        )
        context.argument(
            "instance_name",
            options_list=["--instance", "-i", "-n"],
            help="IoT Operations instance name.",
        )

    with self.argument_context("iot ops schema") as context:
        context.argument(
            "schema_name",
            options_list=["--name", "-n"],
            help="Schema name.",
        )
        context.argument(
            "schema_registry_name",
            options_list=["--registry"],
            help="Schema registry name.",
        )
        context.argument(
            "schema_format", options_list=["--format"], help="Schema format.", arg_type=get_enum_type(SchemaFormat)
        )
        context.argument(
            "schema_type", options_list=["--type"], help="Schema type.", arg_type=get_enum_type(SchemaType)
        )
        context.argument(
            "description",
            options_list=["--desc"],
            help="Description for the schema.",
        )
        context.argument(
            "display_name",
            options_list=["--display-name"],
            help="Display name for the schema.",
        )
        context.argument(
            "schema_version",
            options_list=["--version", "--ver"],
            help="Schema version name.",
            type=int,
            arg_group="Version",
        )
        context.argument(
            "schema_version_content",
            options_list=["--version-content", "--vc"],
            help="File path containing or inline content for the version.",
            arg_group="Version",
        )
        context.argument(
            "schema_version_description",
            options_list=["--version-desc", "--vd"],
            help="Description for the version.",
            arg_group="Version",
        )

    with self.argument_context("iot ops schema show-dataflow-refs") as context:
        context.argument(
            "schema_name",
            options_list=["--schema"],
            help="Schema name. Required if using --version.",
        )
        context.argument(
            "schema_version",
            options_list=["--version", "--ver"],
            help="Schema version name. If used, --latest will be ignored.",
            type=int,
            arg_group=None,
        )
        context.argument(
            "latest",
            options_list=["--latest"],
            help="Flag to show only the latest version(s).",
            arg_type=get_three_state_flag(),
        )

    with self.argument_context("iot ops schema registry") as context:
        context.argument(
            "schema_registry_name",
            options_list=["--name", "-n"],
            help="Schema registry name.",
        )
        context.argument(
            "registry_namespace",
            options_list=["--registry-namespace", "--rn"],
            help="Schema registry namespace. Uniquely identifies a schema registry within a tenant.",
        )
        context.argument(
            "tags",
            options_list=["--tags"],
            arg_type=tags_type,
        )
        context.argument(
            "description",
            options_list=["--desc"],
            help="Description for the schema registry.",
        )
        context.argument(
            "display_name",
            options_list=["--display-name"],
            help="Display name for the schema registry.",
        )
        context.argument(
            "location",
            options_list=["--location", "-l"],
            help="Region to create the schema registry. "
            "If no location is provided the resource group location will be used.",
        )
        context.argument(
            "storage_account_resource_id",
            options_list=["--sa-resource-id"],
            help="Storage account resource Id to be used with the schema registry.",
        )
        context.argument(
            "storage_container_name",
            options_list=["--sa-container"],
            help="Storage account container name where schemas will be stored.",
        )
        context.argument(
            "custom_role_id",
            options_list=["--custom-role-id"],
            help="Fully qualified role definition Id in the following format: "
            "/subscriptions/{subscriptionId}/providers/Microsoft.Authorization/roleDefinitions/{roleId}",
        )

    with self.argument_context("iot ops connector opcua") as context:
        context.argument(
            "instance_name",
            options_list=["--instance", "-i", "-n"],
            help="IoT Operations instance name.",
        )
        context.argument(
            "resource_group",
            options_list=["--resource-group", "-g"],
            help="Instance resource group.",
        )
        context.argument(
            "include_secrets",
            options_list=["--include-secrets"],
            help="Indicates the command should remove the key vault secrets "
            "associated with the certificate(s). This option will delete and "
            "purge the secrets.",
            arg_type=get_three_state_flag(),
        )
        context.argument(
            "certificate_names",
            options_list=["--certificate-names", "--cn"],
            nargs="+",
            help="Space-separated certificate names to remove. "
            "Note: the names can be found under the corresponding "
            "secretsync resource property 'targetKey'.",
        )
        context.argument(
            "overwrite_secret",
            options_list=["--overwrite-secret"],
            arg_type=get_three_state_flag(),
            help="Confirm [y]es without a prompt to overwrite secret. "
            "if secret name existed in Azure key vault. Useful for "
            "CI and automation scenarios.",
        )

    with self.argument_context("iot ops connector opcua trust") as context:
        context.argument(
            "file",
            options_list=["--certificate-file", "--cf"],
            help="Path to the certificate file in .der or .crt format.",
        )
        context.argument(
            "secret_name",
            options_list=["--secret-name", "-s"],
            help="Secret name in the Key Vault. If not provided, the "
            "certificate file name will be used to generate the secret name.",
        )

    with self.argument_context("iot ops connector opcua issuer") as context:
        context.argument(
            "file",
            options_list=["--certificate-file", "--cf"],
            help="Path to the certificate file in .der, .crt or .crl format.",
        )
        context.argument(
            "secret_name",
            options_list=["--secret-name", "-s"],
            help="Secret name in the Key Vault. If not provided, the "
            "certificate file name will be used to generate the secret name.",
        )

    with self.argument_context("iot ops connector opcua client") as context:
        context.argument(
            "public_key_file",
            options_list=["--public-key-file", "--pkf"],
            help="File that contains the enterprise grade application "
            "instance certificate public key in .der format. File "
            "name will be used to generate the public key secret name.",
        )
        context.argument(
            "private_key_file",
            options_list=["--private-key-file", "--prkf"],
            help="File that contains the enterprise grade application "
            "instance certificate private key in .pem format. File name "
            "will be used to generate the private key secret name.",
        )
        context.argument(
            "subject_name",
            options_list=["--subject-name", "--sn"],
            help="The subject name string embedded in the application instance certificate. "
            "Can be found under public key certificate. When provided, the value will be used to "
            "validate the certificate's subject name.",
        )
        context.argument(
            "application_uri",
            options_list=["--application-uri", "--au"],
            help="The application instance URI embedded in the application instance. "
            "Can be found under public key certificate. When provided, the value will be used to "
            "validate the certificate's application URI.",
        )
        context.argument(
            "public_key_secret_name",
            options_list=["--public-key-secret-name", "--pks"],
            help="Public key secret name in the Key Vault. If not provided, the "
            "certificate file name will be used to generate the secret name.",
        )
        context.argument(
            "private_key_secret_name",
            options_list=["--private-key-secret-name", "--prks"],
            help="Private key secret name in the Key Vault. If not provided, the "
            "certificate file name will be used to generate the secret name.",
        )

    with self.argument_context("iot ops schema version") as context:
        context.argument("version_name", options_list=["--name", "-n"], help="Schema version name.", type=int)
        context.argument(
            "schema_name",
            options_list=["--schema"],
            help="Schema name.",
        )
        context.argument(
            "description",
            options_list=["--desc"],
            help="Description for the schema version.",
        )
        context.argument(
            "schema_version_content",
            options_list=["--content"],
            help="File path containing or inline content for the version.",
            arg_group=None,
        )

    with self.argument_context("iot ops clone") as context:
        context.argument(
            "summary_mode",
            options_list=["--summary"],
            arg_type=get_enum_type(CloneSummaryMode, default=CloneSummaryMode.SIMPLE.value),
            help="Deployment summary option.",
        )
        context.argument(
            "instance_name",
            options_list=["--name", "-n"],
            help="The model instance to clone.",
        )
        context.argument(
            "resource_group_name",
            options_list=["--resource-group", "-g"],
            help="The resource group the model instance to clone resides in.",
        )
        context.argument(
            "to_dir",
            options_list=["--to-dir"],
            help="The local directory the instance clone definitions will be stored in.",
            arg_group="Local Target",
        )
        context.argument(
            "template_mode",
            options_list=["--mode"],
            arg_type=get_enum_type(CloneTemplateMode, default=CloneTemplateMode.NESTED.value),
            help="When mode 'nested' is used, sub-deployments will be self-contained in the root deployment. "
            "When mode 'linked' is used, asset related sub-deployments will be split and stored as separate files "
            "linked by the root deployment.",
        )
        context.argument(
            "linked_base_uri",
            options_list=["--base-uri"],
            help="Base URI to use for template links. If not provided a relative path strategy will be used. "
            "Relevant when --mode is set to 'linked'. "
            "Example: 'https://raw.githubusercontent.com/myorg/myproject/main/myclones/'.",
            arg_group="Local Target",
        )
        context.argument(
            "to_cluster_id",
            options_list=["--to-cluster-id"],
            help="The resource Id of the connected cluster the clone will be applied to.",
            arg_group="Cluster Target",
        )
        context.argument(
            "to_instance_name",
            options_list=["--to-instance"],
            help="The instance name that will be used when replicating the clone. If omitted the "
            "model instance name will be used.",
            arg_group="Cluster Target",
        )
        context.argument(
            "use_self_hosted_issuer",
            options_list=["--self-hosted-issuer"],
            arg_type=get_three_state_flag(),
            help="Use the self-hosted oidc issuer for federation. Only applicable if "
            "user-assigned managed identities are associated to the model instance.",
            arg_group="Cluster Target",
        )