1 # Copyright 2020 Canonical Ltd.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
19 from n2vc
.k8s_juju_conn
import K8sJujuConnector
20 from osm_common
import fslocal
21 from .utils
import kubeconfig
, FakeModel
, FakeFileWrapper
22 from n2vc
.exceptions
import (
25 N2VCBadArgumentsException
,
27 from unittest
.mock
import Mock
28 from .utils
import AsyncMock
31 class K8sJujuConnTestCase(asynctest
.TestCase
):
32 @asynctest.mock
.patch("juju.controller.Controller.update_endpoints")
33 @asynctest.mock
.patch("juju.client.connector.Connector.connect")
34 @asynctest.mock
.patch("juju.controller.Controller.connection")
35 @asynctest.mock
.patch("n2vc.k8s_juju_conn.base64_to_cacert")
36 @asynctest.mock
.patch("n2vc.k8s_juju_conn.Libjuju")
40 mock_base64_to_cacert
=None,
43 mock_update_endpoints
=None,
45 self
.loop
= asyncio
.get_event_loop()
46 mock_libjuju
.return_value
= AsyncMock()
50 "api_proxy": "api_proxy",
52 "k8s_cloud": "k8s_cloud",
59 logging
.disable(logging
.CRITICAL
)
61 self
.k8s_juju_conn
= K8sJujuConnector(
66 vca_config
=vca_config
,
71 class K8sJujuConnInitSuccessTestCase(asynctest
.TestCase
):
75 logging
.disable(logging
.CRITICAL
)
77 @asynctest.mock
.patch("juju.controller.Controller.update_endpoints")
78 @asynctest.mock
.patch("juju.client.connector.Connector.connect")
79 @asynctest.mock
.patch("juju.controller.Controller.connection")
80 @asynctest.mock
.patch("n2vc.k8s_juju_conn.base64_to_cacert")
81 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.__init__")
85 mock_base64_to_cacert
=None,
88 mock_update_endpoints
=None,
90 mock_libjuju
.return_value
= None
91 loop
= asyncio
.get_event_loop()
92 log
= logging
.getLogger()
97 "k8s_cloud": "k8s_cloud",
104 fs
=fslocal
.FsLocal(),
108 vca_config
=vca_config
,
112 mock_libjuju
.assert_called_once_with(
113 endpoint
="1.1.1.1:17070",
114 api_proxy
=None, # Not needed for k8s charms
115 enable_os_upgrade
=True,
119 cacert
=mock_base64_to_cacert
.return_value
,
126 class K8sJujuConnectorInitFailureTestCase(asynctest
.TestCase
):
130 self
.loop
= asyncio
.get_event_loop()
131 logging
.disable(logging
.CRITICAL
)
134 "api_proxy": "api_proxy",
136 "k8s_cloud": "k8s_cloud",
143 def test_missing_vca_config_host(self
):
145 self
.vca_config
.pop("host")
146 with self
.assertRaises(N2VCBadArgumentsException
):
147 self
.k8s_juju_conn
= K8sJujuConnector(
148 fs
=fslocal
.FsLocal(),
152 vca_config
=self
.vca_config
,
156 def test_missing_vca_config_user(self
):
158 self
.vca_config
.pop("user")
159 with self
.assertRaises(N2VCBadArgumentsException
):
160 self
.k8s_juju_conn
= K8sJujuConnector(
161 fs
=fslocal
.FsLocal(),
165 vca_config
=self
.vca_config
,
169 def test_missing_vca_config_secret(self
):
171 self
.vca_config
.pop("secret")
172 with self
.assertRaises(N2VCBadArgumentsException
):
173 self
.k8s_juju_conn
= K8sJujuConnector(
174 fs
=fslocal
.FsLocal(),
178 vca_config
=self
.vca_config
,
182 def test_missing_vca_config_ca_cert(self
):
184 self
.vca_config
.pop("ca_cert")
185 with self
.assertRaises(N2VCBadArgumentsException
):
186 self
.k8s_juju_conn
= K8sJujuConnector(
187 fs
=fslocal
.FsLocal(),
191 vca_config
=self
.vca_config
,
196 class InitEnvTest(K8sJujuConnTestCase
):
198 super(InitEnvTest
, self
).setUp()
199 self
.k8s_juju_conn
.libjuju
.add_k8s
= AsyncMock()
201 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
202 def test_with_cluster_uuid(
204 mock_get_default_storage_class
,
206 reuse_cluster_uuid
= "uuid"
207 uuid
, created
= self
.loop
.run_until_complete(
208 self
.k8s_juju_conn
.init_env(
209 k8s_creds
=kubeconfig
, reuse_cluster_uuid
=reuse_cluster_uuid
213 self
.assertTrue(created
)
214 self
.assertEqual(uuid
, reuse_cluster_uuid
)
215 mock_get_default_storage_class
.assert_called_once()
216 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
218 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
219 def test_with_no_cluster_uuid(self
, mock_get_default_storage_class
):
220 uuid
, created
= self
.loop
.run_until_complete(
221 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
224 self
.assertTrue(created
)
225 self
.assertTrue(isinstance(uuid
, str))
226 mock_get_default_storage_class
.assert_called_once()
227 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
229 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
230 def test_init_env_exception(self
, mock_get_default_storage_class
):
231 self
.k8s_juju_conn
.libjuju
.add_k8s
.side_effect
= Exception()
234 with self
.assertRaises(Exception):
235 uuid
, created
= self
.loop
.run_until_complete(
236 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
239 self
.assertIsNone(created
)
240 self
.assertIsNone(uuid
)
241 mock_get_default_storage_class
.assert_called_once()
242 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
245 class NotImplementedTest(K8sJujuConnTestCase
):
247 super(NotImplementedTest
, self
).setUp()
249 def test_repo_add(self
):
250 with self
.assertRaises(MethodNotImplemented
):
251 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_add("", ""))
253 def test_repo_list(self
):
254 with self
.assertRaises(MethodNotImplemented
):
255 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_list())
257 def test_repo_remove(self
):
258 with self
.assertRaises(MethodNotImplemented
):
259 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_remove(""))
261 def test_synchronize_repos(self
):
263 self
.loop
.run_until_complete(self
.k8s_juju_conn
.synchronize_repos("", ""))
266 def test_upgrade(self
):
267 with self
.assertRaises(MethodNotImplemented
):
268 self
.loop
.run_until_complete(self
.k8s_juju_conn
.upgrade("", ""))
270 def test_rollback(self
):
271 with self
.assertRaises(MethodNotImplemented
):
272 self
.loop
.run_until_complete(self
.k8s_juju_conn
.rollback("", ""))
274 def test_get_namespace(self
):
275 self
.assertIsNone(self
.k8s_juju_conn
.get_namespace(""))
277 def test_instances_list(self
):
278 res
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.instances_list(""))
279 self
.assertEqual(res
, [])
282 class ResetTest(K8sJujuConnTestCase
):
284 super(ResetTest
, self
).setUp()
285 self
.k8s_juju_conn
.libjuju
.remove_cloud
= AsyncMock()
287 def test_success(self
):
288 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
289 self
.assertTrue(removed
)
290 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
292 def test_exception(self
):
294 self
.k8s_juju_conn
.libjuju
.remove_cloud
.side_effect
= Exception()
295 with self
.assertRaises(Exception):
296 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
297 self
.assertIsNone(removed
)
298 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
301 @asynctest.mock
.patch("os.chdir")
302 class InstallTest(K8sJujuConnTestCase
):
304 super(InstallTest
, self
).setUp()
305 self
.db_dict
= {"filter": {"_id": "id"}}
306 self
.local_bundle
= "bundle"
307 self
.cs_bundle
= "cs:bundle"
308 self
.http_bundle
= "https://example.com/bundle.yaml"
309 self
.kdu_name
= "kdu_name"
310 self
.cluster_uuid
= "cluster"
311 self
.k8s_juju_conn
.libjuju
.add_model
= AsyncMock()
312 self
.k8s_juju_conn
.libjuju
.deploy
= AsyncMock()
314 def test_success_local(self
, mock_chdir
):
315 expected_kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
316 kdu_instance
= self
.loop
.run_until_complete(
317 self
.k8s_juju_conn
.install(
321 kdu_name
=self
.kdu_name
,
322 db_dict
=self
.db_dict
,
326 self
.assertEqual(kdu_instance
, expected_kdu_instance
)
327 self
.assertEqual(mock_chdir
.call_count
, 2)
328 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
329 model_name
=expected_kdu_instance
,
330 cloud_name
=self
.cluster_uuid
,
331 credential_name
="cred-{}".format(self
.cluster_uuid
),
333 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
334 "local:{}".format(self
.local_bundle
),
335 model_name
=expected_kdu_instance
,
340 def test_success_cs(self
, mock_chdir
):
341 expected_kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
342 kdu_instance
= self
.loop
.run_until_complete(
343 self
.k8s_juju_conn
.install(
347 kdu_name
=self
.kdu_name
,
348 db_dict
=self
.db_dict
,
352 self
.assertEqual(kdu_instance
, expected_kdu_instance
)
353 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
354 model_name
=expected_kdu_instance
,
355 cloud_name
=self
.cluster_uuid
,
356 credential_name
="cred-{}".format(self
.cluster_uuid
),
358 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
360 model_name
=expected_kdu_instance
,
365 def test_success_http(self
, mock_chdir
):
366 expected_kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
367 kdu_instance
= self
.loop
.run_until_complete(
368 self
.k8s_juju_conn
.install(
372 kdu_name
=self
.kdu_name
,
373 db_dict
=self
.db_dict
,
377 self
.assertEqual(kdu_instance
, expected_kdu_instance
)
378 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
379 model_name
=expected_kdu_instance
,
380 cloud_name
=self
.cluster_uuid
,
381 credential_name
="cred-{}".format(self
.cluster_uuid
),
383 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
385 model_name
=expected_kdu_instance
,
390 def test_success_not_kdu_name(self
, mock_chdir
):
391 expected_kdu_instance
= "id"
392 kdu_instance
= self
.loop
.run_until_complete(
393 self
.k8s_juju_conn
.install(
397 db_dict
=self
.db_dict
,
401 self
.assertEqual(kdu_instance
, expected_kdu_instance
)
402 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
403 model_name
=expected_kdu_instance
,
404 cloud_name
=self
.cluster_uuid
,
405 credential_name
="cred-{}".format(self
.cluster_uuid
),
407 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
409 model_name
=expected_kdu_instance
,
414 def test_missing_db_dict(self
, mock_chdir
):
416 with self
.assertRaises(K8sException
):
417 kdu_instance
= self
.loop
.run_until_complete(
418 self
.k8s_juju_conn
.install(
422 kdu_name
=self
.kdu_name
,
426 self
.assertIsNone(kdu_instance
)
427 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
428 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
430 @asynctest.mock
.patch("os.getcwd")
431 def test_getcwd_exception(self
, mock_getcwd
, mock_chdir
):
432 mock_getcwd
.side_effect
= FileNotFoundError()
433 expected_kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
434 kdu_instance
= self
.loop
.run_until_complete(
435 self
.k8s_juju_conn
.install(
439 kdu_name
=self
.kdu_name
,
440 db_dict
=self
.db_dict
,
444 self
.assertEqual(kdu_instance
, expected_kdu_instance
)
445 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
446 model_name
=expected_kdu_instance
,
447 cloud_name
=self
.cluster_uuid
,
448 credential_name
="cred-{}".format(self
.cluster_uuid
),
450 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
452 model_name
=expected_kdu_instance
,
457 def test_missing_bundle(self
, mock_chdir
):
459 with self
.assertRaises(K8sException
):
460 kdu_instance
= self
.loop
.run_until_complete(
461 self
.k8s_juju_conn
.install(
465 kdu_name
=self
.kdu_name
,
467 db_dict
=self
.db_dict
,
470 self
.assertIsNone(kdu_instance
)
471 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
472 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
474 def test_missing_exception(self
, mock_chdir
):
475 expected_kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
477 self
.k8s_juju_conn
.libjuju
.deploy
.side_effect
= Exception()
478 with self
.assertRaises(Exception):
479 kdu_instance
= self
.loop
.run_until_complete(
480 self
.k8s_juju_conn
.install(
484 kdu_name
=self
.kdu_name
,
485 db_dict
=self
.db_dict
,
489 self
.assertIsNone(kdu_instance
)
490 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
491 model_name
=expected_kdu_instance
,
492 cloud_name
=self
.cluster_uuid
,
493 credential_name
="cred-{}".format(self
.cluster_uuid
),
495 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
496 "local:{}".format(self
.local_bundle
),
497 model_name
=expected_kdu_instance
,
503 class UninstallTest(K8sJujuConnTestCase
):
505 super(UninstallTest
, self
).setUp()
506 self
.k8s_juju_conn
.libjuju
.destroy_model
= AsyncMock()
508 def test_success(self
):
509 destroyed
= self
.loop
.run_until_complete(
510 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
512 self
.assertTrue(destroyed
)
513 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
515 def test_exception(self
):
517 self
.k8s_juju_conn
.libjuju
.destroy_model
.side_effect
= Exception()
518 with self
.assertRaises(Exception):
519 destroyed
= self
.loop
.run_until_complete(
520 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
522 self
.assertIsNone(destroyed
)
523 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
526 class ExecPrimitivesTest(K8sJujuConnTestCase
):
528 super(ExecPrimitivesTest
, self
).setUp()
529 self
.action_name
= "touch"
530 self
.application_name
= "myapp"
531 self
.model_name
= "model"
532 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
533 self
.k8s_juju_conn
.libjuju
.execute_action
= AsyncMock()
535 def test_success(self
):
536 params
= {"application-name": self
.application_name
}
537 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
538 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
543 output
= self
.loop
.run_until_complete(
544 self
.k8s_juju_conn
.exec_primitive(
545 "cluster", self
.model_name
, self
.action_name
, params
=params
549 self
.assertEqual(output
, "success")
550 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
551 self
.application_name
, self
.model_name
553 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
554 self
.application_name
, self
.model_name
, self
.action_name
, **params
557 def test_exception(self
):
558 params
= {"application-name": self
.application_name
}
559 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
560 self
.k8s_juju_conn
.libjuju
.execute_action
.side_effect
= Exception()
563 with self
.assertRaises(Exception):
564 output
= self
.loop
.run_until_complete(
565 self
.k8s_juju_conn
.exec_primitive(
566 "cluster", self
.model_name
, self
.action_name
, params
=params
570 self
.assertIsNone(output
)
571 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
572 self
.application_name
, self
.model_name
574 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
575 self
.application_name
, self
.model_name
, self
.action_name
, **params
578 def test_missing_application_name_in_params(self
):
582 with self
.assertRaises(K8sException
):
583 output
= self
.loop
.run_until_complete(
584 self
.k8s_juju_conn
.exec_primitive(
585 "cluster", self
.model_name
, self
.action_name
, params
=params
589 self
.assertIsNone(output
)
590 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
591 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
593 def test_missing_params(self
):
595 with self
.assertRaises(K8sException
):
596 output
= self
.loop
.run_until_complete(
597 self
.k8s_juju_conn
.exec_primitive(
598 "cluster", self
.model_name
, self
.action_name
602 self
.assertIsNone(output
)
603 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
604 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
606 def test_missing_action(self
):
608 params
= {"application-name": self
.application_name
}
609 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
610 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
614 with self
.assertRaises(K8sException
):
615 output
= self
.loop
.run_until_complete(
616 self
.k8s_juju_conn
.exec_primitive(
617 "cluster", self
.model_name
, "non-existing-action", params
=params
621 self
.assertIsNone(output
)
622 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
623 self
.application_name
, self
.model_name
625 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
627 def test_missing_not_completed(self
):
629 params
= {"application-name": self
.application_name
}
630 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
631 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (None, "failed")
632 with self
.assertRaises(K8sException
):
633 output
= self
.loop
.run_until_complete(
634 self
.k8s_juju_conn
.exec_primitive(
635 "cluster", self
.model_name
, self
.action_name
, params
=params
639 self
.assertIsNone(output
)
640 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
641 self
.application_name
, self
.model_name
643 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
644 self
.application_name
, self
.model_name
, self
.action_name
, **params
648 class InspectKduTest(K8sJujuConnTestCase
):
650 super(InspectKduTest
, self
).setUp()
652 @asynctest.mock
.patch("builtins.open")
653 @asynctest.mock
.patch("os.path.exists")
654 def test_existing_file(self
, mock_exists
, mock_open
):
655 mock_exists
.return_value
= True
657 'description': 'test bundle',
658 'bundle': 'kubernetes',
659 'applications': {'app':{ }, 'app2': { }}
661 mock_open
.return_value
= FakeFileWrapper(content
=content
)
662 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
663 self
.assertEqual(kdu
, {"app": {}, "app2": {}})
664 mock_exists
.assert_called_once()
665 mock_open
.assert_called_once()
667 @asynctest.mock
.patch("builtins.open")
668 @asynctest.mock
.patch("os.path.exists")
669 def test_not_existing_file(self
, mock_exists
, mock_open
):
671 mock_exists
.return_value
= False
672 with self
.assertRaises(K8sException
):
673 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
674 self
.assertEqual(kdu
, None)
675 mock_exists
.assert_called_once_with("model")
676 mock_open
.assert_not_called()
679 class HelpKduTest(K8sJujuConnTestCase
):
681 super(HelpKduTest
, self
).setUp()
683 @asynctest.mock
.patch("builtins.open")
684 @asynctest.mock
.patch("os.listdir")
685 def test_existing_file(self
, mock_listdir
, mock_open
):
686 content
= "Readme file content"
687 mock_open
.return_value
= FakeFileWrapper(content
=content
)
688 for file in ["README.md", "README.txt", "README"]:
689 mock_listdir
.return_value
= [file]
690 help = self
.loop
.run_until_complete(
691 self
.k8s_juju_conn
.help_kdu("kdu_instance")
693 self
.assertEqual(help, content
)
695 self
.assertEqual(mock_listdir
.call_count
, 3)
696 self
.assertEqual(mock_open
.call_count
, 3)
698 @asynctest.mock
.patch("builtins.open")
699 @asynctest.mock
.patch("os.listdir")
700 def test_not_existing_file(self
, mock_listdir
, mock_open
):
701 for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
702 mock_listdir
.return_value
= [file]
703 help = self
.loop
.run_until_complete(
704 self
.k8s_juju_conn
.help_kdu("kdu_instance")
706 self
.assertEqual(help, None)
708 self
.assertEqual(mock_listdir
.call_count
, 3)
709 self
.assertEqual(mock_open
.call_count
, 0)
712 class StatusKduTest(K8sJujuConnTestCase
):
714 super(StatusKduTest
, self
).setUp()
715 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
717 def test_success(self
):
718 applications
= {"app": {"status": {"status": "active"}}}
719 model
= FakeModel(applications
=applications
)
720 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
721 status
= self
.loop
.run_until_complete(
722 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
724 self
.assertEqual(status
, {"app": {"status": "active"}})
725 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
727 def test_exception(self
):
728 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
730 with self
.assertRaises(Exception):
731 status
= self
.loop
.run_until_complete(
732 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
734 self
.assertIsNone(status
)
735 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
738 class GetServicesTest(K8sJujuConnTestCase
):
740 super(GetServicesTest
, self
).setUp()
742 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_services")
743 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
744 def test_success(self
, mock_get_credentials
, mock_get_services
):
745 mock_get_credentials
.return_value
= kubeconfig
746 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_services("", "", ""))
747 mock_get_credentials
.assert_called_once()
748 mock_get_services
.assert_called_once()
751 class GetServiceTest(K8sJujuConnTestCase
):
753 super(GetServiceTest
, self
).setUp()
755 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_services")
756 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
757 def test_success(self
, mock_get_credentials
, mock_get_services
):
758 mock_get_credentials
.return_value
= kubeconfig
759 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_service("", "", ""))
760 mock_get_credentials
.assert_called_once()
761 mock_get_services
.assert_called_once()
764 class GetCredentialsTest(K8sJujuConnTestCase
):
766 super(GetCredentialsTest
, self
).setUp()
768 @asynctest.mock
.patch("yaml.safe_dump")
769 def test_success(self
, mock_safe_dump
):
770 self
.k8s_juju_conn
.db
.get_one
.return_value
= {
772 "credentials": "credentials",
773 "schema_version": "2",
775 self
.k8s_juju_conn
.get_credentials("cluster_uuid")
776 self
.k8s_juju_conn
.db
.get_one
.assert_called_once()
777 self
.k8s_juju_conn
.db
.encrypt_decrypt_fields
.assert_called_once()
778 mock_safe_dump
.assert_called_once()