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