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