Add Temporal adapter BDD tests with persistent event loop

- Add comprehensive Temporal adapter feature tests covering workflows, signals, queries, workers, and schedules
- Implement persistent event loop per scenario to keep workers alive across steps
- Add test workflows and activities for various scenarios (greeting, signal/query, timeout, retry, etc.)
- Add Temporal container support with dev mode SQLite backend
- Fix workflow execution to use run_async helper for proper async handling

Author: HosseinNejatiJavaremi

.env.test

@@ -19,7 +19,7 @@ FASTAPI__SWAGGER_UI_PARAMS={}
 TESTCONTAINERS_RYUK_CONTAINER_IMAGE=testcontainers/ryuk:0.14.0
 
 # Temporal Configuration
-TEMPORAL__IMAGE=temporalio/auto-setup:1.29.2
+TEMPORAL__IMAGE=temporalio/temporal:1.5.1
 TEMPORAL__HOST=localhost
 TEMPORAL__PORT=7233
 TEMPORAL__NAMESPACE=default

archipy/adapters/temporal/adapters.py

@@ -211,9 +211,13 @@ async def start_workflow(
         workflow_id = workflow_id or str(uuid4())
         task_queue = task_queue or self.config.TASK_QUEUE
 
+        # Build positional args: only include arg if it's not None,
+        # so workflows that take no parameters aren't given an extra argument.
+        positional_args: list[Any] = [] if arg is None else [arg]
+
         return await client.start_workflow(
             workflow,
-            arg,
+            *positional_args,
             id=workflow_id,
             task_queue=task_queue,
             execution_timeout=timedelta(seconds=execution_timeout or self.config.WORKFLOW_EXECUTION_TIMEOUT),
@@ -259,9 +263,13 @@ async def execute_workflow(
         workflow_id = workflow_id or str(uuid4())
         task_queue = task_queue or self.config.TASK_QUEUE
 
+        # Build positional args: only include arg if it's not None,
+        # so workflows that take no parameters aren't given an extra argument.
+        positional_args: list[Any] = [] if arg is None else [arg]
+
         return await client.execute_workflow(
             workflow,
-            arg,
+            *positional_args,
             id=workflow_id,
             task_queue=task_queue,
             execution_timeout=timedelta(seconds=execution_timeout or self.config.WORKFLOW_EXECUTION_TIMEOUT),
@@ -329,7 +337,9 @@ async def signal_workflow(
                 If None, signals the latest run. Defaults to None.
         """
         handle = await self.get_workflow_handle(workflow_id, run_id)
-        await handle.signal(signal_name, arg)
+        # Only pass arg if it's not None, so signal handlers with no parameters work correctly.
+        positional_args: list[Any] = [] if arg is None else [arg]
+        await handle.signal(signal_name, *positional_args)
 
     @override
     async def query_workflow(
@@ -352,7 +362,9 @@ async def query_workflow(
             Any: The query result from the workflow.
         """
         handle = await self.get_workflow_handle(workflow_id, run_id)
-        return await handle.query(query_name, arg)
+        # Only pass arg if it's not None, so query handlers with no parameters work correctly.
+        positional_args: list[Any] = [] if arg is None else [arg]
+        return await handle.query(query_name, *positional_args)
 
     @override
     async def list_workflows(

archipy/adapters/temporal/worker.py

@@ -283,6 +283,14 @@ async def start_worker(
         Raises:
             WorkerConnectionError: If the worker fails to start.
         """
+        if not task_queue or not task_queue.strip():
+            raise WorkerConnectionError(
+                additional_data={
+                    "message": "Task queue name cannot be empty",
+                    "task_queue": task_queue,
+                },
+            )
+
         client = await self._get_client()
         worker_id = str(uuid4())
         worker_identity = identity or f"worker-{worker_id[:8]}"

features/environment.py

@@ -35,6 +35,7 @@ class TestConfig(BaseConfig):
     KEYCLOAK__IMAGE: str
     SCYLLADB__IMAGE: str
     STARROCKS__IMAGE: str
+    TEMPORAL__IMAGE: str
     TESTCONTAINERS_RYUK_CONTAINER_IMAGE: str | None = None
 
     def __init__(self, **kwargs) -> None:
@@ -173,6 +174,14 @@ def after_scenario(context: Context, scenario: Scenario) -> None:
     scenario_id = getattr(scenario, "id", "unknown")
     logger.info(f"Cleaning up scenario: {scenario.name} (ID: {scenario_id})")
 
+    # Clean up Temporal event loop and workers (must happen before context cleanup)
+    try:
+        from features.steps.temporal_adapter_steps import cleanup_event_loop
+
+        cleanup_event_loop(context)
+    except Exception:
+        pass
+
     # Clean up the scenario context and remove from pool
     if hasattr(context, "scenario_context_pool"):
         context.scenario_context_pool.cleanup_context(scenario_id)