n2vc/tests/unit/test_k8s_juju_conn.py

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