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