):
self.libjuju.endpoints = []
mock__update_api_endpoints_db.side_effect = Exception()
+ controller = None
with self.assertRaises(JujuControllerFailedConnecting):
controller = self.loop.run_until_complete(self.libjuju.get_controller())
- self.assertIsNone(controller)
- mock_disconnect_controller.assert_called_once()
+ self.assertIsNone(controller)
+ mock_disconnect_controller.assert_called_once()
def test_same_endpoint_get_controller(
self, mock__update_api_endpoints_db, mock_api_endpoints, mock_connect
self.libjuju.add_model("existing_model", "cloud")
)
- mock_disconnect_controller.assert_called()
+ mock_disconnect_controller.assert_called()
# TODO Check two job executing at the same time and one returning without doing anything.
self.assertEqual(status, {"status"})
- def test_excpetion(
+ def test_exception(
self,
mock_get_status,
mock_disconnect_controller,
):
mock_get_model.return_value = juju.model.Model()
mock_get_status.side_effect = Exception()
-
+ status = None
with self.assertRaises(Exception):
status = self.loop.run_until_complete(
self.libjuju.get_model_status("model")
)
- mock_disconnect_controller.assert_called_once()
- mock_disconnect_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
- self.assertIsNone(status)
+ self.assertIsNone(status)
@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
mock_get_model,
mock_get_controller,
):
+ machine = None
+ bool_res = None
mock_get_model.return_value = juju.model.Model()
with self.assertRaises(JujuMachineNotFound):
machine, bool_res = self.loop.run_until_complete(
self.libjuju.create_machine("model", "non_existing_machine")
)
- self.assertIsNone(machine)
- self.assertIsNone(bool_res)
+ self.assertIsNone(machine)
+ self.assertIsNone(bool_res)
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
def test_no_machine(
self,
mock_get_model.return_value = juju.model.Model()
mock_applications.return_value = {"existing_app"}
+ application = None
with self.assertRaises(JujuApplicationExists):
application = self.loop.run_until_complete(
self.libjuju.deploy_charm("existing_app", "path", "model", "machine",)
)
- self.assertIsNone(application)
+ self.assertIsNone(application)
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
def test_non_existing_machine(
self,
):
mock_get_model.return_value = juju.model.Model()
mock_machines.return_value = {"existing_machine": FakeMachine()}
+ application = None
with self.assertRaises(JujuMachineNotFound):
application = self.loop.run_until_complete(
self.libjuju.deploy_charm("app", "path", "model", "machine",)
)
- self.assertIsNone(application)
+ self.assertIsNone(application)
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
def test_2_units(
self,
):
mock__get_application.return_value = None
mock_get_model.return_value = juju.model.Model()
-
+ output = None
+ status = None
with self.assertRaises(JujuApplicationNotFound):
output, status = self.loop.run_until_complete(
self.libjuju.execute_action("app", "model", "action",)
)
- self.assertIsNone(output)
- self.assertIsNone(status)
+ self.assertIsNone(output)
+ self.assertIsNone(status)
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
def test_no_action(
self,
mock_get_model.return_value = juju.model.Model()
mock__get_application.return_value = FakeApplication()
+ output = None
+ status = None
with self.assertRaises(JujuActionNotFound):
output, status = self.loop.run_until_complete(
self.libjuju.execute_action("app", "model", "action",)
)
- self.assertIsNone(output)
- self.assertIsNone(status)
+ self.assertIsNone(output)
+ self.assertIsNone(status)
- mock_disconnect_controller.assert_called()
- mock_disconnect_model.assert_called()
+ mock_disconnect_controller.assert_called()
+ mock_disconnect_model.assert_called()
# TODO no leader unit found exception
mock_get_controller,
):
mock_get_application.side_effect = Exception()
-
+ actions = None
with self.assertRaises(Exception):
actions = self.loop.run_until_complete(
self.libjuju.get_actions("app", "model")
)
- self.assertIsNone(actions)
- mock_disconnect_controller.assert_called_once()
- mock_disconnect_model.assert_called_once()
+ self.assertIsNone(actions)
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
def test_success(
self,
mock_add_relation.side_effect = JujuAPIError(result)
self.loop.run_until_complete(
- self.libjuju.add_relation(
- "model", "app1", "app2", "relation1", "relation2",
- )
+ self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",)
)
mock_warning.assert_called_with("Relation not found: not found")
mock_add_relation.side_effect = JujuAPIError(result)
self.loop.run_until_complete(
- self.libjuju.add_relation(
- "model", "app1", "app2", "relation1", "relation2",
- )
+ self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",)
)
mock_warning.assert_called_with("Relation already exists: already exists")
with self.assertRaises(JujuAPIError):
self.loop.run_until_complete(
- self.libjuju.add_relation(
- "model", "app1", "app2", "relation1", "relation2",
- )
+ self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",)
)
- mock_disconnect_controller.assert_called_once()
- mock_disconnect_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
def test_success(
self,
mock_get_model.return_value = juju.model.Model()
self.loop.run_until_complete(
- self.libjuju.add_relation(
- "model", "app1", "app2", "relation1", "relation2",
- )
+ self.libjuju.add_relation("model", "app1:relation1", "app2:relation2",)
)
- mock_add_relation.assert_called_with(
- relation1="app1:relation1", relation2="app2:relation2"
+ mock_add_relation.assert_called_with("app1:relation1", "app2:relation2")
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+
+ def test_saas(
+ self,
+ mock_add_relation,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_model.return_value = juju.model.Model()
+
+ self.loop.run_until_complete(
+ self.libjuju.add_relation("model", "app1:relation1", "saas_name",)
)
+
+ mock_add_relation.assert_called_with("app1:relation1", "saas_name")
mock_disconnect_controller.assert_called_once()
mock_disconnect_model.assert_called_once()
self.loop.run_until_complete(
self.libjuju.configure_application("model", "app", {"config"},)
)
- mock_disconnect_controller.assert_called_once()
- mock_disconnect_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+ mock_disconnect_model.assert_called_once()
# TODO _get_api_endpoints_db test case
offers = self.loop.run_until_complete(self.libjuju.list_offers("model"))
self.assertEqual(offers, ["offer"])
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("juju.model.Model.consume")
+class ConsumeTest(LibjujuTestCase):
+ def setUp(self):
+ super(ConsumeTest, self).setUp()
+
+ def test_consume(
+ self,
+ mock_consume,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.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()
+ mock_disconnect_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+
+ def test_parsing_error_exception(
+ self,
+ mock_consume,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_get_model.return_value = juju.model.Model()
+ mock_consume.side_effect = juju.offerendpoints.ParseError("")
+
+ with self.assertRaises(juju.offerendpoints.ParseError):
+ self.loop.run_until_complete(
+ self.libjuju.consume("offer_url", "model_name")
+ )
+ mock_consume.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+
+ def test_juju_error_exception(
+ self,
+ mock_consume,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_get_model.return_value = juju.model.Model()
+ mock_consume.side_effect = juju.errors.JujuError("")
+
+ with self.assertRaises(juju.errors.JujuError):
+ self.loop.run_until_complete(
+ self.libjuju.consume("offer_url", "model_name")
+ )
+ mock_consume.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+
+ def test_juju_api_error_exception(
+ self,
+ mock_consume,
+ mock_disconnect_controller,
+ mock_disconnect_model,
+ mock_get_model,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_get_model.return_value = juju.model.Model()
+ mock_consume.side_effect = juju.errors.JujuAPIError(
+ {"error": "", "response": "", "request-id": ""}
+ )
+
+ with self.assertRaises(juju.errors.JujuAPIError):
+ self.loop.run_until_complete(
+ self.libjuju.consume("offer_url", "model_name")
+ )
+ mock_consume.assert_called_once()
+ mock_disconnect_model.assert_called_once()
+ mock_disconnect_controller.assert_called_once()
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.add_cloud")
+class AddK8sTest(LibjujuTestCase):
+ def setUp(self):
+ super(AddK8sTest, self).setUp()
+ self.auth_data = {
+ "server": "https://192.168.0.21:16443",
+ "token": "1234",
+ "cacert": "cacert",
+ }
+
+ def test_add_k8s(self, mock_add_cloud):
+ self.loop.run_until_complete(
+ self.libjuju.add_k8s("cloud", self.auth_data, "storage_class")
+ )
+ mock_add_cloud.assert_called_once()
+
+ def test_add_k8s_exception(self, mock_add_cloud):
+ mock_add_cloud.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.add_k8s("cloud", self.auth_data, "storage_class")
+ )
+ mock_add_cloud.assert_called_once()
+
+ def test_add_k8s_missing_name(self, mock_add_cloud):
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.add_k8s("", self.auth_data, "storage_class")
+ )
+ mock_add_cloud.assert_not_called()
+
+ def test_add_k8s_missing_storage_name(self, mock_add_cloud):
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.add_k8s("cloud", self.auth_data, "")
+ )
+ mock_add_cloud.assert_not_called()
+
+ def test_add_k8s_missing_auth_data_keys(self, mock_add_cloud):
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.add_k8s("cloud", {}, "")
+ )
+ mock_add_cloud.assert_not_called()
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("juju.controller.Controller.add_cloud")
+@asynctest.mock.patch("juju.controller.Controller.add_credential")
+class AddCloudTest(LibjujuTestCase):
+ def setUp(self):
+ super(AddCloudTest, self).setUp()
+ self.cloud = juju.client.client.Cloud()
+ self.credential = juju.client.client.CloudCredential()
+
+ def test_add_cloud_with_credential(
+ self,
+ mock_add_credential,
+ mock_add_cloud,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+
+ cloud = self.loop.run_until_complete(
+ self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
+ )
+ self.assertEqual(cloud, self.cloud)
+ mock_add_cloud.assert_called_once_with("cloud", self.cloud)
+ mock_add_credential.assert_called_once_with(
+ "cloud", credential=self.credential, cloud="cloud"
+ )
+ mock_disconnect_controller.assert_called_once()
+
+ def test_add_cloud_no_credential(
+ self,
+ mock_add_credential,
+ mock_add_cloud,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+
+ cloud = self.loop.run_until_complete(
+ self.libjuju.add_cloud("cloud", self.cloud)
+ )
+ self.assertEqual(cloud, self.cloud)
+ mock_add_cloud.assert_called_once_with("cloud", self.cloud)
+ mock_add_credential.assert_not_called()
+ mock_disconnect_controller.assert_called_once()
+
+ def test_add_cloud_exception(
+ self,
+ mock_add_credential,
+ mock_add_cloud,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_add_cloud.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
+ )
+
+ mock_add_cloud.assert_called_once_with("cloud", self.cloud)
+ mock_add_credential.assert_not_called()
+ mock_disconnect_controller.assert_called_once()
+
+ def test_add_credential_exception(
+ self,
+ mock_add_credential,
+ mock_add_cloud,
+ mock_disconnect_controller,
+ mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_add_credential.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.libjuju.add_cloud("cloud", self.cloud, credential=self.credential)
+ )
+
+ mock_add_cloud.assert_called_once_with("cloud", self.cloud)
+ mock_add_credential.assert_called_once_with(
+ "cloud", credential=self.credential, cloud="cloud"
+ )
+ mock_disconnect_controller.assert_called_once()
+
+
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.get_controller")
+@asynctest.mock.patch("n2vc.libjuju.Libjuju.disconnect_controller")
+@asynctest.mock.patch("juju.controller.Controller.remove_cloud")
+class RemoveCloudTest(LibjujuTestCase):
+ def setUp(self):
+ super(RemoveCloudTest, self).setUp()
+
+ def test_remove_cloud(
+ self, mock_remove_cloud, mock_disconnect_controller, mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+
+ self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
+ mock_remove_cloud.assert_called_once_with("cloud")
+ mock_disconnect_controller.assert_called_once()
+
+ def test_remove_cloud_exception(
+ self, mock_remove_cloud, mock_disconnect_controller, mock_get_controller,
+ ):
+ mock_get_controller.return_value = juju.controller.Controller()
+ mock_remove_cloud.side_effect = Exception()
+
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(self.libjuju.remove_cloud("cloud"))
+ mock_remove_cloud.assert_called_once_with("cloud")
+ mock_disconnect_controller.assert_called_once()