n2vc/tests/unit/test_k8s_juju_conn.py

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