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