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