Pin chardet to 3.0.4
[osm/N2VC.git] / n2vc / tests / unit / test_k8s_juju_conn.py
1 # Copyright 2020 Canonical Ltd.
2 #
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
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
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.
14
15
16 import asyncio
17 import logging
18 import asynctest
19 from unittest.mock import Mock
20 from n2vc.k8s_juju_conn import K8sJujuConnector, RBAC_LABEL_KEY_NAME
21 from osm_common import fslocal
22 from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock
23 from n2vc.exceptions import (
24 MethodNotImplemented,
25 K8sException,
26 N2VCBadArgumentsException,
27 )
28
29
30 class K8sJujuConnTestCase(asynctest.TestCase):
31 @asynctest.mock.patch("juju.controller.Controller.update_endpoints")
32 @asynctest.mock.patch("juju.client.connector.Connector.connect")
33 @asynctest.mock.patch("juju.controller.Controller.connection")
34 @asynctest.mock.patch("n2vc.k8s_juju_conn.base64_to_cacert")
35 @asynctest.mock.patch("n2vc.k8s_juju_conn.Libjuju")
36 def setUp(
37 self,
38 mock_libjuju=None,
39 mock_base64_to_cacert=None,
40 mock_connection=None,
41 mock_connect=None,
42 mock_update_endpoints=None,
43 ):
44 self.loop = asyncio.get_event_loop()
45 mock_libjuju.return_value = AsyncMock()
46 db = Mock()
47 vca_config = {
48 "secret": "secret",
49 "api_proxy": "api_proxy",
50 "cloud": "cloud",
51 "k8s_cloud": "k8s_cloud",
52 "user": "user",
53 "host": "1.1.1.1",
54 "port": 17070,
55 "ca_cert": "cacert",
56 }
57
58 logging.disable(logging.CRITICAL)
59
60 self.k8s_juju_conn = K8sJujuConnector(
61 fs=fslocal.FsLocal(),
62 db=db,
63 log=None,
64 loop=self.loop,
65 vca_config=vca_config,
66 on_update_db=None,
67 )
68
69
70 class K8sJujuConnInitSuccessTestCase(asynctest.TestCase):
71 def setUp(
72 self,
73 ):
74 logging.disable(logging.CRITICAL)
75
76 @asynctest.mock.patch("juju.controller.Controller.update_endpoints")
77 @asynctest.mock.patch("juju.client.connector.Connector.connect")
78 @asynctest.mock.patch("juju.controller.Controller.connection")
79 @asynctest.mock.patch("n2vc.k8s_juju_conn.base64_to_cacert")
80 @asynctest.mock.patch("n2vc.libjuju.Libjuju.__init__")
81 def test_success(
82 self,
83 mock_libjuju=None,
84 mock_base64_to_cacert=None,
85 mock_connection=None,
86 mock_connect=None,
87 mock_update_endpoints=None,
88 ):
89 mock_libjuju.return_value = None
90 loop = asyncio.get_event_loop()
91 log = logging.getLogger()
92 db = Mock()
93 vca_config = {
94 "secret": "secret",
95 "cloud": "cloud",
96 "k8s_cloud": "k8s_cloud",
97 "user": "user",
98 "host": "1.1.1.1",
99 "port": 17070,
100 "ca_cert": "cacert",
101 }
102 K8sJujuConnector(
103 fs=fslocal.FsLocal(),
104 db=db,
105 log=log,
106 loop=self.loop,
107 vca_config=vca_config,
108 on_update_db=None,
109 )
110
111 mock_libjuju.assert_called_once_with(
112 endpoint="1.1.1.1:17070",
113 api_proxy=None, # Not needed for k8s charms
114 enable_os_upgrade=True,
115 apt_mirror=None,
116 username="user",
117 password="secret",
118 cacert=mock_base64_to_cacert.return_value,
119 loop=loop,
120 log=log,
121 db=db,
122 )
123
124
125 class K8sJujuConnectorInitFailureTestCase(asynctest.TestCase):
126 def setUp(
127 self,
128 ):
129 self.loop = asyncio.get_event_loop()
130 logging.disable(logging.CRITICAL)
131 self.vca_config = {
132 "secret": "secret",
133 "api_proxy": "api_proxy",
134 "cloud": "cloud",
135 "k8s_cloud": "k8s_cloud",
136 "user": "user",
137 "host": "1.1.1.1",
138 "port": 17070,
139 "ca_cert": "cacert",
140 }
141
142 def test_missing_vca_config_host(self):
143 db = Mock()
144 self.vca_config.pop("host")
145 with self.assertRaises(N2VCBadArgumentsException):
146 self.k8s_juju_conn = K8sJujuConnector(
147 fs=fslocal.FsLocal(),
148 db=db,
149 log=None,
150 loop=self.loop,
151 vca_config=self.vca_config,
152 on_update_db=None,
153 )
154
155 def test_missing_vca_config_user(self):
156 db = Mock()
157 self.vca_config.pop("user")
158 with self.assertRaises(N2VCBadArgumentsException):
159 self.k8s_juju_conn = K8sJujuConnector(
160 fs=fslocal.FsLocal(),
161 db=db,
162 log=None,
163 loop=self.loop,
164 vca_config=self.vca_config,
165 on_update_db=None,
166 )
167
168 def test_missing_vca_config_secret(self):
169 db = Mock()
170 self.vca_config.pop("secret")
171 with self.assertRaises(N2VCBadArgumentsException):
172 self.k8s_juju_conn = K8sJujuConnector(
173 fs=fslocal.FsLocal(),
174 db=db,
175 log=None,
176 loop=self.loop,
177 vca_config=self.vca_config,
178 on_update_db=None,
179 )
180
181 def test_missing_vca_config_ca_cert(self):
182 db = Mock()
183 self.vca_config.pop("ca_cert")
184 with self.assertRaises(N2VCBadArgumentsException):
185 self.k8s_juju_conn = K8sJujuConnector(
186 fs=fslocal.FsLocal(),
187 db=db,
188 log=None,
189 loop=self.loop,
190 vca_config=self.vca_config,
191 on_update_db=None,
192 )
193
194
195 @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
196 class InitEnvTest(K8sJujuConnTestCase):
197 def setUp(self):
198 super(InitEnvTest, self).setUp()
199 self.k8s_juju_conn.libjuju.add_k8s = AsyncMock()
200 self.k8s_juju_conn._create_cluster_role = Mock()
201 self.k8s_juju_conn._create_service_account = Mock()
202 self.k8s_juju_conn._create_cluster_role_binding = Mock()
203 self.k8s_juju_conn._delete_cluster_role = Mock()
204 self.k8s_juju_conn._delete_service_account = Mock()
205 self.k8s_juju_conn._delete_cluster_role_binding = Mock()
206 self.k8s_juju_conn._get_secret_data = AsyncMock()
207 self.k8s_juju_conn._get_secret_data.return_value = ("token", "cacert")
208
209 def test_with_cluster_uuid(
210 self,
211 mock_get_default_storage_class,
212 ):
213 reuse_cluster_uuid = "uuid"
214 uuid, created = self.loop.run_until_complete(
215 self.k8s_juju_conn.init_env(
216 k8s_creds=kubeconfig, reuse_cluster_uuid=reuse_cluster_uuid
217 )
218 )
219
220 self.assertTrue(created)
221 self.assertEqual(uuid, reuse_cluster_uuid)
222 mock_get_default_storage_class.assert_called_once()
223 self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
224
225 def test_with_no_cluster_uuid(
226 self,
227 mock_get_default_storage_class,
228 ):
229 uuid, created = self.loop.run_until_complete(
230 self.k8s_juju_conn.init_env(k8s_creds=kubeconfig)
231 )
232
233 self.assertTrue(created)
234 self.assertTrue(isinstance(uuid, str))
235 mock_get_default_storage_class.assert_called_once()
236 self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
237
238 def test_init_env_exception(
239 self,
240 mock_get_default_storage_class,
241 ):
242 self.k8s_juju_conn.libjuju.add_k8s.side_effect = Exception()
243 created = None
244 uuid = None
245 with self.assertRaises(Exception):
246 uuid, created = self.loop.run_until_complete(
247 self.k8s_juju_conn.init_env(k8s_creds=kubeconfig)
248 )
249
250 self.assertIsNone(created)
251 self.assertIsNone(uuid)
252 mock_get_default_storage_class.assert_called_once()
253 self.k8s_juju_conn.libjuju.add_k8s.assert_called_once()
254
255
256 class NotImplementedTest(K8sJujuConnTestCase):
257 def setUp(self):
258 super(NotImplementedTest, self).setUp()
259
260 def test_repo_add(self):
261 with self.assertRaises(MethodNotImplemented):
262 self.loop.run_until_complete(self.k8s_juju_conn.repo_add("", ""))
263
264 def test_repo_list(self):
265 with self.assertRaises(MethodNotImplemented):
266 self.loop.run_until_complete(self.k8s_juju_conn.repo_list())
267
268 def test_repo_remove(self):
269 with self.assertRaises(MethodNotImplemented):
270 self.loop.run_until_complete(self.k8s_juju_conn.repo_remove(""))
271
272 def test_synchronize_repos(self):
273 self.assertIsNone(
274 self.loop.run_until_complete(self.k8s_juju_conn.synchronize_repos("", ""))
275 )
276
277 def test_upgrade(self):
278 with self.assertRaises(MethodNotImplemented):
279 self.loop.run_until_complete(self.k8s_juju_conn.upgrade("", ""))
280
281 def test_rollback(self):
282 with self.assertRaises(MethodNotImplemented):
283 self.loop.run_until_complete(self.k8s_juju_conn.rollback("", ""))
284
285 def test_get_namespace(self):
286 self.assertIsNone(self.k8s_juju_conn.get_namespace(""))
287
288 def test_instances_list(self):
289 res = self.loop.run_until_complete(self.k8s_juju_conn.instances_list(""))
290 self.assertEqual(res, [])
291
292
293 class ResetTest(K8sJujuConnTestCase):
294 def setUp(self):
295 super(ResetTest, self).setUp()
296 self.k8s_juju_conn.libjuju.remove_cloud = AsyncMock()
297 self.k8s_juju_conn.libjuju.get_cloud_credentials = AsyncMock()
298 cloud_creds = Mock()
299 cloud_creds.result = {"attrs": {RBAC_LABEL_KEY_NAME: "asd"}}
300 self.k8s_juju_conn.libjuju.get_cloud_credentials.return_value = [cloud_creds]
301 self.k8s_juju_conn._delete_cluster_role_binding = Mock()
302 self.k8s_juju_conn._delete_service_account = Mock()
303 self.k8s_juju_conn._delete_cluster_role = Mock()
304 self.k8s_juju_conn.get_credentials = Mock()
305 self.k8s_juju_conn.get_credentials.return_value = kubeconfig
306
307 def test_success(self):
308 removed = self.loop.run_until_complete(self.k8s_juju_conn.reset("uuid"))
309 self.assertTrue(removed)
310 self.k8s_juju_conn.libjuju.remove_cloud.assert_called_once()
311
312 def test_exception(self):
313 removed = None
314 self.k8s_juju_conn.libjuju.remove_cloud.side_effect = Exception()
315 with self.assertRaises(Exception):
316 removed = self.loop.run_until_complete(self.k8s_juju_conn.reset("uuid"))
317 self.assertIsNone(removed)
318 self.k8s_juju_conn.libjuju.remove_cloud.assert_called_once()
319
320
321 @asynctest.mock.patch("os.chdir")
322 class InstallTest(K8sJujuConnTestCase):
323 def setUp(self):
324 super(InstallTest, self).setUp()
325 self.db_dict = {"filter": {"_id": "id"}}
326 self.local_bundle = "bundle"
327 self.cs_bundle = "cs:bundle"
328 self.http_bundle = "https://example.com/bundle.yaml"
329 self.kdu_name = "kdu_name"
330 self.cluster_uuid = "cluster"
331 self.kdu_instance = "{}-{}".format(self.kdu_name, "id")
332 self.k8s_juju_conn.libjuju.add_model = AsyncMock()
333 self.k8s_juju_conn.libjuju.deploy = AsyncMock()
334
335 def test_success_local(self, mock_chdir):
336 self.loop.run_until_complete(
337 self.k8s_juju_conn.install(
338 self.cluster_uuid,
339 self.local_bundle,
340 self.kdu_instance,
341 atomic=True,
342 kdu_name=self.kdu_name,
343 db_dict=self.db_dict,
344 timeout=1800,
345 )
346 )
347 self.assertEqual(mock_chdir.call_count, 2)
348 self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
349 model_name=self.kdu_instance,
350 cloud_name=self.cluster_uuid,
351 credential_name="cred-{}".format(self.cluster_uuid),
352 )
353 self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
354 "local:{}".format(self.local_bundle),
355 model_name=self.kdu_instance,
356 wait=True,
357 timeout=1800,
358 )
359
360 def test_success_cs(self, mock_chdir):
361 self.loop.run_until_complete(
362 self.k8s_juju_conn.install(
363 self.cluster_uuid,
364 self.cs_bundle,
365 self.kdu_instance,
366 atomic=True,
367 kdu_name=self.kdu_name,
368 db_dict=self.db_dict,
369 timeout=1800,
370 )
371 )
372 self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
373 model_name=self.kdu_instance,
374 cloud_name=self.cluster_uuid,
375 credential_name="cred-{}".format(self.cluster_uuid),
376 )
377 self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
378 self.cs_bundle,
379 model_name=self.kdu_instance,
380 wait=True,
381 timeout=1800,
382 )
383
384 def test_success_http(self, mock_chdir):
385 self.loop.run_until_complete(
386 self.k8s_juju_conn.install(
387 self.cluster_uuid,
388 self.http_bundle,
389 self.kdu_instance,
390 atomic=True,
391 kdu_name=self.kdu_name,
392 db_dict=self.db_dict,
393 timeout=1800,
394 )
395 )
396 self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
397 model_name=self.kdu_instance,
398 cloud_name=self.cluster_uuid,
399 credential_name="cred-{}".format(self.cluster_uuid),
400 )
401 self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
402 self.http_bundle,
403 model_name=self.kdu_instance,
404 wait=True,
405 timeout=1800,
406 )
407
408 def test_success_not_kdu_name(self, mock_chdir):
409 self.loop.run_until_complete(
410 self.k8s_juju_conn.install(
411 self.cluster_uuid,
412 self.cs_bundle,
413 self.kdu_instance,
414 atomic=True,
415 db_dict=self.db_dict,
416 timeout=1800,
417 )
418 )
419 self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
420 model_name=self.kdu_instance,
421 cloud_name=self.cluster_uuid,
422 credential_name="cred-{}".format(self.cluster_uuid),
423 )
424 self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
425 self.cs_bundle,
426 model_name=self.kdu_instance,
427 wait=True,
428 timeout=1800,
429 )
430
431 def test_missing_db_dict(self, mock_chdir):
432 kdu_instance = None
433 with self.assertRaises(K8sException):
434 self.loop.run_until_complete(
435 self.k8s_juju_conn.install(
436 self.cluster_uuid,
437 self.cs_bundle,
438 self.kdu_instance,
439 atomic=True,
440 kdu_name=self.kdu_name,
441 timeout=1800,
442 )
443 )
444 self.assertIsNone(kdu_instance)
445 self.k8s_juju_conn.libjuju.add_model.assert_not_called()
446 self.k8s_juju_conn.libjuju.deploy.assert_not_called()
447
448 @asynctest.mock.patch("os.getcwd")
449 def test_getcwd_exception(self, mock_getcwd, mock_chdir):
450 mock_getcwd.side_effect = FileNotFoundError()
451 self.loop.run_until_complete(
452 self.k8s_juju_conn.install(
453 self.cluster_uuid,
454 self.cs_bundle,
455 self.kdu_instance,
456 atomic=True,
457 kdu_name=self.kdu_name,
458 db_dict=self.db_dict,
459 timeout=1800,
460 )
461 )
462 self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
463 model_name=self.kdu_instance,
464 cloud_name=self.cluster_uuid,
465 credential_name="cred-{}".format(self.cluster_uuid),
466 )
467 self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
468 self.cs_bundle,
469 model_name=self.kdu_instance,
470 wait=True,
471 timeout=1800,
472 )
473
474 def test_missing_bundle(self, mock_chdir):
475 with self.assertRaises(K8sException):
476 self.loop.run_until_complete(
477 self.k8s_juju_conn.install(
478 self.cluster_uuid,
479 "",
480 self.kdu_instance,
481 atomic=True,
482 kdu_name=self.kdu_name,
483 timeout=1800,
484 db_dict=self.db_dict,
485 )
486 )
487 self.k8s_juju_conn.libjuju.add_model.assert_not_called()
488 self.k8s_juju_conn.libjuju.deploy.assert_not_called()
489
490 def test_missing_exception(self, mock_chdir):
491 self.k8s_juju_conn.libjuju.deploy.side_effect = Exception()
492 with self.assertRaises(Exception):
493 self.loop.run_until_complete(
494 self.k8s_juju_conn.install(
495 self.cluster_uuid,
496 self.local_bundle,
497 self.kdu_instance,
498 atomic=True,
499 kdu_name=self.kdu_name,
500 db_dict=self.db_dict,
501 timeout=1800,
502 )
503 )
504 self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
505 model_name=self.kdu_instance,
506 cloud_name=self.cluster_uuid,
507 credential_name="cred-{}".format(self.cluster_uuid),
508 )
509 self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
510 "local:{}".format(self.local_bundle),
511 model_name=self.kdu_instance,
512 wait=True,
513 timeout=1800,
514 )
515
516
517 class UninstallTest(K8sJujuConnTestCase):
518 def setUp(self):
519 super(UninstallTest, self).setUp()
520 self.k8s_juju_conn.libjuju.destroy_model = AsyncMock()
521
522 def test_success(self):
523 destroyed = self.loop.run_until_complete(
524 self.k8s_juju_conn.uninstall("cluster_uuid", "model_name")
525 )
526 self.assertTrue(destroyed)
527 self.k8s_juju_conn.libjuju.destroy_model.assert_called_once()
528
529 def test_exception(self):
530 destroyed = None
531 self.k8s_juju_conn.libjuju.destroy_model.side_effect = Exception()
532 with self.assertRaises(Exception):
533 destroyed = self.loop.run_until_complete(
534 self.k8s_juju_conn.uninstall("cluster_uuid", "model_name")
535 )
536 self.assertIsNone(destroyed)
537 self.k8s_juju_conn.libjuju.destroy_model.assert_called_once()
538
539
540 class ExecPrimitivesTest(K8sJujuConnTestCase):
541 def setUp(self):
542 super(ExecPrimitivesTest, self).setUp()
543 self.action_name = "touch"
544 self.application_name = "myapp"
545 self.model_name = "model"
546 self.k8s_juju_conn.libjuju.get_actions = AsyncMock()
547 self.k8s_juju_conn.libjuju.execute_action = AsyncMock()
548
549 def test_success(self):
550 params = {"application-name": self.application_name}
551 self.k8s_juju_conn.libjuju.get_actions.return_value = [self.action_name]
552 self.k8s_juju_conn.libjuju.execute_action.return_value = (
553 "success",
554 "completed",
555 )
556
557 output = self.loop.run_until_complete(
558 self.k8s_juju_conn.exec_primitive(
559 "cluster", self.model_name, self.action_name, params=params
560 )
561 )
562
563 self.assertEqual(output, "success")
564 self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
565 self.application_name, self.model_name
566 )
567 self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
568 self.application_name, self.model_name, self.action_name, **params
569 )
570
571 def test_exception(self):
572 params = {"application-name": self.application_name}
573 self.k8s_juju_conn.libjuju.get_actions.return_value = [self.action_name]
574 self.k8s_juju_conn.libjuju.execute_action.side_effect = Exception()
575 output = None
576
577 with self.assertRaises(Exception):
578 output = self.loop.run_until_complete(
579 self.k8s_juju_conn.exec_primitive(
580 "cluster", self.model_name, self.action_name, params=params
581 )
582 )
583
584 self.assertIsNone(output)
585 self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
586 self.application_name, self.model_name
587 )
588 self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
589 self.application_name, self.model_name, self.action_name, **params
590 )
591
592 def test_missing_application_name_in_params(self):
593 params = {}
594 output = None
595
596 with self.assertRaises(K8sException):
597 output = self.loop.run_until_complete(
598 self.k8s_juju_conn.exec_primitive(
599 "cluster", self.model_name, self.action_name, params=params
600 )
601 )
602
603 self.assertIsNone(output)
604 self.k8s_juju_conn.libjuju.get_actions.assert_not_called()
605 self.k8s_juju_conn.libjuju.execute_action.assert_not_called()
606
607 def test_missing_params(self):
608 output = None
609 with self.assertRaises(K8sException):
610 output = self.loop.run_until_complete(
611 self.k8s_juju_conn.exec_primitive(
612 "cluster", self.model_name, self.action_name
613 )
614 )
615
616 self.assertIsNone(output)
617 self.k8s_juju_conn.libjuju.get_actions.assert_not_called()
618 self.k8s_juju_conn.libjuju.execute_action.assert_not_called()
619
620 def test_missing_action(self):
621 output = None
622 params = {"application-name": self.application_name}
623 self.k8s_juju_conn.libjuju.get_actions.return_value = [self.action_name]
624 self.k8s_juju_conn.libjuju.execute_action.return_value = (
625 "success",
626 "completed",
627 )
628 with self.assertRaises(K8sException):
629 output = self.loop.run_until_complete(
630 self.k8s_juju_conn.exec_primitive(
631 "cluster", self.model_name, "non-existing-action", params=params
632 )
633 )
634
635 self.assertIsNone(output)
636 self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
637 self.application_name, self.model_name
638 )
639 self.k8s_juju_conn.libjuju.execute_action.assert_not_called()
640
641 def test_missing_not_completed(self):
642 output = None
643 params = {"application-name": self.application_name}
644 self.k8s_juju_conn.libjuju.get_actions.return_value = [self.action_name]
645 self.k8s_juju_conn.libjuju.execute_action.return_value = (None, "failed")
646 with self.assertRaises(K8sException):
647 output = self.loop.run_until_complete(
648 self.k8s_juju_conn.exec_primitive(
649 "cluster", self.model_name, self.action_name, params=params
650 )
651 )
652
653 self.assertIsNone(output)
654 self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
655 self.application_name, self.model_name
656 )
657 self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
658 self.application_name, self.model_name, self.action_name, **params
659 )
660
661
662 class InspectKduTest(K8sJujuConnTestCase):
663 def setUp(self):
664 super(InspectKduTest, self).setUp()
665
666 @asynctest.mock.patch("builtins.open")
667 @asynctest.mock.patch("os.path.exists")
668 def test_existing_file(self, mock_exists, mock_open):
669 mock_exists.return_value = True
670 content = """{
671 'description': 'test bundle',
672 'bundle': 'kubernetes',
673 'applications': {'app':{ }, 'app2': { }}
674 }"""
675 mock_open.return_value = FakeFileWrapper(content=content)
676 kdu = self.loop.run_until_complete(self.k8s_juju_conn.inspect_kdu("model"))
677 self.assertEqual(kdu, {"app": {}, "app2": {}})
678 mock_exists.assert_called_once()
679 mock_open.assert_called_once()
680
681 @asynctest.mock.patch("builtins.open")
682 @asynctest.mock.patch("os.path.exists")
683 def test_not_existing_file(self, mock_exists, mock_open):
684 kdu = None
685 mock_exists.return_value = False
686 with self.assertRaises(K8sException):
687 kdu = self.loop.run_until_complete(self.k8s_juju_conn.inspect_kdu("model"))
688 self.assertEqual(kdu, None)
689 mock_exists.assert_called_once_with("model")
690 mock_open.assert_not_called()
691
692
693 class HelpKduTest(K8sJujuConnTestCase):
694 def setUp(self):
695 super(HelpKduTest, self).setUp()
696
697 @asynctest.mock.patch("builtins.open")
698 @asynctest.mock.patch("os.listdir")
699 def test_existing_file(self, mock_listdir, mock_open):
700 content = "Readme file content"
701 mock_open.return_value = FakeFileWrapper(content=content)
702 for file in ["README.md", "README.txt", "README"]:
703 mock_listdir.return_value = [file]
704 help = self.loop.run_until_complete(
705 self.k8s_juju_conn.help_kdu("kdu_instance")
706 )
707 self.assertEqual(help, content)
708
709 self.assertEqual(mock_listdir.call_count, 3)
710 self.assertEqual(mock_open.call_count, 3)
711
712 @asynctest.mock.patch("builtins.open")
713 @asynctest.mock.patch("os.listdir")
714 def test_not_existing_file(self, mock_listdir, mock_open):
715 for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
716 mock_listdir.return_value = [file]
717 help = self.loop.run_until_complete(
718 self.k8s_juju_conn.help_kdu("kdu_instance")
719 )
720 self.assertEqual(help, None)
721
722 self.assertEqual(mock_listdir.call_count, 3)
723 self.assertEqual(mock_open.call_count, 0)
724
725
726 class StatusKduTest(K8sJujuConnTestCase):
727 def setUp(self):
728 super(StatusKduTest, self).setUp()
729 self.k8s_juju_conn.libjuju.get_model_status = AsyncMock()
730
731 def test_success(self):
732 applications = {"app": {"status": {"status": "active"}}}
733 model = FakeModel(applications=applications)
734 self.k8s_juju_conn.libjuju.get_model_status.return_value = model
735 status = self.loop.run_until_complete(
736 self.k8s_juju_conn.status_kdu("cluster", "kdu_instance")
737 )
738 self.assertEqual(status, {"app": {"status": "active"}})
739 self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()
740
741 def test_exception(self):
742 self.k8s_juju_conn.libjuju.get_model_status.side_effect = Exception()
743 status = None
744 with self.assertRaises(Exception):
745 status = self.loop.run_until_complete(
746 self.k8s_juju_conn.status_kdu("cluster", "kdu_instance")
747 )
748 self.assertIsNone(status)
749 self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()
750
751
752 class GetServicesTest(K8sJujuConnTestCase):
753 def setUp(self):
754 super(GetServicesTest, self).setUp()
755
756 @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_services")
757 @asynctest.mock.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
758 def test_success(self, mock_get_credentials, mock_get_services):
759 mock_get_credentials.return_value = kubeconfig
760 self.loop.run_until_complete(self.k8s_juju_conn.get_services("", "", ""))
761 mock_get_credentials.assert_called_once()
762 mock_get_services.assert_called_once()
763
764
765 class GetServiceTest(K8sJujuConnTestCase):
766 def setUp(self):
767 super(GetServiceTest, self).setUp()
768
769 @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_services")
770 @asynctest.mock.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
771 def test_success(self, mock_get_credentials, mock_get_services):
772 mock_get_credentials.return_value = kubeconfig
773 self.loop.run_until_complete(self.k8s_juju_conn.get_service("", "", ""))
774 mock_get_credentials.assert_called_once()
775 mock_get_services.assert_called_once()
776
777
778 class GetCredentialsTest(K8sJujuConnTestCase):
779 def setUp(self):
780 super(GetCredentialsTest, self).setUp()
781
782 @asynctest.mock.patch("yaml.safe_dump")
783 def test_success(self, mock_safe_dump):
784 self.k8s_juju_conn.db.get_one.return_value = {
785 "_id": "id",
786 "credentials": "credentials",
787 "schema_version": "2",
788 }
789 self.k8s_juju_conn.get_credentials("cluster_uuid")
790 self.k8s_juju_conn.db.get_one.assert_called_once()
791 self.k8s_juju_conn.db.encrypt_decrypt_fields.assert_called_once()
792 mock_safe_dump.assert_called_once()