Jenkins refresh
[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.mock import Mock, patch
19 import juju
20 import kubernetes
21 from juju.errors import JujuAPIError
22 import logging
23
24 from n2vc.definitions import Offer, RelationEndpoint
25 from .utils import (
26 FakeApplication,
27 FakeMachine,
28 FakeManualMachine,
29 FakeUnit,
30 )
31 from n2vc.libjuju import Libjuju
32 from n2vc.exceptions import (
33 JujuControllerFailedConnecting,
34 JujuMachineNotFound,
35 JujuApplicationNotFound,
36 JujuActionNotFound,
37 JujuApplicationExists,
38 JujuInvalidK8sConfiguration,
39 JujuLeaderUnitNotFound,
40 JujuError,
41 )
42 from n2vc.k8s_juju_conn import generate_rbac_id
43 from n2vc.tests.unit.utils import AsyncMock
44 from n2vc.vca.connection import Connection
45 from n2vc.vca.connection_data import ConnectionData
46
47
48 cacert = """-----BEGIN CERTIFICATE-----
49 SOMECERT
50 -----END CERTIFICATE-----"""
51
52
53 @asynctest.mock.patch("n2vc.libjuju.Controller")
54 class LibjujuTestCase(asynctest.TestCase):
55 @asynctest.mock.patch("n2vc.vca.connection_data.base64_to_cacert")
56 def setUp(
57 self,
58 mock_base64_to_cacert=None,
59 ):
60 self.loop = asyncio.get_event_loop()
61 self.db = Mock()
62 mock_base64_to_cacert.return_value = cacert
63 # Connection._load_vca_connection_data = Mock()
64 vca_connection = Connection(AsyncMock())
65 vca_connection._data = ConnectionData(
66 **{
67 "endpoints": ["1.2.3.4:17070"],
68 "user": "user",
69 "secret": "secret",
70 "cacert": "cacert",
71 "pubkey": "pubkey",
72 "lxd-cloud": "cloud",
73 "lxd-credentials": "credentials",
74 "k8s-cloud": "k8s_cloud",
75 "k8s-credentials": "k8s_credentials",
76 "model-config": {},
77 "api-proxy": "api_proxy",
78 }
79 )
80 logging.disable(logging.CRITICAL)
81 self.libjuju = Libjuju(vca_connection, self.loop)
82 self.loop.run_until_complete(self.libjuju.disconnect())
83
84
85 @asynctest.mock.patch("juju.controller.Controller.connect")
86 @asynctest.mock.patch(
87 "juju.controller.Controller.api_endpoints",
88 new_callable=asynctest.CoroutineMock(return_value=["127.0.0.1:17070"]),
89 )
90 class GetControllerTest(LibjujuTestCase):
91 def setUp(self):
92 super(GetControllerTest, self).setUp()
93
94 def test_diff_endpoint(self, mock_api_endpoints, mock_connect):
95 self.libjuju.endpoints = []
96 controller = self.loop.run_until_complete(self.libjuju.get_controller())
97 self.assertIsInstance(controller, juju.controller.Controller)
98
99 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
100 def test_exception(
101 self,
102 mock_disconnect_controller,
103 mock_api_endpoints,
104 mock_connect,
105 ):
106 self.libjuju.endpoints = []
107
108 mock_connect.side_effect = Exception()
109 controller = None
110 with self.assertRaises(JujuControllerFailedConnecting):
111 controller = self.loop.run_until_complete(self.libjuju.get_controller())
112 self.assertIsNone(controller)
113 mock_disconnect_controller.assert_called()
114
115 def test_same_endpoint_get_controller(self, mock_api_endpoints, mock_connect):
116 self.libjuju.endpoints = ["127.0.0.1:17070"]
117 controller = self.loop.run_until_complete(self.libjuju.get_controller())
118 self.assertIsInstance(controller, juju.controller.Controller)
119
120
121 class DisconnectTest(LibjujuTestCase):
122 def setUp(self):
123 super(DisconnectTest, self).setUp()
124
125 @asynctest.mock.patch("juju.model.Model.disconnect")
126 def test_disconnect_model(self, mock_disconnect):
127 self.loop.run_until_complete(self.libjuju.disconnect_model(juju.model.Model()))
128 mock_disconnect.assert_called_once()
129
130 @asynctest.mock.patch("juju.controller.Controller.disconnect")
131 def test_disconnect_controller(self, mock_disconnect):
132 self.loop.run_until_complete(
133 self.libjuju.disconnect_controller(juju.controller.Controller())
134 )
135 mock_disconnect.assert_called_once()
136
137
138 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
139 @asynctest.mock.patch("n2vc.libjuju.Libjuju.model_exists")
140 @asynctest.mock.patch("juju.controller.Controller.add_model")
141 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
142 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
143 class AddModelTest(LibjujuTestCase):
144 def setUp(self):
145 super(AddModelTest, self).setUp()
146
147 def test_existing_model(
148 self,
149 mock_disconnect_model,
150 mock_disconnect_controller,
151 mock_add_model,
152 mock_model_exists,
153 mock_get_controller,
154 ):
155 mock_model_exists.return_value = True
156
157 # This should not raise an exception
158 self.loop.run_until_complete(self.libjuju.add_model("existing_model", "cloud"))
159
160 mock_disconnect_controller.assert_called()
161
162 # TODO Check two job executing at the same time and one returning without doing anything.
163
164 def test_non_existing_model(
165 self,
166 mock_disconnect_model,
167 mock_disconnect_controller,
168 mock_add_model,
169 mock_model_exists,
170 mock_get_controller,
171 ):
172 mock_model_exists.return_value = False
173 mock_get_controller.return_value = juju.controller.Controller()
174
175 self.loop.run_until_complete(
176 self.libjuju.add_model("nonexisting_model", Mock())
177 )
178
179 mock_add_model.assert_called_once()
180 mock_disconnect_controller.assert_called()
181 mock_disconnect_model.assert_called()
182
183
184 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
185 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
186 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
187 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
188 @asynctest.mock.patch(
189 "juju.model.Model.applications", new_callable=asynctest.PropertyMock
190 )
191 @asynctest.mock.patch("juju.model.Model.get_action_status")
192 @asynctest.mock.patch("juju.model.Model.get_action_output")
193 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_actions")
194 class GetExecutedActionsTest(LibjujuTestCase):
195 def setUp(self):
196 super(GetExecutedActionsTest, self).setUp()
197
198 def test_exception(
199 self,
200 mock_get_actions,
201 mock_get_action_output,
202 mock_get_action_status,
203 mock_applications,
204 mock_disconnect_controller,
205 mock_disconnect_model,
206 mock_get_model,
207 mock_get_controller,
208 ):
209 mock_get_model.return_value = None
210 with self.assertRaises(JujuError):
211 self.loop.run_until_complete(self.libjuju.get_executed_actions("model"))
212
213 mock_get_controller.assert_called_once()
214 mock_disconnect_controller.assert_called_once()
215 mock_get_model.assert_called_once()
216 mock_disconnect_model.assert_not_called()
217
218 def test_success(
219 self,
220 mock_get_actions,
221 mock_get_action_output,
222 mock_get_action_status,
223 mock_applications,
224 mock_disconnect_controller,
225 mock_disconnect_model,
226 mock_get_model,
227 mock_get_controller,
228 ):
229 mock_get_model.return_value = juju.model.Model()
230 mock_applications.return_value = {"existing_app"}
231 mock_get_actions.return_value = {"action_name": "description"}
232 mock_get_action_status.return_value = {"id": "status"}
233 mock_get_action_output.return_value = {"output": "completed"}
234
235 executed_actions = self.loop.run_until_complete(
236 self.libjuju.get_executed_actions("model")
237 )
238 expected_result = [
239 {
240 "id": "id",
241 "action": "action_name",
242 "status": "status",
243 "output": "completed",
244 }
245 ]
246 self.assertListEqual(expected_result, executed_actions)
247 self.assertIsInstance(executed_actions, list)
248
249 mock_get_controller.assert_called_once()
250 mock_get_model.assert_called_once()
251 mock_disconnect_controller.assert_called_once()
252 mock_disconnect_model.assert_called_once()
253
254
255 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
256 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
257 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
258 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
259 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
260 class GetApplicationConfigsTest(LibjujuTestCase):
261 def setUp(self):
262 super(GetApplicationConfigsTest, self).setUp()
263
264 def test_exception(
265 self,
266 mock_get_application,
267 mock_disconnect_controller,
268 mock_disconnect_model,
269 mock_get_model,
270 mock_get_controller,
271 ):
272 mock_get_model.return_value = None
273 with self.assertRaises(JujuError):
274 self.loop.run_until_complete(
275 self.libjuju.get_application_configs("model", "app")
276 )
277
278 mock_get_controller.assert_called_once()
279 mock_disconnect_controller.assert_called_once()
280 mock_get_model.assert_called_once()
281 mock_disconnect_model.assert_not_called()
282
283 def test_success(
284 self,
285 mock_get_application,
286 mock_disconnect_controller,
287 mock_disconnect_model,
288 mock_get_model,
289 mock_get_controller,
290 ):
291 mock_get_application.return_value = FakeApplication()
292 application_configs = self.loop.run_until_complete(
293 self.libjuju.get_application_configs("model", "app")
294 )
295
296 self.assertEqual(application_configs, ["app_config"])
297
298 mock_get_controller.assert_called_once()
299 mock_get_model.assert_called_once()
300 mock_disconnect_controller.assert_called_once()
301 mock_disconnect_model.assert_called_once()
302
303
304 @asynctest.mock.patch("juju.controller.Controller.get_model")
305 class GetModelTest(LibjujuTestCase):
306 def setUp(self):
307 super(GetModelTest, self).setUp()
308
309 def test_get_model(
310 self,
311 mock_get_model,
312 ):
313 mock_get_model.return_value = juju.model.Model()
314 model = self.loop.run_until_complete(
315 self.libjuju.get_model(juju.controller.Controller(), "model")
316 )
317 self.assertIsInstance(model, juju.model.Model)
318
319
320 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
321 @asynctest.mock.patch("juju.controller.Controller.list_models")
322 class ModelExistsTest(LibjujuTestCase):
323 def setUp(self):
324 super(ModelExistsTest, self).setUp()
325
326 async def test_existing_model(
327 self,
328 mock_list_models,
329 mock_get_controller,
330 ):
331 mock_list_models.return_value = ["existing_model"]
332 self.assertTrue(
333 await self.libjuju.model_exists(
334 "existing_model", juju.controller.Controller()
335 )
336 )
337
338 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
339 async def test_no_controller(
340 self,
341 mock_disconnect_controller,
342 mock_list_models,
343 mock_get_controller,
344 ):
345 mock_list_models.return_value = ["existing_model"]
346 mock_get_controller.return_value = juju.controller.Controller()
347 self.assertTrue(await self.libjuju.model_exists("existing_model"))
348 mock_disconnect_controller.assert_called_once()
349
350 async def test_non_existing_model(
351 self,
352 mock_list_models,
353 mock_get_controller,
354 ):
355 mock_list_models.return_value = ["existing_model"]
356 self.assertFalse(
357 await self.libjuju.model_exists(
358 "not_existing_model", juju.controller.Controller()
359 )
360 )
361
362
363 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
364 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
365 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
366 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
367 @asynctest.mock.patch("juju.model.Model.get_status")
368 class GetModelStatusTest(LibjujuTestCase):
369 def setUp(self):
370 super(GetModelStatusTest, self).setUp()
371
372 def test_success(
373 self,
374 mock_get_status,
375 mock_disconnect_controller,
376 mock_disconnect_model,
377 mock_get_model,
378 mock_get_controller,
379 ):
380 mock_get_model.return_value = juju.model.Model()
381 mock_get_status.return_value = {"status"}
382
383 status = self.loop.run_until_complete(self.libjuju.get_model_status("model"))
384
385 mock_get_status.assert_called_once()
386 mock_disconnect_controller.assert_called_once()
387 mock_disconnect_model.assert_called_once()
388
389 self.assertEqual(status, {"status"})
390
391 def test_exception(
392 self,
393 mock_get_status,
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_status.side_effect = Exception()
401 status = None
402 with self.assertRaises(Exception):
403 status = self.loop.run_until_complete(
404 self.libjuju.get_model_status("model")
405 )
406
407 mock_disconnect_controller.assert_called_once()
408 mock_disconnect_model.assert_called_once()
409
410 self.assertIsNone(status)
411
412
413 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
414 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
415 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
416 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
417 @asynctest.mock.patch("juju.model.Model.get_machines")
418 @asynctest.mock.patch("juju.model.Model.add_machine")
419 @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
420 class CreateMachineTest(LibjujuTestCase):
421 def setUp(self):
422 super(CreateMachineTest, self).setUp()
423
424 def test_existing_machine(
425 self,
426 mock_wait_for,
427 mock_add_machine,
428 mock_get_machines,
429 mock_disconnect_controller,
430 mock_disconnect_model,
431 mock_get_model,
432 mock_get_controller,
433 ):
434 mock_get_model.return_value = juju.model.Model()
435 mock_get_machines.return_value = {"existing_machine": FakeMachine()}
436 machine, bool_res = self.loop.run_until_complete(
437 self.libjuju.create_machine("model", "existing_machine")
438 )
439
440 self.assertIsInstance(machine, FakeMachine)
441 self.assertFalse(bool_res)
442
443 mock_disconnect_controller.assert_called()
444 mock_disconnect_model.assert_called()
445
446 def test_non_existing_machine(
447 self,
448 mock_wait_for,
449 mock_add_machine,
450 mock_get_machines,
451 mock_disconnect_controller,
452 mock_disconnect_model,
453 mock_get_model,
454 mock_get_controller,
455 ):
456 machine = None
457 bool_res = None
458 mock_get_model.return_value = juju.model.Model()
459 with self.assertRaises(JujuMachineNotFound):
460 machine, bool_res = self.loop.run_until_complete(
461 self.libjuju.create_machine("model", "non_existing_machine")
462 )
463 self.assertIsNone(machine)
464 self.assertIsNone(bool_res)
465
466 mock_disconnect_controller.assert_called()
467 mock_disconnect_model.assert_called()
468
469 def test_no_machine(
470 self,
471 mock_wait_for,
472 mock_add_machine,
473 mock_get_machines,
474 mock_disconnect_controller,
475 mock_disconnect_model,
476 mock_get_model,
477 mock_get_controller,
478 ):
479 mock_get_model.return_value = juju.model.Model()
480 mock_add_machine.return_value = FakeMachine()
481
482 machine, bool_res = self.loop.run_until_complete(
483 self.libjuju.create_machine("model")
484 )
485
486 self.assertIsInstance(machine, FakeMachine)
487 self.assertTrue(bool_res)
488
489 mock_wait_for.assert_called_once()
490 mock_add_machine.assert_called_once()
491
492 mock_disconnect_controller.assert_called()
493 mock_disconnect_model.assert_called()
494
495
496 # TODO test provision machine
497
498
499 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
500 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
501 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
502 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
503 @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
504 @asynctest.mock.patch("juju.model.Model.deploy")
505 class DeployTest(LibjujuTestCase):
506 def setUp(self):
507 super(DeployTest, self).setUp()
508
509 def test_deploy(
510 self,
511 mock_deploy,
512 mock_wait_for_model,
513 mock_disconnect_controller,
514 mock_disconnect_model,
515 mock_get_model,
516 mock_get_controller,
517 ):
518 mock_get_model.return_value = juju.model.Model()
519 self.loop.run_until_complete(
520 self.libjuju.deploy("cs:osm", "model", wait=True, timeout=0)
521 )
522 mock_deploy.assert_called_once()
523 mock_wait_for_model.assert_called_once()
524 mock_disconnect_controller.assert_called_once()
525 mock_disconnect_model.assert_called_once()
526
527 def test_deploy_no_wait(
528 self,
529 mock_deploy,
530 mock_wait_for_model,
531 mock_disconnect_controller,
532 mock_disconnect_model,
533 mock_get_model,
534 mock_get_controller,
535 ):
536 mock_get_model.return_value = juju.model.Model()
537 self.loop.run_until_complete(
538 self.libjuju.deploy("cs:osm", "model", wait=False, timeout=0)
539 )
540 mock_deploy.assert_called_once()
541 mock_wait_for_model.assert_not_called()
542 mock_disconnect_controller.assert_called_once()
543 mock_disconnect_model.assert_called_once()
544
545 def test_deploy_exception(
546 self,
547 mock_deploy,
548 mock_wait_for_model,
549 mock_disconnect_controller,
550 mock_disconnect_model,
551 mock_get_model,
552 mock_get_controller,
553 ):
554 mock_deploy.side_effect = Exception()
555 mock_get_model.return_value = juju.model.Model()
556 with self.assertRaises(Exception):
557 self.loop.run_until_complete(self.libjuju.deploy("cs:osm", "model"))
558 mock_deploy.assert_called_once()
559 mock_wait_for_model.assert_not_called()
560 mock_disconnect_controller.assert_called_once()
561 mock_disconnect_model.assert_called_once()
562
563
564 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
565 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
566 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
567 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
568 @asynctest.mock.patch(
569 "juju.model.Model.applications", new_callable=asynctest.PropertyMock
570 )
571 @asynctest.mock.patch("juju.model.Model.machines", new_callable=asynctest.PropertyMock)
572 @asynctest.mock.patch("juju.model.Model.deploy")
573 @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
574 @asynctest.mock.patch("n2vc.libjuju.Libjuju.create_machine")
575 class DeployCharmTest(LibjujuTestCase):
576 def setUp(self):
577 super(DeployCharmTest, self).setUp()
578
579 def test_existing_app(
580 self,
581 mock_create_machine,
582 mock_wait_for,
583 mock_deploy,
584 mock_machines,
585 mock_applications,
586 mock_disconnect_controller,
587 mock_disconnect_model,
588 mock_get_model,
589 mock_get_controller,
590 ):
591 mock_get_model.return_value = juju.model.Model()
592 mock_applications.return_value = {"existing_app"}
593
594 application = None
595 with self.assertRaises(JujuApplicationExists):
596 application = self.loop.run_until_complete(
597 self.libjuju.deploy_charm(
598 "existing_app",
599 "path",
600 "model",
601 "machine",
602 )
603 )
604 self.assertIsNone(application)
605
606 mock_disconnect_controller.assert_called()
607 mock_disconnect_model.assert_called()
608
609 def test_non_existing_machine(
610 self,
611 mock_create_machine,
612 mock_wait_for,
613 mock_deploy,
614 mock_machines,
615 mock_applications,
616 mock_disconnect_controller,
617 mock_disconnect_model,
618 mock_get_model,
619 mock_get_controller,
620 ):
621 mock_get_model.return_value = juju.model.Model()
622 mock_machines.return_value = {"existing_machine": FakeMachine()}
623 application = None
624 with self.assertRaises(JujuMachineNotFound):
625 application = self.loop.run_until_complete(
626 self.libjuju.deploy_charm(
627 "app",
628 "path",
629 "model",
630 "machine",
631 )
632 )
633
634 self.assertIsNone(application)
635
636 mock_disconnect_controller.assert_called()
637 mock_disconnect_model.assert_called()
638
639 def test_2_units(
640 self,
641 mock_create_machine,
642 mock_wait_for,
643 mock_deploy,
644 mock_machines,
645 mock_applications,
646 mock_disconnect_controller,
647 mock_disconnect_model,
648 mock_get_model,
649 mock_get_controller,
650 ):
651 mock_get_model.return_value = juju.model.Model()
652 mock_machines.return_value = {"existing_machine": FakeMachine()}
653 mock_create_machine.return_value = (FakeMachine(), "other")
654 mock_deploy.return_value = FakeApplication()
655 application = self.loop.run_until_complete(
656 self.libjuju.deploy_charm(
657 "app",
658 "path",
659 "model",
660 "existing_machine",
661 num_units=2,
662 )
663 )
664
665 self.assertIsInstance(application, FakeApplication)
666
667 mock_deploy.assert_called_once()
668 mock_wait_for.assert_called_once()
669
670 mock_create_machine.assert_called_once()
671
672 mock_disconnect_controller.assert_called()
673 mock_disconnect_model.assert_called()
674
675 def test_1_unit(
676 self,
677 mock_create_machine,
678 mock_wait_for,
679 mock_deploy,
680 mock_machines,
681 mock_applications,
682 mock_disconnect_controller,
683 mock_disconnect_model,
684 mock_get_model,
685 mock_get_controller,
686 ):
687 mock_get_model.return_value = juju.model.Model()
688 mock_machines.return_value = {"existing_machine": FakeMachine()}
689 mock_deploy.return_value = FakeApplication()
690 application = self.loop.run_until_complete(
691 self.libjuju.deploy_charm("app", "path", "model", "existing_machine")
692 )
693
694 self.assertIsInstance(application, FakeApplication)
695
696 mock_deploy.assert_called_once()
697 mock_wait_for.assert_called_once()
698
699 mock_disconnect_controller.assert_called()
700 mock_disconnect_model.assert_called()
701
702
703 @asynctest.mock.patch(
704 "juju.model.Model.applications", new_callable=asynctest.PropertyMock
705 )
706 class GetApplicationTest(LibjujuTestCase):
707 def setUp(self):
708 super(GetApplicationTest, self).setUp()
709
710 def test_existing_application(
711 self,
712 mock_applications,
713 ):
714 mock_applications.return_value = {"existing_app": "exists"}
715 model = juju.model.Model()
716 result = self.libjuju._get_application(model, "existing_app")
717 self.assertEqual(result, "exists")
718
719 def test_non_existing_application(
720 self,
721 mock_applications,
722 ):
723 mock_applications.return_value = {"existing_app": "exists"}
724 model = juju.model.Model()
725 result = self.libjuju._get_application(model, "nonexisting_app")
726 self.assertIsNone(result)
727
728
729 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
730 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
731 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
732 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
733 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
734 @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
735 @asynctest.mock.patch("juju.model.Model.get_action_output")
736 @asynctest.mock.patch("juju.model.Model.get_action_status")
737 class ExecuteActionTest(LibjujuTestCase):
738 def setUp(self):
739 super(ExecuteActionTest, self).setUp()
740
741 def test_no_application(
742 self,
743 mock_get_action_status,
744 mock_get_action_output,
745 mock_wait_for,
746 mock__get_application,
747 mock_disconnect_controller,
748 mock_disconnect_model,
749 mock_get_model,
750 mock_get_controller,
751 ):
752 mock__get_application.return_value = None
753 mock_get_model.return_value = juju.model.Model()
754 output = None
755 status = None
756 with self.assertRaises(JujuApplicationNotFound):
757 output, status = self.loop.run_until_complete(
758 self.libjuju.execute_action(
759 "app",
760 "model",
761 "action",
762 )
763 )
764 self.assertIsNone(output)
765 self.assertIsNone(status)
766
767 mock_disconnect_controller.assert_called()
768 mock_disconnect_model.assert_called()
769
770 def test_no_action(
771 self,
772 mock_get_action_status,
773 mock_get_action_output,
774 mock_wait_for,
775 mock__get_application,
776 mock_disconnect_controller,
777 mock_disconnect_model,
778 mock_get_model,
779 mock_get_controller,
780 ):
781
782 mock_get_model.return_value = juju.model.Model()
783 mock__get_application.return_value = FakeApplication()
784 output = None
785 status = None
786 with self.assertRaises(JujuActionNotFound):
787 output, status = self.loop.run_until_complete(
788 self.libjuju.execute_action(
789 "app",
790 "model",
791 "action",
792 )
793 )
794 self.assertIsNone(output)
795 self.assertIsNone(status)
796
797 mock_disconnect_controller.assert_called()
798 mock_disconnect_model.assert_called()
799
800 @asynctest.mock.patch("asyncio.sleep")
801 @asynctest.mock.patch("n2vc.tests.unit.utils.FakeUnit.is_leader_from_status")
802 def test_no_leader(
803 self,
804 mock_is_leader_from_status,
805 mock_sleep,
806 mock_get_action_status,
807 mock_get_action_output,
808 mock_wait_for,
809 mock__get_application,
810 mock_disconnect_controller,
811 mock_disconnect_model,
812 mock_get_model,
813 mock_get_controller,
814 ):
815 mock_get_model.return_value = juju.model.Model()
816 mock__get_application.return_value = FakeApplication()
817 mock_is_leader_from_status.return_value = False
818 output = None
819 status = None
820 with self.assertRaises(JujuLeaderUnitNotFound):
821 output, status = self.loop.run_until_complete(
822 self.libjuju.execute_action(
823 "app",
824 "model",
825 "action",
826 )
827 )
828 self.assertIsNone(output)
829 self.assertIsNone(status)
830
831 mock_disconnect_controller.assert_called()
832 mock_disconnect_model.assert_called()
833
834 def test_successful_exec(
835 self,
836 mock_get_action_status,
837 mock_get_action_output,
838 mock_wait_for,
839 mock__get_application,
840 mock_disconnect_controller,
841 mock_disconnect_model,
842 mock_get_model,
843 mock_get_controller,
844 ):
845 mock_get_model.return_value = juju.model.Model()
846 mock__get_application.return_value = FakeApplication()
847 mock_get_action_output.return_value = "output"
848 mock_get_action_status.return_value = {"id": "status"}
849 output, status = self.loop.run_until_complete(
850 self.libjuju.execute_action("app", "model", "existing_action")
851 )
852 self.assertEqual(output, "output")
853 self.assertEqual(status, "status")
854
855 mock_wait_for.assert_called_once()
856
857 mock_disconnect_controller.assert_called()
858 mock_disconnect_model.assert_called()
859
860
861 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
862 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
863 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
864 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
865 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
866 class GetActionTest(LibjujuTestCase):
867 def setUp(self):
868 super(GetActionTest, self).setUp()
869
870 def test_exception(
871 self,
872 mock_get_application,
873 mock_disconnect_controller,
874 mock_disconnect_model,
875 mock_get_model,
876 mock_get_controller,
877 ):
878 mock_get_application.side_effect = Exception()
879 actions = None
880 with self.assertRaises(Exception):
881 actions = self.loop.run_until_complete(
882 self.libjuju.get_actions("app", "model")
883 )
884
885 self.assertIsNone(actions)
886 mock_disconnect_controller.assert_called_once()
887 mock_disconnect_model.assert_called_once()
888
889 def test_success(
890 self,
891 mock_get_application,
892 mock_disconnect_controller,
893 mock_disconnect_model,
894 mock_get_model,
895 mock_get_controller,
896 ):
897 mock_get_application.return_value = FakeApplication()
898
899 actions = self.loop.run_until_complete(self.libjuju.get_actions("app", "model"))
900
901 self.assertEqual(actions, ["existing_action"])
902
903 mock_get_controller.assert_called_once()
904 mock_get_model.assert_called_once()
905 mock_disconnect_controller.assert_called_once()
906 mock_disconnect_model.assert_called_once()
907
908
909 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
910 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
911 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
912 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
913 @asynctest.mock.patch("juju.application.Application.get_metrics")
914 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
915 class GetMetricsTest(LibjujuTestCase):
916 def setUp(self):
917 super(GetMetricsTest, self).setUp()
918
919 def test_get_metrics_success(
920 self,
921 mock_get_application,
922 mock_get_metrics,
923 mock_disconnect_controller,
924 mock_disconnect_model,
925 mock_get_model,
926 mock_get_controller,
927 ):
928 mock_get_application.return_value = FakeApplication()
929 mock_get_model.return_value = juju.model.Model()
930
931 self.loop.run_until_complete(self.libjuju.get_metrics("model", "app1"))
932
933 mock_disconnect_controller.assert_called_once()
934 mock_disconnect_model.assert_called_once()
935
936 def test_get_metrics_exception(
937 self,
938 mock_get_application,
939 mock_get_metrics,
940 mock_disconnect_controller,
941 mock_disconnect_model,
942 mock_get_model,
943 mock_get_controller,
944 ):
945 mock_get_model.return_value = juju.model.Model()
946 mock_get_metrics.side_effect = Exception()
947 with self.assertRaises(Exception):
948 self.loop.run_until_complete(self.libjuju.get_metrics("model", "app1"))
949
950 mock_disconnect_controller.assert_called_once()
951 mock_disconnect_model.assert_called_once()
952
953 def test_missing_args_exception(
954 self,
955 mock_get_application,
956 mock_get_metrics,
957 mock_disconnect_controller,
958 mock_disconnect_model,
959 mock_get_model,
960 mock_get_controller,
961 ):
962 mock_get_model.return_value = juju.model.Model()
963
964 with self.assertRaises(Exception):
965 self.loop.run_until_complete(self.libjuju.get_metrics("", ""))
966
967 mock_get_controller.assert_not_called()
968 mock_get_model.assert_not_called()
969 mock_disconnect_controller.assert_not_called()
970 mock_disconnect_model.assert_not_called()
971
972
973 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
974 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
975 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
976 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
977 @asynctest.mock.patch("juju.model.Model.add_relation")
978 class AddRelationTest(LibjujuTestCase):
979 def setUp(self):
980 super(AddRelationTest, self).setUp()
981
982 @asynctest.mock.patch("logging.Logger.warning")
983 def test_not_found(
984 self,
985 mock_warning,
986 mock_add_relation,
987 mock_disconnect_controller,
988 mock_disconnect_model,
989 mock_get_model,
990 mock_get_controller,
991 ):
992 # TODO in libjuju.py should this fail only with a log message?
993 result = {"error": "not found", "response": "response", "request-id": 1}
994
995 mock_get_model.return_value = juju.model.Model()
996 mock_add_relation.side_effect = JujuAPIError(result)
997
998 self.loop.run_until_complete(
999 self.libjuju.add_relation(
1000 "model",
1001 "app1:relation1",
1002 "app2:relation2",
1003 )
1004 )
1005
1006 mock_warning.assert_called_with("Relation not found: not found")
1007 mock_disconnect_controller.assert_called_once()
1008 mock_disconnect_model.assert_called_once()
1009
1010 @asynctest.mock.patch("logging.Logger.warning")
1011 def test_already_exists(
1012 self,
1013 mock_warning,
1014 mock_add_relation,
1015 mock_disconnect_controller,
1016 mock_disconnect_model,
1017 mock_get_model,
1018 mock_get_controller,
1019 ):
1020 # TODO in libjuju.py should this fail silently?
1021 result = {"error": "already exists", "response": "response", "request-id": 1}
1022
1023 mock_get_model.return_value = juju.model.Model()
1024 mock_add_relation.side_effect = JujuAPIError(result)
1025
1026 self.loop.run_until_complete(
1027 self.libjuju.add_relation(
1028 "model",
1029 "app1:relation1",
1030 "app2:relation2",
1031 )
1032 )
1033
1034 mock_warning.assert_called_with("Relation already exists: already exists")
1035 mock_disconnect_controller.assert_called_once()
1036 mock_disconnect_model.assert_called_once()
1037
1038 def test_exception(
1039 self,
1040 mock_add_relation,
1041 mock_disconnect_controller,
1042 mock_disconnect_model,
1043 mock_get_model,
1044 mock_get_controller,
1045 ):
1046 mock_get_model.return_value = juju.model.Model()
1047 result = {"error": "", "response": "response", "request-id": 1}
1048 mock_add_relation.side_effect = JujuAPIError(result)
1049
1050 with self.assertRaises(JujuAPIError):
1051 self.loop.run_until_complete(
1052 self.libjuju.add_relation(
1053 "model",
1054 "app1:relation1",
1055 "app2:relation2",
1056 )
1057 )
1058
1059 mock_disconnect_controller.assert_called_once()
1060 mock_disconnect_model.assert_called_once()
1061
1062 def test_success(
1063 self,
1064 mock_add_relation,
1065 mock_disconnect_controller,
1066 mock_disconnect_model,
1067 mock_get_model,
1068 mock_get_controller,
1069 ):
1070 mock_get_model.return_value = juju.model.Model()
1071
1072 self.loop.run_until_complete(
1073 self.libjuju.add_relation(
1074 "model",
1075 "app1:relation1",
1076 "app2:relation2",
1077 )
1078 )
1079
1080 mock_add_relation.assert_called_with("app1:relation1", "app2:relation2")
1081 mock_disconnect_controller.assert_called_once()
1082 mock_disconnect_model.assert_called_once()
1083
1084 def test_saas(
1085 self,
1086 mock_add_relation,
1087 mock_disconnect_controller,
1088 mock_disconnect_model,
1089 mock_get_model,
1090 mock_get_controller,
1091 ):
1092 mock_get_model.return_value = juju.model.Model()
1093
1094 self.loop.run_until_complete(
1095 self.libjuju.add_relation(
1096 "model",
1097 "app1:relation1",
1098 "saas_name",
1099 )
1100 )
1101
1102 mock_add_relation.assert_called_with("app1:relation1", "saas_name")
1103 mock_disconnect_controller.assert_called_once()
1104 mock_disconnect_model.assert_called_once()
1105
1106
1107 # TODO destroy_model testcase
1108
1109
1110 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1111 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
1112 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1113 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
1114 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
1115 class DestroyApplicationTest(LibjujuTestCase):
1116 def setUp(self):
1117 super(DestroyApplicationTest, self).setUp()
1118
1119 def test_success(
1120 self,
1121 mock_get_controller,
1122 mock_get_model,
1123 mock_disconnect_controller,
1124 mock_get_application,
1125 mock_disconnect_model,
1126 ):
1127 mock_get_application.return_value = FakeApplication()
1128 mock_get_model.return_value = None
1129 self.loop.run_until_complete(
1130 self.libjuju.destroy_application(
1131 "existing_model",
1132 "existing_app",
1133 3600,
1134 )
1135 )
1136 mock_get_application.assert_called()
1137 mock_disconnect_controller.assert_called_once()
1138 mock_disconnect_model.assert_called_once()
1139
1140 def test_no_application(
1141 self,
1142 mock_get_controller,
1143 mock_get_model,
1144 mock_disconnect_controller,
1145 mock_get_application,
1146 mock_disconnect_model,
1147 ):
1148 mock_get_model.return_value = None
1149 mock_get_application.return_value = None
1150
1151 self.loop.run_until_complete(
1152 self.libjuju.destroy_application(
1153 "existing_model",
1154 "existing_app",
1155 3600,
1156 )
1157 )
1158 mock_get_application.assert_called()
1159
1160 def test_exception(
1161 self,
1162 mock_get_controller,
1163 mock_get_model,
1164 mock_disconnect_controller,
1165 mock_get_application,
1166 mock_disconnect_model,
1167 ):
1168 mock_get_application.return_value = FakeApplication
1169 mock_get_model.return_value = None
1170
1171 with self.assertRaises(Exception):
1172 self.loop.run_until_complete(
1173 self.libjuju.destroy_application(
1174 "existing_model",
1175 "existing_app",
1176 0,
1177 )
1178 )
1179 mock_get_application.assert_called_once()
1180
1181
1182 # @asynctest.mock.patch("juju.model.Model.get_machines")
1183 # @asynctest.mock.patch("logging.Logger.debug")
1184 # class DestroyMachineTest(LibjujuTestCase):
1185 # def setUp(self):
1186 # super(DestroyMachineTest, self).setUp()
1187
1188 # def test_success_manual_machine(
1189 # self, mock_debug, mock_get_machines,
1190 # ):
1191 # mock_get_machines.side_effect = [
1192 # {"machine": FakeManualMachine()},
1193 # {"machine": FakeManualMachine()},
1194 # {},
1195 # ]
1196 # self.loop.run_until_complete(
1197 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2,)
1198 # )
1199 # calls = [
1200 # asynctest.call("Waiting for machine machine is destroyed"),
1201 # asynctest.call("Machine destroyed: machine"),
1202 # ]
1203 # mock_debug.assert_has_calls(calls)
1204
1205 # def test_no_machine(
1206 # self, mock_debug, mock_get_machines,
1207 # ):
1208 # mock_get_machines.return_value = {}
1209 # self.loop.run_until_complete(
1210 # self.libjuju.destroy_machine(juju.model.Model(), "machine", 2)
1211 # )
1212 # mock_debug.assert_called_with("Machine not found: machine")
1213
1214
1215 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1216 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
1217 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
1218 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1219 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
1220 class ConfigureApplicationTest(LibjujuTestCase):
1221 def setUp(self):
1222 super(ConfigureApplicationTest, self).setUp()
1223
1224 def test_success(
1225 self,
1226 mock_get_application,
1227 mock_disconnect_controller,
1228 mock_disconnect_model,
1229 mock_get_model,
1230 mock_get_controller,
1231 ):
1232
1233 mock_get_application.return_value = FakeApplication()
1234
1235 self.loop.run_until_complete(
1236 self.libjuju.configure_application(
1237 "model",
1238 "app",
1239 {"config"},
1240 )
1241 )
1242 mock_get_application.assert_called_once()
1243 mock_disconnect_controller.assert_called_once()
1244 mock_disconnect_model.assert_called_once()
1245
1246 def test_exception(
1247 self,
1248 mock_get_application,
1249 mock_disconnect_controller,
1250 mock_disconnect_model,
1251 mock_get_model,
1252 mock_get_controller,
1253 ):
1254
1255 mock_get_application.side_effect = Exception()
1256
1257 with self.assertRaises(Exception):
1258 self.loop.run_until_complete(
1259 self.libjuju.configure_application(
1260 "model",
1261 "app",
1262 {"config"},
1263 )
1264 )
1265 mock_disconnect_controller.assert_called_once()
1266 mock_disconnect_model.assert_called_once()
1267
1268 def test_controller_exception(
1269 self,
1270 mock_get_application,
1271 mock_disconnect_controller,
1272 mock_disconnect_model,
1273 mock_get_model,
1274 mock_get_controller,
1275 ):
1276
1277 result = {"error": "not found", "response": "response", "request-id": 1}
1278
1279 mock_get_controller.side_effect = JujuAPIError(result)
1280
1281 with self.assertRaises(JujuAPIError):
1282 self.loop.run_until_complete(
1283 self.libjuju.configure_application(
1284 "model",
1285 "app",
1286 {"config"},
1287 )
1288 )
1289 mock_get_model.assert_not_called()
1290 mock_disconnect_controller.assert_not_called()
1291 mock_disconnect_model.assert_not_called()
1292
1293 def test_get_model_exception(
1294 self,
1295 mock_get_application,
1296 mock_disconnect_controller,
1297 mock_disconnect_model,
1298 mock_get_model,
1299 mock_get_controller,
1300 ):
1301
1302 result = {"error": "not found", "response": "response", "request-id": 1}
1303 mock_get_model.side_effect = JujuAPIError(result)
1304
1305 with self.assertRaises(JujuAPIError):
1306 self.loop.run_until_complete(
1307 self.libjuju.configure_application(
1308 "model",
1309 "app",
1310 {"config"},
1311 )
1312 )
1313 mock_get_model.assert_called_once()
1314 mock_disconnect_controller.assert_called_once()
1315 mock_disconnect_model.assert_not_called()
1316
1317
1318 # TODO _get_api_endpoints_db test case
1319 # TODO _update_api_endpoints_db test case
1320 # TODO healthcheck test case
1321
1322
1323 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1324 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1325 @asynctest.mock.patch("juju.controller.Controller.list_models")
1326 class ListModelsTest(LibjujuTestCase):
1327 def setUp(self):
1328 super(ListModelsTest, self).setUp()
1329
1330 def test_containing(
1331 self,
1332 mock_list_models,
1333 mock_disconnect_controller,
1334 mock_get_controller,
1335 ):
1336 mock_get_controller.return_value = juju.controller.Controller()
1337 mock_list_models.return_value = ["existingmodel"]
1338 models = self.loop.run_until_complete(self.libjuju.list_models("existing"))
1339
1340 mock_disconnect_controller.assert_called_once()
1341 self.assertEquals(models, ["existingmodel"])
1342
1343 def test_not_containing(
1344 self,
1345 mock_list_models,
1346 mock_disconnect_controller,
1347 mock_get_controller,
1348 ):
1349 mock_get_controller.return_value = juju.controller.Controller()
1350 mock_list_models.return_value = ["existingmodel", "model"]
1351 models = self.loop.run_until_complete(self.libjuju.list_models("mdl"))
1352
1353 mock_disconnect_controller.assert_called_once()
1354 self.assertEquals(models, [])
1355
1356 def test_no_contains_arg(
1357 self,
1358 mock_list_models,
1359 mock_disconnect_controller,
1360 mock_get_controller,
1361 ):
1362 mock_get_controller.return_value = juju.controller.Controller()
1363 mock_list_models.return_value = ["existingmodel", "model"]
1364 models = self.loop.run_until_complete(self.libjuju.list_models())
1365
1366 mock_disconnect_controller.assert_called_once()
1367 self.assertEquals(models, ["existingmodel", "model"])
1368
1369
1370 @asynctest.mock.patch("n2vc.libjuju.Libjuju.list_models")
1371 class ModelsExistTest(LibjujuTestCase):
1372 def setUp(self):
1373 super(ModelsExistTest, self).setUp()
1374
1375 def test_model_names_none(self, mock_list_models):
1376 mock_list_models.return_value = []
1377 with self.assertRaises(Exception):
1378 self.loop.run_until_complete(self.libjuju.models_exist(None))
1379
1380 def test_model_names_empty(self, mock_list_models):
1381 mock_list_models.return_value = []
1382 with self.assertRaises(Exception):
1383 (exist, non_existing_models) = self.loop.run_until_complete(
1384 self.libjuju.models_exist([])
1385 )
1386
1387 def test_model_names_not_existing(self, mock_list_models):
1388 mock_list_models.return_value = ["prometheus", "grafana"]
1389 (exist, non_existing_models) = self.loop.run_until_complete(
1390 self.libjuju.models_exist(["prometheus2", "grafana"])
1391 )
1392 self.assertFalse(exist)
1393 self.assertEqual(non_existing_models, ["prometheus2"])
1394
1395 def test_model_names_exist(self, mock_list_models):
1396 mock_list_models.return_value = ["prometheus", "grafana"]
1397 (exist, non_existing_models) = self.loop.run_until_complete(
1398 self.libjuju.models_exist(["prometheus", "grafana"])
1399 )
1400 self.assertTrue(exist)
1401 self.assertEqual(non_existing_models, [])
1402
1403
1404 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1405 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1406 @asynctest.mock.patch("juju.controller.Controller.list_offers")
1407 class ListOffers(LibjujuTestCase):
1408 def setUp(self):
1409 super(ListOffers, self).setUp()
1410
1411 def test_disconnect_controller(
1412 self,
1413 mock_list_offers,
1414 mock_disconnect_controller,
1415 mock_get_controller,
1416 ):
1417 mock_get_controller.return_value = juju.controller.Controller()
1418 mock_list_offers.side_effect = Exception()
1419 with self.assertRaises(Exception):
1420 self.loop.run_until_complete(self.libjuju._list_offers("model"))
1421 mock_disconnect_controller.assert_called_once()
1422
1423 def test_empty_list(
1424 self,
1425 mock_list_offers,
1426 mock_disconnect_controller,
1427 mock_get_controller,
1428 ):
1429 mock_get_controller.return_value = juju.controller.Controller()
1430 offer_results = Mock()
1431 offer_results.results = []
1432 mock_list_offers.return_value = offer_results
1433 offers = self.loop.run_until_complete(self.libjuju._list_offers("model"))
1434 self.assertEqual(offers, [])
1435 mock_disconnect_controller.assert_called_once()
1436
1437 def test_non_empty_list(
1438 self,
1439 mock_list_offers,
1440 mock_disconnect_controller,
1441 mock_get_controller,
1442 ):
1443 mock_get_controller.return_value = juju.controller.Controller()
1444 offer = Mock()
1445 offer_results = Mock()
1446 offer_results.results = [offer]
1447 mock_list_offers.return_value = offer_results
1448 offers = self.loop.run_until_complete(self.libjuju._list_offers("model"))
1449 self.assertEqual(offers, [offer])
1450 mock_disconnect_controller.assert_called_once()
1451
1452 def test_matching_offer_name(
1453 self,
1454 mock_list_offers,
1455 mock_disconnect_controller,
1456 mock_get_controller,
1457 ):
1458 mock_get_controller.return_value = juju.controller.Controller()
1459 offer_1 = Mock()
1460 offer_1.offer_name = "offer1"
1461 offer_2 = Mock()
1462 offer_2.offer_name = "offer2"
1463 offer_results = Mock()
1464 offer_results.results = [offer_1, offer_2]
1465 mock_list_offers.return_value = offer_results
1466 offers = self.loop.run_until_complete(
1467 self.libjuju._list_offers("model", offer_name="offer2")
1468 )
1469 self.assertEqual(offers, [offer_2])
1470 mock_disconnect_controller.assert_called_once()
1471
1472 def test_not_matching_offer_name(
1473 self,
1474 mock_list_offers,
1475 mock_disconnect_controller,
1476 mock_get_controller,
1477 ):
1478 mock_get_controller.return_value = juju.controller.Controller()
1479 offer_1 = Mock()
1480 offer_1.offer_name = "offer1"
1481 offer_2 = Mock()
1482 offer_2.offer_name = "offer2"
1483 offer_results = Mock()
1484 offer_results.results = [offer_1, offer_2]
1485 mock_list_offers.return_value = offer_results
1486 offers = self.loop.run_until_complete(
1487 self.libjuju._list_offers("model", offer_name="offer3")
1488 )
1489 self.assertEqual(offers, [])
1490 mock_disconnect_controller.assert_called_once()
1491
1492
1493 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1494 @asynctest.mock.patch("juju.controller.Controller.get_model")
1495 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
1496 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1497 @asynctest.mock.patch("n2vc.libjuju.Libjuju._list_offers")
1498 @asynctest.mock.patch("juju.model.Model.create_offer")
1499 class OfferTest(LibjujuTestCase):
1500 def setUp(self):
1501 super(OfferTest, self).setUp()
1502
1503 def test_offer(
1504 self,
1505 mock_create_offer,
1506 mock__list_offers,
1507 mock_disconnect_controller,
1508 mock_disconnect_model,
1509 mock_get_model,
1510 mock_get_controller,
1511 ):
1512 controller = juju.controller.Controller()
1513 model = juju.model.Model()
1514 mock_get_controller.return_value = controller
1515 mock_get_model.return_value = model
1516 endpoint = RelationEndpoint("model.app-name.0", "vca", "endpoint")
1517 self.loop.run_until_complete(self.libjuju.offer(endpoint))
1518 mock_create_offer.assert_called_with(
1519 "app-name:endpoint", offer_name="app-name-endpoint"
1520 )
1521 mock_disconnect_model.assert_called_once_with(model)
1522 mock_disconnect_controller.assert_called_once_with(controller)
1523
1524 def test_offer_exception(
1525 self,
1526 mock_create_offer,
1527 mock__list_offers,
1528 mock_disconnect_controller,
1529 mock_disconnect_model,
1530 mock_get_model,
1531 mock_get_controller,
1532 ):
1533 controller = juju.controller.Controller()
1534 model = juju.model.Model()
1535 mock_get_controller.return_value = controller
1536 mock_get_model.return_value = model
1537 mock__list_offers.return_value = []
1538 endpoint = RelationEndpoint("model.app-name.0", "vca", "endpoint")
1539 with self.assertRaises(Exception):
1540 self.loop.run_until_complete(self.libjuju.offer(endpoint))
1541 mock_create_offer.assert_called_with(
1542 "app-name:endpoint", offer_name="app-name-endpoint"
1543 )
1544 mock_disconnect_model.assert_called_once_with(model)
1545 mock_disconnect_controller.assert_called_once_with(controller)
1546
1547
1548 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1549 @asynctest.mock.patch("juju.controller.Controller.get_model")
1550 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
1551 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1552 @asynctest.mock.patch("juju.model.Model.consume")
1553 class ConsumeTest(LibjujuTestCase):
1554 def setUp(self):
1555 self.offer_url = "admin/model.offer_name"
1556 super(ConsumeTest, self).setUp()
1557 self.provider_libjuju = self.libjuju
1558
1559 def test_consume(
1560 self,
1561 mock_consume,
1562 mock_disconnect_controller,
1563 mock_disconnect_model,
1564 mock_get_model,
1565 mock_get_controller,
1566 ):
1567 self_controller = juju.controller.Controller()
1568 provider_controller = juju.controller.Controller()
1569 mock_get_controller.side_effect = [self_controller, provider_controller]
1570 mock_get_model.return_value = juju.model.Model()
1571
1572 self.loop.run_until_complete(
1573 self.libjuju.consume(
1574 "model_name",
1575 Offer(self.offer_url, vca_id="vca-id"),
1576 self.provider_libjuju,
1577 )
1578 )
1579 mock_consume.assert_called_once_with(
1580 "admin/model.offer_name",
1581 application_alias="offer_name-model-vca-id",
1582 controller=provider_controller,
1583 )
1584 mock_disconnect_model.assert_called_once()
1585 self.assertEqual(mock_disconnect_controller.call_count, 2)
1586
1587 def test_parsing_error_exception(
1588 self,
1589 mock_consume,
1590 mock_disconnect_controller,
1591 mock_disconnect_model,
1592 mock_get_model,
1593 mock_get_controller,
1594 ):
1595 mock_get_controller.return_value = juju.controller.Controller()
1596 mock_get_model.return_value = juju.model.Model()
1597 mock_consume.side_effect = juju.offerendpoints.ParseError("")
1598
1599 with self.assertRaises(juju.offerendpoints.ParseError):
1600 self.loop.run_until_complete(
1601 self.libjuju.consume(
1602 "model_name", Offer(self.offer_url), self.provider_libjuju
1603 )
1604 )
1605 mock_consume.assert_called_once()
1606 mock_disconnect_model.assert_called_once()
1607 self.assertEqual(mock_disconnect_controller.call_count, 2)
1608
1609 def test_juju_error_exception(
1610 self,
1611 mock_consume,
1612 mock_disconnect_controller,
1613 mock_disconnect_model,
1614 mock_get_model,
1615 mock_get_controller,
1616 ):
1617 mock_get_controller.return_value = juju.controller.Controller()
1618 mock_get_model.return_value = juju.model.Model()
1619 mock_consume.side_effect = juju.errors.JujuError("")
1620
1621 with self.assertRaises(juju.errors.JujuError):
1622 self.loop.run_until_complete(
1623 self.libjuju.consume(
1624 "model_name", Offer(self.offer_url), self.provider_libjuju
1625 )
1626 )
1627 mock_consume.assert_called_once()
1628 mock_disconnect_model.assert_called_once()
1629 self.assertEqual(mock_disconnect_controller.call_count, 2)
1630
1631 def test_juju_api_error_exception(
1632 self,
1633 mock_consume,
1634 mock_disconnect_controller,
1635 mock_disconnect_model,
1636 mock_get_model,
1637 mock_get_controller,
1638 ):
1639 mock_get_controller.return_value = juju.controller.Controller()
1640 mock_get_model.return_value = juju.model.Model()
1641 mock_consume.side_effect = juju.errors.JujuAPIError(
1642 {"error": "", "response": "", "request-id": ""}
1643 )
1644
1645 with self.assertRaises(juju.errors.JujuAPIError):
1646 self.loop.run_until_complete(
1647 self.libjuju.consume(
1648 "model_name", Offer(self.offer_url), self.provider_libjuju
1649 )
1650 )
1651 mock_consume.assert_called_once()
1652 mock_disconnect_model.assert_called_once()
1653 self.assertEqual(mock_disconnect_controller.call_count, 2)
1654
1655
1656 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
1657 @asynctest.mock.patch("n2vc.libjuju.Libjuju.add_cloud")
1658 class AddK8sTest(LibjujuTestCase):
1659 def setUp(self):
1660 super(AddK8sTest, self).setUp()
1661 name = "cloud"
1662 rbac_id = generate_rbac_id()
1663 token = "token"
1664 client_cert_data = "cert"
1665 configuration = kubernetes.client.configuration.Configuration()
1666 storage_class = "storage_class"
1667 credential_name = name
1668
1669 self._add_k8s_args = {
1670 "name": name,
1671 "rbac_id": rbac_id,
1672 "token": token,
1673 "client_cert_data": client_cert_data,
1674 "configuration": configuration,
1675 "storage_class": storage_class,
1676 "credential_name": credential_name,
1677 }
1678
1679 def test_add_k8s(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1680 self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1681 mock_add_cloud.assert_called_once()
1682 mock_get_k8s_cloud_credential.assert_called_once()
1683
1684 def test_add_k8s_exception(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1685 mock_add_cloud.side_effect = Exception()
1686 with self.assertRaises(Exception):
1687 self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1688 mock_add_cloud.assert_called_once()
1689 mock_get_k8s_cloud_credential.assert_called_once()
1690
1691 def test_add_k8s_missing_name(self, mock_add_cloud, mock_get_k8s_cloud_credential):
1692 self._add_k8s_args["name"] = ""
1693 with self.assertRaises(Exception):
1694 self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1695 mock_add_cloud.assert_not_called()
1696
1697 def test_add_k8s_missing_storage_name(
1698 self, mock_add_cloud, mock_get_k8s_cloud_credential
1699 ):
1700 self._add_k8s_args["storage_class"] = ""
1701 with self.assertRaises(Exception):
1702 self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1703 mock_add_cloud.assert_not_called()
1704
1705 def test_add_k8s_missing_configuration_keys(
1706 self, mock_add_cloud, mock_get_k8s_cloud_credential
1707 ):
1708 self._add_k8s_args["configuration"] = None
1709 with self.assertRaises(Exception):
1710 self.loop.run_until_complete(self.libjuju.add_k8s(**self._add_k8s_args))
1711 mock_add_cloud.assert_not_called()
1712
1713
1714 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1715 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1716 @asynctest.mock.patch("juju.controller.Controller.add_cloud")
1717 @asynctest.mock.patch("juju.controller.Controller.add_credential")
1718 class AddCloudTest(LibjujuTestCase):
1719 def setUp(self):
1720 super(AddCloudTest, self).setUp()
1721 self.cloud = juju.client.client.Cloud()
1722 self.credential = juju.client.client.CloudCredential()
1723
1724 def test_add_cloud_with_credential(
1725 self,
1726 mock_add_credential,
1727 mock_add_cloud,
1728 mock_disconnect_controller,
1729 mock_get_controller,
1730 ):
1731 mock_get_controller.return_value = juju.controller.Controller()
1732
1733 cloud = self.loop.run_until_complete(
1734 self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1735 )
1736 self.assertEqual(cloud, self.cloud)
1737 mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1738 mock_add_credential.assert_called_once_with(
1739 "cloud", credential=self.credential, cloud="cloud"
1740 )
1741 mock_disconnect_controller.assert_called_once()
1742
1743 def test_add_cloud_no_credential(
1744 self,
1745 mock_add_credential,
1746 mock_add_cloud,
1747 mock_disconnect_controller,
1748 mock_get_controller,
1749 ):
1750 mock_get_controller.return_value = juju.controller.Controller()
1751
1752 cloud = self.loop.run_until_complete(
1753 self.libjuju.add_cloud("cloud", self.cloud)
1754 )
1755 self.assertEqual(cloud, self.cloud)
1756 mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1757 mock_add_credential.assert_not_called()
1758 mock_disconnect_controller.assert_called_once()
1759
1760 def test_add_cloud_exception(
1761 self,
1762 mock_add_credential,
1763 mock_add_cloud,
1764 mock_disconnect_controller,
1765 mock_get_controller,
1766 ):
1767 mock_get_controller.return_value = juju.controller.Controller()
1768 mock_add_cloud.side_effect = Exception()
1769 with self.assertRaises(Exception):
1770 self.loop.run_until_complete(
1771 self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1772 )
1773
1774 mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1775 mock_add_credential.assert_not_called()
1776 mock_disconnect_controller.assert_called_once()
1777
1778 def test_add_credential_exception(
1779 self,
1780 mock_add_credential,
1781 mock_add_cloud,
1782 mock_disconnect_controller,
1783 mock_get_controller,
1784 ):
1785 mock_get_controller.return_value = juju.controller.Controller()
1786 mock_add_credential.side_effect = Exception()
1787 with self.assertRaises(Exception):
1788 self.loop.run_until_complete(
1789 self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
1790 )
1791
1792 mock_add_cloud.assert_called_once_with("cloud", self.cloud)
1793 mock_add_credential.assert_called_once_with(
1794 "cloud", credential=self.credential, cloud="cloud"
1795 )
1796 mock_disconnect_controller.assert_called_once()
1797
1798
1799 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
1800 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
1801 @asynctest.mock.patch("juju.controller.Controller.remove_cloud")
1802 class RemoveCloudTest(LibjujuTestCase):
1803 def setUp(self):
1804 super(RemoveCloudTest, self).setUp()
1805
1806 def test_remove_cloud(
1807 self,
1808 mock_remove_cloud,
1809 mock_disconnect_controller,
1810 mock_get_controller,
1811 ):
1812 mock_get_controller.return_value = juju.controller.Controller()
1813
1814 self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
1815 mock_remove_cloud.assert_called_once_with("cloud")
1816 mock_disconnect_controller.assert_called_once()
1817
1818 def test_remove_cloud_exception(
1819 self,
1820 mock_remove_cloud,
1821 mock_disconnect_controller,
1822 mock_get_controller,
1823 ):
1824 mock_get_controller.return_value = juju.controller.Controller()
1825 mock_remove_cloud.side_effect = Exception()
1826
1827 with self.assertRaises(Exception):
1828 self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
1829 mock_remove_cloud.assert_called_once_with("cloud")
1830 mock_disconnect_controller.assert_called_once()
1831
1832
1833 @asynctest.mock.patch("kubernetes.client.configuration.Configuration")
1834 class GetK8sCloudCredentials(LibjujuTestCase):
1835 def setUp(self):
1836 super(GetK8sCloudCredentials, self).setUp()
1837 self.cert_data = "cert"
1838 self.token = "token"
1839
1840 @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1841 def test_not_supported(self, mock_exception, mock_configuration):
1842 mock_configuration.username = ""
1843 mock_configuration.password = ""
1844 mock_configuration.ssl_ca_cert = None
1845 mock_configuration.cert_file = None
1846 mock_configuration.key_file = None
1847 exception_raised = False
1848 self.token = None
1849 self.cert_data = None
1850 try:
1851 _ = self.libjuju.get_k8s_cloud_credential(
1852 mock_configuration,
1853 self.cert_data,
1854 self.token,
1855 )
1856 except JujuInvalidK8sConfiguration as e:
1857 exception_raised = True
1858 self.assertEqual(
1859 e.message,
1860 "authentication method not supported",
1861 )
1862 self.assertTrue(exception_raised)
1863
1864 def test_user_pass(self, mock_configuration):
1865 mock_configuration.username = "admin"
1866 mock_configuration.password = "admin"
1867 mock_configuration.ssl_ca_cert = None
1868 mock_configuration.cert_file = None
1869 mock_configuration.key_file = None
1870 self.token = None
1871 self.cert_data = None
1872 credential = self.libjuju.get_k8s_cloud_credential(
1873 mock_configuration,
1874 self.cert_data,
1875 self.token,
1876 )
1877 self.assertEqual(
1878 credential,
1879 juju.client._definitions.CloudCredential(
1880 attrs={"username": "admin", "password": "admin"}, auth_type="userpass"
1881 ),
1882 )
1883
1884 def test_user_pass_with_cert(self, mock_configuration):
1885 mock_configuration.username = "admin"
1886 mock_configuration.password = "admin"
1887 mock_configuration.ssl_ca_cert = None
1888 mock_configuration.cert_file = None
1889 mock_configuration.key_file = None
1890 self.token = None
1891 credential = self.libjuju.get_k8s_cloud_credential(
1892 mock_configuration,
1893 self.cert_data,
1894 self.token,
1895 )
1896 self.assertEqual(
1897 credential,
1898 juju.client._definitions.CloudCredential(
1899 attrs={
1900 "ClientCertificateData": self.cert_data,
1901 "username": "admin",
1902 "password": "admin",
1903 },
1904 auth_type="userpasswithcert",
1905 ),
1906 )
1907
1908 def test_user_no_pass(self, mock_configuration):
1909 mock_configuration.username = "admin"
1910 mock_configuration.password = ""
1911 mock_configuration.ssl_ca_cert = None
1912 mock_configuration.cert_file = None
1913 mock_configuration.key_file = None
1914 self.token = None
1915 self.cert_data = None
1916 with patch.object(self.libjuju.log, "debug") as mock_debug:
1917 credential = self.libjuju.get_k8s_cloud_credential(
1918 mock_configuration,
1919 self.cert_data,
1920 self.token,
1921 )
1922 self.assertEqual(
1923 credential,
1924 juju.client._definitions.CloudCredential(
1925 attrs={"username": "admin", "password": ""}, auth_type="userpass"
1926 ),
1927 )
1928 mock_debug.assert_called_once_with(
1929 "credential for user admin has empty password"
1930 )
1931
1932 def test_cert(self, mock_configuration):
1933 mock_configuration.username = ""
1934 mock_configuration.password = ""
1935 mock_configuration.api_key = {"authorization": "Bearer Token"}
1936 ssl_ca_cert = tempfile.NamedTemporaryFile()
1937 with open(ssl_ca_cert.name, "w") as ssl_ca_cert_file:
1938 ssl_ca_cert_file.write("cacert")
1939 mock_configuration.ssl_ca_cert = ssl_ca_cert.name
1940 mock_configuration.cert_file = None
1941 mock_configuration.key_file = None
1942 credential = self.libjuju.get_k8s_cloud_credential(
1943 mock_configuration,
1944 self.cert_data,
1945 self.token,
1946 )
1947 self.assertEqual(
1948 credential,
1949 juju.client._definitions.CloudCredential(
1950 attrs={"ClientCertificateData": self.cert_data, "Token": self.token},
1951 auth_type="certificate",
1952 ),
1953 )
1954
1955 # TODO: Fix this test when oauth authentication is supported
1956 # def test_oauth2(self, mock_configuration):
1957 # mock_configuration.username = ""
1958 # mock_configuration.password = ""
1959 # mock_configuration.api_key = {"authorization": "Bearer Token"}
1960 # key = tempfile.NamedTemporaryFile()
1961 # with open(key.name, "w") as key_file:
1962 # key_file.write("key")
1963 # mock_configuration.ssl_ca_cert = None
1964 # mock_configuration.cert_file = None
1965 # mock_configuration.key_file = key.name
1966 # credential = self.libjuju.get_k8s_cloud_credential(
1967 # mock_configuration,
1968 # self.cert_data,
1969 # self.token,
1970 # )
1971 # self.assertEqual(
1972 # credential,
1973 # juju.client._definitions.CloudCredential(
1974 # attrs={"ClientKeyData": "key", "Token": "Token"},
1975 # auth_type="oauth2",
1976 # ),
1977 # )
1978
1979 # @asynctest.mock.patch("n2vc.exceptions.JujuInvalidK8sConfiguration")
1980 # def test_oauth2_missing_token(self, mock_exception, mock_configuration):
1981 # mock_configuration.username = ""
1982 # mock_configuration.password = ""
1983 # key = tempfile.NamedTemporaryFile()
1984 # with open(key.name, "w") as key_file:
1985 # key_file.write("key")
1986 # mock_configuration.ssl_ca_cert = None
1987 # mock_configuration.cert_file = None
1988 # mock_configuration.key_file = key.name
1989 # exception_raised = False
1990 # try:
1991 # _ = self.libjuju.get_k8s_cloud_credential(
1992 # mock_configuration,
1993 # self.cert_data,
1994 # self.token,
1995 # )
1996 # except JujuInvalidK8sConfiguration as e:
1997 # exception_raised = True
1998 # self.assertEqual(
1999 # e.message,
2000 # "missing token for auth type oauth2",
2001 # )
2002 # self.assertTrue(exception_raised)
2003
2004 def test_exception_cannot_set_token_and_userpass(self, mock_configuration):
2005 mock_configuration.username = "admin"
2006 mock_configuration.password = "pass"
2007 mock_configuration.api_key = {"authorization": "No_bearer_token"}
2008 mock_configuration.ssl_ca_cert = None
2009 mock_configuration.cert_file = None
2010 mock_configuration.key_file = None
2011 exception_raised = False
2012 try:
2013 _ = self.libjuju.get_k8s_cloud_credential(
2014 mock_configuration,
2015 self.cert_data,
2016 self.token,
2017 )
2018 except JujuInvalidK8sConfiguration as e:
2019 exception_raised = True
2020 self.assertEqual(
2021 e.message,
2022 "Cannot set both token and user/pass",
2023 )
2024 self.assertTrue(exception_raised)
2025
2026
2027 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
2028 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
2029 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
2030 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
2031 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2032 @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
2033 class ScaleApplicationTest(LibjujuTestCase):
2034 def setUp(self):
2035 super(ScaleApplicationTest, self).setUp()
2036
2037 @asynctest.mock.patch("asyncio.sleep")
2038 def test_scale_application(
2039 self,
2040 mock_sleep,
2041 mock_wait_for_model,
2042 mock_disconnect_controller,
2043 mock_disconnect_model,
2044 mock_get_application,
2045 mock_get_model,
2046 mock_get_controller,
2047 ):
2048 mock_get_model.return_value = juju.model.Model()
2049 mock_get_application.return_value = FakeApplication()
2050 self.loop.run_until_complete(self.libjuju.scale_application("model", "app", 2))
2051 mock_wait_for_model.assert_called_once()
2052 mock_disconnect_controller.assert_called_once()
2053 mock_disconnect_model.assert_called_once()
2054
2055 def test_no_application(
2056 self,
2057 mock_wait_for,
2058 mock_disconnect_controller,
2059 mock_disconnect_model,
2060 mock_get_application,
2061 mock_get_model,
2062 mock_get_controller,
2063 ):
2064 mock_get_application.return_value = None
2065 mock_get_model.return_value = juju.model.Model()
2066 with self.assertRaises(JujuApplicationNotFound):
2067 self.loop.run_until_complete(
2068 self.libjuju.scale_application("model", "app", 2)
2069 )
2070 mock_disconnect_controller.assert_called()
2071 mock_disconnect_model.assert_called()
2072
2073 def test_exception(
2074 self,
2075 mock_wait_for,
2076 mock_disconnect_controller,
2077 mock_disconnect_model,
2078 mock_get_application,
2079 mock_get_model,
2080 mock_get_controller,
2081 ):
2082 mock_get_model.return_value = None
2083 mock_get_application.return_value = FakeApplication()
2084 with self.assertRaises(Exception):
2085 self.loop.run_until_complete(
2086 self.libjuju.scale_application("model", "app", 2, total_timeout=0)
2087 )
2088 mock_disconnect_controller.assert_called_once()
2089
2090
2091 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
2092 class GetUnitNumberTest(LibjujuTestCase):
2093 def setUp(self):
2094 super(GetUnitNumberTest, self).setUp()
2095
2096 def test_successful_get_unit_number(
2097 self,
2098 mock_get_applications,
2099 ):
2100 mock_get_applications.return_value = FakeApplication()
2101 model = juju.model.Model()
2102 result = self.libjuju._get_application_count(model, "app")
2103 self.assertEqual(result, 2)
2104
2105 def test_non_existing_application(
2106 self,
2107 mock_get_applications,
2108 ):
2109 mock_get_applications.return_value = None
2110 model = juju.model.Model()
2111 result = self.libjuju._get_application_count(model, "app")
2112 self.assertEqual(result, None)
2113
2114
2115 @asynctest.mock.patch("juju.model.Model.machines", new_callable=asynctest.PropertyMock)
2116 class GetMachineInfoTest(LibjujuTestCase):
2117 def setUp(self):
2118 super(GetMachineInfoTest, self).setUp()
2119
2120 def test_successful(
2121 self,
2122 mock_machines,
2123 ):
2124 machine_id = "existing_machine"
2125 model = juju.model.Model()
2126 mock_machines.return_value = {"existing_machine": FakeManualMachine()}
2127 machine, series = self.libjuju._get_machine_info(
2128 machine_id=machine_id,
2129 model=model,
2130 )
2131 self.assertIsNotNone(machine, series)
2132
2133 def test_exception(
2134 self,
2135 mock_machines,
2136 ):
2137 machine_id = "not_existing_machine"
2138 machine = series = None
2139 model = juju.model.Model()
2140 mock_machines.return_value = {"existing_machine": FakeManualMachine()}
2141 with self.assertRaises(JujuMachineNotFound):
2142 machine, series = self.libjuju._get_machine_info(
2143 machine_id=machine_id,
2144 model=model,
2145 )
2146 self.assertIsNone(machine, series)
2147
2148
2149 class GetUnitTest(LibjujuTestCase):
2150 def setUp(self):
2151 super(GetUnitTest, self).setUp()
2152
2153 def test_successful(self):
2154 result = self.libjuju._get_unit(FakeApplication(), "existing_machine_id")
2155 self.assertIsInstance(result, FakeUnit)
2156
2157 def test_return_none(self):
2158 result = self.libjuju._get_unit(FakeApplication(), "not_existing_machine_id")
2159 self.assertIsNone(result)
2160
2161
2162 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
2163 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
2164 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
2165 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2166 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
2167 class CheckApplicationExists(LibjujuTestCase):
2168 def setUp(self):
2169 super(CheckApplicationExists, self).setUp()
2170
2171 def test_successful(
2172 self,
2173 mock_get_application,
2174 mock_disconnect_controller,
2175 mock_disconnect_model,
2176 mock_get_model,
2177 mock_get_controller,
2178 ):
2179 mock_get_model.return_value = juju.model.Model()
2180 mock_get_application.return_value = FakeApplication()
2181 result = self.loop.run_until_complete(
2182 self.libjuju.check_application_exists(
2183 "model",
2184 "app",
2185 )
2186 )
2187 self.assertEqual(result, True)
2188
2189 mock_get_application.assert_called_once()
2190 mock_get_controller.assert_called_once()
2191 mock_get_model.assert_called_once()
2192 mock_disconnect_controller.assert_called_once()
2193 mock_disconnect_model.assert_called_once()
2194
2195 def test_no_application(
2196 self,
2197 mock_get_application,
2198 mock_disconnect_controller,
2199 mock_disconnect_model,
2200 mock_get_model,
2201 mock_get_controller,
2202 ):
2203 mock_get_model.return_value = juju.model.Model()
2204 mock_get_application.return_value = None
2205 result = self.loop.run_until_complete(
2206 self.libjuju.check_application_exists(
2207 "model",
2208 "app",
2209 )
2210 )
2211 self.assertEqual(result, False)
2212
2213 mock_get_application.assert_called_once()
2214 mock_get_controller.assert_called_once()
2215 mock_get_model.assert_called_once()
2216 mock_disconnect_controller.assert_called_once()
2217 mock_disconnect_model.assert_called_once()
2218
2219
2220 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
2221 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
2222 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
2223 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2224 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
2225 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_machine_info")
2226 class AddUnitTest(LibjujuTestCase):
2227 def setUp(self):
2228 super(AddUnitTest, self).setUp()
2229
2230 @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for")
2231 @asynctest.mock.patch("asyncio.sleep")
2232 def test_successful(
2233 self,
2234 mock_sleep,
2235 mock_wait_for,
2236 mock_get_machine_info,
2237 mock_get_application,
2238 mock_disconnect_controller,
2239 mock_disconnect_model,
2240 mock_get_model,
2241 mock_get_controller,
2242 ):
2243 mock_get_model.return_value = juju.model.Model()
2244 mock_get_application.return_value = FakeApplication()
2245 mock_get_machine_info.return_value = FakeMachine(), "series"
2246 self.loop.run_until_complete(
2247 self.libjuju.add_unit(
2248 "existing_app",
2249 "model",
2250 "machine",
2251 )
2252 )
2253
2254 mock_wait_for.assert_called_once()
2255 mock_get_application.assert_called_once()
2256 mock_get_controller.assert_called_once()
2257 mock_get_model.assert_called_once()
2258 mock_disconnect_controller.assert_called_once()
2259 mock_disconnect_model.assert_called_once()
2260
2261 def test_no_app(
2262 self,
2263 mock_get_machine_info,
2264 mock_get_application,
2265 mock_disconnect_controller,
2266 mock_disconnect_model,
2267 mock_get_model,
2268 mock_get_controller,
2269 ):
2270 mock_get_model.return_value = juju.model.Model()
2271 mock_get_application.return_value = None
2272 with self.assertRaises(JujuApplicationNotFound):
2273 self.loop.run_until_complete(
2274 self.libjuju.add_unit(
2275 "existing_app",
2276 "model",
2277 "machine",
2278 )
2279 )
2280
2281 mock_get_application.assert_called_once()
2282 mock_get_controller.assert_called_once()
2283 mock_get_model.assert_called_once()
2284 mock_disconnect_controller.assert_called_once()
2285 mock_disconnect_model.assert_called_once()
2286
2287 def test_no_machine(
2288 self,
2289 mock_get_machine_info,
2290 mock_get_application,
2291 mock_disconnect_controller,
2292 mock_disconnect_model,
2293 mock_get_model,
2294 mock_get_controller,
2295 ):
2296 mock_get_model.return_value = juju.model.Model()
2297 mock_get_application.return_value = FakeApplication()
2298 mock_get_machine_info.side_effect = JujuMachineNotFound()
2299 with self.assertRaises(JujuMachineNotFound):
2300 self.loop.run_until_complete(
2301 self.libjuju.add_unit(
2302 "existing_app",
2303 "model",
2304 "machine",
2305 )
2306 )
2307
2308 mock_get_application.assert_called_once()
2309 mock_get_controller.assert_called_once()
2310 mock_get_model.assert_called_once()
2311 mock_disconnect_controller.assert_called_once()
2312 mock_disconnect_model.assert_called_once()
2313
2314
2315 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
2316 @asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
2317 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
2318 @asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
2319 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
2320 @asynctest.mock.patch("n2vc.libjuju.Libjuju._get_unit")
2321 class DestroyUnitTest(LibjujuTestCase):
2322 def setUp(self):
2323 super(DestroyUnitTest, self).setUp()
2324
2325 @asynctest.mock.patch("asyncio.sleep")
2326 def test_successful(
2327 self,
2328 mock_sleep,
2329 mock_get_unit,
2330 mock_get_application,
2331 mock_disconnect_controller,
2332 mock_disconnect_model,
2333 mock_get_model,
2334 mock_get_controller,
2335 ):
2336 mock_get_model.return_value = juju.model.Model()
2337 mock_get_application.return_value = FakeApplication()
2338
2339 self.loop.run_until_complete(
2340 self.libjuju.destroy_unit("app", "model", "machine", 0)
2341 )
2342
2343 mock_get_unit.assert_called()
2344 mock_get_application.assert_called_once()
2345 mock_get_controller.assert_called_once()
2346 mock_get_model.assert_called_once()
2347 mock_disconnect_controller.assert_called_once()
2348 mock_disconnect_model.assert_called_once()
2349
2350 def test_no_app(
2351 self,
2352 mock_get_unit,
2353 mock_get_application,
2354 mock_disconnect_controller,
2355 mock_disconnect_model,
2356 mock_get_model,
2357 mock_get_controller,
2358 ):
2359 mock_get_model.return_value = juju.model.Model()
2360 mock_get_application.return_value = None
2361
2362 with self.assertRaises(JujuApplicationNotFound):
2363 self.loop.run_until_complete(
2364 self.libjuju.destroy_unit("app", "model", "machine")
2365 )
2366
2367 mock_get_application.assert_called_once()
2368 mock_get_controller.assert_called_once()
2369 mock_get_model.assert_called_once()
2370 mock_disconnect_controller.assert_called_once()
2371 mock_disconnect_model.assert_called_once()
2372
2373 def test_no_unit(
2374 self,
2375 mock_get_unit,
2376 mock_get_application,
2377 mock_disconnect_controller,
2378 mock_disconnect_model,
2379 mock_get_model,
2380 mock_get_controller,
2381 ):
2382 mock_get_model.return_value = juju.model.Model()
2383 mock_get_application.return_value = FakeApplication()
2384 mock_get_unit.return_value = None
2385
2386 with self.assertRaises(JujuError):
2387 self.loop.run_until_complete(
2388 self.libjuju.destroy_unit("app", "model", "machine")
2389 )
2390
2391 mock_get_unit.assert_called_once()
2392 mock_get_application.assert_called_once()
2393 mock_get_controller.assert_called_once()
2394 mock_get_model.assert_called_once()
2395 mock_disconnect_controller.assert_called_once()
2396 mock_disconnect_model.assert_called_once()