Update env to use different eopf processors

Author: jgaucher-cs

services/staging/rs_server_staging/processors/processor_staging.py

@@ -734,14 +734,14 @@ def dask_cluster_connect(
 
         2. **Kubernetes Mode**:
         - If `self.cluster` is not already defined, the method attempts to connect to a Dask Gateway
-            (using environment variables `DASK_GATEWAY__ADDRESS` and `DASK_GATEWAY__AUTH__TYPE`) to
+            (using environment variables `DASK_GATEWAY_ADDRESS` and `DASK_GATEWAY__AUTH__TYPE`) to
             retrieve a list of existing clusters.
         - If no clusters are available, it attempts to create a new cluster scheduler.
 
         Raises:
             RuntimeError: Raised if the cluster name is None, required environment variables are missing,
                         cluster creation fails or authentication errors occur.
-            KeyError: Raised if the necessary Dask Gateway environment variables (`DASK_GATEWAY__ADDRESS`,
+            KeyError: Raised if the necessary Dask Gateway environment variables (`DASK_GATEWAY_ADDRESS`,
                 `DASK_GATEWAY__AUTH__TYPE`, `RSPY_DASK_STAGING_CLUSTER_NAME`, `JUPYTERHUB_API_TOKEN` ) are not set.
             IndexError: Raised if no clusters are found in the Dask Gateway and new cluster creation is attempted.
             dask_gateway.exceptions.GatewayServerError: Raised when there is a server-side error in Dask Gateway.
@@ -802,13 +802,13 @@ def dask_cluster_connect(
                         raise RuntimeError(f"Unsupported authentication type: {auth_type}")
 
                 gateway = Gateway(
-                    address=os.environ["DASK_GATEWAY__ADDRESS"],
+                    address=os.environ["DASK_GATEWAY_ADDRESS"],
                     auth=gateway_auth,
                 )
 
                 # Sort the clusters by newest first
                 clusters = sorted(gateway.list_clusters(), key=lambda cluster: cluster.start_time, reverse=True)
-                self.logger.debug(f"Cluster list for gateway {os.environ['DASK_GATEWAY__ADDRESS']!r}: {clusters}")
+                self.logger.debug(f"Cluster list for gateway {os.environ['DASK_GATEWAY_ADDRESS']!r}: {clusters}")
 
                 # In local mode, get the first cluster from the gateway.
                 cluster_id = None
@@ -847,7 +847,7 @@ def dask_cluster_connect(
                 self.logger.exception(
                     "Failed to retrieve the required connection details for "
                     "the Dask Gateway from one or more of the following environment variables: "
-                    "DASK_GATEWAY__ADDRESS, RSPY_DASK_STAGING_CLUSTER_NAME, "
+                    "DASK_GATEWAY_ADDRESS, RSPY_DASK_STAGING_CLUSTER_NAME, "
                     f"JUPYTERHUB_API_TOKEN, DASK_GATEWAY__AUTH__TYPE. {e}",
                 )
 

services/staging/tests/test_processors/test_processor_staging.py

@@ -315,7 +315,7 @@ def test_dask_cluster_connect(
         mocker.patch.dict(
             os.environ,
             {
-                "DASK_GATEWAY__ADDRESS": "gateway-address",
+                "DASK_GATEWAY_ADDRESS": "gateway-address",
                 "DASK_GATEWAY__AUTH__TYPE": "jupyterhub",
                 "JUPYTERHUB_API_TOKEN": "mock_api_token",
                 "RSPY_DASK_STAGING_CLUSTER_NAME": str(
@@ -375,7 +375,7 @@ def test_dask_cluster_connect_failure_no_cluster_name(
         mocker.patch.dict(
             os.environ,
             {
-                "DASK_GATEWAY__ADDRESS": "gateway-address",
+                "DASK_GATEWAY_ADDRESS": "gateway-address",
                 "DASK_GATEWAY__AUTH__TYPE": "jupyterhub",
                 "JUPYTERHUB_API_TOKEN": "mock_api_token",
                 "RSPY_DASK_STAGING_CLUSTER_NAME": non_existent_cluster,
@@ -411,7 +411,7 @@ def test_dask_cluster_connect_failure_no_envs(
         mocker.patch.dict(
             os.environ,
             {
-                "DASK_GATEWAY__ADDRESS": "gateway-address",
+                "DASK_GATEWAY_ADDRESS": "gateway-address",
             },
         )
 
@@ -606,7 +606,7 @@ async def test_process_rspy_features_success(
         mocker.patch.dict(
             os.environ,
             {
-                "DASK_GATEWAY__ADDRESS": "gateway-address",
+                "DASK_GATEWAY_ADDRESS": "gateway-address",
                 "DASK_GATEWAY__AUTH__TYPE": "jupyterhub",
                 "JUPYTERHUB_API_TOKEN": "mock_api_token",
                 "RSPY_DASK_STAGING_CLUSTER_NAME": str(