re-pre-commit

Author: Carreau

enterprise_gateway/client/gateway_client.py

@@ -111,7 +111,9 @@ def __init__(
         self.response_reader = Thread(target=self._read_responses)
         self.response_queues = {}
         self.interrupt_thread = None
-        self.log.debug("Initializing kernel client (%s) to %s", kernel_id, self.kernel_ws_api_endpoint)
+        self.log.debug(
+            "Initializing kernel client (%s) to %s", kernel_id, self.kernel_ws_api_endpoint
+        )
 
         try:
             self.kernel_socket = websocket.create_connection(

enterprise_gateway/services/processproxies/distributed.py

@@ -101,7 +101,9 @@ async def launch_process(
             result_pid = self._launch_remote_process(kernel_cmd, **kwargs)
             self.pid = int(result_pid)
         except Exception as e:
-            error_message = f"Failure occurred starting kernel on '{self.ip}'.  Returned result: {e}"
+            error_message = (
+                f"Failure occurred starting kernel on '{self.ip}'.  Returned result: {e}"
+            )
             self.log_and_raise(http_status_code=500, reason=error_message)
 
         self.log.info(

enterprise_gateway/services/processproxies/k8s.py

@@ -382,7 +382,9 @@ def _create_kernel_namespace(self, service_account_name: str) -> str:
                 self.log.info("Re-using kernel namespace: %s", namespace)
             else:
                 if self.delete_kernel_namespace:
-                    reason = f"Error occurred creating role binding for namespace '{namespace}': {err}"
+                    reason = (
+                        f"Error occurred creating role binding for namespace '{namespace}': {err}"
+                    )
                     # delete the namespace since we'll be using the EG namespace...
                     body = client.V1DeleteOptions(
                         grace_period_seconds=0, propagation_policy="Background"

enterprise_gateway/services/processproxies/processproxy.py

@@ -223,7 +223,9 @@ def _prepare_response_socket(self) -> None:
                 s.bind((bind_ip, port))
             except OSError as e:
                 if e.errno == errno.EADDRINUSE:
-                    self.log.info("Response port %s is already in use, trying another port...", port)
+                    self.log.info(
+                        "Response port %s is already in use, trying another port...", port
+                    )
                     continue
                 elif e.errno in (errno.EACCES, getattr(errno, "WSAEACCES", errno.EACCES)):
                     self.log.warning(
@@ -710,7 +712,9 @@ def _get_ssh_client(self, host: str) -> SSHClient | None:
                 error_message = error_message_prefix + (
                     " provided" if self.remote_pwd else "-less SSH"
                 )
-                error_message = error_message + f"and EG_REMOTE_GSS_SSH={self._use_gss_raw!r} ({self.use_gss})"
+                error_message = (
+                    error_message + f"and EG_REMOTE_GSS_SSH={self._use_gss_raw!r} ({self.use_gss})"
+                )
 
             self.log_and_raise(http_status_code=http_status_code, reason=error_message)
         return ssh

enterprise_gateway/tests/test_process_proxy.py

@@ -61,9 +61,12 @@ def test_missing_variables_fallback(self):
             }
         }
 
-        with patch.object(self.proxy, 'log'), patch(
-            'enterprise_gateway.services.processproxies.k8s.KernelSessionManager'
-        ) as mock_session_manager:
+        with (
+            patch.object(self.proxy, 'log'),
+            patch(
+                'enterprise_gateway.services.processproxies.k8s.KernelSessionManager'
+            ) as mock_session_manager,
+        ):
             mock_session_manager.get_kernel_username.return_value = "testuser"
             result = self.proxy._determine_kernel_pod_name(**kwargs)
             # Should fall back to default naming: kernel_username + "-" + kernel_id
@@ -97,9 +100,10 @@ def test_malicious_template_injection_prevention(self):
         variables = {"kernel_id": "test-123"}
 
         for malicious_template in malicious_templates:
-            with self.subTest(template=malicious_template), patch.object(
-                self.proxy, 'log'
-            ) as mock_log:
+            with (
+                self.subTest(template=malicious_template),
+                patch.object(self.proxy, 'log') as mock_log,
+            ):
                 result = self.proxy._safe_template_substitute(malicious_template, variables)
                 # All malicious templates should be treated as invalid and return None
                 self.assertIsNone(result)
@@ -131,9 +135,12 @@ def test_pod_name_determination_with_malicious_template(self):
             }
         }
 
-        with patch.object(self.proxy, 'log'), patch(
-            'enterprise_gateway.services.processproxies.k8s.KernelSessionManager'
-        ) as mock_session_manager:
+        with (
+            patch.object(self.proxy, 'log'),
+            patch(
+                'enterprise_gateway.services.processproxies.k8s.KernelSessionManager'
+            ) as mock_session_manager,
+        ):
             mock_session_manager.get_kernel_username.return_value = "testuser"
             result = self.proxy._determine_kernel_pod_name(**kwargs)
             # Should fall back to default naming
@@ -148,9 +155,12 @@ def test_pod_name_determination_with_missing_variables(self):
             }
         }
 
-        with patch.object(self.proxy, 'log'), patch(
-            'enterprise_gateway.services.processproxies.k8s.KernelSessionManager'
-        ) as mock_session_manager:
+        with (
+            patch.object(self.proxy, 'log'),
+            patch(
+                'enterprise_gateway.services.processproxies.k8s.KernelSessionManager'
+            ) as mock_session_manager,
+        ):
             mock_session_manager.get_kernel_username.return_value = "testuser"
             result = self.proxy._determine_kernel_pod_name(**kwargs)
             # Should fall back to default naming