PR Feedback

Author: andystaples

durabletask/client.py

@@ -397,7 +397,7 @@ def __init__(self, *,
         )
         self._channel = channel
         self._stub = stubs.TaskHubSidecarServiceStub(channel)
-        self._logger = shared.get_logger("client", log_handler, log_formatter)
+        self._logger = shared.get_logger("async_client", log_handler, log_formatter)
         self.default_version = default_version
 
     async def close(self) -> None:

durabletask/internal/grpc_interceptor.py

@@ -25,7 +25,7 @@ class _AsyncClientCallDetails(
             ['method', 'timeout', 'metadata', 'credentials', 'wait_for_ready']),
         grpc.aio.ClientCallDetails):
     """This is an implementation of the aio ClientCallDetails interface needed for async interceptors.
-    This class takes six named values and inherits the ClientCallDetails from grpc.aio package.
+    This class takes five named values and inherits the ClientCallDetails from grpc.aio package.
     This class encloses the values that describe a RPC to be invoked.
     """
     pass

tests/durabletask/test_orchestration_async_e2e.py

@@ -7,11 +7,18 @@
 import pytest
 
 from durabletask import client, task, worker
+from durabletask.testing import create_test_backend
 
-# NOTE: These tests assume a sidecar process is running. Example command:
-#       go install github.com/microsoft/durabletask-go@main
-#       durabletask-go --port 4001
-pytestmark = pytest.mark.e2e
+HOST = "localhost:50060"
+
+
+@pytest.fixture(autouse=True)
+def backend():
+    """Create an in-memory backend for testing."""
+    b = create_test_backend(port=50060)
+    yield b
+    b.stop()
+    b.reset()
 
 
 @pytest.mark.asyncio
@@ -23,11 +30,11 @@ def empty_orchestrator(ctx: task.OrchestrationContext, _):
         nonlocal invoked  # don't do this in a real app!
         invoked = True
 
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
         w.add_orchestrator(empty_orchestrator)
         w.start()
 
-        c = client.AsyncTaskHubGrpcClient()
+        c = client.AsyncTaskHubGrpcClient(host_address=HOST)
         id = await c.schedule_new_orchestration(empty_orchestrator, tags={'Tagged': 'true'})
         state = await c.wait_for_orchestration_completion(id, timeout=30)
 
@@ -56,12 +63,12 @@ def sequence(ctx: task.OrchestrationContext, start_val: int):
             numbers.append(current)
         return numbers
 
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
         w.add_orchestrator(sequence)
         w.add_activity(plus_one)
         w.start()
 
-        c = client.AsyncTaskHubGrpcClient()
+        c = client.AsyncTaskHubGrpcClient(host_address=HOST)
         id = await c.schedule_new_orchestration(sequence, input=1)
         state = await c.wait_for_orchestration_completion(id, timeout=30)
 
@@ -82,11 +89,11 @@ def orchestrator(ctx: task.OrchestrationContext, _):
         c = yield ctx.wait_for_external_event('C')
         return [a, b, c]
 
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
         w.add_orchestrator(orchestrator)
         w.start()
 
-        c = client.AsyncTaskHubGrpcClient()
+        c = client.AsyncTaskHubGrpcClient(host_address=HOST)
         id = await c.schedule_new_orchestration(orchestrator)
         await c.raise_orchestration_event(id, 'A', data='a')
         await c.raise_orchestration_event(id, 'B', data='b')
@@ -104,11 +111,11 @@ def orchestrator(ctx: task.OrchestrationContext, _):
         result = yield ctx.wait_for_external_event("my_event")
         return result
 
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
         w.add_orchestrator(orchestrator)
         w.start()
 
-        c = client.AsyncTaskHubGrpcClient()
+        c = client.AsyncTaskHubGrpcClient(host_address=HOST)
         id = await c.schedule_new_orchestration(orchestrator)
         state = await c.wait_for_orchestration_start(id, timeout=30)
         assert state is not None
@@ -143,11 +150,11 @@ def orchestrator(ctx: task.OrchestrationContext, _):
         result = yield ctx.wait_for_external_event("my_event")
         return result
 
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
         w.add_orchestrator(orchestrator)
         w.start()
 
-        c = client.AsyncTaskHubGrpcClient()
+        c = client.AsyncTaskHubGrpcClient(host_address=HOST)
         id = await c.schedule_new_orchestration(orchestrator)
         state = await c.wait_for_orchestration_start(id, timeout=30)
         assert state is not None
@@ -165,11 +172,11 @@ async def test_async_purge_orchestration():
     def orchestrator(ctx: task.OrchestrationContext, _):
         pass
 
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
         w.add_orchestrator(orchestrator)
         w.start()
 
-        c = client.AsyncTaskHubGrpcClient()
+        c = client.AsyncTaskHubGrpcClient(host_address=HOST)
         id = await c.schedule_new_orchestration(orchestrator)
         await c.wait_for_orchestration_completion(id, timeout=30)
 
@@ -181,7 +188,6 @@ def orchestrator(ctx: task.OrchestrationContext, _):
 
 
 @pytest.mark.asyncio
-@pytest.mark.skip(reason="durabletask-go does not yet support RestartInstance")
 async def test_async_restart_with_same_instance_id():
     def orchestrator(ctx: task.OrchestrationContext, _):
         result = yield ctx.call_activity(say_hello, input="World")
@@ -190,12 +196,12 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     def say_hello(ctx: task.ActivityContext, input: str):
         return f"Hello, {input}!"
 
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
         w.add_orchestrator(orchestrator)
         w.add_activity(say_hello)
         w.start()
 
-        c = client.AsyncTaskHubGrpcClient()
+        c = client.AsyncTaskHubGrpcClient(host_address=HOST)
         id = await c.schedule_new_orchestration(orchestrator)
         state = await c.wait_for_orchestration_completion(id, timeout=30)
         assert state is not None
@@ -213,7 +219,6 @@ def say_hello(ctx: task.ActivityContext, input: str):
 
 
 @pytest.mark.asyncio
-@pytest.mark.skip(reason="durabletask-go does not yet support RestartInstance")
 async def test_async_restart_with_new_instance_id():
     def orchestrator(ctx: task.OrchestrationContext, _):
         result = yield ctx.call_activity(say_hello, input="World")
@@ -222,12 +227,12 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     def say_hello(ctx: task.ActivityContext, input: str):
         return f"Hello, {input}!"
 
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
         w.add_orchestrator(orchestrator)
         w.add_activity(say_hello)
         w.start()
 
-        c = client.AsyncTaskHubGrpcClient()
+        c = client.AsyncTaskHubGrpcClient(host_address=HOST)
         id = await c.schedule_new_orchestration(orchestrator)
         state = await c.wait_for_orchestration_completion(id, timeout=30)
         assert state is not None