n2vc/tests/unit/test_k8s_juju_conn.py

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