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