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