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