n2vc/tests/unit/test_k8s_juju_conn.py

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