1 # Copyright 2020 Canonical Ltd.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
15 from unittest
import TestCase
, mock
16 from n2vc
.kubectl
import Kubectl
, CORE_CLIENT
17 from n2vc
.utils
import Dict
18 from kubernetes
.client
.rest
import ApiException
19 from kubernetes
.client
import (
27 class FakeK8sResourceMetadata
:
31 namespace
: str = None,
32 annotations
: dict = {},
35 self
._annotations
= annotations
36 self
._name
= name
or "name"
37 self
._namespace
= namespace
or "namespace"
38 self
._labels
= labels
or {"juju-app": "squid"}
46 return self
._namespace
53 def annotations(self
):
54 return self
._annotations
57 class FakeK8sStorageClass
:
58 def __init__(self
, metadata
=None):
59 self
._metadata
= metadata
or FakeK8sResourceMetadata()
66 class FakeK8sStorageClassesList
:
67 def __init__(self
, items
=[]):
75 class FakeK8sServiceAccountsList
:
76 def __init__(self
, items
=[]):
84 class FakeK8sSecretList
:
85 def __init__(self
, items
=[]):
93 class FakeK8sVersionApiCode
:
94 def __init__(self
, major
: str, minor
: str):
107 fake_list_services
= Dict(
116 "labels": {"juju-app": "squid"},
121 "cluster_ip": "10.152.183.79",
122 "type": "LoadBalancer",
130 "target_port": 30666,
138 "load_balancer": Dict(
141 Dict({"hostname": None, "ip": "192.168.0.201"})
154 class KubectlTestCase(TestCase
):
162 def list_service_for_all_namespaces(self
, **kwargs
):
163 return fake_list_services
166 class GetServices(TestCase
):
167 @mock.patch("n2vc.kubectl.config.load_kube_config")
168 @mock.patch("n2vc.kubectl.client.CoreV1Api")
169 def setUp(self
, mock_core
, mock_config
):
170 mock_core
.return_value
= mock
.MagicMock()
171 mock_config
.return_value
= mock
.MagicMock()
172 self
.kubectl
= Kubectl()
174 @mock.patch("n2vc.kubectl.client.CoreV1Api")
175 def test_get_service(self
, mock_corev1api
):
176 mock_corev1api
.return_value
= FakeCoreV1Api()
177 services
= self
.kubectl
.get_services(
178 field_selector
="metadata.namespace", label_selector
="juju-operator=squid"
180 keys
= ["name", "cluster_ip", "type", "ports", "external_ip"]
181 self
.assertTrue(k
in service
for service
in services
for k
in keys
)
183 def test_get_service_exception(self
):
184 self
.kubectl
.clients
[
186 ].list_service_for_all_namespaces
.side_effect
= ApiException()
187 with self
.assertRaises(ApiException
):
188 self
.kubectl
.get_services()
191 @mock.patch("n2vc.kubectl.client")
192 @mock.patch("n2vc.kubectl.config.kube_config.Configuration.get_default_copy")
193 @mock.patch("n2vc.kubectl.config.load_kube_config")
194 class GetConfiguration(KubectlTestCase
):
196 super(GetConfiguration
, self
).setUp()
198 def test_get_configuration(
200 mock_load_kube_config
,
205 kubectl
.configuration
206 mock_configuration
.assert_called_once()
207 mock_load_kube_config
.assert_called_once()
208 mock_client
.CoreV1Api
.assert_called_once()
209 mock_client
.RbacAuthorizationV1Api
.assert_called_once()
210 mock_client
.StorageV1Api
.assert_called_once()
213 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
214 @mock.patch("kubernetes.config.load_kube_config")
215 class GetDefaultStorageClass(KubectlTestCase
):
217 super(GetDefaultStorageClass
, self
).setUp()
219 # Default Storage Class
220 self
.default_sc_name
= "default-sc"
221 default_sc_metadata
= FakeK8sResourceMetadata(
222 name
=self
.default_sc_name
,
223 annotations
={"storageclass.kubernetes.io/is-default-class": "true"},
225 self
.default_sc
= FakeK8sStorageClass(metadata
=default_sc_metadata
)
227 # Default Storage Class with old annotation
228 self
.default_sc_old_name
= "default-sc-old"
229 default_sc_old_metadata
= FakeK8sResourceMetadata(
230 name
=self
.default_sc_old_name
,
231 annotations
={"storageclass.beta.kubernetes.io/is-default-class": "true"},
233 self
.default_sc_old
= FakeK8sStorageClass(metadata
=default_sc_old_metadata
)
235 # Storage class - not default
236 self
.sc_name
= "default-sc-old"
237 self
.sc
= FakeK8sStorageClass(
238 metadata
=FakeK8sResourceMetadata(name
=self
.sc_name
)
241 def test_get_default_storage_class_exists_default(
242 self
, mock_load_kube_config
, mock_list_storage_class
245 items
= [self
.default_sc
]
246 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
247 sc_name
= kubectl
.get_default_storage_class()
248 self
.assertEqual(sc_name
, self
.default_sc_name
)
249 mock_list_storage_class
.assert_called_once()
251 def test_get_default_storage_class_exists_default_old(
252 self
, mock_load_kube_config
, mock_list_storage_class
255 items
= [self
.default_sc_old
]
256 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
257 sc_name
= kubectl
.get_default_storage_class()
258 self
.assertEqual(sc_name
, self
.default_sc_old_name
)
259 mock_list_storage_class
.assert_called_once()
261 def test_get_default_storage_class_none(
262 self
, mock_load_kube_config
, mock_list_storage_class
265 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=[])
266 sc_name
= kubectl
.get_default_storage_class()
267 self
.assertEqual(sc_name
, None)
268 mock_list_storage_class
.assert_called_once()
270 def test_get_default_storage_class_exists_not_default(
271 self
, mock_load_kube_config
, mock_list_storage_class
275 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
276 sc_name
= kubectl
.get_default_storage_class()
277 self
.assertEqual(sc_name
, self
.sc_name
)
278 mock_list_storage_class
.assert_called_once()
280 def test_get_default_storage_class_choose(
281 self
, mock_load_kube_config
, mock_list_storage_class
284 items
= [self
.sc
, self
.default_sc
]
285 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
286 sc_name
= kubectl
.get_default_storage_class()
287 self
.assertEqual(sc_name
, self
.default_sc_name
)
288 mock_list_storage_class
.assert_called_once()
291 @mock.patch("kubernetes.client.VersionApi.get_code")
292 @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_secret")
293 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_secret")
294 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_service_account")
295 @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_service_account")
296 class CreateServiceAccountClass(KubectlTestCase
):
297 @mock.patch("kubernetes.config.load_kube_config")
298 def setUp(self
, mock_load_kube_config
):
299 super(CreateServiceAccountClass
, self
).setUp()
300 self
.service_account_name
= "Service_account"
301 self
.labels
= {"Key1": "Value1", "Key2": "Value2"}
302 self
.namespace
= "kubernetes"
303 self
.token_id
= "abc12345"
304 self
.kubectl
= Kubectl()
306 def assert_create_secret(self
, mock_create_secret
, secret_name
):
307 annotations
= {"kubernetes.io/service-account.name": self
.service_account_name
}
308 secret_metadata
= V1ObjectMeta(
309 name
=secret_name
, namespace
=self
.namespace
, annotations
=annotations
311 secret_type
= "kubernetes.io/service-account-token"
312 secret
= V1Secret(metadata
=secret_metadata
, type=secret_type
)
313 mock_create_secret
.assert_called_once_with(self
.namespace
, secret
)
315 def assert_create_service_account_v_1_24(
316 self
, mock_create_service_account
, secret_name
318 sevice_account_metadata
= V1ObjectMeta(
319 name
=self
.service_account_name
, labels
=self
.labels
, namespace
=self
.namespace
321 secrets
= [V1SecretReference(name
=secret_name
, namespace
=self
.namespace
)]
322 service_account
= V1ServiceAccount(
323 metadata
=sevice_account_metadata
, secrets
=secrets
325 mock_create_service_account
.assert_called_once_with(
326 self
.namespace
, service_account
329 def assert_create_service_account_v_1_23(self
, mock_create_service_account
):
330 metadata
= V1ObjectMeta(
331 name
=self
.service_account_name
, labels
=self
.labels
, namespace
=self
.namespace
333 service_account
= V1ServiceAccount(metadata
=metadata
)
334 mock_create_service_account
.assert_called_once_with(
335 self
.namespace
, service_account
338 @mock.patch("n2vc.kubectl.uuid.uuid4")
339 def test_secret_is_created_when_k8s_1_24(
342 mock_list_service_account
,
343 mock_create_service_account
,
348 mock_list_service_account
.return_value
= FakeK8sServiceAccountsList(items
=[])
349 mock_list_secret
.return_value
= FakeK8sSecretList(items
=[])
350 mock_version
.return_value
= FakeK8sVersionApiCode("1", "24")
351 mock_uuid4
.return_value
= self
.token_id
352 self
.kubectl
.create_service_account(
353 self
.service_account_name
, self
.labels
, self
.namespace
355 secret_name
= "{}-token-{}".format(self
.service_account_name
, self
.token_id
[:5])
356 self
.assert_create_service_account_v_1_24(
357 mock_create_service_account
, secret_name
359 self
.assert_create_secret(mock_create_secret
, secret_name
)
361 def test_secret_is_not_created_when_k8s_1_23(
363 mock_list_service_account
,
364 mock_create_service_account
,
369 mock_list_service_account
.return_value
= FakeK8sServiceAccountsList(items
=[])
370 mock_version
.return_value
= FakeK8sVersionApiCode("1", "23+")
371 self
.kubectl
.create_service_account(
372 self
.service_account_name
, self
.labels
, self
.namespace
374 self
.assert_create_service_account_v_1_23(mock_create_service_account
)
375 mock_create_secret
.assert_not_called()
376 mock_list_secret
.assert_not_called()
378 def test_raise_exception_if_service_account_already_exists(
380 mock_list_service_account
,
381 mock_create_service_account
,
386 mock_list_service_account
.return_value
= FakeK8sServiceAccountsList(items
=[1])
387 with self
.assertRaises(Exception) as context
:
388 self
.kubectl
.create_service_account(
389 self
.service_account_name
, self
.labels
, self
.namespace
392 "Service account with metadata.name={} already exists".format(
393 self
.service_account_name
395 in str(context
.exception
)
397 mock_create_service_account
.assert_not_called()
398 mock_create_secret
.assert_not_called()
400 @mock.patch("n2vc.kubectl.uuid.uuid4")
401 def test_raise_exception_if_secret_already_exists(
404 mock_list_service_account
,
405 mock_create_service_account
,
410 mock_list_service_account
.return_value
= FakeK8sServiceAccountsList(items
=[])
411 mock_list_secret
.return_value
= FakeK8sSecretList(items
=[1])
412 mock_version
.return_value
= FakeK8sVersionApiCode("1", "24+")
413 mock_uuid4
.return_value
= self
.token_id
414 with self
.assertRaises(Exception) as context
:
415 self
.kubectl
.create_service_account(
416 self
.service_account_name
, self
.labels
, self
.namespace
419 "Secret with metadata.name={}-token-{} already exists".format(
420 self
.service_account_name
, self
.token_id
[:5]
422 in str(context
.exception
)
424 mock_create_service_account
.assert_called()
425 mock_create_secret
.assert_not_called()