linter fixes

Author: berndverst

tests/durabletask-azuremanaged/test_dts_activity_sequence.py

@@ -2,15 +2,15 @@
 that calls an activity function in a sequence and prints the outputs."""
 import os
 
+import pytest
+
 from durabletask import client, task
 from durabletask.azuremanaged.client import DurableTaskSchedulerClient
 from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker
 
-import pytest
-
-
 pytestmark = pytest.mark.dts
 
+
 def hello(ctx: task.ActivityContext, name: str) -> str:
     """Activity function that returns a greeting"""
     return f'Hello {name}!'

tests/durabletask-azuremanaged/test_dts_orchestration_e2e.py

@@ -2,9 +2,8 @@
 # Licensed under the MIT License.
 
 import json
-import threading
-import time
 import os
+import threading
 from datetime import timedelta
 
 import pytest
@@ -21,6 +20,7 @@
 taskhub_name = os.getenv("TASKHUB", "default")
 endpoint = os.getenv("ENDPOINT", "http://localhost:8080")
 
+
 def test_empty_orchestration():
 
     invoked = False
@@ -31,12 +31,12 @@ def empty_orchestrator(ctx: task.OrchestrationContext, _):
 
     # Start a worker, which will connect to the sidecar in a background thread
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
-                                taskhub=taskhub_name, token_credential=None) as w:
+                                    taskhub=taskhub_name, token_credential=None) as w:
         w.add_orchestrator(empty_orchestrator)
         w.start()
 
         c = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
-                                   taskhub=taskhub_name, token_credential=None)
+                                       taskhub=taskhub_name, token_credential=None)
         id = c.schedule_new_orchestration(empty_orchestrator)
         state = c.wait_for_orchestration_completion(id, timeout=30)
 
@@ -66,13 +66,13 @@ def sequence(ctx: task.OrchestrationContext, start_val: int):
 
     # Start a worker, which will connect to the sidecar in a background thread
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
-                                taskhub=taskhub_name, token_credential=None) as w:
+                                    taskhub=taskhub_name, token_credential=None) as w:
         w.add_orchestrator(sequence)
         w.add_activity(plus_one)
         w.start()
 
         task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
-                                   taskhub=taskhub_name, token_credential=None)
+                                                     taskhub=taskhub_name, token_credential=None)
         id = task_hub_client.schedule_new_orchestration(sequence, input=1)
         state = task_hub_client.wait_for_orchestration_completion(
             id, timeout=30)
@@ -113,14 +113,14 @@ def orchestrator(ctx: task.OrchestrationContext, input: int):
 
     # Start a worker, which will connect to the sidecar in a background thread
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
-                                taskhub=taskhub_name, token_credential=None) as w:
+                                    taskhub=taskhub_name, token_credential=None) as w:
         w.add_orchestrator(orchestrator)
         w.add_activity(throw)
         w.add_activity(increment_counter)
         w.start()
 
         task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
-                                   taskhub=taskhub_name, token_credential=None)
+                                                     taskhub=taskhub_name, token_credential=None)
         id = task_hub_client.schedule_new_orchestration(orchestrator, input=1)
         state = task_hub_client.wait_for_orchestration_completion(id, timeout=30)
 
@@ -158,14 +158,14 @@ def parent_orchestrator(ctx: task.OrchestrationContext, count: int):
 
     # Start a worker, which will connect to the sidecar in a background thread
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
-                                taskhub=taskhub_name, token_credential=None) as w:
+                                    taskhub=taskhub_name, token_credential=None) as w:
         w.add_activity(increment)
         w.add_orchestrator(orchestrator_child)
         w.add_orchestrator(parent_orchestrator)
         w.start()
 
         task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
-                                   taskhub=taskhub_name, token_credential=None)
+                                                     taskhub=taskhub_name, token_credential=None)
         id = task_hub_client.schedule_new_orchestration(parent_orchestrator, input=10)
         state = task_hub_client.wait_for_orchestration_completion(id, timeout=30)
 
@@ -184,13 +184,13 @@ def orchestrator(ctx: task.OrchestrationContext, _):
 
     # Start a worker, which will connect to the sidecar in a background thread
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
-                                taskhub=taskhub_name, token_credential=None) as w:
+                                    taskhub=taskhub_name, token_credential=None) as w:
         w.add_orchestrator(orchestrator)
         w.start()
 
         # Start the orchestration and immediately raise events to it.
         task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
-                                   taskhub=taskhub_name, token_credential=None)
+                                                     taskhub=taskhub_name, token_credential=None)
         id = task_hub_client.schedule_new_orchestration(orchestrator)
         task_hub_client.raise_orchestration_event(id, 'A', data='a')
         task_hub_client.raise_orchestration_event(id, 'B', data='b')
@@ -285,12 +285,12 @@ def orchestrator(ctx: task.OrchestrationContext, _):
 
     # Start a worker, which will connect to the sidecar in a background thread
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
-                                taskhub=taskhub_name, token_credential=None) as w:
+                                    taskhub=taskhub_name, token_credential=None) as w:
         w.add_orchestrator(orchestrator)
         w.start()
 
         task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
-                                   taskhub=taskhub_name, token_credential=None)
+                                                     taskhub=taskhub_name, token_credential=None)
         id = task_hub_client.schedule_new_orchestration(orchestrator)
         state = task_hub_client.wait_for_orchestration_start(id, timeout=30)
         assert state is not None
@@ -302,23 +302,25 @@ def orchestrator(ctx: task.OrchestrationContext, _):
         assert state.runtime_status == client.OrchestrationStatus.TERMINATED
         assert state.serialized_output == json.dumps("some reason for termination")
 
+
 def test_terminate_recursive():
     def root(ctx: task.OrchestrationContext, _):
         result = yield ctx.call_sub_orchestrator(child)
         return result
+
     def child(ctx: task.OrchestrationContext, _):
         result = yield ctx.wait_for_external_event("my_event")
         return result
 
     # Start a worker, which will connect to the sidecar in a background thread
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
-                                taskhub=taskhub_name, token_credential=None) as w:
+                                    taskhub=taskhub_name, token_credential=None) as w:
         w.add_orchestrator(root)
         w.add_orchestrator(child)
         w.start()
 
         task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
-                                   taskhub=taskhub_name, token_credential=None)
+                                                     taskhub=taskhub_name, token_credential=None)
         id = task_hub_client.schedule_new_orchestration(root)
         state = task_hub_client.wait_for_orchestration_start(id, timeout=30)
         assert state is not None
@@ -331,7 +333,7 @@ def child(ctx: task.OrchestrationContext, _):
         assert state.runtime_status == client.OrchestrationStatus.TERMINATED
 
         # Verify that child orchestration is also terminated
-        c = task_hub_client.wait_for_orchestration_completion(id, timeout=30)
+        task_hub_client.wait_for_orchestration_completion(id, timeout=30)
         assert state is not None
         assert state.runtime_status == client.OrchestrationStatus.TERMINATED
 
@@ -417,14 +419,14 @@ def throw_activity_with_retry(ctx: task.ActivityContext, _):
         raise RuntimeError("Kah-BOOOOM!!!")
 
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
-                                taskhub=taskhub_name, token_credential=None) as w:
+                                    taskhub=taskhub_name, token_credential=None) as w:
         w.add_orchestrator(parent_orchestrator_with_retry)
         w.add_orchestrator(child_orchestrator_with_retry)
         w.add_activity(throw_activity_with_retry)
         w.start()
 
         task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
-                                   taskhub=taskhub_name, token_credential=None)
+                                                     taskhub=taskhub_name, token_credential=None)
         id = task_hub_client.schedule_new_orchestration(parent_orchestrator_with_retry)
         state = task_hub_client.wait_for_orchestration_completion(id, timeout=30)
         assert state is not None
@@ -460,13 +462,13 @@ def throw_activity(ctx: task.ActivityContext, _):
         raise RuntimeError("Kah-BOOOOM!!!")
 
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
-                                taskhub=taskhub_name, token_credential=None) as w:
+                                    taskhub=taskhub_name, token_credential=None) as w:
         w.add_orchestrator(mock_orchestrator)
         w.add_activity(throw_activity)
         w.start()
 
         task_hub_client = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
-                                   taskhub=taskhub_name, token_credential=None)
+                                                     taskhub=taskhub_name, token_credential=None)
         id = task_hub_client.schedule_new_orchestration(mock_orchestrator)
         state = task_hub_client.wait_for_orchestration_completion(id, timeout=30)
         assert state is not None
@@ -477,19 +479,20 @@ def throw_activity(ctx: task.ActivityContext, _):
         assert state.failure_details.stack_trace is not None
         assert throw_activity_counter == 4
 
+
 def test_custom_status():
 
     def empty_orchestrator(ctx: task.OrchestrationContext, _):
         ctx.set_custom_status("foobaz")
 
     # Start a worker, which will connect to the sidecar in a background thread
     with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=True,
-                                taskhub=taskhub_name, token_credential=None) as w:
+                                    taskhub=taskhub_name, token_credential=None) as w:
         w.add_orchestrator(empty_orchestrator)
         w.start()
 
         c = DurableTaskSchedulerClient(host_address=endpoint, secure_channel=True,
-                                   taskhub=taskhub_name, token_credential=None)
+                                       taskhub=taskhub_name, token_credential=None)
         id = c.schedule_new_orchestration(empty_orchestrator)
         state = c.wait_for_orchestration_completion(id, timeout=30)
 

tests/durabletask-azuremanaged/test_durabletask_grpc_interceptor.py

@@ -1,28 +1,24 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
-import threading
 import unittest
 from concurrent import futures
 from importlib.metadata import version
 
 import grpc
 
 from durabletask.azuremanaged.client import DurableTaskSchedulerClient
-from durabletask.azuremanaged.internal.durabletask_grpc_interceptor import (
-    DTSDefaultClientInterceptorImpl,
-)
 from durabletask.internal import orchestrator_service_pb2 as pb
 from durabletask.internal import orchestrator_service_pb2_grpc as stubs
 
 
 class MockTaskHubSidecarServiceServicer(stubs.TaskHubSidecarServiceServicer):
     """Mock implementation of the TaskHubSidecarService for testing."""
-    
+
     def __init__(self):
         self.captured_metadata = {}
         self.requests_received = 0
-    
+
     def GetInstance(self, request, context):
         """Implementation of GetInstance that captures the metadata."""
         # Store all metadata key-value pairs from the context
@@ -38,7 +34,7 @@ def GetInstance(self, request, context):
 
 class TestDurableTaskGrpcInterceptor(unittest.TestCase):
     """Tests for the DTSDefaultClientInterceptorImpl class."""
-    
+
     @classmethod
     def setUpClass(cls):
         # Start a real gRPC server on a free port
@@ -52,11 +48,11 @@ def setUpClass(cls):
 
         # Start the server in a background thread
         cls.server.start()
-    
+
     @classmethod
     def tearDownClass(cls):
         cls.server.stop(grace=None)
-    
+
     def test_user_agent_metadata_passed_in_request(self):
         """Test that the user agent metadata is correctly passed in gRPC requests."""
         # Create a client that connects to our mock server

tests/durabletask/test_orchestration_e2e.py

@@ -278,10 +278,12 @@ def orchestrator(ctx: task.OrchestrationContext, _):
         assert state.runtime_status == client.OrchestrationStatus.TERMINATED
         assert state.serialized_output == json.dumps("some reason for termination")
 
+
 def test_terminate_recursive():
     def root(ctx: task.OrchestrationContext, _):
         result = yield ctx.call_sub_orchestrator(child)
         return result
+
     def child(ctx: task.OrchestrationContext, _):
         result = yield ctx.wait_for_external_event("my_event")
         return result
@@ -305,7 +307,7 @@ def child(ctx: task.OrchestrationContext, _):
         assert state.runtime_status == client.OrchestrationStatus.TERMINATED
 
         # Verify that child orchestration is also terminated
-        c = task_hub_client.wait_for_orchestration_completion(id, timeout=30)
+        task_hub_client.wait_for_orchestration_completion(id, timeout=30)
         assert state is not None
         assert state.runtime_status == client.OrchestrationStatus.TERMINATED
 
@@ -445,6 +447,7 @@ def throw_activity(ctx: task.ActivityContext, _):
         assert state.failure_details.stack_trace is not None
         assert throw_activity_counter == 4
 
+
 def test_custom_status():
 
     def empty_orchestrator(ctx: task.OrchestrationContext, _):