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.k8s_juju_conn.libjuju.add_model = AsyncMock()
 329         self.k8s_juju_conn.libjuju.deploy = AsyncMock()
 330
 331     def test_success_local(self, mock_chdir):
 332         expected_kdu_instance = "{}-{}".format(self.kdu_name, "id")
 333         kdu_instance = self.loop.run_until_complete(
 334             self.k8s_juju_conn.install(
 335                 self.cluster_uuid,
 336                 self.local_bundle,
 337                 atomic=True,
 338                 kdu_name=self.kdu_name,
 339                 db_dict=self.db_dict,
 340                 timeout=1800,
 341             )
 342         )
 343         self.assertEqual(kdu_instance, expected_kdu_instance)
 344         self.assertEqual(mock_chdir.call_count, 2)
 345         self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
 346             model_name=expected_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=expected_kdu_instance,
 353             wait=True,
 354             timeout=1800,
 355         )
 356
 357     def test_success_cs(self, mock_chdir):
 358         expected_kdu_instance = "{}-{}".format(self.kdu_name, "id")
 359         kdu_instance = self.loop.run_until_complete(
 360             self.k8s_juju_conn.install(
 361                 self.cluster_uuid,
 362                 self.cs_bundle,
 363                 atomic=True,
 364                 kdu_name=self.kdu_name,
 365                 db_dict=self.db_dict,
 366                 timeout=1800,
 367             )
 368         )
 369         self.assertEqual(kdu_instance, expected_kdu_instance)
 370         self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
 371             model_name=expected_kdu_instance,
 372             cloud_name=self.cluster_uuid,
 373             credential_name="cred-{}".format(self.cluster_uuid),
 374         )
 375         self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
 376             self.cs_bundle,
 377             model_name=expected_kdu_instance,
 378             wait=True,
 379             timeout=1800,
 380         )
 381
 382     def test_success_http(self, mock_chdir):
 383         expected_kdu_instance = "{}-{}".format(self.kdu_name, "id")
 384         kdu_instance = self.loop.run_until_complete(
 385             self.k8s_juju_conn.install(
 386                 self.cluster_uuid,
 387                 self.http_bundle,
 388                 atomic=True,
 389                 kdu_name=self.kdu_name,
 390                 db_dict=self.db_dict,
 391                 timeout=1800,
 392             )
 393         )
 394         self.assertEqual(kdu_instance, expected_kdu_instance)
 395         self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
 396             model_name=expected_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=expected_kdu_instance,
 403             wait=True,
 404             timeout=1800,
 405         )
 406
 407     def test_success_not_kdu_name(self, mock_chdir):
 408         expected_kdu_instance = "id"
 409         kdu_instance = self.loop.run_until_complete(
 410             self.k8s_juju_conn.install(
 411                 self.cluster_uuid,
 412                 self.cs_bundle,
 413                 atomic=True,
 414                 db_dict=self.db_dict,
 415                 timeout=1800,
 416             )
 417         )
 418         self.assertEqual(kdu_instance, expected_kdu_instance)
 419         self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
 420             model_name=expected_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=expected_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             kdu_instance = self.loop.run_until_complete(
 435                 self.k8s_juju_conn.install(
 436                     self.cluster_uuid,
 437                     self.cs_bundle,
 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         expected_kdu_instance = "{}-{}".format(self.kdu_name, "id")
 451         kdu_instance = self.loop.run_until_complete(
 452             self.k8s_juju_conn.install(
 453                 self.cluster_uuid,
 454                 self.cs_bundle,
 455                 atomic=True,
 456                 kdu_name=self.kdu_name,
 457                 db_dict=self.db_dict,
 458                 timeout=1800,
 459             )
 460         )
 461         self.assertEqual(kdu_instance, expected_kdu_instance)
 462         self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
 463             model_name=expected_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=expected_kdu_instance,
 470             wait=True,
 471             timeout=1800,
 472         )
 473
 474     def test_missing_bundle(self, mock_chdir):
 475         kdu_instance = None
 476         with self.assertRaises(K8sException):
 477             kdu_instance = self.loop.run_until_complete(
 478                 self.k8s_juju_conn.install(
 479                     self.cluster_uuid,
 480                     "",
 481                     atomic=True,
 482                     kdu_name=self.kdu_name,
 483                     timeout=1800,
 484                     db_dict=self.db_dict,
 485                 )
 486             )
 487         self.assertIsNone(kdu_instance)
 488         self.k8s_juju_conn.libjuju.add_model.assert_not_called()
 489         self.k8s_juju_conn.libjuju.deploy.assert_not_called()
 490
 491     def test_missing_exception(self, mock_chdir):
 492         expected_kdu_instance = "{}-{}".format(self.kdu_name, "id")
 493         kdu_instance = None
 494         self.k8s_juju_conn.libjuju.deploy.side_effect = Exception()
 495         with self.assertRaises(Exception):
 496             kdu_instance = self.loop.run_until_complete(
 497                 self.k8s_juju_conn.install(
 498                     self.cluster_uuid,
 499                     self.local_bundle,
 500                     atomic=True,
 501                     kdu_name=self.kdu_name,
 502                     db_dict=self.db_dict,
 503                     timeout=1800,
 504                 )
 505             )
 506         self.assertIsNone(kdu_instance)
 507         self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
 508             model_name=expected_kdu_instance,
 509             cloud_name=self.cluster_uuid,
 510             credential_name="cred-{}".format(self.cluster_uuid),
 511         )
 512         self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
 513             "local:{}".format(self.local_bundle),
 514             model_name=expected_kdu_instance,
 515             wait=True,
 516             timeout=1800,
 517         )
 518
 519
 520 class UninstallTest(K8sJujuConnTestCase):
 521     def setUp(self):
 522         super(UninstallTest, self).setUp()
 523         self.k8s_juju_conn.libjuju.destroy_model = AsyncMock()
 524
 525     def test_success(self):
 526         destroyed = self.loop.run_until_complete(
 527             self.k8s_juju_conn.uninstall("cluster_uuid", "model_name")
 528         )
 529         self.assertTrue(destroyed)
 530         self.k8s_juju_conn.libjuju.destroy_model.assert_called_once()
 531
 532     def test_exception(self):
 533         destroyed = None
 534         self.k8s_juju_conn.libjuju.destroy_model.side_effect = Exception()
 535         with self.assertRaises(Exception):
 536             destroyed = self.loop.run_until_complete(
 537                 self.k8s_juju_conn.uninstall("cluster_uuid", "model_name")
 538             )
 539         self.assertIsNone(destroyed)
 540         self.k8s_juju_conn.libjuju.destroy_model.assert_called_once()
 541
 542
 543 class ExecPrimitivesTest(K8sJujuConnTestCase):
 544     def setUp(self):
 545         super(ExecPrimitivesTest, self).setUp()
 546         self.action_name = "touch"
 547         self.application_name = "myapp"
 548         self.model_name = "model"
 549         self.k8s_juju_conn.libjuju.get_actions = AsyncMock()
 550         self.k8s_juju_conn.libjuju.execute_action = AsyncMock()
 551
 552     def test_success(self):
 553         params = {"application-name": self.application_name}
 554         self.k8s_juju_conn.libjuju.get_actions.return_value = [self.action_name]
 555         self.k8s_juju_conn.libjuju.execute_action.return_value = (
 556             "success",
 557             "completed",
 558         )
 559
 560         output = self.loop.run_until_complete(
 561             self.k8s_juju_conn.exec_primitive(
 562                 "cluster", self.model_name, self.action_name, params=params
 563             )
 564         )
 565
 566         self.assertEqual(output, "success")
 567         self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
 568             self.application_name, self.model_name
 569         )
 570         self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
 571             self.application_name, self.model_name, self.action_name, **params
 572         )
 573
 574     def test_exception(self):
 575         params = {"application-name": self.application_name}
 576         self.k8s_juju_conn.libjuju.get_actions.return_value = [self.action_name]
 577         self.k8s_juju_conn.libjuju.execute_action.side_effect = Exception()
 578         output = None
 579
 580         with self.assertRaises(Exception):
 581             output = self.loop.run_until_complete(
 582                 self.k8s_juju_conn.exec_primitive(
 583                     "cluster", self.model_name, self.action_name, params=params
 584                 )
 585             )
 586
 587         self.assertIsNone(output)
 588         self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
 589             self.application_name, self.model_name
 590         )
 591         self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
 592             self.application_name, self.model_name, self.action_name, **params
 593         )
 594
 595     def test_missing_application_name_in_params(self):
 596         params = {}
 597         output = None
 598
 599         with self.assertRaises(K8sException):
 600             output = self.loop.run_until_complete(
 601                 self.k8s_juju_conn.exec_primitive(
 602                     "cluster", self.model_name, self.action_name, params=params
 603                 )
 604             )
 605
 606         self.assertIsNone(output)
 607         self.k8s_juju_conn.libjuju.get_actions.assert_not_called()
 608         self.k8s_juju_conn.libjuju.execute_action.assert_not_called()
 609
 610     def test_missing_params(self):
 611         output = None
 612         with self.assertRaises(K8sException):
 613             output = self.loop.run_until_complete(
 614                 self.k8s_juju_conn.exec_primitive(
 615                     "cluster", self.model_name, self.action_name
 616                 )
 617             )
 618
 619         self.assertIsNone(output)
 620         self.k8s_juju_conn.libjuju.get_actions.assert_not_called()
 621         self.k8s_juju_conn.libjuju.execute_action.assert_not_called()
 622
 623     def test_missing_action(self):
 624         output = None
 625         params = {"application-name": self.application_name}
 626         self.k8s_juju_conn.libjuju.get_actions.return_value = [self.action_name]
 627         self.k8s_juju_conn.libjuju.execute_action.return_value = (
 628             "success",
 629             "completed",
 630         )
 631         with self.assertRaises(K8sException):
 632             output = self.loop.run_until_complete(
 633                 self.k8s_juju_conn.exec_primitive(
 634                     "cluster", self.model_name, "non-existing-action", params=params
 635                 )
 636             )
 637
 638         self.assertIsNone(output)
 639         self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
 640             self.application_name, self.model_name
 641         )
 642         self.k8s_juju_conn.libjuju.execute_action.assert_not_called()
 643
 644     def test_missing_not_completed(self):
 645         output = None
 646         params = {"application-name": self.application_name}
 647         self.k8s_juju_conn.libjuju.get_actions.return_value = [self.action_name]
 648         self.k8s_juju_conn.libjuju.execute_action.return_value = (None, "failed")
 649         with self.assertRaises(K8sException):
 650             output = self.loop.run_until_complete(
 651                 self.k8s_juju_conn.exec_primitive(
 652                     "cluster", self.model_name, self.action_name, params=params
 653                 )
 654             )
 655
 656         self.assertIsNone(output)
 657         self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
 658             self.application_name, self.model_name
 659         )
 660         self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
 661             self.application_name, self.model_name, self.action_name, **params
 662         )
 663
 664
 665 class InspectKduTest(K8sJujuConnTestCase):
 666     def setUp(self):
 667         super(InspectKduTest, self).setUp()
 668
 669     @asynctest.mock.patch("builtins.open")
 670     @asynctest.mock.patch("os.path.exists")
 671     def test_existing_file(self, mock_exists, mock_open):
 672         mock_exists.return_value = True
 673         content = """{
 674             'description': 'test bundle',
 675             'bundle': 'kubernetes',
 676             'applications': {'app':{ }, 'app2': { }}
 677         }"""
 678         mock_open.return_value = FakeFileWrapper(content=content)
 679         kdu = self.loop.run_until_complete(self.k8s_juju_conn.inspect_kdu("model"))
 680         self.assertEqual(kdu, {"app": {}, "app2": {}})
 681         mock_exists.assert_called_once()
 682         mock_open.assert_called_once()
 683
 684     @asynctest.mock.patch("builtins.open")
 685     @asynctest.mock.patch("os.path.exists")
 686     def test_not_existing_file(self, mock_exists, mock_open):
 687         kdu = None
 688         mock_exists.return_value = False
 689         with self.assertRaises(K8sException):
 690             kdu = self.loop.run_until_complete(self.k8s_juju_conn.inspect_kdu("model"))
 691         self.assertEqual(kdu, None)
 692         mock_exists.assert_called_once_with("model")
 693         mock_open.assert_not_called()
 694
 695
 696 class HelpKduTest(K8sJujuConnTestCase):
 697     def setUp(self):
 698         super(HelpKduTest, self).setUp()
 699
 700     @asynctest.mock.patch("builtins.open")
 701     @asynctest.mock.patch("os.listdir")
 702     def test_existing_file(self, mock_listdir, mock_open):
 703         content = "Readme file content"
 704         mock_open.return_value = FakeFileWrapper(content=content)
 705         for file in ["README.md", "README.txt", "README"]:
 706             mock_listdir.return_value = [file]
 707             help = self.loop.run_until_complete(
 708                 self.k8s_juju_conn.help_kdu("kdu_instance")
 709             )
 710             self.assertEqual(help, content)
 711
 712         self.assertEqual(mock_listdir.call_count, 3)
 713         self.assertEqual(mock_open.call_count, 3)
 714
 715     @asynctest.mock.patch("builtins.open")
 716     @asynctest.mock.patch("os.listdir")
 717     def test_not_existing_file(self, mock_listdir, mock_open):
 718         for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
 719             mock_listdir.return_value = [file]
 720             help = self.loop.run_until_complete(
 721                 self.k8s_juju_conn.help_kdu("kdu_instance")
 722             )
 723             self.assertEqual(help, None)
 724
 725         self.assertEqual(mock_listdir.call_count, 3)
 726         self.assertEqual(mock_open.call_count, 0)
 727
 728
 729 class StatusKduTest(K8sJujuConnTestCase):
 730     def setUp(self):
 731         super(StatusKduTest, self).setUp()
 732         self.k8s_juju_conn.libjuju.get_model_status = AsyncMock()
 733
 734     def test_success(self):
 735         applications = {"app": {"status": {"status": "active"}}}
 736         model = FakeModel(applications=applications)
 737         self.k8s_juju_conn.libjuju.get_model_status.return_value = model
 738         status = self.loop.run_until_complete(
 739             self.k8s_juju_conn.status_kdu("cluster", "kdu_instance")
 740         )
 741         self.assertEqual(status, {"app": {"status": "active"}})
 742         self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()
 743
 744     def test_exception(self):
 745         self.k8s_juju_conn.libjuju.get_model_status.side_effect = Exception()
 746         status = None
 747         with self.assertRaises(Exception):
 748             status = self.loop.run_until_complete(
 749                 self.k8s_juju_conn.status_kdu("cluster", "kdu_instance")
 750             )
 751         self.assertIsNone(status)
 752         self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()
 753
 754
 755 class GetServicesTest(K8sJujuConnTestCase):
 756     def setUp(self):
 757         super(GetServicesTest, self).setUp()
 758
 759     @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_services")
 760     @asynctest.mock.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
 761     def test_success(self, mock_get_credentials, mock_get_services):
 762         mock_get_credentials.return_value = kubeconfig
 763         self.loop.run_until_complete(self.k8s_juju_conn.get_services("", "", ""))
 764         mock_get_credentials.assert_called_once()
 765         mock_get_services.assert_called_once()
 766
 767
 768 class GetServiceTest(K8sJujuConnTestCase):
 769     def setUp(self):
 770         super(GetServiceTest, self).setUp()
 771
 772     @asynctest.mock.patch("n2vc.kubectl.Kubectl.get_services")
 773     @asynctest.mock.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
 774     def test_success(self, mock_get_credentials, mock_get_services):
 775         mock_get_credentials.return_value = kubeconfig
 776         self.loop.run_until_complete(self.k8s_juju_conn.get_service("", "", ""))
 777         mock_get_credentials.assert_called_once()
 778         mock_get_services.assert_called_once()
 779
 780
 781 class GetCredentialsTest(K8sJujuConnTestCase):
 782     def setUp(self):
 783         super(GetCredentialsTest, self).setUp()
 784
 785     @asynctest.mock.patch("yaml.safe_dump")
 786     def test_success(self, mock_safe_dump):
 787         self.k8s_juju_conn.db.get_one.return_value = {
 788             "_id": "id",
 789             "credentials": "credentials",
 790             "schema_version": "2",
 791         }
 792         self.k8s_juju_conn.get_credentials("cluster_uuid")
 793         self.k8s_juju_conn.db.get_one.assert_called_once()
 794         self.k8s_juju_conn.db.encrypt_decrypt_fields.assert_called_once()
 795         mock_safe_dump.assert_called_once()