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