+
+
+class UpdateVcaStatusTest(K8sJujuConnTestCase):
+ def setUp(self):
+ super(UpdateVcaStatusTest, self).setUp()
+ self.vcaStatus = {"model": {"applications": {"app": {"actions": {}}}}}
+ self.kdu_name = "kdu_name"
+ self.kdu_instance = "{}-{}".format(self.kdu_name, "id")
+ self.k8s_juju_conn.libjuju.get_executed_actions = AsyncMock()
+ self.k8s_juju_conn.libjuju.get_actions = AsyncMock()
+ self.k8s_juju_conn.libjuju.get_application_configs = AsyncMock()
+
+ def test_success(self):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.update_vca_status(self.vcaStatus, self.kdu_instance)
+ )
+ self.k8s_juju_conn.libjuju.get_executed_actions.assert_called_once()
+ self.k8s_juju_conn.libjuju.get_actions.assert_called_once()
+ self.k8s_juju_conn.libjuju.get_application_configs.assert_called_once()
+
+ def test_exception(self):
+ self.k8s_juju_conn.libjuju.get_model.return_value = None
+ self.k8s_juju_conn.libjuju.get_executed_actions.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.update_vca_status(self.vcaStatus, self.kdu_instance)
+ )
+ self.k8s_juju_conn.libjuju.get_executed_actions.assert_not_called()
+ self.k8s_juju_conn.libjuju.get_actions.assert_not_called_once()
+ self.k8s_juju_conn.libjuju.get_application_configs.assert_not_called_once()
+
+
+class ScaleTest(K8sJujuConnTestCase):
+ def setUp(self):
+ super(ScaleTest, self).setUp()
+ self.application_name = "app"
+ self.kdu_name = "kdu-instance"
+ self._scale = 2
+ self.k8s_juju_conn.libjuju.scale_application = AsyncMock()
+
+ def test_success(self):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.scale(self.kdu_name, self._scale, self.application_name)
+ )
+ self.k8s_juju_conn.libjuju.scale_application.assert_called_once()
+
+ def test_exception(self):
+ self.k8s_juju_conn.libjuju.scale_application.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.scale(
+ self.kdu_name, self._scale, self.application_name
+ )
+ )
+ self.k8s_juju_conn.libjuju.scale_application.assert_called_once()
+
+
+class GetScaleCount(K8sJujuConnTestCase):
+ def setUp(self):
+ super(GetScaleCount, self).setUp()
+ self.k8s_juju_conn.libjuju.get_model_status = AsyncMock()
+
+ def test_success(self):
+ applications = {"app": FakeApplication()}
+ model = FakeModel(applications=applications)
+ self.k8s_juju_conn.libjuju.get_model_status.return_value = model
+ status = self.loop.run_until_complete(
+ self.k8s_juju_conn.get_scale_count("app", "kdu_instance")
+ )
+ self.assertEqual(status, 2)
+ self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()
+
+ def test_exception(self):
+ self.k8s_juju_conn.libjuju.get_model_status.side_effect = Exception()
+ status = None
+ with self.assertRaises(Exception):
+ status = self.loop.run_until_complete(
+ self.k8s_juju_conn.status_kdu("app", "kdu_instance")
+ )
+ self.assertIsNone(status)
+ self.k8s_juju_conn.libjuju.get_model_status.assert_called_once()
+
+
+class AddRelationTest(K8sJujuConnTestCase):
+ def setUp(self):
+ super(AddRelationTest, self).setUp()
+ self.k8s_juju_conn.libjuju.add_relation = AsyncMock()
+ self.k8s_juju_conn.libjuju.offer = AsyncMock()
+ self.k8s_juju_conn.libjuju.get_controller = AsyncMock()
+ self.k8s_juju_conn.libjuju.consume = AsyncMock()
+
+ def test_standard_relation(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint")
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with(
+ model_name="model-1", endpoint_1="app1:endpoint", endpoint_2="app2:endpoint"
+ )
+ self.k8s_juju_conn.libjuju.offer.assert_not_called()
+ self.k8s_juju_conn.libjuju.consume.assert_not_called()
+
+ def test_cmr_relation_same_controller(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
+ offer = Offer("admin/model-1.app1")
+ self.k8s_juju_conn.libjuju.offer.return_value = offer
+ self.k8s_juju_conn.libjuju.consume.return_value = "saas"
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.k8s_juju_conn.libjuju.offer.assert_called_once_with(relation_endpoint_1)
+ self.k8s_juju_conn.libjuju.consume.assert_called_once()
+ self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with(
+ "model-2", "app2:endpoint", "saas"
+ )
+
+ def test_relation_exception(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
+ self.k8s_juju_conn.libjuju.offer.side_effect = Exception()
+ with self.assertRaises(Exception):
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(
+ relation_endpoint_1, relation_endpoint_2
+ )
+ )