Enable call suspend-resume from durable client (#483)

* Add API syntax support for suspend-resume

* initial commit

* add new line at the end of file

* add unit tests for suspend-resume

* add suspend/resume history event

* remove empty line

* remove duplicate caused by emrge

* update by comment

---------

Co-authored-by: Julio Arroyo <[email protected]>

Author: nytian

azure/durable_functions/models/DurableOrchestrationClient.py

@@ -709,3 +709,73 @@ async def rewind(self,
         else:
             ex_msg = "Received unexpected payload from the durable-extension: " + str(response)
             raise Exception(ex_msg)
+
+    async def suspend(self, instance_id: str, reason: str) -> None:
+        """Suspend the specified orchestration instance.
+
+        Parameters
+        ----------
+        instance_id : str
+            The ID of the orchestration instance to suspend.
+        reason: str
+            The reason for suspending the instance.
+
+        Raises
+        ------
+        Exception:
+            When the suspend call failed with an unexpected status code
+
+        Returns
+        -------
+        None
+        """
+        request_url = f"{self._orchestration_bindings.rpc_base_url}instances/{instance_id}/" \
+                      f"suspend?reason={quote(reason)}"
+        response = await self._post_async_request(request_url, None)
+        switch_statement = {
+            202: lambda: None,  # instance is suspended
+            410: lambda: None,  # instance completed
+            404: lambda: f"No instance with ID '{instance_id}' found.",
+        }
+
+        has_error_message = switch_statement.get(
+            response[0],
+            lambda: f"The operation failed with an unexpected status code {response[0]}")
+        error_message = has_error_message()
+        if error_message:
+            raise Exception(error_message)
+
+    async def resume(self, instance_id: str, reason: str) -> None:
+        """Resume the specified orchestration instance.
+
+        Parameters
+        ----------
+        instance_id : str
+            The ID of the orchestration instance to query.
+        reason: str
+            The reason for resuming the instance.
+
+        Raises
+        ------
+        Exception:
+            When the resume call failed with an unexpected status code
+
+        Returns
+        -------
+        None
+        """
+        request_url = f"{self._orchestration_bindings.rpc_base_url}instances/{instance_id}/" \
+                      f"resume?reason={quote(reason)}"
+        response = await self._post_async_request(request_url, None)
+        switch_statement = {
+            202: lambda: None,  # instance is resumed
+            410: lambda: None,  # instance completed
+            404: lambda: f"No instance with ID '{instance_id}' found.",
+        }
+
+        has_error_message = switch_statement.get(
+            response[0],
+            lambda: f"The operation failed with an unexpected status code {response[0]}")
+        error_message = has_error_message()
+        if error_message:
+            raise Exception(error_message)

azure/durable_functions/models/OrchestrationRuntimeStatus.py

@@ -27,3 +27,6 @@ class OrchestrationRuntimeStatus(Enum):
 
     Pending = 'Pending'
     """The orchestration instance has been scheduled but has not yet started running."""
+
+    Suspended = 'Suspended'
+    """The orchestration instance has been suspended and may go back to running at a later time."""

azure/durable_functions/models/history/HistoryEventType.py

@@ -23,3 +23,5 @@ class HistoryEventType(IntEnum):
     CONTINUE_AS_NEW = 16
     GENERIC_EVENT = 17
     HISTORY_STATE = 18
+    EXECUTION_SUSPENDED = 19
+    EXECUTION_RESUMED = 20

tests/conftest.py

@@ -39,7 +39,13 @@ def get_binding_string():
                              "{text}&taskHub="
                              f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
             "purgeHistoryDeleteUri": f"{BASE_URL}/instances/INSTANCEID?taskHub="
-                                     f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}"
+                                     f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
+            "suspendPostUri": f"{BASE_URL}/instances/INSTANCEID/suspend?reason="
+                                "{text}&taskHub="
+                                f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
+            "resumePostUri": f"{BASE_URL}/instances/INSTANCEID/resume?reason="
+                                "{text}&taskHub="
+                                f"{TASK_HUB_NAME}&connection=Storage&code={AUTH_CODE}",
         },
         "rpcBaseUrl": RPC_BASE_URL
     }

tests/models/test_DurableOrchestrationBindings.py

@@ -54,3 +54,15 @@ def test_extracts_purge_history_delete_uri(binding_info):
         "=Storage&code=AUTH_CODE")
     assert expected_url == binding_info.management_urls[
         "purgeHistoryDeleteUri"]
+
+def test_extracts_suspend_post_uri(binding_info):
+    expected_url = replace_stand_in_bits(
+        "BASE_URL/instances/INSTANCEID/suspend?reason={"
+        "text}&taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE")
+    assert expected_url == binding_info.management_urls["suspendPostUri"]
+
+def test_extracts_resume_post_uri(binding_info):
+    expected_url = replace_stand_in_bits(
+        "BASE_URL/instances/INSTANCEID/resume?reason={"
+        "text}&taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE")
+    assert expected_url == binding_info.management_urls["resumePostUri"]

tests/models/test_DurableOrchestrationClient.py

@@ -137,7 +137,15 @@ def test_create_check_status_response(binding_string):
         "purgeHistoryDeleteUri":
             r"http://test_azure.net/runtime/webhooks/durabletask/instances/"
             r"2e2568e7-a906-43bd-8364-c81733c5891e"
-            r"?taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE"
+            r"?taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE",
+        "suspendPostUri":
+            r"http://test_azure.net/runtime/webhooks/durabletask/instances/"
+            r"2e2568e7-a906-43bd-8364-c81733c5891e/suspend"
+            r"?reason={text}&taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE",
+        "resumePostUri":
+            r"http://test_azure.net/runtime/webhooks/durabletask/instances/"
+            r"2e2568e7-a906-43bd-8364-c81733c5891e/resume"
+            r"?reason={text}&taskHub=TASK_HUB_NAME&connection=Storage&code=AUTH_CODE"
     }
     for key, _ in http_management_payload.items():
         http_management_payload[key] = replace_stand_in_bits(http_management_payload[key])
@@ -610,3 +618,113 @@ async def test_rewind_with_no_rpc_endpoint(binding_string):
         await client.rewind(INSTANCE_ID, REASON)
     ex_message = str(ex.value)
     assert ex_message == expected_exception_str
+
+@pytest.mark.asyncio
+async def test_post_202_suspend(binding_string):
+    raw_reason = 'stuff and things'
+    reason = 'stuff%20and%20things'
+    mock_request = MockRequest(
+        expected_url=f"{RPC_BASE_URL}instances/{TEST_INSTANCE_ID}/suspend?reason={reason}",
+        response=[202, None])
+    client = DurableOrchestrationClient(binding_string)
+    client._post_async_request = mock_request.post
+
+    result = await client.suspend(TEST_INSTANCE_ID, raw_reason)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_post_410_suspend(binding_string):
+    raw_reason = 'stuff and things'
+    reason = 'stuff%20and%20things'
+    mock_request = MockRequest(
+        expected_url=f"{RPC_BASE_URL}instances/{TEST_INSTANCE_ID}/suspend?reason={reason}",
+        response=[410, None])
+    client = DurableOrchestrationClient(binding_string)
+    client._post_async_request = mock_request.post
+
+    result = await client.suspend(TEST_INSTANCE_ID, raw_reason)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_post_404_suspend(binding_string):
+    raw_reason = 'stuff and things'
+    reason = 'stuff%20and%20things'
+    mock_request = MockRequest(
+        expected_url=f"{RPC_BASE_URL}instances/{TEST_INSTANCE_ID}/suspend?reason={reason}",
+        response=[404, MESSAGE_404])
+    client = DurableOrchestrationClient(binding_string)
+    client._post_async_request = mock_request.post
+
+    with pytest.raises(Exception):
+        await client.suspend(TEST_INSTANCE_ID, raw_reason)
+
+
+@pytest.mark.asyncio
+async def test_post_500_suspend(binding_string):
+    raw_reason = 'stuff and things'
+    reason = 'stuff%20and%20things'
+    mock_request = MockRequest(
+        expected_url=f"{RPC_BASE_URL}instances/{TEST_INSTANCE_ID}/suspend?reason={reason}",
+        response=[500, MESSAGE_500])
+    client = DurableOrchestrationClient(binding_string)
+    client._post_async_request = mock_request.post
+
+    with pytest.raises(Exception):
+        await client.suspend(TEST_INSTANCE_ID, raw_reason)
+
+@pytest.mark.asyncio
+async def test_post_202_resume(binding_string):
+    raw_reason = 'stuff and things'
+    reason = 'stuff%20and%20things'
+    mock_request = MockRequest(
+        expected_url=f"{RPC_BASE_URL}instances/{TEST_INSTANCE_ID}/resume?reason={reason}",
+        response=[202, None])
+    client = DurableOrchestrationClient(binding_string)
+    client._post_async_request = mock_request.post
+
+    result = await client.resume(TEST_INSTANCE_ID, raw_reason)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_post_410_resume(binding_string):
+    raw_reason = 'stuff and things'
+    reason = 'stuff%20and%20things'
+    mock_request = MockRequest(
+        expected_url=f"{RPC_BASE_URL}instances/{TEST_INSTANCE_ID}/resume?reason={reason}",
+        response=[410, None])
+    client = DurableOrchestrationClient(binding_string)
+    client._post_async_request = mock_request.post
+
+    result = await client.resume(TEST_INSTANCE_ID, raw_reason)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_post_404_resume(binding_string):
+    raw_reason = 'stuff and things'
+    reason = 'stuff%20and%20things'
+    mock_request = MockRequest(
+        expected_url=f"{RPC_BASE_URL}instances/{TEST_INSTANCE_ID}/resume?reason={reason}",
+        response=[404, MESSAGE_404])
+    client = DurableOrchestrationClient(binding_string)
+    client._post_async_request = mock_request.post
+
+    with pytest.raises(Exception):
+        await client.resume(TEST_INSTANCE_ID, raw_reason)
+
+
+@pytest.mark.asyncio
+async def test_post_500_resume(binding_string):
+    raw_reason = 'stuff and things'
+    reason = 'stuff%20and%20things'
+    mock_request = MockRequest(
+        expected_url=f"{RPC_BASE_URL}instances/{TEST_INSTANCE_ID}/resume?reason={reason}",
+        response=[500, MESSAGE_500])
+    client = DurableOrchestrationClient(binding_string)
+    client._post_async_request = mock_request.post
+
+    with pytest.raises(Exception):
+        await client.resume(TEST_INSTANCE_ID, raw_reason)