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