Merge pull request #326 from Azure/dev

Promote dev to main for 1.1.2

Author: davidmrdavid

azure/durable_functions/models/DurableOrchestrationContext.py

@@ -1,7 +1,7 @@
 from collections import defaultdict
 from azure.durable_functions.models.actions.SignalEntityAction import SignalEntityAction
 from azure.durable_functions.models.actions.CallEntityAction import CallEntityAction
-from azure.durable_functions.models.Task import TaskBase
+from azure.durable_functions.models.Task import TaskBase, TimerTask
 from azure.durable_functions.models.actions.CallHttpAction import CallHttpAction
 from azure.durable_functions.models.DurableHttpRequest import DurableHttpRequest
 from azure.durable_functions.models.actions.CallSubOrchestratorWithRetryAction import \
@@ -100,7 +100,8 @@ def from_json(cls, json_string: str):
     def _generate_task(self, action: Action,
                        retry_options: Optional[RetryOptions] = None,
                        id_: Optional[Union[int, str]] = None,
-                       parent: Optional[TaskBase] = None) -> Union[AtomicTask, RetryAbleTask]:
+                       parent: Optional[TaskBase] = None,
+                       task_constructor=AtomicTask) -> Union[AtomicTask, RetryAbleTask, TimerTask]:
         """Generate an atomic or retryable Task based on an input.
 
         Parameters
@@ -124,7 +125,7 @@ def _generate_task(self, action: Action,
             action_payload = [action]
         else:
             action_payload = action
-        task = AtomicTask(id_, action_payload)
+        task = task_constructor(id_, action_payload)
         task.parent = parent
 
         # if task is retryable, provide the retryable wrapper class
@@ -517,7 +518,7 @@ def create_timer(self, fire_at: datetime.datetime) -> TaskBase:
             A Durable Timer Task that schedules the timer to wake up the activity
         """
         action = CreateTimerAction(fire_at)
-        task = self._generate_task(action)
+        task = self._generate_task(action, task_constructor=TimerTask)
         return task
 
     def wait_for_external_event(self, name: str) -> TaskBase:

azure/durable_functions/models/Task.py

@@ -5,6 +5,7 @@
 from azure.durable_functions.models.actions.Action import Action
 from azure.durable_functions.models.actions.WhenAnyAction import WhenAnyAction
 from azure.durable_functions.models.actions.WhenAllAction import WhenAllAction
+from azure.durable_functions.models.actions.CreateTimerAction import CreateTimerAction
 
 import enum
 from typing import Any, List, Optional, Set, Type, Union
@@ -56,6 +57,14 @@ def __init__(self, id_: Union[int, str], actions: Union[List[Action], Action]):
         self.action_repr: Union[List[Action], Action] = actions
         self.is_played = False
 
+    @property
+    def is_completed(self) -> bool:
+        """Get indicator of whether the task completed.
+
+        Note that completion is not equivalent to success.
+        """
+        return not(self.state is TaskState.RUNNING)
+
     def set_is_played(self, is_played: bool):
         """Set the is_played flag for the Task.
 
@@ -208,7 +217,47 @@ def try_set_value(self, child: TaskBase):
 class AtomicTask(TaskBase):
     """A Task with no subtasks."""
 
-    pass
+    def _get_action(self) -> Action:
+        action: Action
+        if isinstance(self.action_repr, list):
+            action = self.action_repr[0]
+        else:
+            action = self.action_repr
+        return action
+
+
+class TimerTask(AtomicTask):
+    """A Timer Task."""
+
+    def __init__(self, id_: Union[int, str], action: CreateTimerAction):
+        super().__init__(id_, action)
+        self.action_repr: Union[List[CreateTimerAction], CreateTimerAction]
+
+    @property
+    def is_cancelled(self) -> bool:
+        """Check if the Timer is cancelled.
+
+        Returns
+        -------
+        bool
+            Returns whether a timer has been cancelled or not
+        """
+        action: CreateTimerAction = self._get_action()
+        return action.is_cancelled
+
+    def cancel(self):
+        """Cancel a timer.
+
+        Raises
+        ------
+        ValueError
+            Raises an error if the task is already completed and an attempt is made to cancel it
+        """
+        if not self.is_completed:
+            action: CreateTimerAction = self._get_action()
+            action.is_cancelled = True
+        else:
+            raise ValueError("Cannot cancel a completed task.")
 
 
 class WhenAllTask(CompoundTask):

azure/durable_functions/models/TaskOrchestrationExecutor.py

@@ -180,8 +180,12 @@ def parse_history_event(directive_result):
             # retrieve result
             new_value = parse_history_event(event)
             if task._api_name == "CallEntityAction":
-                new_value = ResponseMessage.from_dict(new_value)
-                new_value = json.loads(new_value.result)
+                event_payload = ResponseMessage.from_dict(new_value)
+                new_value = json.loads(event_payload.result)
+
+                if event_payload.is_exception:
+                    new_value = Exception(new_value)
+                    is_success = False
         else:
             # generate exception
             new_value = Exception(f"{event.Reason} \n {event.Details}")

azure/durable_functions/models/entities/ResponseMessage.py

@@ -7,7 +7,7 @@ class ResponseMessage:
     Specifies the response of an entity, as processed by the durable-extension.
     """
 
-    def __init__(self, result: str):
+    def __init__(self, result: str, is_exception: bool = False):
         """Instantiate a ResponseMessage.
 
         Specifies the response of an entity, as processed by the durable-extension.
@@ -18,6 +18,7 @@ def __init__(self, result: str):
             The result provided by the entity
         """
         self.result = result
+        self.is_exception = is_exception
         # TODO: JS has an additional exceptionType field, but does not use it
 
     @classmethod
@@ -34,5 +35,6 @@ def from_dict(cls, d: Dict[str, Any]) -> 'ResponseMessage':
         ResponseMessage:
             The ResponseMessage built from the provided dictionary
         """
-        result = cls(d["result"])
+        is_error = "exceptionType" in d.keys()
+        result = cls(d["result"], is_error)
         return result

tests/orchestrator/orchestrator_test_utils.py

@@ -30,8 +30,9 @@ def assert_entity_state_equals(expected, result):
     assert_attribute_equal(expected, result, "signals")
 
 def assert_results_are_equal(expected: Dict[str, Any], result: Dict[str, Any]) -> bool:
-    assert_attribute_equal(expected, result, "result")
-    assert_attribute_equal(expected, result, "isError")
+    for (payload_expected, payload_result) in zip(expected, result):
+        assert_attribute_equal(payload_expected, payload_result, "result")
+        assert_attribute_equal(payload_expected, payload_result, "isError")
 
 def assert_attribute_equal(expected, result, attribute):
     if attribute in expected:

tests/orchestrator/test_create_timer.py

@@ -28,6 +28,20 @@ def generator_function(context):
     yield context.create_timer(fire_at)
     return "Done!"
 
+def generator_function_timer_can_be_cancelled(context):
+    time_limit1 = context.current_utc_datetime + timedelta(minutes=5)
+    timer_task1 = context.create_timer(time_limit1)
+
+    time_limit2 = context.current_utc_datetime + timedelta(minutes=10)
+    timer_task2 = context.create_timer(time_limit2)
+
+    winner = yield context.task_any([timer_task1, timer_task2])
+    if winner == timer_task1:
+        timer_task2.cancel()
+        return "Done!"
+    else:
+        raise Exception("timer task 1 should complete before timer task 2")
+
 def add_timer_action(state: OrchestratorState, fire_at: datetime):
     action = CreateTimerAction(fire_at=fire_at)
     state._actions.append([action])
@@ -64,4 +78,25 @@ def test_timers_comparison_with_relaxed_precision():
     #assert_valid_schema(result)
     # TODO: getting the following error when validating the schema
     # "Additional properties are not allowed ('fireAt', 'isCanceled' were unexpected)">
-    assert_orchestration_state_equals(expected, result)
+    assert_orchestration_state_equals(expected, result)
+
+def test_timers_can_be_cancelled():
+
+    context_builder = ContextBuilder("test_timers_can_be_cancelled")
+    fire_at1 = context_builder.current_datetime + timedelta(minutes=5)
+    fire_at2 = context_builder.current_datetime + timedelta(minutes=10)
+    add_timer_fired_events(context_builder, 0, str(fire_at1))
+    add_timer_fired_events(context_builder, 1, str(fire_at2))
+
+    result = get_orchestration_state_result(
+        context_builder, generator_function_timer_can_be_cancelled)
+
+    expected_state = base_expected_state(output='Done!')
+    expected_state._actions.append(
+        [CreateTimerAction(fire_at=fire_at1), CreateTimerAction(fire_at=fire_at2, is_cancelled=True)])
+
+    expected_state._is_done = True
+    expected = expected_state.to_json()
+
+    assert_orchestration_state_equals(expected, result)
+    assert result["actions"][0][1]["isCanceled"]

tests/orchestrator/test_entity.py

@@ -1,6 +1,6 @@
 from azure.durable_functions.models.ReplaySchema import ReplaySchema
 from .orchestrator_test_utils \
-    import assert_orchestration_state_equals, get_orchestration_state_result, assert_valid_schema, \
+    import assert_orchestration_state_equals, assert_results_are_equal, get_orchestration_state_result, assert_valid_schema, \
             get_entity_state_result, assert_entity_state_equals
 from tests.test_utils.ContextBuilder import ContextBuilder
 from tests.test_utils.EntityContextBuilder import EntityContextBuilder
@@ -23,6 +23,14 @@ def generator_function_call_entity(context):
     outputs.append(x)
     return outputs
 
+def generator_function_catch_entity_exception(context):
+    entityId = df.EntityId("Counter", "myCounter")
+    try:
+        yield context.call_entity(entityId, "add", 3)
+        return "No exception thrown"
+    except:
+        return "Exception thrown"
+
 def generator_function_signal_entity(context):
     outputs = []
     entityId = df.EntityId("Counter", "myCounter")
@@ -53,6 +61,29 @@ def counter_entity_function(context):
     context.set_state(current_value)
     context.set_result(result)
 
+def counter_entity_function_raises_exception(context):
+    raise Exception("boom!")
+
+def test_entity_raises_exception():
+    # Create input batch
+    batch = []
+    add_to_batch(batch, name="get")
+    context_builder = EntityContextBuilder(batch=batch)
+
+    # Run the entity, get observed result
+    result = get_entity_state_result(
+        context_builder,
+        counter_entity_function_raises_exception,
+        )
+
+    # Construct expected result
+    expected_state = entity_base_expected_state()
+    apply_operation(expected_state, result="boom!", state=None, is_error=True)
+    expected = expected_state.to_json()
+
+    # Ensure expectation matches observed behavior
+    #assert_valid_schema(result)
+    assert_entity_state_equals(expected, result)
 
 def test_entity_signal_then_call():
     """Tests that a simple counter entity outputs the correct value
@@ -161,11 +192,11 @@ def add_signal_entity_action(state: OrchestratorState, id_: df.EntityId, op: str
     state.actions.append([action])
 
 def add_call_entity_completed_events(
-        context_builder: ContextBuilder, op: str, instance_id=str, input_=None, event_id=0):
+        context_builder: ContextBuilder, op: str, instance_id=str, input_=None, event_id=0, is_error=False):
     context_builder.add_event_sent_event(instance_id, event_id)
     context_builder.add_orchestrator_completed_event()
     context_builder.add_orchestrator_started_event()
-    context_builder.add_event_raised_event(name="0000", id_=0, input_=input_, is_entity=True)
+    context_builder.add_event_raised_event(name="0000", id_=0, input_=input_, is_entity=True, is_error=is_error)
 
 def test_call_entity_sent():
     context_builder = ContextBuilder('test_simple_function')
@@ -233,4 +264,29 @@ def test_call_entity_raised():
 
     #assert_valid_schema(result)
 
+    assert_orchestration_state_equals(expected, result)
+
+def test_call_entity_catch_exception():
+    entityId = df.EntityId("Counter", "myCounter")
+    context_builder = ContextBuilder('catch exceptions')
+    add_call_entity_completed_events(
+        context_builder,
+        "add",
+        df.EntityId.get_scheduler_id(entityId),
+        input_="I am an error!",
+        event_id=0,
+        is_error=True
+    )
+
+    result = get_orchestration_state_result(
+        context_builder, generator_function_catch_entity_exception)
+
+    expected_state = base_expected_state(
+        "Exception thrown"
+    )
+
+    add_call_entity_action(expected_state, entityId, "add", 3)
+    expected_state._is_done = True
+    expected = expected_state.to_json()
+
     assert_orchestration_state_equals(expected, result)

tests/orchestrator/test_sequential_orchestrator.py

@@ -136,10 +136,14 @@ def generator_function_new_guid(context):
     outputs.append(str(output3))
     return outputs
 
-
 def base_expected_state(output=None, replay_schema: ReplaySchema = ReplaySchema.V1) -> OrchestratorState:
     return OrchestratorState(is_done=False, actions=[], output=output, replay_schema=replay_schema)
 
+def add_timer_fired_events(context_builder: ContextBuilder, id_: int, timestamp: str):
+    fire_at: str = context_builder.add_timer_created_event(id_, timestamp)
+    context_builder.add_orchestrator_completed_event()
+    context_builder.add_orchestrator_started_event()
+    context_builder.add_timer_fired_event(id_=id_, fire_at=fire_at)
 
 def add_hello_action(state: OrchestratorState, input_: str):
     action = CallActivityAction(function_name='Hello', input_=input_)

tests/test_utils/ContextBuilder.py

@@ -125,11 +125,14 @@ def add_execution_started_event(
         event.Input = input_
         self.history_events.append(event)
 
-    def add_event_raised_event(self, name:str, id_: int, input_=None, timestamp=None, is_entity=False):
+    def add_event_raised_event(self, name:str, id_: int, input_=None, timestamp=None, is_entity=False, is_error = False):
         event = self.get_base_event(HistoryEventType.EVENT_RAISED, id_=id_, timestamp=timestamp)
         event.Name = name
         if is_entity:
-            event.Input = json.dumps({ "result": json.dumps(input_) })
+            if is_error:
+                event.Input = json.dumps({ "result": json.dumps(input_), "exceptionType": "True" })
+            else:
+                event.Input = json.dumps({ "result": json.dumps(input_) })
         else:
             event.Input = input_
         # event.timestamp = timestamp