import kubernetes
from juju.errors import JujuAPIError
import logging
-from .utils import FakeMachine, FakeApplication
+
+from n2vc.definitions import Offer, RelationEndpoint
+from .utils import (
+ FakeApplication,
+ FakeMachine,
+ FakeManualMachine,
+ FakeUnit,
+)
from n2vc.libjuju import Libjuju
from n2vc.exceptions import (
JujuControllerFailedConnecting,
with self.assertRaises(JujuControllerFailedConnecting):
controller = self.loop.run_until_complete(self.libjuju.get_controller())
self.assertIsNone(controller)
- mock_disconnect_controller.assert_called_once()
+ mock_disconnect_controller.assert_called()
def test_same_endpoint_get_controller(self, mock_api_endpoints, mock_connect):
self.libjuju.endpoints = ["127.0.0.1:17070"]
executed_actions = self.loop.run_until_complete(
self.libjuju.get_executed_actions("model")
)
- expected_result = [{'id': 'id', 'action': 'action_name',
- 'status': 'status', 'output': 'completed'}]
+ expected_result = [
+ {
+ "id": "id",
+ "action": "action_name",
+ "status": "status",
+ "output": "completed",
+ }
+ ]
self.assertListEqual(expected_result, executed_actions)
self.assertIsInstance(executed_actions, list)
mock_get_model.return_value = None
with self.assertRaises(JujuError):
self.loop.run_until_complete(
- self.libjuju.get_application_configs("model", "app"))
+ self.libjuju.get_application_configs("model", "app")
+ )
mock_get_controller.assert_called_once()
mock_disconnect_controller.assert_called_once()
mock_get_controller,
):
mock_get_application.return_value = FakeApplication()
- application_configs = self.loop.run_until_complete(self.libjuju
- .get_application_configs("model", "app"))
+ application_configs = self.loop.run_until_complete(
+ self.libjuju.get_application_configs("model", "app")
+ )
self.assertEqual(application_configs, ["app_config"])
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,
super(DestroyApplicationTest, self).setUp()
def test_success(
- self,
- mock_get_controller,
- mock_get_model,
- mock_disconnect_controller,
- mock_get_application,
- mock_disconnect_model,
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
):
mock_get_application.return_value = FakeApplication()
mock_get_model.return_value = None
mock_disconnect_model.assert_called_once()
def test_no_application(
- self,
- mock_get_controller,
- mock_get_model,
- mock_disconnect_controller,
- mock_get_application,
- mock_disconnect_model,
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
):
mock_get_model.return_value = None
mock_get_application.return_value = None
mock_get_application.assert_called()
def test_exception(
- self,
- mock_get_controller,
- mock_get_model,
- mock_disconnect_controller,
- mock_get_application,
- mock_disconnect_model,
+ self,
+ mock_get_controller,
+ mock_get_model,
+ mock_disconnect_controller,
+ mock_get_application,
+ mock_disconnect_model,
):
mock_get_application.return_value = FakeApplication
mock_get_model.return_value = None
mock_get_controller.return_value = juju.controller.Controller()
mock_list_offers.side_effect = Exception()
with self.assertRaises(Exception):
- self.loop.run_until_complete(self.libjuju.list_offers("model"))
+ self.loop.run_until_complete(self.libjuju._list_offers("model"))
mock_disconnect_controller.assert_called_once()
def test_empty_list(
mock_get_controller,
):
mock_get_controller.return_value = juju.controller.Controller()
- mock_list_offers.return_value = []
- offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
+ offer_results = Mock()
+ offer_results.results = []
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(self.libjuju._list_offers("model"))
self.assertEqual(offers, [])
mock_disconnect_controller.assert_called_once()
mock_get_controller,
):
mock_get_controller.return_value = juju.controller.Controller()
- mock_list_offers.return_value = ["offer"]
- offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
- self.assertEqual(offers, ["offer"])
+ offer = Mock()
+ offer_results = Mock()
+ offer_results.results = [offer]
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(self.libjuju._list_offers("model"))
+ self.assertEqual(offers, [offer])
+ mock_disconnect_controller.assert_called_once()
+
+ def test_matching_offer_name(
+ self,
+ mock_list_offers,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ offer_1 = Mock()
+ offer_1.offer_name = "offer1"
+ offer_2 = Mock()
+ offer_2.offer_name = "offer2"
+ offer_results = Mock()
+ offer_results.results = [offer_1, offer_2]
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(
+ self.libjuju._list_offers("model", offer_name="offer2")
+ )
+ self.assertEqual(offers, [offer_2])
mock_disconnect_controller.assert_called_once()
+ def test_not_matching_offer_name(
+ self,
+ mock_list_offers,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ offer_1 = Mock()
+ offer_1.offer_name = "offer1"
+ offer_2 = Mock()
+ offer_2.offer_name = "offer2"
+ offer_results = Mock()
+ offer_results.results = [offer_1, offer_2]
+ mock_list_offers.return_value = offer_results
+ offers = self.loop.run_until_complete(
+ self.libjuju._list_offers("model", offer_name="offer3")
+ )
+ self.assertEqual(offers, [])
+ mock_disconnect_controller.assert_called_once()
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("juju.controller.Controller.get_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._list_offers")
+@asynctest.mock.patch("juju.model.Model.create_offer")
+class OfferTest(LibjujuTestCase):
+ def setUp(self):
+ super(OfferTest, self).setUp()
+
+ def test_offer(
+ self,
+ mock_create_offer,
+ mock__list_offers,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ controller = juju.controller.Controller()
+ model = juju.model.Model()
+ mock_get_controller.return_value = controller
+ mock_get_model.return_value = model
+ endpoint = RelationEndpoint("model.app-name.0", "vca", "endpoint")
+ self.loop.run_until_complete(self.libjuju.offer(endpoint))
+ mock_create_offer.assert_called_with(
+ "app-name:endpoint", offer_name="app-name-endpoint"
+ )
+ mock_disconnect_model.assert_called_once_with(model)
+ mock_disconnect_controller.assert_called_once_with(controller)
+
+ def test_offer_exception(
+ self,
+ mock_create_offer,
+ mock__list_offers,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ controller = juju.controller.Controller()
+ model = juju.model.Model()
+ mock_get_controller.return_value = controller
+ mock_get_model.return_value = model
+ mock__list_offers.return_value = []
+ endpoint = RelationEndpoint("model.app-name.0", "vca", "endpoint")
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(self.libjuju.offer(endpoint))
+ mock_create_offer.assert_called_with(
+ "app-name:endpoint", offer_name="app-name-endpoint"
+ )
+ mock_disconnect_model.assert_called_once_with(model)
+ mock_disconnect_controller.assert_called_once_with(controller)
+
@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
@asynctest.mock.patch("juju.controller.Controller.get_model")
@asynctest.mock.patch("juju.model.Model.consume")
class ConsumeTest(LibjujuTestCase):
def setUp(self):
+ self.offer_url = "admin/model.offer_name"
super(ConsumeTest, self).setUp()
+ self.provider_libjuju = self.libjuju
def test_consume(
self,
mock_get_model,
mock_get_controller,
):
- mock_get_controller.return_value = juju.controller.Controller()
+ self_controller = juju.controller.Controller()
+ provider_controller = juju.controller.Controller()
+ mock_get_controller.side_effect = [self_controller, provider_controller]
mock_get_model.return_value = juju.model.Model()
- self.loop.run_until_complete(self.libjuju.consume("offer_url", "model_name"))
- mock_consume.assert_called_once()
+ self.loop.run_until_complete(
+ self.libjuju.consume(
+ "model_name",
+ Offer(self.offer_url, vca_id="vca-id"),
+ self.provider_libjuju,
+ )
+ )
+ mock_consume.assert_called_once_with(
+ "admin/model.offer_name",
+ application_alias="offer_name-model-vca-id",
+ controller=provider_controller,
+ )
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
def test_parsing_error_exception(
self,
with self.assertRaises(juju.offerendpoints.ParseError):
self.loop.run_until_complete(
- self.libjuju.consume("offer_url", "model_name")
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
)
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
def test_juju_error_exception(
self,
with self.assertRaises(juju.errors.JujuError):
self.loop.run_until_complete(
- self.libjuju.consume("offer_url", "model_name")
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
)
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
def test_juju_api_error_exception(
self,
with self.assertRaises(juju.errors.JujuAPIError):
self.loop.run_until_complete(
- self.libjuju.consume("offer_url", "model_name")
+ self.libjuju.consume(
+ "model_name", Offer(self.offer_url), self.provider_libjuju
+ )
)
mock_consume.assert_called_once()
mock_disconnect_model.assert_called_once()
- mock_disconnect_controller.assert_called_once()
+ self.assertEqual(mock_disconnect_controller.call_count, 2)
@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_k8s_cloud_credential")
"Cannot set both token and user/pass",
)
self.assertTrue(exception_raised)
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_model")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("n2vc.juju_watcher.JujuModelWatcher.wait_for_model")
+class ScaleApplicationTest(LibjujuTestCase):
+ def setUp(self):
+ super(ScaleApplicationTest, self).setUp()
+
+ @asynctest.mock.patch("asyncio.sleep")
+ def test_scale_application(
+ self,
+ mock_sleep,
+ mock_wait_for_model,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_application,
+ 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.scale_application("model", "app", 2))
+ mock_wait_for_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_no_application(
+ self,
+ mock_wait_for,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_application,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_application.return_value = None
+ mock_get_model.return_value = juju.model.Model()
+ with self.assertRaises(JujuApplicationNotFound):
+ self.loop.run_until_complete(
+ self.libjuju.scale_application("model", "app", 2)
+ )
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
+
+ def test_exception(
+ self,
+ mock_wait_for,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_application,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = None
+ mock_get_application.return_value = FakeApplication()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.scale_application("model", "app", 2, total_timeout=0)
+ )
+ mock_disconnect_controller.assert_called_once()
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju._get_application")
+class GetUnitNumberTest(LibjujuTestCase):
+ def setUp(self):
+ super(GetUnitNumberTest, self).setUp()
+
+ def test_successful_get_unit_number(
+ self,
+ mock_get_applications,
+ ):
+ mock_get_applications.return_value = FakeApplication()
+ model = juju.model.Model()
+ result = self.libjuju._get_application_count(model, "app")
+ self.assertEqual(result, 2)
+
+ def test_non_existing_application(
+ self,
+ mock_get_applications,
+ ):
+ mock_get_applications.return_value = None
+ 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()