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