to be refactored

src/acat/azext_acat/aaz/latest/app_compliance_automation/__init__.py

@@ -1,12 +1,13 @@
-# --------------------------------------------------------------------------------------------
-# Copyright (c) Microsoft Corporation. All rights reserved.
-# Licensed under the MIT License. See License.txt in the project root for license information.
-#
-# Code generated by aaz-dev-tools
-# --------------------------------------------------------------------------------------------
-
-# pylint: skip-file
-# flake8: noqa
-
-from .__cmd_group import *
-from ._onboard import *
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+#
+# Code generated by aaz-dev-tools
+# --------------------------------------------------------------------------------------------
+
+# pylint: skip-file
+# flake8: noqa
+
+from .__cmd_group import *
+from ._onboard import *
+from ._trigger_evaluation import *

src/acat/azext_acat/aaz/latest/app_compliance_automation/_trigger_evaluation.py

@@ -0,0 +1,235 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+#
+# Code generated by aaz-dev-tools
+# --------------------------------------------------------------------------------------------
+
+# pylint: skip-file
+# flake8: noqa
+
+from azure.cli.core.aaz import *
+from azure.cli.core._profile import Profile
+
+
+@register_command(
+    "acat quick-evaluation",
+)
+class TriggerEvaluation(AAZCommand):
+    """Trigger an evaluation for a list of resources.
+    """
+
+    _aaz_info = {
+        "version": "2024-06-27",
+        "resources": [
+            ["mgmt-plane", "/providers/microsoft.appcomplianceautomation/triggerevaluation", "2024-06-27"],
+        ]
+    }
+
+    AZ_SUPPORT_NO_WAIT = True
+
+    def _handler(self, command_args):
+        super()._handler(command_args)
+        return self.build_lro_poller(self._execute_operations, self._output)
+
+    _args_schema = None
+
+    @classmethod
+    def _build_arguments_schema(cls, *args, **kwargs):
+        if cls._args_schema is not None:
+            return cls._args_schema
+        cls._args_schema = super()._build_arguments_schema(*args, **kwargs)
+
+        # define Arg Group "Body"
+
+        _args_schema = cls._args_schema
+        _args_schema.resource_ids = AAZListArg(
+            options=["--resource-ids"],
+            arg_group="Body",
+            help="List of resource ids to be evaluated",
+            required=True,
+        )
+
+        resource_ids = cls._args_schema.resource_ids
+        resource_ids.Element = AAZStrArg()
+        return cls._args_schema
+
+    def _execute_operations(self):
+        profile = Profile(cli_ctx=self.cli_ctx)
+        creds,_,_=profile.get_raw_token();
+        self.pre_operations()
+        yield self.ProviderActionsTriggerEvaluation(ctx=self.ctx)(creds[1])
+        self.post_operations()
+
+    @register_callback
+    def pre_operations(self):
+        pass
+
+    @register_callback
+    def post_operations(self):
+        pass
+
+    def _output(self, *args, **kwargs):
+        result = self.deserialize_output(self.ctx.vars.instance, client_flatten=True)
+        return result
+
+    class ProviderActionsTriggerEvaluation(AAZHttpOperation):
+        CLIENT_TYPE = "MgmtClient"
+
+        def __call__(self, *args, **kwargs):
+            self.token="Bearer "+args[0]
+            request = self.make_request()
+            session = self.client.send_request(request=request, stream=False, **kwargs)
+            if session.http_response.status_code in [202]:
+                return self.client.build_lro_polling(
+                    self.ctx.args.no_wait,
+                    session,
+                    self.on_200,
+                    self.on_error,
+                    lro_options={"final-state-via": "location"},
+                    path_format_arguments=self.url_parameters,
+                )
+            if session.http_response.status_code in [200]:
+                return self.client.build_lro_polling(
+                    self.ctx.args.no_wait,
+                    session,
+                    self.on_200,
+                    self.on_error,
+                    lro_options={"final-state-via": "location"},
+                    path_format_arguments=self.url_parameters,
+                )
+
+            return self.on_error(session.http_response)
+
+        @property
+        def url(self):
+            return self.client.format_url(
+                "/providers/Microsoft.AppComplianceAutomation/triggerEvaluation",
+                **self.url_parameters
+            )
+
+        @property
+        def method(self):
+            return "POST"
+
+        @property
+        def error_format(self):
+            return "MgmtErrorFormat"
+
+        @property
+        def query_parameters(self):
+            parameters = {
+                **self.serialize_query_param(
+                    "api-version", "2024-06-27",
+                    required=True,
+                ),
+            }
+            return parameters
+
+        @property
+        def header_parameters(self):
+            parameters = {
+                **self.serialize_header_param(
+                    "Content-Type", "application/json",
+                ),
+                **self.serialize_header_param(
+                    "Accept", "application/json",
+                ),
+                **self.serialize_header_param(
+                    "x-ms-aad-user-token", self.token,
+                ),
+            }
+            return parameters
+
+        @property
+        def content(self):
+            _content_value, _builder = self.new_content_builder(
+                self.ctx.args,
+                typ=AAZObjectType,
+                typ_kwargs={"flags": {"required": True, "client_flatten": True}}
+            )
+            _builder.set_prop("resourceIds", AAZListType, ".resource_ids", typ_kwargs={"flags": {"required": True}})
+
+            resource_ids = _builder.get(".resourceIds")
+            if resource_ids is not None:
+                resource_ids.set_elements(AAZStrType, ".")
+
+            return self.serialize_content(_content_value)
+
+        def on_200(self, session):
+            data = self.deserialize_http_content(session)
+            self.ctx.set_var(
+                "instance",
+                data,
+                schema_builder=self._build_schema_on_200
+            )
+
+        _schema_on_200 = None
+
+        @classmethod
+        def _build_schema_on_200(cls):
+            if cls._schema_on_200 is not None:
+                return cls._schema_on_200
+
+            cls._schema_on_200 = AAZObjectType()
+
+            _schema_on_200 = cls._schema_on_200
+            _schema_on_200.properties = AAZObjectType()
+
+            properties = cls._schema_on_200.properties
+            properties.evaluation_end_time = AAZStrType(
+                serialized_name="evaluationEndTime",
+                flags={"read_only": True},
+            )
+            properties.quick_assessments = AAZListType(
+                serialized_name="quickAssessments",
+            )
+            properties.resource_ids = AAZListType(
+                serialized_name="resourceIds",
+            )
+            properties.trigger_time = AAZStrType(
+                serialized_name="triggerTime",
+                flags={"read_only": True},
+            )
+
+            quick_assessments = cls._schema_on_200.properties.quick_assessments
+            quick_assessments.Element = AAZObjectType()
+
+            _element = cls._schema_on_200.properties.quick_assessments.Element
+            _element.description = AAZStrType(
+                flags={"read_only": True},
+            )
+            _element.display_name = AAZStrType(
+                serialized_name="displayName",
+                flags={"read_only": True},
+            )
+            _element.remediation_link = AAZStrType(
+                serialized_name="remediationLink",
+                flags={"read_only": True},
+            )
+            _element.resource_id = AAZStrType(
+                serialized_name="resourceId",
+                flags={"read_only": True},
+            )
+            _element.resource_status = AAZStrType(
+                serialized_name="resourceStatus",
+            )
+            _element.responsibility_id = AAZStrType(
+                serialized_name="responsibilityId",
+                flags={"read_only": True},
+            )
+            _element.timestamp = AAZStrType(
+                flags={"read_only": True},
+            )
+
+            resource_ids = cls._schema_on_200.properties.resource_ids
+            resource_ids.Element = AAZStrType()
+
+            return cls._schema_on_200
+
+
+class _TriggerEvaluationHelper:
+    """Helper class for TriggerEvaluation"""
+
+
+__all__ = ["TriggerEvaluation"]

src/acat/azext_acat/aaz/latest/app_compliance_automation/polling.py

@@ -0,0 +1,100 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+#
+# Code generated by aaz-dev-tools
+# --------------------------------------------------------------------------------------------
+
+from azure.core.exceptions import HttpResponseError
+from azure.core.pipeline._tools import is_rest
+from azure.core.polling import NoPolling
+from azure.core.polling.base_polling import (LocationPolling, LROBasePolling,
+                                             StatusCheckPolling)
+
+
+class AAZNoPolling(NoPolling):
+    pass
+
+
+class MyBasePolling(LROBasePolling):
+    def __init__(self, *args, **kwargs):
+        self.token = kwargs.pop("token", None)
+        super().__init__(*args, **kwargs)
+
+    def request_status(self, status_link):
+        if self._path_format_arguments:
+            status_link = self._client.format_url(status_link, **self._path_format_arguments)
+        # Re-inject 'x-ms-client-request-id' while polling
+        if "request_id" not in self._operation_config:
+            self._operation_config["request_id"] = self._get_request_id()
+        if is_rest(self._initial_response.http_response):
+            # if I am a azure.core.rest.HttpResponse
+            # want to keep making azure.core.rest calls
+            from azure.core.rest import HttpRequest as RestHttpRequest
+            request = RestHttpRequest("GET", status_link, headers={"x-ms-aad-user-token": self.token})
+            return self._client.send_request(
+                request, _return_pipeline_response=True, **self._operation_config
+            )
+        # if I am a azure.core.pipeline.transport.HttpResponse
+        request = self._client.get(status_link, headers={"x-ms-aad-user-token": self.token})
+        return self._client._pipeline.run(  # pylint: disable=protected-access
+            request, stream=False, **self._operation_config
+        )
+
+
+class AAZBasePolling(MyBasePolling):
+
+    def __init__(self, *args, http_response_error_callback=None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._http_response_error_callback = http_response_error_callback
+
+    def initialize(self, client, initial_response, deserialization_callback):
+        try:
+            super().initialize(client, initial_response, deserialization_callback)
+        except HttpResponseError as err:
+            if self._http_response_error_callback:
+                # the HttpResponseError raise by LROBasePolling missing detailed error message
+                # _http_response_error_callback will build HttpResponseError with detailed error message
+                self._http_response_error_callback(err.response)
+            else:
+                raise err
+
+    def run(self):
+        try:
+            super().run()
+        except HttpResponseError as err:
+            if self._http_response_error_callback:
+                # the HttpResponseError raise by LROBasePolling missing detailed message
+                # _http_response_error_callback will build HttpResponseError with detailed error message
+                self._http_response_error_callback(err.response)
+            else:
+                raise err
+
+
+def PollOperationResult(client, no_wait, initial_session, deserialization_callback, error_callback, lro_options=None, path_format_arguments=None, user_token=None):
+    from azure.mgmt.core.polling.arm_polling import (
+        AzureAsyncOperationPolling, BodyContentPolling)
+
+    # from azure.cli.core._poller import AAZNoPolling, AAZBasePolling
+    if no_wait == True:  # noqa: E712, pylint: disable=singleton-comparison
+        polling = AAZNoPolling()
+    else:
+        polling = AAZBasePolling(
+            lro_options=lro_options,
+            lro_algorithms=[
+                AzureAsyncOperationPolling(lro_options=lro_options),
+                LocationPolling(),
+                BodyContentPolling(),
+                StatusCheckPolling(),
+            ],
+            path_format_arguments=path_format_arguments,
+            http_response_error_callback=error_callback,
+            token=user_token,
+        )
+
+    polling.initialize(
+        client,
+        initial_response=initial_session,
+        deserialization_callback=deserialization_callback
+    )
+    return polling

src/acat/azext_acat/aaz/latest/app_compliance_automation/report/snapshot/__cmd_group.py

@@ -11,9 +11,6 @@
 from azure.cli.core.aaz import *
 
 
-@register_command_group(
-    "app-compliance-automation report snapshot",
-)
 class __CMDGroup(AAZCommandGroup):
     """manage ACAT report snapshots
     """