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