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.
18 from unittest
import TestCase
, mock
19 from n2vc
.kubectl
import Kubectl
, CORE_CLIENT
, CUSTOM_OBJECT_CLIENT
20 from n2vc
.utils
import Dict
21 from kubernetes
.client
.rest
import ApiException
22 from kubernetes
.client
import (
30 class FakeK8sResourceMetadata
:
34 namespace
: str = None,
35 annotations
: dict = {},
38 self
._annotations
= annotations
39 self
._name
= name
or "name"
40 self
._namespace
= namespace
or "namespace"
41 self
._labels
= labels
or {"juju-app": "squid"}
49 return self
._namespace
56 def annotations(self
):
57 return self
._annotations
60 class FakeK8sStorageClass
:
61 def __init__(self
, metadata
=None):
62 self
._metadata
= metadata
or FakeK8sResourceMetadata()
69 class FakeK8sStorageClassesList
:
70 def __init__(self
, items
=[]):
78 class FakeK8sServiceAccountsList
:
79 def __init__(self
, items
=[]):
87 class FakeK8sSecretList
:
88 def __init__(self
, items
=[]):
96 class FakeK8sVersionApiCode
:
97 def __init__(self
, major
: str, minor
: str):
110 fake_list_services
= Dict(
119 "labels": {"juju-app": "squid"},
124 "cluster_ip": "10.152.183.79",
125 "type": "LoadBalancer",
133 "target_port": 30666,
141 "load_balancer": Dict(
144 Dict({"hostname": None, "ip": "192.168.0.201"})
157 class KubectlTestCase(TestCase
):
165 def list_service_for_all_namespaces(self
, **kwargs
):
166 return fake_list_services
169 class GetServices(TestCase
):
170 @mock.patch("n2vc.kubectl.config.load_kube_config")
171 @mock.patch("n2vc.kubectl.client.CoreV1Api")
172 def setUp(self
, mock_core
, mock_config
):
173 mock_core
.return_value
= mock
.MagicMock()
174 mock_config
.return_value
= mock
.MagicMock()
175 self
.kubectl
= Kubectl()
177 @mock.patch("n2vc.kubectl.client.CoreV1Api")
178 def test_get_service(self
, mock_corev1api
):
179 mock_corev1api
.return_value
= FakeCoreV1Api()
180 services
= self
.kubectl
.get_services(
181 field_selector
="metadata.namespace", label_selector
="juju-operator=squid"
183 keys
= ["name", "cluster_ip", "type", "ports", "external_ip"]
184 self
.assertTrue(k
in service
for service
in services
for k
in keys
)
186 def test_get_service_exception(self
):
187 self
.kubectl
.clients
[
189 ].list_service_for_all_namespaces
.side_effect
= ApiException()
190 with self
.assertRaises(ApiException
):
191 self
.kubectl
.get_services()
194 @mock.patch("n2vc.kubectl.client")
195 @mock.patch("n2vc.kubectl.config.kube_config.Configuration.get_default_copy")
196 @mock.patch("n2vc.kubectl.config.load_kube_config")
197 class GetConfiguration(KubectlTestCase
):
199 super(GetConfiguration
, self
).setUp()
201 def test_get_configuration(
203 mock_load_kube_config
,
208 kubectl
.configuration
209 mock_configuration
.assert_called_once()
210 mock_load_kube_config
.assert_called_once()
211 mock_client
.CoreV1Api
.assert_called_once()
212 mock_client
.RbacAuthorizationV1Api
.assert_called_once()
213 mock_client
.StorageV1Api
.assert_called_once()
216 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
217 @mock.patch("kubernetes.config.load_kube_config")
218 class GetDefaultStorageClass(KubectlTestCase
):
220 super(GetDefaultStorageClass
, self
).setUp()
222 # Default Storage Class
223 self
.default_sc_name
= "default-sc"
224 default_sc_metadata
= FakeK8sResourceMetadata(
225 name
=self
.default_sc_name
,
226 annotations
={"storageclass.kubernetes.io/is-default-class": "true"},
228 self
.default_sc
= FakeK8sStorageClass(metadata
=default_sc_metadata
)
230 # Default Storage Class with old annotation
231 self
.default_sc_old_name
= "default-sc-old"
232 default_sc_old_metadata
= FakeK8sResourceMetadata(
233 name
=self
.default_sc_old_name
,
234 annotations
={"storageclass.beta.kubernetes.io/is-default-class": "true"},
236 self
.default_sc_old
= FakeK8sStorageClass(metadata
=default_sc_old_metadata
)
238 # Storage class - not default
239 self
.sc_name
= "default-sc-old"
240 self
.sc
= FakeK8sStorageClass(
241 metadata
=FakeK8sResourceMetadata(name
=self
.sc_name
)
244 def test_get_default_storage_class_exists_default(
245 self
, mock_load_kube_config
, mock_list_storage_class
248 items
= [self
.default_sc
]
249 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
250 sc_name
= kubectl
.get_default_storage_class()
251 self
.assertEqual(sc_name
, self
.default_sc_name
)
252 mock_list_storage_class
.assert_called_once()
254 def test_get_default_storage_class_exists_default_old(
255 self
, mock_load_kube_config
, mock_list_storage_class
258 items
= [self
.default_sc_old
]
259 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
260 sc_name
= kubectl
.get_default_storage_class()
261 self
.assertEqual(sc_name
, self
.default_sc_old_name
)
262 mock_list_storage_class
.assert_called_once()
264 def test_get_default_storage_class_none(
265 self
, mock_load_kube_config
, mock_list_storage_class
268 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=[])
269 sc_name
= kubectl
.get_default_storage_class()
270 self
.assertEqual(sc_name
, None)
271 mock_list_storage_class
.assert_called_once()
273 def test_get_default_storage_class_exists_not_default(
274 self
, mock_load_kube_config
, mock_list_storage_class
278 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
279 sc_name
= kubectl
.get_default_storage_class()
280 self
.assertEqual(sc_name
, self
.sc_name
)
281 mock_list_storage_class
.assert_called_once()
283 def test_get_default_storage_class_choose(
284 self
, mock_load_kube_config
, mock_list_storage_class
287 items
= [self
.sc
, self
.default_sc
]
288 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
289 sc_name
= kubectl
.get_default_storage_class()
290 self
.assertEqual(sc_name
, self
.default_sc_name
)
291 mock_list_storage_class
.assert_called_once()
294 @mock.patch("kubernetes.client.VersionApi.get_code")
295 @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_secret")
296 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_secret")
297 @mock.patch("kubernetes.client.CoreV1Api.create_namespaced_service_account")
298 @mock.patch("kubernetes.client.CoreV1Api.list_namespaced_service_account")
299 class CreateServiceAccountClass(KubectlTestCase
):
300 @mock.patch("kubernetes.config.load_kube_config")
301 def setUp(self
, mock_load_kube_config
):
302 super(CreateServiceAccountClass
, self
).setUp()
303 self
.service_account_name
= "Service_account"
304 self
.labels
= {"Key1": "Value1", "Key2": "Value2"}
305 self
.namespace
= "kubernetes"
306 self
.token_id
= "abc12345"
307 self
.kubectl
= Kubectl()
309 def assert_create_secret(self
, mock_create_secret
, secret_name
):
310 annotations
= {"kubernetes.io/service-account.name": self
.service_account_name
}
311 secret_metadata
= V1ObjectMeta(
312 name
=secret_name
, namespace
=self
.namespace
, annotations
=annotations
314 secret_type
= "kubernetes.io/service-account-token"
315 secret
= V1Secret(metadata
=secret_metadata
, type=secret_type
)
316 mock_create_secret
.assert_called_once_with(self
.namespace
, secret
)
318 def assert_create_service_account_v_1_24(
319 self
, mock_create_service_account
, secret_name
321 sevice_account_metadata
= V1ObjectMeta(
322 name
=self
.service_account_name
, labels
=self
.labels
, namespace
=self
.namespace
324 secrets
= [V1SecretReference(name
=secret_name
, namespace
=self
.namespace
)]
325 service_account
= V1ServiceAccount(
326 metadata
=sevice_account_metadata
, secrets
=secrets
328 mock_create_service_account
.assert_called_once_with(
329 self
.namespace
, service_account
332 def assert_create_service_account_v_1_23(self
, mock_create_service_account
):
333 metadata
= V1ObjectMeta(
334 name
=self
.service_account_name
, labels
=self
.labels
, namespace
=self
.namespace
336 service_account
= V1ServiceAccount(metadata
=metadata
)
337 mock_create_service_account
.assert_called_once_with(
338 self
.namespace
, service_account
341 @mock.patch("n2vc.kubectl.uuid.uuid4")
342 def test_secret_is_created_when_k8s_1_24(
345 mock_list_service_account
,
346 mock_create_service_account
,
351 mock_list_service_account
.return_value
= FakeK8sServiceAccountsList(items
=[])
352 mock_list_secret
.return_value
= FakeK8sSecretList(items
=[])
353 mock_version
.return_value
= FakeK8sVersionApiCode("1", "24")
354 mock_uuid4
.return_value
= self
.token_id
355 self
.kubectl
.create_service_account(
356 self
.service_account_name
, self
.labels
, self
.namespace
358 secret_name
= "{}-token-{}".format(self
.service_account_name
, self
.token_id
[:5])
359 self
.assert_create_service_account_v_1_24(
360 mock_create_service_account
, secret_name
362 self
.assert_create_secret(mock_create_secret
, secret_name
)
364 def test_secret_is_not_created_when_k8s_1_23(
366 mock_list_service_account
,
367 mock_create_service_account
,
372 mock_list_service_account
.return_value
= FakeK8sServiceAccountsList(items
=[])
373 mock_version
.return_value
= FakeK8sVersionApiCode("1", "23+")
374 self
.kubectl
.create_service_account(
375 self
.service_account_name
, self
.labels
, self
.namespace
377 self
.assert_create_service_account_v_1_23(mock_create_service_account
)
378 mock_create_secret
.assert_not_called()
379 mock_list_secret
.assert_not_called()
381 def test_raise_exception_if_service_account_already_exists(
383 mock_list_service_account
,
384 mock_create_service_account
,
389 mock_list_service_account
.return_value
= FakeK8sServiceAccountsList(items
=[1])
390 with self
.assertRaises(Exception) as context
:
391 self
.kubectl
.create_service_account(
392 self
.service_account_name
, self
.labels
, self
.namespace
395 "Service account with metadata.name={} already exists".format(
396 self
.service_account_name
398 in str(context
.exception
)
400 mock_create_service_account
.assert_not_called()
401 mock_create_secret
.assert_not_called()
403 @mock.patch("n2vc.kubectl.uuid.uuid4")
404 def test_raise_exception_if_secret_already_exists(
407 mock_list_service_account
,
408 mock_create_service_account
,
413 mock_list_service_account
.return_value
= FakeK8sServiceAccountsList(items
=[])
414 mock_list_secret
.return_value
= FakeK8sSecretList(items
=[1])
415 mock_version
.return_value
= FakeK8sVersionApiCode("1", "24+")
416 mock_uuid4
.return_value
= self
.token_id
417 with self
.assertRaises(Exception) as context
:
418 self
.kubectl
.create_service_account(
419 self
.service_account_name
, self
.labels
, self
.namespace
422 "Secret with metadata.name={}-token-{} already exists".format(
423 self
.service_account_name
, self
.token_id
[:5]
425 in str(context
.exception
)
427 mock_create_service_account
.assert_called()
428 mock_create_secret
.assert_not_called()
431 @mock.patch("kubernetes.client.CustomObjectsApi.create_namespaced_custom_object")
432 class CreateCertificateClass(asynctest
.TestCase
):
433 @mock.patch("kubernetes.config.load_kube_config")
434 def setUp(self
, mock_load_kube_config
):
435 super(CreateCertificateClass
, self
).setUp()
436 self
.namespace
= "osm"
437 self
.name
= "test-cert"
438 self
.dns_prefix
= "*"
439 self
.secret_name
= "test-cert-secret"
440 self
.usages
= ["server auth"]
441 self
.issuer_name
= "ca-issuer"
442 self
.kubectl
= Kubectl()
444 @asynctest.fail_on(active_handles
=True)
445 async def test_certificate_is_created(
447 mock_create_certificate
,
451 os
.path
.dirname(__file__
), "testdata", "test_certificate.yaml"
454 ) as test_certificate
:
455 certificate_body
= yaml
.safe_load(test_certificate
.read())
456 print(certificate_body
)
457 await self
.kubectl
.create_certificate(
458 namespace
=self
.namespace
,
460 dns_prefix
=self
.dns_prefix
,
461 secret_name
=self
.secret_name
,
463 issuer_name
=self
.issuer_name
,
465 mock_create_certificate
.assert_called_once_with(
466 group
="cert-manager.io",
467 plural
="certificates",
469 body
=certificate_body
,
470 namespace
=self
.namespace
,
473 @asynctest.fail_on(active_handles
=True)
474 async def test_no_exception_if_alreadyexists(
476 mock_create_certificate
,
478 api_exception
= ApiException()
479 api_exception
.body
= '{"reason": "AlreadyExists"}'
480 self
.kubectl
.clients
[
482 ].create_namespaced_custom_object
.side_effect
= api_exception
485 await self
.kubectl
.create_certificate(
486 namespace
=self
.namespace
,
488 dns_prefix
=self
.dns_prefix
,
489 secret_name
=self
.secret_name
,
491 issuer_name
=self
.issuer_name
,
495 self
.assertFalse(raised
, "An exception was raised")
497 @asynctest.fail_on(active_handles
=True)
498 async def test_other_exceptions(
500 mock_create_certificate
,
502 self
.kubectl
.clients
[
504 ].create_namespaced_custom_object
.side_effect
= Exception()
505 with self
.assertRaises(Exception):
506 await self
.kubectl
.create_certificate(
507 namespace
=self
.namespace
,
509 dns_prefix
=self
.dns_prefix
,
510 secret_name
=self
.secret_name
,
512 issuer_name
=self
.issuer_name
,
516 @mock.patch("kubernetes.client.CustomObjectsApi.delete_namespaced_custom_object")
517 class DeleteCertificateClass(asynctest
.TestCase
):
518 @mock.patch("kubernetes.config.load_kube_config")
519 def setUp(self
, mock_load_kube_config
):
520 super(DeleteCertificateClass
, self
).setUp()
521 self
.namespace
= "osm"
522 self
.object_name
= "test-cert"
523 self
.kubectl
= Kubectl()
525 @asynctest.fail_on(active_handles
=True)
526 async def test_no_exception_if_notfound(
528 mock_create_certificate
,
530 api_exception
= ApiException()
531 api_exception
.body
= '{"reason": "NotFound"}'
532 self
.kubectl
.clients
[
534 ].delete_namespaced_custom_object
.side_effect
= api_exception
537 await self
.kubectl
.delete_certificate(
538 namespace
=self
.namespace
,
539 object_name
=self
.object_name
,
543 self
.assertFalse(raised
, "An exception was raised")
545 @asynctest.fail_on(active_handles
=True)
546 async def test_other_exceptions(
548 mock_create_certificate
,
550 self
.kubectl
.clients
[
552 ].delete_namespaced_custom_object
.side_effect
= Exception()
553 with self
.assertRaises(Exception):
554 await self
.kubectl
.delete_certificate(
555 namespace
=self
.namespace
,
556 object_name
=self
.object_name
,