6
6
from unittest .mock import Mock , patch
7
7
8
8
# Mock Kubernetes configuration before importing the module
9
- with patch ('kubernetes.config.load_incluster_config' ), \
10
- patch ('kubernetes.config.load_kube_config' ):
9
+ with patch ('kubernetes.config.load_incluster_config' ), patch ('kubernetes.config.load_kube_config' ):
11
10
from enterprise_gateway .services .processproxies .k8s import KubernetesProcessProxy
12
11
13
12
@@ -19,15 +18,14 @@ def setUp(self):
19
18
self .mock_kernel_manager = Mock ()
20
19
self .mock_kernel_manager .get_kernel_username .return_value = "testuser"
21
20
self .mock_kernel_manager .port_range = "0..0" # Mock port range
22
-
21
+
23
22
# Mock proxy config
24
- self .proxy_config = {
25
- "kernel_id" : "test-kernel-id" ,
26
- "kernel_name" : "python3"
27
- }
23
+ self .proxy_config = {"kernel_id" : "test-kernel-id" , "kernel_name" : "python3" }
28
24
29
25
# Mock KernelSessionManager methods
30
- with patch ('enterprise_gateway.services.processproxies.k8s.KernelSessionManager' ) as mock_session_manager :
26
+ with patch (
27
+ 'enterprise_gateway.services.processproxies.k8s.KernelSessionManager'
28
+ ) as mock_session_manager :
31
29
mock_session_manager .get_kernel_username .return_value = "testuser"
32
30
self .proxy = KubernetesProcessProxy (self .mock_kernel_manager , self .proxy_config )
33
31
self .proxy .kernel_id = "test-kernel-id"
@@ -54,7 +52,7 @@ def test_valid_template_substitution(self):
54
52
result = self .proxy ._safe_template_substitute (template , variables )
55
53
self .assertEqual (result , expected )
56
54
57
- def test_missing_variables_fallback (self ):
55
+ def test_missing_variables_fallback (self ):
58
56
# Test the full pod name determination process
59
57
kwargs = {
60
58
"env" : {
@@ -63,8 +61,9 @@ def test_missing_variables_fallback(self):
63
61
}
64
62
}
65
63
66
- with patch .object (self .proxy , 'log' ), \
67
- patch ('enterprise_gateway.services.processproxies.k8s.KernelSessionManager' ) as mock_session_manager :
64
+ with patch .object (self .proxy , 'log' ), patch (
65
+ 'enterprise_gateway.services.processproxies.k8s.KernelSessionManager'
66
+ ) as mock_session_manager :
68
67
mock_session_manager .get_kernel_username .return_value = "testuser"
69
68
result = self .proxy ._determine_kernel_pod_name (** kwargs )
70
69
# Should fall back to default naming: kernel_username + "-" + kernel_id
@@ -132,8 +131,9 @@ def test_pod_name_determination_with_malicious_template(self):
132
131
}
133
132
}
134
133
135
- with patch .object (self .proxy , 'log' ), \
136
- patch ('enterprise_gateway.services.processproxies.k8s.KernelSessionManager' ) as mock_session_manager :
134
+ with patch .object (self .proxy , 'log' ), patch (
135
+ 'enterprise_gateway.services.processproxies.k8s.KernelSessionManager'
136
+ ) as mock_session_manager :
137
137
mock_session_manager .get_kernel_username .return_value = "testuser"
138
138
result = self .proxy ._determine_kernel_pod_name (** kwargs )
139
139
# Should fall back to default naming
@@ -148,8 +148,9 @@ def test_pod_name_determination_with_missing_variables(self):
148
148
}
149
149
}
150
150
151
- with patch .object (self .proxy , 'log' ), \
152
- patch ('enterprise_gateway.services.processproxies.k8s.KernelSessionManager' ) as mock_session_manager :
151
+ with patch .object (self .proxy , 'log' ), patch (
152
+ 'enterprise_gateway.services.processproxies.k8s.KernelSessionManager'
153
+ ) as mock_session_manager :
153
154
mock_session_manager .get_kernel_username .return_value = "testuser"
154
155
result = self .proxy ._determine_kernel_pod_name (** kwargs )
155
156
# Should fall back to default naming
0 commit comments