Fix tests

Author: andystaples

tests/durabletask/test_worker_concurrency_loop.py

@@ -128,8 +128,8 @@ def fn(*args, **kwargs):
 
     # Submit more work than concurrency allows
     for i in range(5):
-        manager.submit_orchestration(make_work("orch", i))
-        manager.submit_activity(make_work("act", i))
+        manager.submit_orchestration(make_work("orch", i), lambda *a, **k: None)
+        manager.submit_activity(make_work("act", i), lambda *a, **k: None)
 
     # Run the manager loop in a thread (sync context)
     def run_manager():
@@ -139,6 +139,11 @@ def run_manager():
     t.start()
     time.sleep(1.5)  # Let work process
     manager.shutdown()
+
+    # Ensure the queues have been started
+    if (manager.activity_queue is None or manager.orchestration_queue is None):
+        raise RuntimeError("Worker manager queues not initialized")
+
     # Unblock the consumers by putting dummy items in the queues
     manager.activity_queue.put_nowait((lambda: None, (), {}))
     manager.orchestration_queue.put_nowait((lambda: None, (), {}))

tests/durabletask/test_worker_concurrency_loop_async.py

@@ -46,25 +46,25 @@ def test_worker_concurrency_loop_async():
     grpc_worker = TaskHubGrpcWorker(concurrency_options=options)
     stub = DummyStub()
 
-    async def dummy_orchestrator(req, stub, completionToken):
+    async def dummy_orchestrator(self, req, stub, completionToken):
         await asyncio.sleep(0.1)
         stub.CompleteOrchestratorTask('ok')
 
-    async def cancel_dummy_orchestrator(req, stub, completionToken):
+    async def cancel_dummy_orchestrator(self, req, stub, completionToken):
         pass
 
-    async def dummy_activity(req, stub, completionToken):
+    async def dummy_activity(self, req, stub, completionToken):
         await asyncio.sleep(0.1)
         stub.CompleteActivityTask('ok')
 
-    async def cancel_dummy_activity(req, stub, completionToken):
+    async def cancel_dummy_activity(self, req, stub, completionToken):
         pass
 
     # Patch the worker's _execute_orchestrator and _execute_activity
-    grpc_worker._execute_orchestrator = dummy_orchestrator
-    grpc_worker._cancel_orchestrator = cancel_dummy_orchestrator
-    grpc_worker._execute_activity = dummy_activity
-    grpc_worker._cancel_activity = cancel_dummy_activity
+    grpc_worker._execute_orchestrator = dummy_orchestrator.__get__(grpc_worker, TaskHubGrpcWorker)
+    grpc_worker._cancel_orchestrator = cancel_dummy_orchestrator.__get__(grpc_worker, TaskHubGrpcWorker)
+    grpc_worker._execute_activity = dummy_activity.__get__(grpc_worker, TaskHubGrpcWorker)
+    grpc_worker._cancel_activity = cancel_dummy_activity.__get__(grpc_worker, TaskHubGrpcWorker)
 
     orchestrator_requests = [DummyRequest('orchestrator', f'orch{i}') for i in range(3)]
     activity_requests = [DummyRequest('activity', f'act{i}') for i in range(4)]