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     JujuMachineNotFound,
  28     JujuApplicationNotFound,
  29     JujuActionNotFound,
  30     JujuApplicationExists,
  31     JujuInvalidK8sConfiguration,
  32     JujuLeaderUnitNotFound,
  33 )
  34 from n2vc.k8s_juju_conn import generate_rbac_id
  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         # This should not raise an exception
 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         name = "cloud"
1435         rbac_id = generate_rbac_id()
1436         token = "token"
1437         client_cert_data = "cert"
1438         configuration = kubernetes.client.configuration.Configuration()
1439         storage_class = "storage_class"
1440         credential_name = name
1441
1442         self._add_k8s_args = {
1443             "name": name,
1444             "rbac_id": rbac_id,
1445             "token": token,
1446             "client_cert_data": client_cert_data,
1447             "configuration": configuration,
1448             "storage_class": storage_class,
1449             "credential_name": credential_name,
1450         }
1451
1452     def test_add_k8s(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1453         self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1454         mock_add_cloud.assert_called_once()
1455         mock_get_k8s_cloud_credential.assert_called_once()
1456
1457     def test_add_k8s_exception(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1458         mock_add_cloud.side_effect = Exception()
1459         with self.assertRaises(Exception):
1460             self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1461         mock_add_cloud.assert_called_once()
1462         mock_get_k8s_cloud_credential.assert_called_once()
1463
1464     def test_add_k8s_missing_name(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1465         self._add_k8s_args["name"] = ""
1466         with self.assertRaises(Exception):
1467             self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1468         mock_add_cloud.assert_not_called()
1469
1470     def test_add_k8s_missing_storage_name(
1471         self, mock_add_cloud, mock_get_k8s_cloud_credential
1472     ):
1473         self._add_k8s_args["storage_class"] = ""
1474         with self.assertRaises(Exception):
1475             self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1476         mock_add_cloud.assert_not_called()
1477
1478     def test_add_k8s_missing_configuration_keys(
1479         self, mock_add_cloud, mock_get_k8s_cloud_credential
1480     ):
1481         self._add_k8s_args["configuration"] = None
1482         with self.assertRaises(Exception):
1483             self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1484         mock_add_cloud.assert_not_called()
1485
1486
1487 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1488 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1489 @asynctest.mock.patch("juju.controller.Controller.add_cloud")
1490 @asynctest.mock.patch("juju.controller.Controller.add_credential")
1491 class AddCloudTest(LibjujuTestCase):
1492     def setUp(self):
1493         super(AddCloudTest, self).setUp()
1494         self.cloud = juju.client.client.Cloud()
1495         self.credential = juju.client.client.CloudCredential()
1496
1497     def test_add_cloud_with_credential(
1498         self,
1499         mock_add_credential,
1500         mock_add_cloud,
1501         mock_disconnect_controller,
1502         mock_get_controller,
1503     ):
1504         mock_get_controller.return_value = juju.controller.Controller()
1505
1506         cloud = self.loop.run_until_complete(
1507             self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1508         )
1509         self.assertEqual(cloud, self.cloud)
1510         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1511         mock_add_credential.assert_called_once_with(
1512             "cloud", credential=self.credential, cloud="cloud"
1513         )
1514         mock_disconnect_controller.assert_called_once()
1515
1516     def test_add_cloud_no_credential(
1517         self,
1518         mock_add_credential,
1519         mock_add_cloud,
1520         mock_disconnect_controller,
1521         mock_get_controller,
1522     ):
1523         mock_get_controller.return_value = juju.controller.Controller()
1524
1525         cloud = self.loop.run_until_complete(
1526             self.libjuju.add_cloud("cloud", self.cloud)
1527         )
1528         self.assertEqual(cloud, self.cloud)
1529         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1530         mock_add_credential.assert_not_called()
1531         mock_disconnect_controller.assert_called_once()
1532
1533     def test_add_cloud_exception(
1534         self,
1535         mock_add_credential,
1536         mock_add_cloud,
1537         mock_disconnect_controller,
1538         mock_get_controller,
1539     ):
1540         mock_get_controller.return_value = juju.controller.Controller()
1541         mock_add_cloud.side_effect = Exception()
1542         with self.assertRaises(Exception):
1543             self.loop.run_until_complete(
1544                 self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1545             )
1546
1547         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1548         mock_add_credential.assert_not_called()
1549         mock_disconnect_controller.assert_called_once()
1550
1551     def test_add_credential_exception(
1552         self,
1553         mock_add_credential,
1554         mock_add_cloud,
1555         mock_disconnect_controller,
1556         mock_get_controller,
1557     ):
1558         mock_get_controller.return_value = juju.controller.Controller()
1559         mock_add_credential.side_effect = Exception()
1560         with self.assertRaises(Exception):
1561             self.loop.run_until_complete(
1562                 self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1563             )
1564
1565         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1566         mock_add_credential.assert_called_once_with(
1567             "cloud", credential=self.credential, cloud="cloud"
1568         )
1569         mock_disconnect_controller.assert_called_once()
1570
1571
1572 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1573 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1574 @asynctest.mock.patch("juju.controller.Controller.remove_cloud")
1575 class RemoveCloudTest(LibjujuTestCase):
1576     def setUp(self):
1577         super(RemoveCloudTest, self).setUp()
1578
1579     def test_remove_cloud(
1580         self,
1581         mock_remove_cloud,
1582         mock_disconnect_controller,
1583         mock_get_controller,
1584     ):
1585         mock_get_controller.return_value = juju.controller.Controller()
1586
1587         self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
1588         mock_remove_cloud.assert_called_once_with("cloud")
1589         mock_disconnect_controller.assert_called_once()
1590
1591     def test_remove_cloud_exception(
1592         self,
1593         mock_remove_cloud,
1594         mock_disconnect_controller,
1595         mock_get_controller,
1596     ):
1597         mock_get_controller.return_value = juju.controller.Controller()
1598         mock_remove_cloud.side_effect = Exception()
1599
1600         with self.assertRaises(Exception):
1601             self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
1602         mock_remove_cloud.assert_called_once_with("cloud")
1603         mock_disconnect_controller.assert_called_once()
1604
1605
1606 @asynctest.mock.patch("kubernetes.client.configuration.Configuration")
1607 class GetK8sCloudCredentials(LibjujuTestCase):
1608     def setUp(self):
1609         super(GetK8sCloudCredentials, self).setUp()
1610         self.cert_data = "cert"
1611         self.token = "token"
1612
1613     @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1614     def test_not_supported(self, mock_exception, mock_configuration):
1615         mock_configuration.username = ""
1616         mock_configuration.password = ""
1617         mock_configuration.ssl_ca_cert = None
1618         mock_configuration.cert_file = None
1619         mock_configuration.key_file = None
1620         exception_raised = False
1621         self.token = None
1622         self.cert_data = None
1623         try:
1624             _ = self.libjuju.get_k8s_cloud_credential(
1625                 mock_configuration,
1626                 self.cert_data,
1627                 self.token,
1628             )
1629         except JujuInvalidK8sConfiguration as e:
1630             exception_raised = True
1631             self.assertEqual(
1632                 e.message,
1633                 "authentication method not supported",
1634             )
1635         self.assertTrue(exception_raised)
1636
1637     def test_user_pass(self, mock_configuration):
1638         mock_configuration.username = "admin"
1639         mock_configuration.password = "admin"
1640         mock_configuration.ssl_ca_cert = None
1641         mock_configuration.cert_file = None
1642         mock_configuration.key_file = None
1643         self.token = None
1644         self.cert_data = None
1645         credential = self.libjuju.get_k8s_cloud_credential(
1646             mock_configuration,
1647             self.cert_data,
1648             self.token,
1649         )
1650         self.assertEqual(
1651             credential,
1652             juju.client._definitions.CloudCredential(
1653                 attrs={"username": "admin", "password": "admin"}, auth_type="userpass"
1654             ),
1655         )
1656
1657     def test_user_pass_with_cert(self, mock_configuration):
1658         mock_configuration.username = "admin"
1659         mock_configuration.password = "admin"
1660         mock_configuration.ssl_ca_cert = None
1661         mock_configuration.cert_file = None
1662         mock_configuration.key_file = None
1663         self.token = None
1664         credential = self.libjuju.get_k8s_cloud_credential(
1665             mock_configuration,
1666             self.cert_data,
1667             self.token,
1668         )
1669         self.assertEqual(
1670             credential,
1671             juju.client._definitions.CloudCredential(
1672                 attrs={
1673                     "ClientCertificateData": self.cert_data,
1674                     "username": "admin",
1675                     "password": "admin",
1676                 },
1677                 auth_type="userpasswithcert",
1678             ),
1679         )
1680
1681     def test_user_no_pass(self, mock_configuration):
1682         mock_configuration.username = "admin"
1683         mock_configuration.password = ""
1684         mock_configuration.ssl_ca_cert = None
1685         mock_configuration.cert_file = None
1686         mock_configuration.key_file = None
1687         self.token = None
1688         self.cert_data = None
1689         with mock.patch.object(self.libjuju.log, "debug") as mock_debug:
1690             credential = self.libjuju.get_k8s_cloud_credential(
1691                 mock_configuration,
1692                 self.cert_data,
1693                 self.token,
1694             )
1695             self.assertEqual(
1696                 credential,
1697                 juju.client._definitions.CloudCredential(
1698                     attrs={"username": "admin", "password": ""}, auth_type="userpass"
1699                 ),
1700             )
1701             mock_debug.assert_called_once_with(
1702                 "credential for user admin has empty password"
1703             )
1704
1705     def test_cert(self, mock_configuration):
1706         mock_configuration.username = ""
1707         mock_configuration.password = ""
1708         mock_configuration.api_key = {"authorization": "Bearer Token"}
1709         ssl_ca_cert = tempfile.NamedTemporaryFile()
1710         with open(ssl_ca_cert.name, "w") as ssl_ca_cert_file:
1711             ssl_ca_cert_file.write("cacert")
1712         mock_configuration.ssl_ca_cert = ssl_ca_cert.name
1713         mock_configuration.cert_file = None
1714         mock_configuration.key_file = None
1715         credential = self.libjuju.get_k8s_cloud_credential(
1716             mock_configuration,
1717             self.cert_data,
1718             self.token,
1719         )
1720         self.assertEqual(
1721             credential,
1722             juju.client._definitions.CloudCredential(
1723                 attrs={"ClientCertificateData": self.cert_data, "Token": self.token},
1724                 auth_type="certificate",
1725             ),
1726         )
1727
1728     # TODO: Fix this test when oauth authentication is supported
1729     # def test_oauth2(self, mock_configuration):
1730     #     mock_configuration.username = ""
1731     #     mock_configuration.password = ""
1732     #     mock_configuration.api_key = {"authorization": "Bearer Token"}
1733     #     key = tempfile.NamedTemporaryFile()
1734     #     with open(key.name, "w") as key_file:
1735     #         key_file.write("key")
1736     #     mock_configuration.ssl_ca_cert = None
1737     #     mock_configuration.cert_file = None
1738     #     mock_configuration.key_file = key.name
1739     #     credential = self.libjuju.get_k8s_cloud_credential(
1740     #         mock_configuration,
1741     #         self.cert_data,
1742     #         self.token,
1743     #     )
1744     #     self.assertEqual(
1745     #         credential,
1746     #         juju.client._definitions.CloudCredential(
1747     #             attrs={"ClientKeyData": "key", "Token": "Token"},
1748     #             auth_type="oauth2",
1749     #         ),
1750     #     )
1751
1752     # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1753     # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1754     #     mock_configuration.username = ""
1755     #     mock_configuration.password = ""
1756     #     key = tempfile.NamedTemporaryFile()
1757     #     with open(key.name, "w") as key_file:
1758     #         key_file.write("key")
1759     #     mock_configuration.ssl_ca_cert = None
1760     #     mock_configuration.cert_file = None
1761     #     mock_configuration.key_file = key.name
1762     #     exception_raised = False
1763     #     try:
1764     #         _ = self.libjuju.get_k8s_cloud_credential(
1765     #             mock_configuration,
1766     #             self.cert_data,
1767     #             self.token,
1768     #         )
1769     #     except JujuInvalidK8sConfiguration as e:
1770     #         exception_raised = True
1771     #         self.assertEqual(
1772     #             e.message,
1773     #             "missing token for auth type oauth2",
1774     #         )
1775     #     self.assertTrue(exception_raised)
1776
1777     def test_exception_cannot_set_token_and_userpass(self, mock_configuration):
1778         mock_configuration.username = "admin"
1779         mock_configuration.password = "pass"
1780         mock_configuration.api_key = {"authorization": "No_bearer_token"}
1781         mock_configuration.ssl_ca_cert = None
1782         mock_configuration.cert_file = None
1783         mock_configuration.key_file = None
1784         exception_raised = False
1785         try:
1786             _ = self.libjuju.get_k8s_cloud_credential(
1787                 mock_configuration,
1788                 self.cert_data,
1789                 self.token,
1790             )
1791         except JujuInvalidK8sConfiguration as e:
1792             exception_raised = True
1793             self.assertEqual(
1794                 e.message,
1795                 "Cannot set both token and user/pass",
1796             )
1797         self.assertTrue(exception_raised)