Add tests for new event types

Author: andystaples

durabletask/internal/helpers.py

@@ -20,6 +20,11 @@ def new_orchestrator_started_event(timestamp: Optional[datetime] = None) -> pb.H
     return pb.HistoryEvent(eventId=-1, timestamp=ts, orchestratorStarted=pb.OrchestratorStartedEvent())
 
 
+def new_orchestrator_completed_event() -> pb.HistoryEvent:
+    return pb.HistoryEvent(eventId=-1, timestamp=timestamp_pb2.Timestamp(),
+                           orchestratorCompleted=pb.OrchestratorCompletedEvent())
+
+
 def new_execution_started_event(name: str, instance_id: str, encoded_input: Optional[str] = None,
                                 tags: Optional[dict[str, str]] = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
@@ -119,6 +124,18 @@ def new_failure_details(ex: Exception) -> pb.TaskFailureDetails:
     )
 
 
+def new_event_sent_event(event_id: int, instance_id: str, input: str):
+    return pb.HistoryEvent(
+        eventId=event_id,
+        timestamp=timestamp_pb2.Timestamp(),
+        eventSent=pb.EventSentEvent(
+            name="",
+            input=get_string_value(input),
+            instanceId=instance_id
+        )
+    )
+
+
 def new_event_raised_event(name: str, encoded_input: Optional[str] = None) -> pb.HistoryEvent:
     return pb.HistoryEvent(
         eventId=-1,

tests/durabletask/test_orchestration_executor.py

@@ -9,7 +9,7 @@
 
 import durabletask.internal.helpers as helpers
 import durabletask.internal.orchestrator_service_pb2 as pb
-from durabletask import task, worker
+from durabletask import task, worker, entities
 
 logging.basicConfig(
     format='%(asctime)s.%(msecs)03d %(name)s %(levelname)s: %(message)s',
@@ -1183,6 +1183,77 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     assert str(ex) in complete_action.failureDetails.errorMessage
 
 
+def test_orchestrator_completed_no_effect():
+    def dummy_activity(ctx, _):
+        pass
+
+    def orchestrator(ctx: task.OrchestrationContext, orchestrator_input):
+        yield ctx.call_activity(dummy_activity, input=orchestrator_input)
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+
+    encoded_input = json.dumps(42)
+    new_events = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input),
+        helpers.new_orchestrator_completed_event()]
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER)
+    result = executor.execute(TEST_INSTANCE_ID, [], new_events)
+    actions = result.actions
+
+    assert len(actions) == 1
+    assert type(actions[0]) is pb.OrchestratorAction
+    assert actions[0].id == 1
+    assert actions[0].HasField("scheduleTask")
+    assert actions[0].scheduleTask.name == task.get_name(dummy_activity)
+    assert actions[0].scheduleTask.input.value == encoded_input
+
+
+def test_entity_lock_created_as_event():
+    test_entity_id = entities.EntityInstanceId("Counter", "myCounter")
+
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        entity_id = test_entity_id
+        with (yield ctx.lock_entities([entity_id])):
+            return (yield ctx.call_entity(entity_id, "set", 1))
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(orchestrator)
+
+    new_events = [
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, None),
+    ]
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER)
+    result1 = executor.execute(TEST_INSTANCE_ID, [], new_events)
+    actions = result1.actions
+    assert len(actions) == 1
+    assert type(actions[0]) is pb.OrchestratorAction
+    assert actions[0].id == 1
+    assert actions[0].HasField("sendEntityMessage")
+    assert actions[0].sendEntityMessage.HasField("entityLockRequested")
+
+    old_events = new_events
+    event_sent_input = {
+        "id": actions[0].sendEntityMessage.entityLockRequested.criticalSectionId,
+    }
+    new_events = [
+        helpers.new_event_sent_event(1, str(test_entity_id), json.dumps(event_sent_input)),
+        helpers.new_event_raised_event(event_sent_input["id"], None),
+    ]
+    result = executor.execute(TEST_INSTANCE_ID, old_events, new_events)
+    actions = result.actions
+
+    assert len(actions) == 1
+    assert type(actions[0]) is pb.OrchestratorAction
+    assert actions[0].id == 2
+    assert actions[0].HasField("sendEntityMessage")
+    assert actions[0].sendEntityMessage.HasField("entityOperationCalled")
+    assert actions[0].sendEntityMessage.entityOperationCalled.targetInstanceId.value == str(test_entity_id)
+
+
 def get_and_validate_complete_orchestration_action_list(expected_action_count: int, actions: list[pb.OrchestratorAction]) -> pb.CompleteOrchestrationAction:
     assert len(actions) == expected_action_count
     assert type(actions[-1]) is pb.OrchestratorAction