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("n2vc.libjuju.Libjuju.get_controller")
1181 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
1182 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1183 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
1184 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
1185 class DestroyApplicationTest(LibjujuTestCase):
1186     def setUp(self):
1187         super(DestroyApplicationTest, self).setUp()
1188
1189     def test_success(
1190             self,
1191             mock_get_controller,
1192             mock_get_model,
1193             mock_disconnect_controller,
1194             mock_get_application,
1195             mock_disconnect_model,
1196     ):
1197         mock_get_application.return_value = FakeApplication()
1198         mock_get_model.return_value = None
1199         self.loop.run_until_complete(
1200             self.libjuju.destroy_application(
1201                 "existing_model",
1202                 "existing_app",
1203                 3600,
1204             )
1205         )
1206         mock_get_application.assert_called()
1207         mock_disconnect_controller.assert_called_once()
1208         mock_disconnect_model.assert_called_once()
1209
1210     def test_no_application(
1211             self,
1212             mock_get_controller,
1213             mock_get_model,
1214             mock_disconnect_controller,
1215             mock_get_application,
1216             mock_disconnect_model,
1217     ):
1218         mock_get_model.return_value = None
1219         mock_get_application.return_value = None
1220
1221         self.loop.run_until_complete(
1222             self.libjuju.destroy_application(
1223                 "existing_model",
1224                 "existing_app",
1225                 3600,
1226             )
1227         )
1228         mock_get_application.assert_called()
1229
1230     def test_exception(
1231             self,
1232             mock_get_controller,
1233             mock_get_model,
1234             mock_disconnect_controller,
1235             mock_get_application,
1236             mock_disconnect_model,
1237     ):
1238         mock_get_application.return_value = FakeApplication
1239         mock_get_model.return_value = None
1240
1241         with self.assertRaises(Exception):
1242             self.loop.run_until_complete(
1243                 self.libjuju.destroy_application(
1244                     "existing_model",
1245                     "existing_app",
1246                     0,
1247                 )
1248             )
1249             mock_get_application.assert_called_once()
1250
1251
1252 # @asynctest.mock.patch("juju.model.Model.get_machines")
1253 # @asynctest.mock.patch("logging.Logger.debug")
1254 # class DestroyMachineTest(LibjujuTestCase):
1255 #     def setUp(self):
1256 #         super(DestroyMachineTest, self).setUp()
1257
1258 #     def test_success_manual_machine(
1259 #         self, mock_debug, mock_get_machines,
1260 #     ):
1261 #         mock_get_machines.side_effect = [
1262 #             {"machine": FakeManualMachine()},
1263 #             {"machine": FakeManualMachine()},
1264 #             {},
1265 #         ]
1266 #         self.loop.run_until_complete(
1267 #             self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1268 #         )
1269 #         calls = [
1270 #             asynctest.call("Waiting for machine machine is destroyed"),
1271 #             asynctest.call("Machine destroyed: machine"),
1272 #         ]
1273 #         mock_debug.assert_has_calls(calls)
1274
1275 #     def test_no_machine(
1276 #         self, mock_debug, mock_get_machines,
1277 #     ):
1278 #         mock_get_machines.return_value = {}
1279 #         self.loop.run_until_complete(
1280 #             self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1281 #         )
1282 #         mock_debug.assert_called_with("Machine not found: machine")
1283
1284
1285 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1286 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
1287 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
1288 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1289 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
1290 class ConfigureApplicationTest(LibjujuTestCase):
1291     def setUp(self):
1292         super(ConfigureApplicationTest, self).setUp()
1293
1294     def test_success(
1295         self,
1296         mock_get_application,
1297         mock_disconnect_controller,
1298         mock_disconnect_model,
1299         mock_get_model,
1300         mock_get_controller,
1301     ):
1302
1303         mock_get_application.return_value = FakeApplication()
1304
1305         self.loop.run_until_complete(
1306             self.libjuju.configure_application(
1307                 "model",
1308                 "app",
1309                 {"config"},
1310             )
1311         )
1312         mock_get_application.assert_called_once()
1313         mock_disconnect_controller.assert_called_once()
1314         mock_disconnect_model.assert_called_once()
1315
1316     def test_exception(
1317         self,
1318         mock_get_application,
1319         mock_disconnect_controller,
1320         mock_disconnect_model,
1321         mock_get_model,
1322         mock_get_controller,
1323     ):
1324
1325         mock_get_application.side_effect = Exception()
1326
1327         with self.assertRaises(Exception):
1328             self.loop.run_until_complete(
1329                 self.libjuju.configure_application(
1330                     "model",
1331                     "app",
1332                     {"config"},
1333                 )
1334             )
1335         mock_disconnect_controller.assert_called_once()
1336         mock_disconnect_model.assert_called_once()
1337
1338     def test_controller_exception(
1339         self,
1340         mock_get_application,
1341         mock_disconnect_controller,
1342         mock_disconnect_model,
1343         mock_get_model,
1344         mock_get_controller,
1345     ):
1346
1347         result = {"error": "not found", "response": "response", "request-id": 1}
1348
1349         mock_get_controller.side_effect = JujuAPIError(result)
1350
1351         with self.assertRaises(JujuAPIError):
1352             self.loop.run_until_complete(
1353                 self.libjuju.configure_application(
1354                     "model",
1355                     "app",
1356                     {"config"},
1357                 )
1358             )
1359         mock_get_model.assert_not_called()
1360         mock_disconnect_controller.assert_not_called()
1361         mock_disconnect_model.assert_not_called()
1362
1363     def test_get_model_exception(
1364         self,
1365         mock_get_application,
1366         mock_disconnect_controller,
1367         mock_disconnect_model,
1368         mock_get_model,
1369         mock_get_controller,
1370     ):
1371
1372         result = {"error": "not found", "response": "response", "request-id": 1}
1373         mock_get_model.side_effect = JujuAPIError(result)
1374
1375         with self.assertRaises(JujuAPIError):
1376             self.loop.run_until_complete(
1377                 self.libjuju.configure_application(
1378                     "model",
1379                     "app",
1380                     {"config"},
1381                 )
1382             )
1383         mock_get_model.assert_called_once()
1384         mock_disconnect_controller.assert_called_once()
1385         mock_disconnect_model.assert_not_called()
1386
1387
1388 # TODO _get_api_endpoints_db test case
1389 # TODO _update_api_endpoints_db test case
1390 # TODO healthcheck test case
1391
1392
1393 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1394 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1395 @asynctest.mock.patch("juju.controller.Controller.list_models")
1396 class ListModelsTest(LibjujuTestCase):
1397     def setUp(self):
1398         super(ListModelsTest, self).setUp()
1399
1400     def test_containing(
1401         self,
1402         mock_list_models,
1403         mock_disconnect_controller,
1404         mock_get_controller,
1405     ):
1406         mock_get_controller.return_value = juju.controller.Controller()
1407         mock_list_models.return_value = ["existingmodel"]
1408         models = self.loop.run_until_complete(self.libjuju.list_models("existing"))
1409
1410         mock_disconnect_controller.assert_called_once()
1411         self.assertEquals(models, ["existingmodel"])
1412
1413     def test_not_containing(
1414         self,
1415         mock_list_models,
1416         mock_disconnect_controller,
1417         mock_get_controller,
1418     ):
1419         mock_get_controller.return_value = juju.controller.Controller()
1420         mock_list_models.return_value = ["existingmodel", "model"]
1421         models = self.loop.run_until_complete(self.libjuju.list_models("mdl"))
1422
1423         mock_disconnect_controller.assert_called_once()
1424         self.assertEquals(models, [])
1425
1426     def test_no_contains_arg(
1427         self,
1428         mock_list_models,
1429         mock_disconnect_controller,
1430         mock_get_controller,
1431     ):
1432         mock_get_controller.return_value = juju.controller.Controller()
1433         mock_list_models.return_value = ["existingmodel", "model"]
1434         models = self.loop.run_until_complete(self.libjuju.list_models())
1435
1436         mock_disconnect_controller.assert_called_once()
1437         self.assertEquals(models, ["existingmodel", "model"])
1438
1439
1440 @asynctest.mock.patch("n2vc.libjuju.Libjuju.list_models")
1441 class ModelsExistTest(LibjujuTestCase):
1442     def setUp(self):
1443         super(ModelsExistTest, self).setUp()
1444
1445     def test_model_names_none(self, mock_list_models):
1446         mock_list_models.return_value = []
1447         with self.assertRaises(Exception):
1448             self.loop.run_until_complete(self.libjuju.models_exist(None))
1449
1450     def test_model_names_empty(self, mock_list_models):
1451         mock_list_models.return_value = []
1452         with self.assertRaises(Exception):
1453             (exist, non_existing_models) = self.loop.run_until_complete(
1454                 self.libjuju.models_exist([])
1455             )
1456
1457     def test_model_names_not_existing(self, mock_list_models):
1458         mock_list_models.return_value = ["prometheus", "grafana"]
1459         (exist, non_existing_models) = self.loop.run_until_complete(
1460             self.libjuju.models_exist(["prometheus2", "grafana"])
1461         )
1462         self.assertFalse(exist)
1463         self.assertEqual(non_existing_models, ["prometheus2"])
1464
1465     def test_model_names_exist(self, mock_list_models):
1466         mock_list_models.return_value = ["prometheus", "grafana"]
1467         (exist, non_existing_models) = self.loop.run_until_complete(
1468             self.libjuju.models_exist(["prometheus", "grafana"])
1469         )
1470         self.assertTrue(exist)
1471         self.assertEqual(non_existing_models, [])
1472
1473
1474 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1475 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1476 @asynctest.mock.patch("juju.controller.Controller.list_offers")
1477 class ListOffers(LibjujuTestCase):
1478     def setUp(self):
1479         super(ListOffers, self).setUp()
1480
1481     def test_disconnect_controller(
1482         self,
1483         mock_list_offers,
1484         mock_disconnect_controller,
1485         mock_get_controller,
1486     ):
1487         mock_get_controller.return_value = juju.controller.Controller()
1488         mock_list_offers.side_effect = Exception()
1489         with self.assertRaises(Exception):
1490             self.loop.run_until_complete(self.libjuju.list_offers("model"))
1491         mock_disconnect_controller.assert_called_once()
1492
1493     def test_empty_list(
1494         self,
1495         mock_list_offers,
1496         mock_disconnect_controller,
1497         mock_get_controller,
1498     ):
1499         mock_get_controller.return_value = juju.controller.Controller()
1500         mock_list_offers.return_value = []
1501         offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
1502         self.assertEqual(offers, [])
1503         mock_disconnect_controller.assert_called_once()
1504
1505     def test_non_empty_list(
1506         self,
1507         mock_list_offers,
1508         mock_disconnect_controller,
1509         mock_get_controller,
1510     ):
1511         mock_get_controller.return_value = juju.controller.Controller()
1512         mock_list_offers.return_value = ["offer"]
1513         offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
1514         self.assertEqual(offers, ["offer"])
1515         mock_disconnect_controller.assert_called_once()
1516
1517
1518 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1519 @asynctest.mock.patch("juju.controller.Controller.get_model")
1520 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
1521 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1522 @asynctest.mock.patch("juju.model.Model.consume")
1523 class ConsumeTest(LibjujuTestCase):
1524     def setUp(self):
1525         super(ConsumeTest, self).setUp()
1526
1527     def test_consume(
1528         self,
1529         mock_consume,
1530         mock_disconnect_controller,
1531         mock_disconnect_model,
1532         mock_get_model,
1533         mock_get_controller,
1534     ):
1535         mock_get_controller.return_value = juju.controller.Controller()
1536         mock_get_model.return_value = juju.model.Model()
1537
1538         self.loop.run_until_complete(self.libjuju.consume("offer_url", "model_name"))
1539         mock_consume.assert_called_once()
1540         mock_disconnect_model.assert_called_once()
1541         mock_disconnect_controller.assert_called_once()
1542
1543     def test_parsing_error_exception(
1544         self,
1545         mock_consume,
1546         mock_disconnect_controller,
1547         mock_disconnect_model,
1548         mock_get_model,
1549         mock_get_controller,
1550     ):
1551         mock_get_controller.return_value = juju.controller.Controller()
1552         mock_get_model.return_value = juju.model.Model()
1553         mock_consume.side_effect = juju.offerendpoints.ParseError("")
1554
1555         with self.assertRaises(juju.offerendpoints.ParseError):
1556             self.loop.run_until_complete(
1557                 self.libjuju.consume("offer_url", "model_name")
1558             )
1559         mock_consume.assert_called_once()
1560         mock_disconnect_model.assert_called_once()
1561         mock_disconnect_controller.assert_called_once()
1562
1563     def test_juju_error_exception(
1564         self,
1565         mock_consume,
1566         mock_disconnect_controller,
1567         mock_disconnect_model,
1568         mock_get_model,
1569         mock_get_controller,
1570     ):
1571         mock_get_controller.return_value = juju.controller.Controller()
1572         mock_get_model.return_value = juju.model.Model()
1573         mock_consume.side_effect = juju.errors.JujuError("")
1574
1575         with self.assertRaises(juju.errors.JujuError):
1576             self.loop.run_until_complete(
1577                 self.libjuju.consume("offer_url", "model_name")
1578             )
1579         mock_consume.assert_called_once()
1580         mock_disconnect_model.assert_called_once()
1581         mock_disconnect_controller.assert_called_once()
1582
1583     def test_juju_api_error_exception(
1584         self,
1585         mock_consume,
1586         mock_disconnect_controller,
1587         mock_disconnect_model,
1588         mock_get_model,
1589         mock_get_controller,
1590     ):
1591         mock_get_controller.return_value = juju.controller.Controller()
1592         mock_get_model.return_value = juju.model.Model()
1593         mock_consume.side_effect = juju.errors.JujuAPIError(
1594             {"error": "", "response": "", "request-id": ""}
1595         )
1596
1597         with self.assertRaises(juju.errors.JujuAPIError):
1598             self.loop.run_until_complete(
1599                 self.libjuju.consume("offer_url", "model_name")
1600             )
1601         mock_consume.assert_called_once()
1602         mock_disconnect_model.assert_called_once()
1603         mock_disconnect_controller.assert_called_once()
1604
1605
1606 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1607 @asynctest.mock.patch("n2vc.libjuju.Libjuju.add_cloud")
1608 class AddK8sTest(LibjujuTestCase):
1609     def setUp(self):
1610         super(AddK8sTest, self).setUp()
1611         name = "cloud"
1612         rbac_id = generate_rbac_id()
1613         token = "token"
1614         client_cert_data = "cert"
1615         configuration = kubernetes.client.configuration.Configuration()
1616         storage_class = "storage_class"
1617         credential_name = name
1618
1619         self._add_k8s_args = {
1620             "name": name,
1621             "rbac_id": rbac_id,
1622             "token": token,
1623             "client_cert_data": client_cert_data,
1624             "configuration": configuration,
1625             "storage_class": storage_class,
1626             "credential_name": credential_name,
1627         }
1628
1629     def test_add_k8s(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1630         self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1631         mock_add_cloud.assert_called_once()
1632         mock_get_k8s_cloud_credential.assert_called_once()
1633
1634     def test_add_k8s_exception(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1635         mock_add_cloud.side_effect = Exception()
1636         with self.assertRaises(Exception):
1637             self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1638         mock_add_cloud.assert_called_once()
1639         mock_get_k8s_cloud_credential.assert_called_once()
1640
1641     def test_add_k8s_missing_name(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1642         self._add_k8s_args["name"] = ""
1643         with self.assertRaises(Exception):
1644             self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1645         mock_add_cloud.assert_not_called()
1646
1647     def test_add_k8s_missing_storage_name(
1648         self, mock_add_cloud, mock_get_k8s_cloud_credential
1649     ):
1650         self._add_k8s_args["storage_class"] = ""
1651         with self.assertRaises(Exception):
1652             self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1653         mock_add_cloud.assert_not_called()
1654
1655     def test_add_k8s_missing_configuration_keys(
1656         self, mock_add_cloud, mock_get_k8s_cloud_credential
1657     ):
1658         self._add_k8s_args["configuration"] = None
1659         with self.assertRaises(Exception):
1660             self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1661         mock_add_cloud.assert_not_called()
1662
1663
1664 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1665 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1666 @asynctest.mock.patch("juju.controller.Controller.add_cloud")
1667 @asynctest.mock.patch("juju.controller.Controller.add_credential")
1668 class AddCloudTest(LibjujuTestCase):
1669     def setUp(self):
1670         super(AddCloudTest, self).setUp()
1671         self.cloud = juju.client.client.Cloud()
1672         self.credential = juju.client.client.CloudCredential()
1673
1674     def test_add_cloud_with_credential(
1675         self,
1676         mock_add_credential,
1677         mock_add_cloud,
1678         mock_disconnect_controller,
1679         mock_get_controller,
1680     ):
1681         mock_get_controller.return_value = juju.controller.Controller()
1682
1683         cloud = self.loop.run_until_complete(
1684             self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1685         )
1686         self.assertEqual(cloud, self.cloud)
1687         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1688         mock_add_credential.assert_called_once_with(
1689             "cloud", credential=self.credential, cloud="cloud"
1690         )
1691         mock_disconnect_controller.assert_called_once()
1692
1693     def test_add_cloud_no_credential(
1694         self,
1695         mock_add_credential,
1696         mock_add_cloud,
1697         mock_disconnect_controller,
1698         mock_get_controller,
1699     ):
1700         mock_get_controller.return_value = juju.controller.Controller()
1701
1702         cloud = self.loop.run_until_complete(
1703             self.libjuju.add_cloud("cloud", self.cloud)
1704         )
1705         self.assertEqual(cloud, self.cloud)
1706         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1707         mock_add_credential.assert_not_called()
1708         mock_disconnect_controller.assert_called_once()
1709
1710     def test_add_cloud_exception(
1711         self,
1712         mock_add_credential,
1713         mock_add_cloud,
1714         mock_disconnect_controller,
1715         mock_get_controller,
1716     ):
1717         mock_get_controller.return_value = juju.controller.Controller()
1718         mock_add_cloud.side_effect = Exception()
1719         with self.assertRaises(Exception):
1720             self.loop.run_until_complete(
1721                 self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1722             )
1723
1724         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1725         mock_add_credential.assert_not_called()
1726         mock_disconnect_controller.assert_called_once()
1727
1728     def test_add_credential_exception(
1729         self,
1730         mock_add_credential,
1731         mock_add_cloud,
1732         mock_disconnect_controller,
1733         mock_get_controller,
1734     ):
1735         mock_get_controller.return_value = juju.controller.Controller()
1736         mock_add_credential.side_effect = Exception()
1737         with self.assertRaises(Exception):
1738             self.loop.run_until_complete(
1739                 self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1740             )
1741
1742         mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1743         mock_add_credential.assert_called_once_with(
1744             "cloud", credential=self.credential, cloud="cloud"
1745         )
1746         mock_disconnect_controller.assert_called_once()
1747
1748
1749 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1750 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1751 @asynctest.mock.patch("juju.controller.Controller.remove_cloud")
1752 class RemoveCloudTest(LibjujuTestCase):
1753     def setUp(self):
1754         super(RemoveCloudTest, self).setUp()
1755
1756     def test_remove_cloud(
1757         self,
1758         mock_remove_cloud,
1759         mock_disconnect_controller,
1760         mock_get_controller,
1761     ):
1762         mock_get_controller.return_value = juju.controller.Controller()
1763
1764         self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
1765         mock_remove_cloud.assert_called_once_with("cloud")
1766         mock_disconnect_controller.assert_called_once()
1767
1768     def test_remove_cloud_exception(
1769         self,
1770         mock_remove_cloud,
1771         mock_disconnect_controller,
1772         mock_get_controller,
1773     ):
1774         mock_get_controller.return_value = juju.controller.Controller()
1775         mock_remove_cloud.side_effect = Exception()
1776
1777         with self.assertRaises(Exception):
1778             self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
1779         mock_remove_cloud.assert_called_once_with("cloud")
1780         mock_disconnect_controller.assert_called_once()
1781
1782
1783 @asynctest.mock.patch("kubernetes.client.configuration.Configuration")
1784 class GetK8sCloudCredentials(LibjujuTestCase):
1785     def setUp(self):
1786         super(GetK8sCloudCredentials, self).setUp()
1787         self.cert_data = "cert"
1788         self.token = "token"
1789
1790     @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1791     def test_not_supported(self, mock_exception, mock_configuration):
1792         mock_configuration.username = ""
1793         mock_configuration.password = ""
1794         mock_configuration.ssl_ca_cert = None
1795         mock_configuration.cert_file = None
1796         mock_configuration.key_file = None
1797         exception_raised = False
1798         self.token = None
1799         self.cert_data = None
1800         try:
1801             _ = self.libjuju.get_k8s_cloud_credential(
1802                 mock_configuration,
1803                 self.cert_data,
1804                 self.token,
1805             )
1806         except JujuInvalidK8sConfiguration as e:
1807             exception_raised = True
1808             self.assertEqual(
1809                 e.message,
1810                 "authentication method not supported",
1811             )
1812         self.assertTrue(exception_raised)
1813
1814     def test_user_pass(self, mock_configuration):
1815         mock_configuration.username = "admin"
1816         mock_configuration.password = "admin"
1817         mock_configuration.ssl_ca_cert = None
1818         mock_configuration.cert_file = None
1819         mock_configuration.key_file = None
1820         self.token = None
1821         self.cert_data = None
1822         credential = self.libjuju.get_k8s_cloud_credential(
1823             mock_configuration,
1824             self.cert_data,
1825             self.token,
1826         )
1827         self.assertEqual(
1828             credential,
1829             juju.client._definitions.CloudCredential(
1830                 attrs={"username": "admin", "password": "admin"}, auth_type="userpass"
1831             ),
1832         )
1833
1834     def test_user_pass_with_cert(self, mock_configuration):
1835         mock_configuration.username = "admin"
1836         mock_configuration.password = "admin"
1837         mock_configuration.ssl_ca_cert = None
1838         mock_configuration.cert_file = None
1839         mock_configuration.key_file = None
1840         self.token = None
1841         credential = self.libjuju.get_k8s_cloud_credential(
1842             mock_configuration,
1843             self.cert_data,
1844             self.token,
1845         )
1846         self.assertEqual(
1847             credential,
1848             juju.client._definitions.CloudCredential(
1849                 attrs={
1850                     "ClientCertificateData": self.cert_data,
1851                     "username": "admin",
1852                     "password": "admin",
1853                 },
1854                 auth_type="userpasswithcert",
1855             ),
1856         )
1857
1858     def test_user_no_pass(self, mock_configuration):
1859         mock_configuration.username = "admin"
1860         mock_configuration.password = ""
1861         mock_configuration.ssl_ca_cert = None
1862         mock_configuration.cert_file = None
1863         mock_configuration.key_file = None
1864         self.token = None
1865         self.cert_data = None
1866         with mock.patch.object(self.libjuju.log, "debug") as mock_debug:
1867             credential = self.libjuju.get_k8s_cloud_credential(
1868                 mock_configuration,
1869                 self.cert_data,
1870                 self.token,
1871             )
1872             self.assertEqual(
1873                 credential,
1874                 juju.client._definitions.CloudCredential(
1875                     attrs={"username": "admin", "password": ""}, auth_type="userpass"
1876                 ),
1877             )
1878             mock_debug.assert_called_once_with(
1879                 "credential for user admin has empty password"
1880             )
1881
1882     def test_cert(self, mock_configuration):
1883         mock_configuration.username = ""
1884         mock_configuration.password = ""
1885         mock_configuration.api_key = {"authorization": "Bearer Token"}
1886         ssl_ca_cert = tempfile.NamedTemporaryFile()
1887         with open(ssl_ca_cert.name, "w") as ssl_ca_cert_file:
1888             ssl_ca_cert_file.write("cacert")
1889         mock_configuration.ssl_ca_cert = ssl_ca_cert.name
1890         mock_configuration.cert_file = None
1891         mock_configuration.key_file = None
1892         credential = self.libjuju.get_k8s_cloud_credential(
1893             mock_configuration,
1894             self.cert_data,
1895             self.token,
1896         )
1897         self.assertEqual(
1898             credential,
1899             juju.client._definitions.CloudCredential(
1900                 attrs={"ClientCertificateData": self.cert_data, "Token": self.token},
1901                 auth_type="certificate",
1902             ),
1903         )
1904
1905     # TODO: Fix this test when oauth authentication is supported
1906     # def test_oauth2(self, mock_configuration):
1907     #     mock_configuration.username = ""
1908     #     mock_configuration.password = ""
1909     #     mock_configuration.api_key = {"authorization": "Bearer Token"}
1910     #     key = tempfile.NamedTemporaryFile()
1911     #     with open(key.name, "w") as key_file:
1912     #         key_file.write("key")
1913     #     mock_configuration.ssl_ca_cert = None
1914     #     mock_configuration.cert_file = None
1915     #     mock_configuration.key_file = key.name
1916     #     credential = self.libjuju.get_k8s_cloud_credential(
1917     #         mock_configuration,
1918     #         self.cert_data,
1919     #         self.token,
1920     #     )
1921     #     self.assertEqual(
1922     #         credential,
1923     #         juju.client._definitions.CloudCredential(
1924     #             attrs={"ClientKeyData": "key", "Token": "Token"},
1925     #             auth_type="oauth2",
1926     #         ),
1927     #     )
1928
1929     # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1930     # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1931     #     mock_configuration.username = ""
1932     #     mock_configuration.password = ""
1933     #     key = tempfile.NamedTemporaryFile()
1934     #     with open(key.name, "w") as key_file:
1935     #         key_file.write("key")
1936     #     mock_configuration.ssl_ca_cert = None
1937     #     mock_configuration.cert_file = None
1938     #     mock_configuration.key_file = key.name
1939     #     exception_raised = False
1940     #     try:
1941     #         _ = self.libjuju.get_k8s_cloud_credential(
1942     #             mock_configuration,
1943     #             self.cert_data,
1944     #             self.token,
1945     #         )
1946     #     except JujuInvalidK8sConfiguration as e:
1947     #         exception_raised = True
1948     #         self.assertEqual(
1949     #             e.message,
1950     #             "missing token for auth type oauth2",
1951     #         )
1952     #     self.assertTrue(exception_raised)
1953
1954     def test_exception_cannot_set_token_and_userpass(self, mock_configuration):
1955         mock_configuration.username = "admin"
1956         mock_configuration.password = "pass"
1957         mock_configuration.api_key = {"authorization": "No_bearer_token"}
1958         mock_configuration.ssl_ca_cert = None
1959         mock_configuration.cert_file = None
1960         mock_configuration.key_file = None
1961         exception_raised = False
1962         try:
1963             _ = self.libjuju.get_k8s_cloud_credential(
1964                 mock_configuration,
1965                 self.cert_data,
1966                 self.token,
1967             )
1968         except JujuInvalidK8sConfiguration as e:
1969             exception_raised = True
1970             self.assertEqual(
1971                 e.message,
1972                 "Cannot set both token and user/pass",
1973             )
1974         self.assertTrue(exception_raised)