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