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