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