fix lro parameterization with constant params

ChangeLog.md

@@ -1,5 +1,15 @@
 # Change Log
 
+### 2021-06-22 - 5.8.3
+
+min Autorest core version: 3.3.0
+
+min Modelerfour version: 4.19.1
+
+**Bug Fixes**
+
+- Fix LRO path parameterization when we have a constant path parameter #968
+
 ### 2021-06-22 - 5.8.2
 
 min Autorest core version: 3.3.0

autorest/codegen/templates/lro_operation_helper.jinja2

@@ -62,6 +62,11 @@
     {% set operation_name = "begin_"+operation.python_name %}
     {% if operation.parameters.path %}
     {% set path_format_arguments = ", path_format_arguments=path_format_arguments" %}
+    {% if operation.parameters.path|selectattr("constant") %}
+        {% for constant_parameter in operation.parameters.path|selectattr("constant") %}
+    {{ constant_parameter.serialized_name }} = {{ constant_parameter.constant_declaration }}
+        {% endfor %}
+    {% endif %}
     {{ op_tools.path_format_arguments(operation.parameters.path)|indent }}
 
     {% endif %}

package.json

@@ -1,6 +1,6 @@
 {
     "name": "@autorest/python",
-    "version": "5.8.2",
+    "version": "5.8.3",
     "description": "The Python extension for generators in AutoRest.",
     "scripts": {
         "prepare": "node run-python3.js prepare.py",
@@ -28,7 +28,7 @@
     },
     "devDependencies": {
         "@autorest/autorest": "^3.0.0",
-        "@microsoft.azure/autorest.testserver": "^3.0.24"
+        "@microsoft.azure/autorest.testserver": "^3.0.26"
     },
     "files": [
         "autorest/**/*.py",

test/azure/AcceptanceTests/asynctests/test_lro_parameterized_endpoints.py

@@ -25,11 +25,20 @@
 # --------------------------------------------------------------------------
 import pytest
 from lrowithparameterizedendpoints.aio import LROWithParamaterizedEndpoints
+from async_generator import yield_, async_generator
 
-class TestLroWithParameterizedEndpoints:
+@pytest.fixture
+@async_generator
+async def client(credential):
+    async with LROWithParamaterizedEndpoints(credential=credential, host="host:3000") as client:
+        await yield_(client)
 
-    @pytest.mark.asyncio
-    async def test_poll_with_parameterized_endpoints(self, credential):
-        async with LROWithParamaterizedEndpoints(credential=credential, host="host:3000") as client:
-            poller = await client.begin_poll_with_parameterized_endpoints(account_name='local', polling=True, polling_interval=0)
-            assert (await poller.result()) == 'success'
+@pytest.mark.asyncio
+async def test_poll_with_parameterized_endpoints(client):
+    poller = await client.begin_poll_with_parameterized_endpoints(account_name='local', polling_interval=0)
+    assert (await poller.result()) == 'success'
+
+@pytest.mark.asyncio
+async def test_poll_with_constant_parameterized_endpoints(client):
+    poller = await client.begin_poll_with_constant_parameterized_endpoints(account_name='local', polling_interval=0)
+    assert (await poller.result()) == 'success'

test/azure/AcceptanceTests/test_lro_parameterized_endpoints.py

@@ -23,12 +23,19 @@
 # IN THE SOFTWARE.
 #
 # --------------------------------------------------------------------------
-
+import pytest
 from lrowithparameterizedendpoints import LROWithParamaterizedEndpoints
 
-class TestLroWithParameterizedEndpoints:
+@pytest.fixture
+def client(credential):
+    with LROWithParamaterizedEndpoints(credential=credential, host="host:3000") as client:
+        yield client
+
+
+def test_poll_with_parameterized_endpoints(client):
+    poller = client.begin_poll_with_parameterized_endpoints(account_name='local', polling_interval=0)
+    assert poller.result() == 'success'
 
-    def test_poll_with_parameterized_endpoints(self, credential):
-        client = LROWithParamaterizedEndpoints(credential=credential, host="host:3000")
-        poller = client.begin_poll_with_parameterized_endpoints(account_name='local', polling=True, polling_interval=0)
-        assert poller.result() == 'success'
+def test_poll_with_constant_parameterized_endpoints(client):
+    poller = client.begin_poll_with_constant_parameterized_endpoints(account_name='local', polling_interval=0)
+    assert poller.result() == 'success'

test/azure/Expected/AcceptanceTests/LroWithParameterizedEndpoints/lrowithparameterizedendpoints/aio/operations/_lro_with_paramaterized_endpoints_operations.py

@@ -135,3 +135,114 @@ def get_long_running_output(pipeline_response):
             return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method)
 
     begin_poll_with_parameterized_endpoints.metadata = {"url": "/lroParameterizedEndpoints"}  # type: ignore
+
+    async def _poll_with_constant_parameterized_endpoints_initial(
+        self, account_name: str, **kwargs: Any
+    ) -> Optional[str]:
+        cls = kwargs.pop("cls", None)  # type: ClsType[Optional[str]]
+        error_map = {401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError}
+        error_map.update(kwargs.pop("error_map", {}))
+        constant_parameter = "iAmConstant"
+        accept = "application/json"
+
+        # Construct URL
+        url = self._poll_with_constant_parameterized_endpoints_initial.metadata["url"]  # type: ignore
+        path_format_arguments = {
+            "accountName": self._serialize.url("account_name", account_name, "str", skip_quote=True),
+            "host": self._serialize.url("self._config.host", self._config.host, "str", skip_quote=True),
+            "constantParameter": self._serialize.url("constant_parameter", constant_parameter, "str", skip_quote=True),
+        }
+        url = self._client.format_url(url, **path_format_arguments)
+
+        # Construct parameters
+        query_parameters = {}  # type: Dict[str, Any]
+
+        # Construct headers
+        header_parameters = {}  # type: Dict[str, Any]
+        header_parameters["Accept"] = self._serialize.header("accept", accept, "str")
+
+        request = self._client.post(url, query_parameters, header_parameters)
+        pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs)
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200, 202]:
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = self._deserialize.failsafe_deserialize(_models.Error, response)
+            raise HttpResponseError(response=response, model=error)
+
+        response_headers = {}
+        deserialized = None
+        if response.status_code == 200:
+            deserialized = self._deserialize("str", pipeline_response)
+
+        if response.status_code == 202:
+            response_headers["Location"] = self._deserialize("str", response.headers.get("Location"))
+
+        if cls:
+            return cls(pipeline_response, deserialized, response_headers)
+
+        return deserialized
+
+    _poll_with_constant_parameterized_endpoints_initial.metadata = {"url": "/lroConstantParameterizedEndpoints/{constantParameter}"}  # type: ignore
+
+    @distributed_trace_async
+    async def begin_poll_with_constant_parameterized_endpoints(
+        self, account_name: str, **kwargs: Any
+    ) -> AsyncLROPoller[str]:
+        """Poll with method and client level parameters in endpoint, with a constant value.
+
+        :param account_name: Account Name. Pass in 'local' to pass test.
+        :type account_name: str
+        :keyword callable cls: A custom type or function that will be passed the direct response
+        :keyword str continuation_token: A continuation token to restart a poller from a saved state.
+        :keyword polling: By default, your polling method will be AsyncLROBasePolling.
+         Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy.
+        :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod
+        :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present.
+        :return: An instance of AsyncLROPoller that returns either str or the result of cls(response)
+        :rtype: ~azure.core.polling.AsyncLROPoller[str]
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        polling = kwargs.pop("polling", True)  # type: Union[bool, AsyncPollingMethod]
+        cls = kwargs.pop("cls", None)  # type: ClsType[str]
+        lro_delay = kwargs.pop("polling_interval", self._config.polling_interval)
+        cont_token = kwargs.pop("continuation_token", None)  # type: Optional[str]
+        if cont_token is None:
+            raw_result = await self._poll_with_constant_parameterized_endpoints_initial(
+                account_name=account_name, cls=lambda x, y, z: x, **kwargs
+            )
+
+        kwargs.pop("error_map", None)
+        kwargs.pop("content_type", None)
+
+        def get_long_running_output(pipeline_response):
+            deserialized = self._deserialize("str", pipeline_response)
+
+            if cls:
+                return cls(pipeline_response, deserialized, {})
+            return deserialized
+
+        constant_parameter = "iAmConstant"
+        path_format_arguments = {
+            "accountName": self._serialize.url("account_name", account_name, "str", skip_quote=True),
+            "host": self._serialize.url("self._config.host", self._config.host, "str", skip_quote=True),
+            "constantParameter": self._serialize.url("constant_parameter", constant_parameter, "str", skip_quote=True),
+        }
+
+        if polling is True:
+            polling_method = AsyncLROBasePolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
+        elif polling is False:
+            polling_method = AsyncNoPolling()
+        else:
+            polling_method = polling
+        if cont_token:
+            return AsyncLROPoller.from_continuation_token(
+                polling_method=polling_method,
+                continuation_token=cont_token,
+                client=self._client,
+                deserialization_callback=get_long_running_output,
+            )
+        else:
+            return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method)
+
+    begin_poll_with_constant_parameterized_endpoints.metadata = {"url": "/lroConstantParameterizedEndpoints/{constantParameter}"}  # type: ignore

test/azure/Expected/AcceptanceTests/LroWithParameterizedEndpoints/lrowithparameterizedendpoints/operations/_lro_with_paramaterized_endpoints_operations.py

@@ -149,3 +149,120 @@ def get_long_running_output(pipeline_response):
             return LROPoller(self._client, raw_result, get_long_running_output, polling_method)
 
     begin_poll_with_parameterized_endpoints.metadata = {"url": "/lroParameterizedEndpoints"}  # type: ignore
+
+    def _poll_with_constant_parameterized_endpoints_initial(
+        self,
+        account_name,  # type: str
+        **kwargs  # type: Any
+    ):
+        # type: (...) -> Optional[str]
+        cls = kwargs.pop("cls", None)  # type: ClsType[Optional[str]]
+        error_map = {401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError}
+        error_map.update(kwargs.pop("error_map", {}))
+        constant_parameter = "iAmConstant"
+        accept = "application/json"
+
+        # Construct URL
+        url = self._poll_with_constant_parameterized_endpoints_initial.metadata["url"]  # type: ignore
+        path_format_arguments = {
+            "accountName": self._serialize.url("account_name", account_name, "str", skip_quote=True),
+            "host": self._serialize.url("self._config.host", self._config.host, "str", skip_quote=True),
+            "constantParameter": self._serialize.url("constant_parameter", constant_parameter, "str", skip_quote=True),
+        }
+        url = self._client.format_url(url, **path_format_arguments)
+
+        # Construct parameters
+        query_parameters = {}  # type: Dict[str, Any]
+
+        # Construct headers
+        header_parameters = {}  # type: Dict[str, Any]
+        header_parameters["Accept"] = self._serialize.header("accept", accept, "str")
+
+        request = self._client.post(url, query_parameters, header_parameters)
+        pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs)
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200, 202]:
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            error = self._deserialize.failsafe_deserialize(_models.Error, response)
+            raise HttpResponseError(response=response, model=error)
+
+        response_headers = {}
+        deserialized = None
+        if response.status_code == 200:
+            deserialized = self._deserialize("str", pipeline_response)
+
+        if response.status_code == 202:
+            response_headers["Location"] = self._deserialize("str", response.headers.get("Location"))
+
+        if cls:
+            return cls(pipeline_response, deserialized, response_headers)
+
+        return deserialized
+
+    _poll_with_constant_parameterized_endpoints_initial.metadata = {"url": "/lroConstantParameterizedEndpoints/{constantParameter}"}  # type: ignore
+
+    @distributed_trace
+    def begin_poll_with_constant_parameterized_endpoints(
+        self,
+        account_name,  # type: str
+        **kwargs  # type: Any
+    ):
+        # type: (...) -> LROPoller[str]
+        """Poll with method and client level parameters in endpoint, with a constant value.
+
+        :param account_name: Account Name. Pass in 'local' to pass test.
+        :type account_name: str
+        :keyword callable cls: A custom type or function that will be passed the direct response
+        :keyword str continuation_token: A continuation token to restart a poller from a saved state.
+        :keyword polling: By default, your polling method will be LROBasePolling.
+         Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy.
+        :paramtype polling: bool or ~azure.core.polling.PollingMethod
+        :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present.
+        :return: An instance of LROPoller that returns either str or the result of cls(response)
+        :rtype: ~azure.core.polling.LROPoller[str]
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        polling = kwargs.pop("polling", True)  # type: Union[bool, PollingMethod]
+        cls = kwargs.pop("cls", None)  # type: ClsType[str]
+        lro_delay = kwargs.pop("polling_interval", self._config.polling_interval)
+        cont_token = kwargs.pop("continuation_token", None)  # type: Optional[str]
+        if cont_token is None:
+            raw_result = self._poll_with_constant_parameterized_endpoints_initial(
+                account_name=account_name, cls=lambda x, y, z: x, **kwargs
+            )
+
+        kwargs.pop("error_map", None)
+        kwargs.pop("content_type", None)
+
+        def get_long_running_output(pipeline_response):
+            deserialized = self._deserialize("str", pipeline_response)
+
+            if cls:
+                return cls(pipeline_response, deserialized, {})
+            return deserialized
+
+        constant_parameter = "iAmConstant"
+        path_format_arguments = {
+            "accountName": self._serialize.url("account_name", account_name, "str", skip_quote=True),
+            "host": self._serialize.url("self._config.host", self._config.host, "str", skip_quote=True),
+            "constantParameter": self._serialize.url("constant_parameter", constant_parameter, "str", skip_quote=True),
+        }
+
+        if polling is True:
+            polling_method = LROBasePolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs)
+        elif polling is False:
+            polling_method = NoPolling()
+        else:
+            polling_method = polling
+        if cont_token:
+            return LROPoller.from_continuation_token(
+                polling_method=polling_method,
+                continuation_token=cont_token,
+                client=self._client,
+                deserialization_callback=get_long_running_output,
+            )
+        else:
+            return LROPoller(self._client, raw_result, get_long_running_output, polling_method)
+
+    begin_poll_with_constant_parameterized_endpoints.metadata = {"url": "/lroConstantParameterizedEndpoints/{constantParameter}"}  # type: ignore

test/vanilla/AcceptanceTests/test_zzz.py

@@ -45,6 +45,7 @@ def test_ensure_coverage(self):
         # Please add missing features or failing tests here
         missing_features_or_bugs = {
             'ConstantsInBody': 1,  # https://github.com/Azure/autorest.modelerfour/issues/83
+            "LLCRequiredToOptional": 1,  # adding in LLC PR
         }
 
         print("Coverage:")