X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Ftests%2Funit%2Ftest_libjuju.py;h=918a2fb87b2fc39decf78181d10e535a40bf7194;hp=724b5d7a491152beb6adda5be213166e8c966219;hb=fa02f8a90b7fe1e1b7a80feedef4132bef1ca3e4;hpb=ba1d07263821169bc36b3f9f9f93cbed93baad4a diff --git a/n2vc/tests/unit/test_libjuju.py b/n2vc/tests/unit/test_libjuju.py index 724b5d7..918a2fb 100644 --- a/n2vc/tests/unit/test_libjuju.py +++ b/n2vc/tests/unit/test_libjuju.py @@ -20,7 +20,12 @@ import juju import kubernetes from juju.errors import JujuAPIError import logging -from .utils import FakeMachine, FakeApplication +from .utils import ( + FakeApplication, + FakeMachine, + FakeManualMachine, + FakeUnit, +) from n2vc.libjuju import Libjuju from n2vc.exceptions import ( JujuControllerFailedConnecting, @@ -824,7 +829,7 @@ class ExecuteActionTest(LibjujuTestCase): mock_disconnect_controller.assert_called() mock_disconnect_model.assert_called() - def test_succesful_exec( + def test_successful_exec( self, mock_get_action_status, mock_get_action_output, @@ -1966,7 +1971,7 @@ class GetUnitNumberTest(LibjujuTestCase): def setUp(self): super(GetUnitNumberTest, self).setUp() - def test_succesful_get_unit_number( + def test_successful_get_unit_number( self, mock_get_applications, ): @@ -1983,3 +1988,287 @@ class GetUnitNumberTest(LibjujuTestCase): model = juju.model.Model() result = self.libjuju._get_application_count(model, "app") self.assertEqual(result, None) + + +@asynctest.mock.patch("juju.model.Model.machines", new_callable=asynctest.PropertyMock) +class GetMachineInfoTest(LibjujuTestCase): + def setUp(self): + super(GetMachineInfoTest, self).setUp() + + def test_successful( + self, + mock_machines, + ): + machine_id = "existing_machine" + model = juju.model.Model() + mock_machines.return_value = {"existing_machine": FakeManualMachine()} + machine, series = self.libjuju._get_machine_info( + machine_id=machine_id, + model=model, + ) + self.assertIsNotNone(machine, series) + + def test_exception( + self, + mock_machines, + ): + machine_id = "not_existing_machine" + machine = series = None + model = juju.model.Model() + mock_machines.return_value = {"existing_machine": FakeManualMachine()} + with self.assertRaises(JujuMachineNotFound): + machine, series = self.libjuju._get_machine_info( + machine_id=machine_id, + model=model, + ) + self.assertIsNone(machine, series) + + +class GetUnitTest(LibjujuTestCase): + def setUp(self): + super(GetUnitTest, self).setUp() + + def test_successful(self): + result = self.libjuju._get_unit(FakeApplication(), "existing_machine_id") + self.assertIsInstance(result, FakeUnit) + + def test_return_none(self): + result = self.libjuju._get_unit(FakeApplication(), "not_existing_machine_id") + self.assertIsNone(result) + + +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") +class CheckApplicationExists(LibjujuTestCase): + def setUp(self): + super(CheckApplicationExists, self).setUp() + + def test_successful( + self, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + result = self.loop.run_until_complete( + self.libjuju.check_application_exists( + "model", + "app", + ) + ) + self.assertEqual(result, True) + + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_application( + self, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = None + result = self.loop.run_until_complete( + self.libjuju.check_application_exists( + "model", + "app", + ) + ) + self.assertEqual(result, False) + + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_machine_info") +class AddUnitTest(LibjujuTestCase): + def setUp(self): + super(AddUnitTest, self).setUp() + + @asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for") + @asynctest.mock.patch("asyncio.sleep") + def test_successful( + self, + mock_sleep, + mock_wait_for, + mock_get_machine_info, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + mock_get_machine_info.return_value = FakeMachine(), "series" + self.loop.run_until_complete( + self.libjuju.add_unit( + "existing_app", + "model", + "machine", + ) + ) + + mock_wait_for.assert_called_once() + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_app( + self, + mock_get_machine_info, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = None + with self.assertRaises(JujuApplicationNotFound): + self.loop.run_until_complete( + self.libjuju.add_unit( + "existing_app", + "model", + "machine", + ) + ) + + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_machine( + self, + mock_get_machine_info, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + mock_get_machine_info.side_effect = JujuMachineNotFound() + with self.assertRaises(JujuMachineNotFound): + self.loop.run_until_complete( + self.libjuju.add_unit( + "existing_app", + "model", + "machine", + ) + ) + + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model") +@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application") +@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_unit") +class DestroyUnitTest(LibjujuTestCase): + def setUp(self): + super(DestroyUnitTest, self).setUp() + + @asynctest.mock.patch("asyncio.sleep") + def test_successful( + self, + mock_sleep, + mock_get_unit, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + + self.loop.run_until_complete( + self.libjuju.destroy_unit("app", "model", "machine", 0) + ) + + mock_get_unit.assert_called() + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_app( + self, + mock_get_unit, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = None + + with self.assertRaises(JujuApplicationNotFound): + self.loop.run_until_complete( + self.libjuju.destroy_unit("app", "model", "machine") + ) + + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once() + + def test_no_unit( + self, + mock_get_unit, + mock_get_application, + mock_disconnect_controller, + mock_disconnect_model, + mock_get_model, + mock_get_controller, + ): + mock_get_model.return_value = juju.model.Model() + mock_get_application.return_value = FakeApplication() + mock_get_unit.return_value = None + + with self.assertRaises(JujuError): + self.loop.run_until_complete( + self.libjuju.destroy_unit("app", "model", "machine") + ) + + mock_get_unit.assert_called_once() + mock_get_application.assert_called_once() + mock_get_controller.assert_called_once() + mock_get_model.assert_called_once() + mock_disconnect_controller.assert_called_once() + mock_disconnect_model.assert_called_once()