n2vc/tests/unit/test_libjuju.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 import asyncio
  16 import asynctest
  17 import tempfile
  18 from unittest import mock
  19 import juju
  20 import kubernetes
  21 from juju.errors import JujuAPIError
  22 import logging
  23 from .utils import FakeN2VC, FakeMachine, FakeApplication
  24 from n2vc.libjuju import Libjuju
  25 from n2vc.exceptions import (
  26     JujuControllerFailedConnecting,
  27     JujuModelAlreadyExists,
  28     JujuMachineNotFound,
  29     JujuApplicationNotFound,
  30     JujuActionNotFound,
  31     JujuApplicationExists,
  32     JujuInvalidK8sConfiguration,
  33     JujuLeaderUnitNotFound,
  34 )
  35
  36
  37 class LibjujuTestCase(asynctest.TestCase):
  38     @asynctest.mock.patch("juju.controller.Controller.update_endpoints")
  39     @asynctest.mock.patch("juju.client.connector.Connector.connect")
  40     @asynctest.mock.patch("juju.controller.Controller.connection")
  41     @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
  42     def setUp(
  43         self,
  44         mock__get_api_endpoints_db=None,
  45         mock_connection=None,
  46         mock_connect=None,
  47         mock_update_endpoints=None,
  48     ):
  49         loop = asyncio.get_event_loop()
  50         n2vc = FakeN2VC()
  51         mock__get_api_endpoints_db.return_value = ["127.0.0.1:17070"]
  52         endpoints = "127.0.0.1:17070"
  53         username = "admin"
  54         password = "secret"
  55         cacert = """
  56     -----BEGIN CERTIFICATE-----
  57     SOMECERT
  58     -----END CERTIFICATE-----"""
  59         self.libjuju = Libjuju(
  60             endpoints,
  61             "192.168.0.155:17070",
  62             username,
  63             password,
  64             cacert,
  65             loop,
  66             log=None,
  67             db={"get_one": []},
  68             n2vc=n2vc,
  69             apt_mirror="192.168.0.100",
  70             enable_os_upgrade=True,
  71         )
  72         logging.disable(logging.CRITICAL)
  73         loop.run_until_complete(self.libjuju.disconnect())
  74
  75
  76 @asynctest.mock.patch("n2vc.libjuju.Libjuju._create_health_check_task")
  77 @asynctest.mock.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
  78 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_api_endpoints_db")
  79 class LibjujuInitTestCase(asynctest.TestCase):
  80     def setUp(self):
  81         self.loop = asyncio.get_event_loop()
  82         self.n2vc = FakeN2VC()
  83         self.endpoint = "192.168.100.100:17070"
  84         self.username = "admin"
  85         self.password = "secret"
  86         self.cacert = """
  87     -----BEGIN CERTIFICATE-----
  88     SOMECERT
  89     -----END CERTIFICATE-----"""
  90
  91     def test_endpoint_not_in_db(
  92         self,
  93         mock__get_api_endpoints_db,
  94         mock_update_endpoints,
  95         mock_create_health_check_task,
  96     ):
  97         mock__get_api_endpoints_db.return_value = ["another_ip"]
  98         Libjuju(
  99             self.endpoint,
 100             "192.168.0.155:17070",
 101             self.username,
 102             self.password,
 103             self.cacert,
 104             self.loop,
 105             log=None,
 106             db={"get_one": []},
 107             n2vc=self.n2vc,
 108             apt_mirror="192.168.0.100",
 109             enable_os_upgrade=True,
 110         )
 111         mock_update_endpoints.assert_called_once_with([self.endpoint])
 112         mock__get_api_endpoints_db.assert_called_once()
 113
 114     def test_endpoint_in_db(
 115         self,
 116         mock__get_api_endpoints_db,
 117         mock_update_endpoints,
 118         mock_create_health_check_task,
 119     ):
 120         mock__get_api_endpoints_db.return_value = [self.endpoint, "another_ip"]
 121         Libjuju(
 122             self.endpoint,
 123             "192.168.0.155:17070",
 124             self.username,
 125             self.password,
 126             self.cacert,
 127             self.loop,
 128             log=None,
 129             db={"get_one": []},
 130             n2vc=self.n2vc,
 131             apt_mirror="192.168.0.100",
 132             enable_os_upgrade=True,
 133         )
 134         mock_update_endpoints.assert_not_called()
 135         mock__get_api_endpoints_db.assert_called_once()
 136
 137     def test_no_db_endpoints(
 138         self,
 139         mock__get_api_endpoints_db,
 140         mock_update_endpoints,
 141         mock_create_health_check_task,
 142     ):
 143         mock__get_api_endpoints_db.return_value = None
 144         Libjuju(
 145             self.endpoint,
 146             "192.168.0.155:17070",
 147             self.username,
 148             self.password,
 149             self.cacert,
 150             self.loop,
 151             log=None,
 152             db={"get_one": []},
 153             n2vc=self.n2vc,
 154             apt_mirror="192.168.0.100",
 155             enable_os_upgrade=True,
 156         )
 157         mock_update_endpoints.assert_called_once_with([self.endpoint])
 158         mock__get_api_endpoints_db.assert_called_once()
 159
 160
 161 @asynctest.mock.patch("juju.controller.Controller.connect")
 162 @asynctest.mock.patch(
 163     "juju.controller.Controller.api_endpoints",
 164     new_callable=asynctest.CoroutineMock(return_value=["127.0.0.1:17070"]),
 165 )
 166 @asynctest.mock.patch("n2vc.libjuju.Libjuju._update_api_endpoints_db")
 167 class GetControllerTest(LibjujuTestCase):
 168     def setUp(self):
 169         super(GetControllerTest, self).setUp()
 170
 171     def test_diff_endpoint(
 172         self, mock__update_api_endpoints_db, mock_api_endpoints, mock_connect
 173     ):
 174         self.libjuju.endpoints = []
 175         controller = self.loop.run_until_complete(self.libjuju.get_controller())
 176         mock__update_api_endpoints_db.assert_called_once_with(["127.0.0.1:17070"])
 177         self.assertIsInstance(controller, juju.controller.Controller)
 178
 179     @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 180     def test_exception(
 181         self,
 182         mock_disconnect_controller,
 183         mock__update_api_endpoints_db,
 184         mock_api_endpoints,
 185         mock_connect,
 186     ):
 187         self.libjuju.endpoints = []
 188         mock__update_api_endpoints_db.side_effect = Exception()
 189         controller = None
 190         with self.assertRaises(JujuControllerFailedConnecting):
 191             controller = self.loop.run_until_complete(self.libjuju.get_controller())
 192         self.assertIsNone(controller)
 193         mock_disconnect_controller.assert_called_once()
 194
 195     def test_same_endpoint_get_controller(
 196         self, mock__update_api_endpoints_db, mock_api_endpoints, mock_connect
 197     ):
 198         self.libjuju.endpoints = ["127.0.0.1:17070"]
 199         controller = self.loop.run_until_complete(self.libjuju.get_controller())
 200         mock__update_api_endpoints_db.assert_not_called()
 201         self.assertIsInstance(controller, juju.controller.Controller)
 202
 203
 204 class DisconnectTest(LibjujuTestCase):
 205     def setUp(self):
 206         super(DisconnectTest, self).setUp()
 207
 208     @asynctest.mock.patch("juju.model.Model.disconnect")
 209     def test_disconnect_model(self, mock_disconnect):
 210         self.loop.run_until_complete(self.libjuju.disconnect_model(juju.model.Model()))
 211         mock_disconnect.assert_called_once()
 212
 213     @asynctest.mock.patch("juju.controller.Controller.disconnect")
 214     def test_disconnect_controller(self, mock_disconnect):
 215         self.loop.run_until_complete(
 216             self.libjuju.disconnect_controller(juju.controller.Controller())
 217         )
 218         mock_disconnect.assert_called_once()
 219
 220
 221 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
 222 @asynctest.mock.patch("n2vc.libjuju.Libjuju.model_exists")
 223 @asynctest.mock.patch("juju.controller.Controller.add_model")
 224 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 225 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
 226 class AddModelTest(LibjujuTestCase):
 227     def setUp(self):
 228         super(AddModelTest, self).setUp()
 229
 230     def test_existing_model(
 231         self,
 232         mock_disconnect_model,
 233         mock_disconnect_controller,
 234         mock_add_model,
 235         mock_model_exists,
 236         mock_get_controller,
 237     ):
 238         mock_model_exists.return_value = True
 239
 240         with self.assertRaises(JujuModelAlreadyExists):
 241             self.loop.run_until_complete(
 242                 self.libjuju.add_model("existing_model", "cloud")
 243             )
 244
 245         mock_disconnect_controller.assert_called()
 246
 247     # TODO Check two job executing at the same time and one returning without doing anything.
 248
 249     def test_non_existing_model(
 250         self,
 251         mock_disconnect_model,
 252         mock_disconnect_controller,
 253         mock_add_model,
 254         mock_model_exists,
 255         mock_get_controller,
 256     ):
 257         mock_model_exists.return_value = False
 258         mock_get_controller.return_value = juju.controller.Controller()
 259
 260         self.loop.run_until_complete(
 261             self.libjuju.add_model("nonexisting_model", "cloud")
 262         )
 263
 264         mock_add_model.assert_called_once()
 265         mock_disconnect_controller.assert_called()
 266         mock_disconnect_model.assert_called()
 267
 268
 269 @asynctest.mock.patch("juju.controller.Controller.get_model")
 270 class GetModelTest(LibjujuTestCase):
 271     def setUp(self):
 272         super(GetModelTest, self).setUp()
 273
 274     def test_get_model(
 275         self, mock_get_model,
 276     ):
 277         mock_get_model.return_value = juju.model.Model()
 278         model = self.loop.run_until_complete(
 279             self.libjuju.get_model(juju.controller.Controller(), "model")
 280         )
 281         self.assertIsInstance(model, juju.model.Model)
 282
 283
 284 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
 285 @asynctest.mock.patch("juju.controller.Controller.list_models")
 286 class ModelExistsTest(LibjujuTestCase):
 287     def setUp(self):
 288         super(ModelExistsTest, self).setUp()
 289
 290     async def test_existing_model(
 291         self, mock_list_models, mock_get_controller,
 292     ):
 293         mock_list_models.return_value = ["existing_model"]
 294         self.assertTrue(
 295             await self.libjuju.model_exists(
 296                 "existing_model", juju.controller.Controller()
 297             )
 298         )
 299
 300     @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 301     async def test_no_controller(
 302         self, mock_disconnect_controller, mock_list_models, mock_get_controller,
 303     ):
 304         mock_list_models.return_value = ["existing_model"]
 305         mock_get_controller.return_value = juju.controller.Controller()
 306         self.assertTrue(await self.libjuju.model_exists("existing_model"))
 307         mock_disconnect_controller.assert_called_once()
 308
 309     async def test_non_existing_model(
 310         self, mock_list_models, mock_get_controller,
 311     ):
 312         mock_list_models.return_value = ["existing_model"]
 313         self.assertFalse(
 314             await self.libjuju.model_exists(
 315                 "not_existing_model", juju.controller.Controller()
 316             )
 317         )
 318
 319
 320 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
 321 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
 322 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
 323 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 324 @asynctest.mock.patch("juju.model.Model.get_status")
 325 class GetModelStatusTest(LibjujuTestCase):
 326     def setUp(self):
 327         super(GetModelStatusTest, self).setUp()
 328
 329     def test_success(
 330         self,
 331         mock_get_status,
 332         mock_disconnect_controller,
 333         mock_disconnect_model,
 334         mock_get_model,
 335         mock_get_controller,
 336     ):
 337         mock_get_model.return_value = juju.model.Model()
 338         mock_get_status.return_value = {"status"}
 339
 340         status = self.loop.run_until_complete(self.libjuju.get_model_status("model"))
 341
 342         mock_get_status.assert_called_once()
 343         mock_disconnect_controller.assert_called_once()
 344         mock_disconnect_model.assert_called_once()
 345
 346         self.assertEqual(status, {"status"})
 347
 348     def test_exception(
 349         self,
 350         mock_get_status,
 351         mock_disconnect_controller,
 352         mock_disconnect_model,
 353         mock_get_model,
 354         mock_get_controller,
 355     ):
 356         mock_get_model.return_value = juju.model.Model()
 357         mock_get_status.side_effect = Exception()
 358         status = None
 359         with self.assertRaises(Exception):
 360             status = self.loop.run_until_complete(
 361                 self.libjuju.get_model_status("model")
 362             )
 363
 364         mock_disconnect_controller.assert_called_once()
 365         mock_disconnect_model.assert_called_once()
 366
 367         self.assertIsNone(status)
 368
 369
 370 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
 371 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
 372 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
 373 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 374 @asynctest.mock.patch("juju.model.Model.get_machines")
 375 @asynctest.mock.patch("juju.model.Model.add_machine")
 376 @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
 377 class CreateMachineTest(LibjujuTestCase):
 378     def setUp(self):
 379         super(CreateMachineTest, self).setUp()
 380
 381     def test_existing_machine(
 382         self,
 383         mock_wait_for,
 384         mock_add_machine,
 385         mock_get_machines,
 386         mock_disconnect_controller,
 387         mock_disconnect_model,
 388         mock_get_model,
 389         mock_get_controller,
 390     ):
 391         mock_get_model.return_value = juju.model.Model()
 392         mock_get_machines.return_value = {"existing_machine": FakeMachine()}
 393         machine, bool_res = self.loop.run_until_complete(
 394             self.libjuju.create_machine("model", "existing_machine")
 395         )
 396
 397         self.assertIsInstance(machine, FakeMachine)
 398         self.assertFalse(bool_res)
 399
 400         mock_disconnect_controller.assert_called()
 401         mock_disconnect_model.assert_called()
 402
 403     def test_non_existing_machine(
 404         self,
 405         mock_wait_for,
 406         mock_add_machine,
 407         mock_get_machines,
 408         mock_disconnect_controller,
 409         mock_disconnect_model,
 410         mock_get_model,
 411         mock_get_controller,
 412     ):
 413         machine = None
 414         bool_res = None
 415         mock_get_model.return_value = juju.model.Model()
 416         with self.assertRaises(JujuMachineNotFound):
 417             machine, bool_res = self.loop.run_until_complete(
 418                 self.libjuju.create_machine("model", "non_existing_machine")
 419             )
 420         self.assertIsNone(machine)
 421         self.assertIsNone(bool_res)
 422
 423         mock_disconnect_controller.assert_called()
 424         mock_disconnect_model.assert_called()
 425
 426     def test_no_machine(
 427         self,
 428         mock_wait_for,
 429         mock_add_machine,
 430         mock_get_machines,
 431         mock_disconnect_controller,
 432         mock_disconnect_model,
 433         mock_get_model,
 434         mock_get_controller,
 435     ):
 436         mock_get_model.return_value = juju.model.Model()
 437         mock_add_machine.return_value = FakeMachine()
 438
 439         machine, bool_res = self.loop.run_until_complete(
 440             self.libjuju.create_machine("model")
 441         )
 442
 443         self.assertIsInstance(machine, FakeMachine)
 444         self.assertTrue(bool_res)
 445
 446         mock_wait_for.assert_called_once()
 447         mock_add_machine.assert_called_once()
 448
 449         mock_disconnect_controller.assert_called()
 450         mock_disconnect_model.assert_called()
 451
 452
 453 # TODO test provision machine
 454
 455
 456 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
 457 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
 458 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
 459 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 460 @asynctest.mock.patch(
 461     "juju.model.Model.applications", new_callable=asynctest.PropertyMock
 462 )
 463 @asynctest.mock.patch("juju.model.Model.machines", new_callable=asynctest.PropertyMock)
 464 @asynctest.mock.patch("juju.model.Model.deploy")
 465 @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
 466 @asynctest.mock.patch("n2vc.libjuju.Libjuju.create_machine")
 467 class DeployCharmTest(LibjujuTestCase):
 468     def setUp(self):
 469         super(DeployCharmTest, self).setUp()
 470
 471     def test_existing_app(
 472         self,
 473         mock_create_machine,
 474         mock_wait_for,
 475         mock_deploy,
 476         mock_machines,
 477         mock_applications,
 478         mock_disconnect_controller,
 479         mock_disconnect_model,
 480         mock_get_model,
 481         mock_get_controller,
 482     ):
 483         mock_get_model.return_value = juju.model.Model()
 484         mock_applications.return_value = {"existing_app"}
 485
 486         application = None
 487         with self.assertRaises(JujuApplicationExists):
 488             application = self.loop.run_until_complete(
 489                 self.libjuju.deploy_charm("existing_app", "path", "model", "machine",)
 490             )
 491         self.assertIsNone(application)
 492
 493         mock_disconnect_controller.assert_called()
 494         mock_disconnect_model.assert_called()
 495
 496     def test_non_existing_machine(
 497         self,
 498         mock_create_machine,
 499         mock_wait_for,
 500         mock_deploy,
 501         mock_machines,
 502         mock_applications,
 503         mock_disconnect_controller,
 504         mock_disconnect_model,
 505         mock_get_model,
 506         mock_get_controller,
 507     ):
 508         mock_get_model.return_value = juju.model.Model()
 509         mock_machines.return_value = {"existing_machine": FakeMachine()}
 510         application = None
 511         with self.assertRaises(JujuMachineNotFound):
 512             application = self.loop.run_until_complete(
 513                 self.libjuju.deploy_charm("app", "path", "model", "machine",)
 514             )
 515
 516         self.assertIsNone(application)
 517
 518         mock_disconnect_controller.assert_called()
 519         mock_disconnect_model.assert_called()
 520
 521     def test_2_units(
 522         self,
 523         mock_create_machine,
 524         mock_wait_for,
 525         mock_deploy,
 526         mock_machines,
 527         mock_applications,
 528         mock_disconnect_controller,
 529         mock_disconnect_model,
 530         mock_get_model,
 531         mock_get_controller,
 532     ):
 533         mock_get_model.return_value = juju.model.Model()
 534         mock_machines.return_value = {"existing_machine": FakeMachine()}
 535         mock_create_machine.return_value = (FakeMachine(), "other")
 536         mock_deploy.return_value = FakeApplication()
 537         application = self.loop.run_until_complete(
 538             self.libjuju.deploy_charm(
 539                 "app", "path", "model", "existing_machine", num_units=2,
 540             )
 541         )
 542
 543         self.assertIsInstance(application, FakeApplication)
 544
 545         mock_deploy.assert_called_once()
 546         mock_wait_for.assert_called_once()
 547
 548         mock_create_machine.assert_called_once()
 549
 550         mock_disconnect_controller.assert_called()
 551         mock_disconnect_model.assert_called()
 552
 553     def test_1_unit(
 554         self,
 555         mock_create_machine,
 556         mock_wait_for,
 557         mock_deploy,
 558         mock_machines,
 559         mock_applications,
 560         mock_disconnect_controller,
 561         mock_disconnect_model,
 562         mock_get_model,
 563         mock_get_controller,
 564     ):
 565         mock_get_model.return_value = juju.model.Model()
 566         mock_machines.return_value = {"existing_machine": FakeMachine()}
 567         mock_deploy.return_value = FakeApplication()
 568         application = self.loop.run_until_complete(
 569             self.libjuju.deploy_charm("app", "path", "model", "existing_machine")
 570         )
 571
 572         self.assertIsInstance(application, FakeApplication)
 573
 574         mock_deploy.assert_called_once()
 575         mock_wait_for.assert_called_once()
 576
 577         mock_disconnect_controller.assert_called()
 578         mock_disconnect_model.assert_called()
 579
 580
 581 @asynctest.mock.patch(
 582     "juju.model.Model.applications", new_callable=asynctest.PropertyMock
 583 )
 584 class GetApplicationTest(LibjujuTestCase):
 585     def setUp(self):
 586         super(GetApplicationTest, self).setUp()
 587
 588     def test_existing_application(
 589         self, mock_applications,
 590     ):
 591         mock_applications.return_value = {"existing_app": "exists"}
 592         model = juju.model.Model()
 593         result = self.libjuju._get_application(model, "existing_app")
 594         self.assertEqual(result, "exists")
 595
 596     def test_non_existing_application(
 597         self, mock_applications,
 598     ):
 599         mock_applications.return_value = {"existing_app": "exists"}
 600         model = juju.model.Model()
 601         result = self.libjuju._get_application(model, "nonexisting_app")
 602         self.assertIsNone(result)
 603
 604
 605 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
 606 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
 607 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
 608 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 609 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
 610 @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
 611 @asynctest.mock.patch("juju.model.Model.get_action_output")
 612 @asynctest.mock.patch("juju.model.Model.get_action_status")
 613 class ExecuteActionTest(LibjujuTestCase):
 614     def setUp(self):
 615         super(ExecuteActionTest, self).setUp()
 616
 617     def test_no_application(
 618         self,
 619         mock_get_action_status,
 620         mock_get_action_output,
 621         mock_wait_for,
 622         mock__get_application,
 623         mock_disconnect_controller,
 624         mock_disconnect_model,
 625         mock_get_model,
 626         mock_get_controller,
 627     ):
 628         mock__get_application.return_value = None
 629         mock_get_model.return_value = juju.model.Model()
 630         output = None
 631         status = None
 632         with self.assertRaises(JujuApplicationNotFound):
 633             output, status = self.loop.run_until_complete(
 634                 self.libjuju.execute_action("app", "model", "action",)
 635             )
 636         self.assertIsNone(output)
 637         self.assertIsNone(status)
 638
 639         mock_disconnect_controller.assert_called()
 640         mock_disconnect_model.assert_called()
 641
 642     def test_no_action(
 643         self,
 644         mock_get_action_status,
 645         mock_get_action_output,
 646         mock_wait_for,
 647         mock__get_application,
 648         mock_disconnect_controller,
 649         mock_disconnect_model,
 650         mock_get_model,
 651         mock_get_controller,
 652     ):
 653
 654         mock_get_model.return_value = juju.model.Model()
 655         mock__get_application.return_value = FakeApplication()
 656         output = None
 657         status = None
 658         with self.assertRaises(JujuActionNotFound):
 659             output, status = self.loop.run_until_complete(
 660                 self.libjuju.execute_action("app", "model", "action",)
 661             )
 662         self.assertIsNone(output)
 663         self.assertIsNone(status)
 664
 665         mock_disconnect_controller.assert_called()
 666         mock_disconnect_model.assert_called()
 667
 668     @asynctest.mock.patch("asyncio.sleep")
 669     @asynctest.mock.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
 670     def test_no_leader(
 671         self,
 672         mock_is_leader_from_status,
 673         mock_sleep,
 674         mock_get_action_status,
 675         mock_get_action_output,
 676         mock_wait_for,
 677         mock__get_application,
 678         mock_disconnect_controller,
 679         mock_disconnect_model,
 680         mock_get_model,
 681         mock_get_controller,
 682     ):
 683         mock_get_model.return_value = juju.model.Model()
 684         mock__get_application.return_value = FakeApplication()
 685         mock_is_leader_from_status.return_value = False
 686         output = None
 687         status = None
 688         with self.assertRaises(JujuLeaderUnitNotFound):
 689             output, status = self.loop.run_until_complete(
 690                 self.libjuju.execute_action("app", "model", "action",)
 691             )
 692         self.assertIsNone(output)
 693         self.assertIsNone(status)
 694
 695         mock_disconnect_controller.assert_called()
 696         mock_disconnect_model.assert_called()
 697
 698     def test_succesful_exec(
 699         self,
 700         mock_get_action_status,
 701         mock_get_action_output,
 702         mock_wait_for,
 703         mock__get_application,
 704         mock_disconnect_controller,
 705         mock_disconnect_model,
 706         mock_get_model,
 707         mock_get_controller,
 708     ):
 709         mock_get_model.return_value = juju.model.Model()
 710         mock__get_application.return_value = FakeApplication()
 711         mock_get_action_output.return_value = "output"
 712         mock_get_action_status.return_value = {"id": "status"}
 713         output, status = self.loop.run_until_complete(
 714             self.libjuju.execute_action("app", "model", "existing_action")
 715         )
 716         self.assertEqual(output, "output")
 717         self.assertEqual(status, "status")
 718
 719         mock_wait_for.assert_called_once()
 720
 721         mock_disconnect_controller.assert_called()
 722         mock_disconnect_model.assert_called()
 723
 724
 725 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
 726 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
 727 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
 728 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 729 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
 730 class GetActionTest(LibjujuTestCase):
 731     def setUp(self):
 732         super(GetActionTest, self).setUp()
 733
 734     def test_exception(
 735         self,
 736         mock_get_application,
 737         mock_disconnect_controller,
 738         mock_disconnect_model,
 739         mock_get_model,
 740         mock_get_controller,
 741     ):
 742         mock_get_application.side_effect = Exception()
 743         actions = None
 744         with self.assertRaises(Exception):
 745             actions = self.loop.run_until_complete(
 746                 self.libjuju.get_actions("app", "model")
 747             )
 748
 749         self.assertIsNone(actions)
 750         mock_disconnect_controller.assert_called_once()
 751         mock_disconnect_model.assert_called_once()
 752
 753     def test_success(
 754         self,
 755         mock_get_application,
 756         mock_disconnect_controller,
 757         mock_disconnect_model,
 758         mock_get_model,
 759         mock_get_controller,
 760     ):
 761         mock_get_application.return_value = FakeApplication()
 762
 763         actions = self.loop.run_until_complete(self.libjuju.get_actions("app", "model"))
 764
 765         self.assertEqual(actions, ["existing_action"])
 766
 767         mock_get_controller.assert_called_once()
 768         mock_get_model.assert_called_once()
 769         mock_disconnect_controller.assert_called_once()
 770         mock_disconnect_model.assert_called_once()
 771
 772
 773 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
 774 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
 775 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
 776 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 777 @asynctest.mock.patch("juju.application.Application.get_metrics")
 778 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
 779 class GetMetricsTest(LibjujuTestCase):
 780     def setUp(self):
 781         super(GetMetricsTest, self).setUp()
 782
 783     def test_get_metrics_success(
 784         self,
 785         mock_get_application,
 786         mock_get_metrics,
 787         mock_disconnect_controller,
 788         mock_disconnect_model,
 789         mock_get_model,
 790         mock_get_controller,
 791     ):
 792         mock_get_application.return_value = FakeApplication()
 793         mock_get_model.return_value = juju.model.Model()
 794
 795         self.loop.run_until_complete(self.libjuju.get_metrics("model", "app1"))
 796
 797         mock_disconnect_controller.assert_called_once()
 798         mock_disconnect_model.assert_called_once()
 799
 800     def test_get_metrics_exception(
 801         self,
 802         mock_get_application,
 803         mock_get_metrics,
 804         mock_disconnect_controller,
 805         mock_disconnect_model,
 806         mock_get_model,
 807         mock_get_controller,
 808     ):
 809         mock_get_model.return_value = juju.model.Model()
 810         mock_get_metrics.side_effect = Exception()
 811         with self.assertRaises(Exception):
 812             self.loop.run_until_complete(self.libjuju.get_metrics("model", "app1"))
 813
 814         mock_disconnect_controller.assert_called_once()
 815         mock_disconnect_model.assert_called_once()
 816
 817     def test_missing_args_exception(
 818         self,
 819         mock_get_application,
 820         mock_get_metrics,
 821         mock_disconnect_controller,
 822         mock_disconnect_model,
 823         mock_get_model,
 824         mock_get_controller,
 825     ):
 826         mock_get_model.return_value = juju.model.Model()
 827
 828         with self.assertRaises(Exception):
 829             self.loop.run_until_complete(self.libjuju.get_metrics("", ""))
 830
 831         mock_get_controller.assert_not_called()
 832         mock_get_model.assert_not_called()
 833         mock_disconnect_controller.assert_not_called()
 834         mock_disconnect_model.assert_not_called()
 835
 836
 837 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
 838 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
 839 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
 840 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 841 @asynctest.mock.patch("juju.model.Model.add_relation")
 842 class AddRelationTest(LibjujuTestCase):
 843     def setUp(self):
 844         super(AddRelationTest, self).setUp()
 845
 846     @asynctest.mock.patch("logging.Logger.warning")
 847     def test_not_found(
 848         self,
 849         mock_warning,
 850         mock_add_relation,
 851         mock_disconnect_controller,
 852         mock_disconnect_model,
 853         mock_get_model,
 854         mock_get_controller,
 855     ):
 856         # TODO in libjuju.py should this fail only with a log message?
 857         result = {"error": "not found", "response": "response", "request-id": 1}
 858
 859         mock_get_model.return_value = juju.model.Model()
 860         mock_add_relation.side_effect = JujuAPIError(result)
 861
 862         self.loop.run_until_complete(
 863             self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",)
 864         )
 865
 866         mock_warning.assert_called_with("Relation not found: not found")
 867         mock_disconnect_controller.assert_called_once()
 868         mock_disconnect_model.assert_called_once()
 869
 870     @asynctest.mock.patch("logging.Logger.warning")
 871     def test_already_exists(
 872         self,
 873         mock_warning,
 874         mock_add_relation,
 875         mock_disconnect_controller,
 876         mock_disconnect_model,
 877         mock_get_model,
 878         mock_get_controller,
 879     ):
 880         # TODO in libjuju.py should this fail silently?
 881         result = {"error": "already exists", "response": "response", "request-id": 1}
 882
 883         mock_get_model.return_value = juju.model.Model()
 884         mock_add_relation.side_effect = JujuAPIError(result)
 885
 886         self.loop.run_until_complete(
 887             self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",)
 888         )
 889
 890         mock_warning.assert_called_with("Relation already exists: already exists")
 891         mock_disconnect_controller.assert_called_once()
 892         mock_disconnect_model.assert_called_once()
 893
 894     def test_exception(
 895         self,
 896         mock_add_relation,
 897         mock_disconnect_controller,
 898         mock_disconnect_model,
 899         mock_get_model,
 900         mock_get_controller,
 901     ):
 902         mock_get_model.return_value = juju.model.Model()
 903         result = {"error": "", "response": "response", "request-id": 1}
 904         mock_add_relation.side_effect = JujuAPIError(result)
 905
 906         with self.assertRaises(JujuAPIError):
 907             self.loop.run_until_complete(
 908                 self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",)
 909             )
 910
 911         mock_disconnect_controller.assert_called_once()
 912         mock_disconnect_model.assert_called_once()
 913
 914     def test_success(
 915         self,
 916         mock_add_relation,
 917         mock_disconnect_controller,
 918         mock_disconnect_model,
 919         mock_get_model,
 920         mock_get_controller,
 921     ):
 922         mock_get_model.return_value = juju.model.Model()
 923
 924         self.loop.run_until_complete(
 925             self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",)
 926         )
 927
 928         mock_add_relation.assert_called_with("app1:relation1", "app2:relation2")
 929         mock_disconnect_controller.assert_called_once()
 930         mock_disconnect_model.assert_called_once()
 931
 932     def test_saas(
 933         self,
 934         mock_add_relation,
 935         mock_disconnect_controller,
 936         mock_disconnect_model,
 937         mock_get_model,
 938         mock_get_controller,
 939     ):
 940         mock_get_model.return_value = juju.model.Model()
 941
 942         self.loop.run_until_complete(
 943             self.libjuju.add_relation("model", "app1:relation1", "saas_name",)
 944         )
 945
 946         mock_add_relation.assert_called_with("app1:relation1", "saas_name")
 947         mock_disconnect_controller.assert_called_once()
 948         mock_disconnect_model.assert_called_once()
 949
 950
 951 # TODO destroy_model testcase
 952
 953
 954 # @asynctest.mock.patch("juju.model.Model.get_machines")
 955 # @asynctest.mock.patch("logging.Logger.debug")
 956 # class DestroyMachineTest(LibjujuTestCase):
 957 #     def setUp(self):
 958 #         super(DestroyMachineTest, self).setUp()
 959
 960 #     def test_success_manual_machine(
 961 #         self, mock_debug, mock_get_machines,
 962 #     ):
 963 #         mock_get_machines.side_effect = [
 964 #             {"machine": FakeManualMachine()},
 965 #             {"machine": FakeManualMachine()},
 966 #             {},
 967 #         ]
 968 #         self.loop.run_until_complete(
 969 #             self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
 970 #         )
 971 #         calls = [
 972 #             asynctest.call("Waiting for machine machine is destroyed"),
 973 #             asynctest.call("Machine destroyed: machine"),
 974 #         ]
 975 #         mock_debug.assert_has_calls(calls)
 976
 977 #     def test_no_machine(
 978 #         self, mock_debug, mock_get_machines,
 979 #     ):
 980 #         mock_get_machines.return_value = {}
 981 #         self.loop.run_until_complete(
 982 #             self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
 983 #         )
 984 #         mock_debug.assert_called_with("Machine not found: machine")
 985
 986
 987 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
 988 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
 989 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
 990 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
 991 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
 992 class ConfigureApplicationTest(LibjujuTestCase):
 993     def setUp(self):
 994         super(ConfigureApplicationTest, self).setUp()
 995
 996     def test_success(
 997         self,
 998         mock_get_application,
 999         mock_disconnect_controller,
1000         mock_disconnect_model,
1001         mock_get_model,
1002         mock_get_controller,
1003     ):
1004
1005         mock_get_application.return_value = FakeApplication()
1006
1007         self.loop.run_until_complete(
1008             self.libjuju.configure_application("model", "app", {"config"},)
1009         )
1010         mock_get_application.assert_called_once()
1011         mock_disconnect_controller.assert_called_once()
1012         mock_disconnect_model.assert_called_once()
1013
1014     def test_exception(
1015         self,
1016         mock_get_application,
1017         mock_disconnect_controller,
1018         mock_disconnect_model,
1019         mock_get_model,
1020         mock_get_controller,
1021     ):
1022
1023         mock_get_application.side_effect = Exception()
1024
1025         with self.assertRaises(Exception):
1026             self.loop.run_until_complete(
1027                 self.libjuju.configure_application("model", "app", {"config"},)
1028             )
1029         mock_disconnect_controller.assert_called_once()
1030         mock_disconnect_model.assert_called_once()
1031
1032
1033 # TODO _get_api_endpoints_db test case
1034 # TODO _update_api_endpoints_db test case
1035 # TODO healthcheck test case
1036
1037
1038 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1039 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1040 @asynctest.mock.patch("juju.controller.Controller.list_models")
1041 class ListModelsTest(LibjujuTestCase):
1042     def setUp(self):
1043         super(ListModelsTest, self).setUp()
1044
1045     def test_containing(
1046         self, mock_list_models, mock_disconnect_controller, mock_get_controller,
1047     ):
1048         mock_get_controller.return_value = juju.controller.Controller()
1049         mock_list_models.return_value = ["existingmodel"]
1050         models = self.loop.run_until_complete(self.libjuju.list_models("existing"))
1051
1052         mock_disconnect_controller.assert_called_once()
1053         self.assertEquals(models, ["existingmodel"])
1054
1055     def test_not_containing(
1056         self, mock_list_models, mock_disconnect_controller, mock_get_controller,
1057     ):
1058         mock_get_controller.return_value = juju.controller.Controller()
1059         mock_list_models.return_value = ["existingmodel", "model"]
1060         models = self.loop.run_until_complete(self.libjuju.list_models("mdl"))
1061
1062         mock_disconnect_controller.assert_called_once()
1063         self.assertEquals(models, [])
1064
1065     def test_no_contains_arg(
1066         self, mock_list_models, mock_disconnect_controller, mock_get_controller,
1067     ):
1068         mock_get_controller.return_value = juju.controller.Controller()
1069         mock_list_models.return_value = ["existingmodel", "model"]
1070         models = self.loop.run_until_complete(self.libjuju.list_models())
1071
1072         mock_disconnect_controller.assert_called_once()
1073         self.assertEquals(models, ["existingmodel", "model"])
1074
1075
1076 @asynctest.mock.patch("n2vc.libjuju.Libjuju.list_models")
1077 class ModelsExistTest(LibjujuTestCase):
1078     def setUp(self):
1079         super(ModelsExistTest, self).setUp()
1080
1081     def test_model_names_none(self, mock_list_models):
1082         mock_list_models.return_value = []
1083         with self.assertRaises(Exception):
1084             self.loop.run_until_complete(self.libjuju.models_exist(None))
1085
1086     def test_model_names_empty(self, mock_list_models):
1087         mock_list_models.return_value = []
1088         with self.assertRaises(Exception):
1089             (exist, non_existing_models) = self.loop.run_until_complete(
1090                 self.libjuju.models_exist([])
1091             )
1092
1093     def test_model_names_not_existing(self, mock_list_models):
1094         mock_list_models.return_value = ["prometheus", "grafana"]
1095         (exist, non_existing_models) = self.loop.run_until_complete(
1096             self.libjuju.models_exist(["prometheus2", "grafana"])
1097         )
1098         self.assertFalse(exist)
1099         self.assertEqual(non_existing_models, ["prometheus2"])
1100
1101     def test_model_names_exist(self, mock_list_models):
1102         mock_list_models.return_value = ["prometheus", "grafana"]
1103         (exist, non_existing_models) = self.loop.run_until_complete(
1104             self.libjuju.models_exist(["prometheus", "grafana"])
1105         )
1106         self.assertTrue(exist)
1107         self.assertEqual(non_existing_models, [])
1108
1109
1110 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1111 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1112 @asynctest.mock.patch("juju.controller.Controller.list_offers")
1113 class ListOffers(LibjujuTestCase):
1114     def setUp(self):
1115         super(ListOffers, self).setUp()
1116
1117     def test_disconnect_controller(
1118         self, mock_list_offers, mock_disconnect_controller, mock_get_controller,
1119     ):
1120         mock_get_controller.return_value = juju.controller.Controller()
1121         mock_list_offers.side_effect = Exception()
1122         with self.assertRaises(Exception):
1123             self.loop.run_until_complete(self.libjuju.list_offers("model"))
1124         mock_disconnect_controller.assert_called_once()
1125
1126     def test_empty_list(
1127         self, mock_list_offers, mock_disconnect_controller, mock_get_controller,
1128     ):
1129         mock_get_controller.return_value = juju.controller.Controller()
1130         mock_list_offers.return_value = []
1131         offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
1132         self.assertEqual(offers, [])
1133         mock_disconnect_controller.assert_called_once()
1134
1135     def test_non_empty_list(
1136         self, mock_list_offers, mock_disconnect_controller, mock_get_controller,
1137     ):
1138         mock_get_controller.return_value = juju.controller.Controller()
1139         mock_list_offers.return_value = ["offer"]
1140         offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
1141         self.assertEqual(offers, ["offer"])
1142         mock_disconnect_controller.assert_called_once()
1143
1144
1145 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1146 @asynctest.mock.patch("juju.controller.Controller.get_model")
1147 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
1148 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1149 @asynctest.mock.patch("juju.model.Model.consume")
1150 class ConsumeTest(LibjujuTestCase):
1151     def setUp(self):
1152         super(ConsumeTest, self).setUp()
1153
1154     def test_consume(
1155         self,
1156         mock_consume,
1157         mock_disconnect_controller,
1158         mock_disconnect_model,
1159         mock_get_model,
1160         mock_get_controller,
1161     ):
1162         mock_get_controller.return_value = juju.controller.Controller()
1163         mock_get_model.return_value = juju.model.Model()
1164
1165         self.loop.run_until_complete(self.libjuju.consume("offer_url", "model_name"))
1166         mock_consume.assert_called_once()
1167         mock_disconnect_model.assert_called_once()
1168         mock_disconnect_controller.assert_called_once()
1169
1170     def test_parsing_error_exception(
1171         self,
1172         mock_consume,
1173         mock_disconnect_controller,
1174         mock_disconnect_model,
1175         mock_get_model,
1176         mock_get_controller,
1177     ):
1178         mock_get_controller.return_value = juju.controller.Controller()
1179         mock_get_model.return_value = juju.model.Model()
1180         mock_consume.side_effect = juju.offerendpoints.ParseError("")
1181
1182         with self.assertRaises(juju.offerendpoints.ParseError):
1183             self.loop.run_until_complete(
1184                 self.libjuju.consume("offer_url", "model_name")
1185             )
1186         mock_consume.assert_called_once()
1187         mock_disconnect_model.assert_called_once()
1188         mock_disconnect_controller.assert_called_once()
1189
1190     def test_juju_error_exception(
1191         self,
1192         mock_consume,
1193         mock_disconnect_controller,
1194         mock_disconnect_model,
1195         mock_get_model,
1196         mock_get_controller,
1197     ):
1198         mock_get_controller.return_value = juju.controller.Controller()
1199         mock_get_model.return_value = juju.model.Model()
1200         mock_consume.side_effect = juju.errors.JujuError("")
1201
1202         with self.assertRaises(juju.errors.JujuError):
1203             self.loop.run_until_complete(
1204                 self.libjuju.consume("offer_url", "model_name")
1205             )
1206         mock_consume.assert_called_once()
1207         mock_disconnect_model.assert_called_once()
1208         mock_disconnect_controller.assert_called_once()
1209
1210     def test_juju_api_error_exception(
1211         self,
1212         mock_consume,
1213         mock_disconnect_controller,
1214         mock_disconnect_model,
1215         mock_get_model,
1216         mock_get_controller,
1217     ):
1218         mock_get_controller.return_value = juju.controller.Controller()
1219         mock_get_model.return_value = juju.model.Model()
1220         mock_consume.side_effect = juju.errors.JujuAPIError(
1221             {"error": "", "response": "", "request-id": ""}
1222         )
1223
1224         with self.assertRaises(juju.errors.JujuAPIError):
1225             self.loop.run_until_complete(
1226                 self.libjuju.consume("offer_url", "model_name")
1227             )
1228         mock_consume.assert_called_once()
1229         mock_disconnect_model.assert_called_once()
1230         mock_disconnect_controller.assert_called_once()
1231
1232
1233 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1234 @asynctest.mock.patch("n2vc.libjuju.Libjuju.add_cloud")
1235 class AddK8sTest(LibjujuTestCase):
1236     def setUp(self):
1237         super(AddK8sTest, self).setUp()
1238         self.configuration = kubernetes.client.configuration.Configuration()
1239
1240     def test_add_k8s(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1241         self.loop.run_until_complete(
1242             self.libjuju.add_k8s("cloud", self.configuration, "storage_class")
1243         )
1244         mock_add_cloud.assert_called_once()
1245         mock_get_k8s_cloud_credential.assert_called_once()
1246
1247     def test_add_k8s_exception(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1248         mock_add_cloud.side_effect = Exception()
1249         with self.assertRaises(Exception):
1250             self.loop.run_until_complete(
1251                 self.libjuju.add_k8s("cloud", self.configuration, "storage_class")
1252             )
1253         mock_add_cloud.assert_called_once()
1254         mock_get_k8s_cloud_credential.assert_called_once()
1255
1256     def test_add_k8s_missing_name(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1257         with self.assertRaises(Exception):
1258             self.loop.run_until_complete(
1259                 self.libjuju.add_k8s("", self.configuration, "storage_class")
1260             )
1261         mock_add_cloud.assert_not_called()
1262
1263     def test_add_k8s_missing_storage_name(
1264         self, mock_add_cloud, mock_get_k8s_cloud_credential
1265     ):
1266         with self.assertRaises(Exception):
1267             self.loop.run_until_complete(
1268                 self.libjuju.add_k8s("cloud", self.configuration, "")
1269             )
1270         mock_add_cloud.assert_not_called()
1271
1272     def test_add_k8s_missing_configuration_keys(
1273         self, mock_add_cloud, mock_get_k8s_cloud_credential
1274     ):
1275         with self.assertRaises(Exception):
1276             self.loop.run_until_complete(self.libjuju.add_k8s("cloud", None, ""))
1277         mock_add_cloud.assert_not_called()
1278
1279
1280 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1281 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1282 @asynctest.mock.patch("juju.controller.Controller.add_cloud")
1283 @asynctest.mock.patch("juju.controller.Controller.add_credential")
1284 class AddCloudTest(LibjujuTestCase):
1285     def setUp(self):
1286         super(AddCloudTest, self).setUp()
1287         self.cloud = juju.client.client.Cloud()
1288         self.credential = juju.client.client.CloudCredential()
1289
1290     def test_add_cloud_with_credential(
1291         self,
1292         mock_add_credential,
1293         mock_add_cloud,
1294         mock_disconnect_controller,
1295         mock_get_controller,
1296     ):
1297         mock_get_controller.return_value = juju.controller.Controller()
1298
1299         cloud = self.loop.run_until_complete(
1300             self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1301         )
1302         self.assertEqual(cloud, self.cloud)
1303         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1304         mock_add_credential.assert_called_once_with(
1305             "cloud", credential=self.credential, cloud="cloud"
1306         )
1307         mock_disconnect_controller.assert_called_once()
1308
1309     def test_add_cloud_no_credential(
1310         self,
1311         mock_add_credential,
1312         mock_add_cloud,
1313         mock_disconnect_controller,
1314         mock_get_controller,
1315     ):
1316         mock_get_controller.return_value = juju.controller.Controller()
1317
1318         cloud = self.loop.run_until_complete(
1319             self.libjuju.add_cloud("cloud", self.cloud)
1320         )
1321         self.assertEqual(cloud, self.cloud)
1322         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1323         mock_add_credential.assert_not_called()
1324         mock_disconnect_controller.assert_called_once()
1325
1326     def test_add_cloud_exception(
1327         self,
1328         mock_add_credential,
1329         mock_add_cloud,
1330         mock_disconnect_controller,
1331         mock_get_controller,
1332     ):
1333         mock_get_controller.return_value = juju.controller.Controller()
1334         mock_add_cloud.side_effect = Exception()
1335         with self.assertRaises(Exception):
1336             self.loop.run_until_complete(
1337                 self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1338             )
1339
1340         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1341         mock_add_credential.assert_not_called()
1342         mock_disconnect_controller.assert_called_once()
1343
1344     def test_add_credential_exception(
1345         self,
1346         mock_add_credential,
1347         mock_add_cloud,
1348         mock_disconnect_controller,
1349         mock_get_controller,
1350     ):
1351         mock_get_controller.return_value = juju.controller.Controller()
1352         mock_add_credential.side_effect = Exception()
1353         with self.assertRaises(Exception):
1354             self.loop.run_until_complete(
1355                 self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1356             )
1357
1358         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1359         mock_add_credential.assert_called_once_with(
1360             "cloud", credential=self.credential, cloud="cloud"
1361         )
1362         mock_disconnect_controller.assert_called_once()
1363
1364
1365 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1366 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1367 @asynctest.mock.patch("juju.controller.Controller.remove_cloud")
1368 class RemoveCloudTest(LibjujuTestCase):
1369     def setUp(self):
1370         super(RemoveCloudTest, self).setUp()
1371
1372     def test_remove_cloud(
1373         self, mock_remove_cloud, mock_disconnect_controller, mock_get_controller,
1374     ):
1375         mock_get_controller.return_value = juju.controller.Controller()
1376
1377         self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
1378         mock_remove_cloud.assert_called_once_with("cloud")
1379         mock_disconnect_controller.assert_called_once()
1380
1381     def test_remove_cloud_exception(
1382         self, mock_remove_cloud, mock_disconnect_controller, mock_get_controller,
1383     ):
1384         mock_get_controller.return_value = juju.controller.Controller()
1385         mock_remove_cloud.side_effect = Exception()
1386
1387         with self.assertRaises(Exception):
1388             self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
1389         mock_remove_cloud.assert_called_once_with("cloud")
1390         mock_disconnect_controller.assert_called_once()
1391
1392
1393 @asynctest.mock.patch("kubernetes.client.configuration.Configuration")
1394 class GetK8sCloudCredentials(LibjujuTestCase):
1395     def setUp(self):
1396         super(GetK8sCloudCredentials, self).setUp()
1397
1398     @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1399     def test_not_supported(self, mock_exception, mock_configuration):
1400         mock_configuration.username = ""
1401         mock_configuration.password = ""
1402         mock_configuration.ssl_ca_cert = None
1403         mock_configuration.cert_file = None
1404         mock_configuration.key_file = None
1405         exception_raised = False
1406         try:
1407             _ = self.libjuju.get_k8s_cloud_credential(mock_configuration)
1408         except JujuInvalidK8sConfiguration as e:
1409             exception_raised = True
1410             self.assertEqual(
1411                 e.message, "authentication method not supported",
1412             )
1413         self.assertTrue(exception_raised)
1414
1415     def test_user_pass(self, mock_configuration):
1416         mock_configuration.username = "admin"
1417         mock_configuration.password = "admin"
1418         mock_configuration.ssl_ca_cert = None
1419         mock_configuration.cert_file = None
1420         mock_configuration.key_file = None
1421         credential = self.libjuju.get_k8s_cloud_credential(mock_configuration)
1422         self.assertEqual(
1423             credential,
1424             juju.client._definitions.CloudCredential(
1425                 attrs={"username": "admin", "password": "admin"}, auth_type="userpass"
1426             ),
1427         )
1428
1429     def test_user_no_pass(self, mock_configuration):
1430         mock_configuration.username = "admin"
1431         mock_configuration.password = ""
1432         mock_configuration.ssl_ca_cert = None
1433         mock_configuration.cert_file = None
1434         mock_configuration.key_file = None
1435         with mock.patch.object(self.libjuju.log, "debug") as mock_debug:
1436             credential = self.libjuju.get_k8s_cloud_credential(mock_configuration)
1437             self.assertEqual(
1438                 credential,
1439                 juju.client._definitions.CloudCredential(
1440                     attrs={"username": "admin", "password": ""}, auth_type="userpass"
1441                 ),
1442             )
1443             mock_debug.assert_called_once_with(
1444                 "credential for user admin has empty password"
1445             )
1446
1447     def test_user_pass_with_cert(self, mock_configuration):
1448         mock_configuration.username = "admin"
1449         mock_configuration.password = "admin"
1450         ssl_ca_cert = tempfile.NamedTemporaryFile()
1451         with open(ssl_ca_cert.name, "w") as ssl_ca_cert_file:
1452             ssl_ca_cert_file.write("cacert")
1453         mock_configuration.ssl_ca_cert = ssl_ca_cert.name
1454         mock_configuration.cert_file = None
1455         mock_configuration.key_file = None
1456         credential = self.libjuju.get_k8s_cloud_credential(mock_configuration)
1457         self.assertEqual(
1458             credential,
1459             juju.client._definitions.CloudCredential(
1460                 attrs={
1461                     "username": "admin",
1462                     "password": "admin",
1463                     "ClientCertificateData": "cacert",
1464                 },
1465                 auth_type="userpasswithcert",
1466             ),
1467         )
1468
1469     def test_cert(self, mock_configuration):
1470         mock_configuration.username = ""
1471         mock_configuration.password = ""
1472         mock_configuration.api_key = {"authorization": "Bearer Token"}
1473         ssl_ca_cert = tempfile.NamedTemporaryFile()
1474         with open(ssl_ca_cert.name, "w") as ssl_ca_cert_file:
1475             ssl_ca_cert_file.write("cacert")
1476         mock_configuration.ssl_ca_cert = ssl_ca_cert.name
1477         mock_configuration.cert_file = None
1478         mock_configuration.key_file = None
1479         credential = self.libjuju.get_k8s_cloud_credential(mock_configuration)
1480         self.assertEqual(
1481             credential,
1482             juju.client._definitions.CloudCredential(
1483                 attrs={"ClientCertificateData": "cacert", "Token": "Token"},
1484                 auth_type="certificate",
1485             ),
1486         )
1487
1488     def test_oauth2(self, mock_configuration):
1489         mock_configuration.username = ""
1490         mock_configuration.password = ""
1491         mock_configuration.api_key = {"authorization": "Bearer Token"}
1492         key = tempfile.NamedTemporaryFile()
1493         with open(key.name, "w") as key_file:
1494             key_file.write("key")
1495         mock_configuration.ssl_ca_cert = None
1496         mock_configuration.cert_file = None
1497         mock_configuration.key_file = key.name
1498         credential = self.libjuju.get_k8s_cloud_credential(mock_configuration)
1499         self.assertEqual(
1500             credential,
1501             juju.client._definitions.CloudCredential(
1502                 attrs={"ClientKeyData": "key", "Token": "Token"}, auth_type="oauth2",
1503             ),
1504         )
1505
1506     @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1507     def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1508         mock_configuration.username = ""
1509         mock_configuration.password = ""
1510         key = tempfile.NamedTemporaryFile()
1511         with open(key.name, "w") as key_file:
1512             key_file.write("key")
1513         mock_configuration.ssl_ca_cert = None
1514         mock_configuration.cert_file = None
1515         mock_configuration.key_file = key.name
1516         exception_raised = False
1517         try:
1518             _ = self.libjuju.get_k8s_cloud_credential(mock_configuration)
1519         except JujuInvalidK8sConfiguration as e:
1520             exception_raised = True
1521             self.assertEqual(
1522                 e.message, "missing token for auth type oauth2",
1523             )
1524         self.assertTrue(exception_raised)
1525
1526     def test_unknown_api_key(self, mock_configuration):
1527         mock_configuration.username = ""
1528         mock_configuration.password = ""
1529         mock_configuration.api_key = {"authorization": "Bearer Token Wrong"}
1530         mock_configuration.ssl_ca_cert = None
1531         mock_configuration.cert_file = None
1532         mock_configuration.key_file = None
1533         exception_raised = False
1534         try:
1535             _ = self.libjuju.get_k8s_cloud_credential(mock_configuration)
1536         except JujuInvalidK8sConfiguration as e:
1537             exception_raised = True
1538             self.assertEqual(
1539                 e.message, "unknown format of api_key",
1540             )
1541         self.assertTrue(exception_raised)
1542
1543     def test_exception_cannot_set_token_and_userpass(self, mock_configuration):
1544         mock_configuration.username = "admin"
1545         mock_configuration.password = "pass"
1546         mock_configuration.api_key = {"authorization": "No_bearer_token"}
1547         mock_configuration.ssl_ca_cert = None
1548         mock_configuration.cert_file = None
1549         mock_configuration.key_file = None
1550         exception_raised = False
1551         try:
1552             _ = self.libjuju.get_k8s_cloud_credential(mock_configuration)
1553         except JujuInvalidK8sConfiguration as e:
1554             exception_raised = True
1555             self.assertEqual(
1556                 e.message, "Cannot set both token and user/pass",
1557             )
1558         self.assertTrue(exception_raised)