)
logging.disable(logging.CRITICAL)
+ self.kdu_name = "kdu_name"
+ self.kdu_instance = "{}-{}".format(self.kdu_name, "id")
+ self.default_namespace = self.kdu_instance
+
self.k8s_juju_conn = K8sJujuConnector(
fs=fslocal.FsLocal(),
db=self.db,
self.kubectl.get_services.return_value = [{}]
self.k8s_juju_conn._get_kubectl = Mock()
self.k8s_juju_conn._get_kubectl.return_value = self.kubectl
+ self.k8s_juju_conn._obtain_namespace_from_db = Mock(
+ return_value=self.default_namespace
+ )
class InitEnvTest(K8sJujuConnTestCase):
self.local_bundle = "bundle"
self.cs_bundle = "cs:bundle"
self.http_bundle = "https://example.com/bundle.yaml"
- self.kdu_name = "kdu_name"
self.cluster_uuid = "cluster"
- self.kdu_instance = "{}-{}".format(self.kdu_name, "id")
self.k8s_juju_conn.libjuju.add_model = AsyncMock()
self.k8s_juju_conn.libjuju.deploy = AsyncMock()
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
"local:{}".format(self.local_bundle),
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.cs_bundle,
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.http_bundle,
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.cs_bundle,
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.cs_bundle,
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
"local:{}".format(self.local_bundle),
- model_name=self.kdu_instance,
+ model_name=self.default_namespace,
wait=True,
timeout=1800,
)
super(ExecPrimitivesTest, self).setUp()
self.action_name = "touch"
self.application_name = "myapp"
- self.model_name = "model"
self.k8s_juju_conn.libjuju.get_actions = AsyncMock()
self.k8s_juju_conn.libjuju.execute_action = AsyncMock()
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, self.action_name, params=params
+ "cluster", self.kdu_instance, self.action_name, params=params
)
)
self.assertEqual(output, "success")
+ self.k8s_juju_conn._obtain_namespace_from_db.assert_called_once_with(
+ kdu_instance=self.kdu_instance
+ )
self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
- self.application_name, self.model_name
+ application_name=self.application_name, model_name=self.default_namespace
)
self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
- self.application_name, self.model_name, self.action_name, **params
+ application_name=self.application_name,
+ model_name=self.default_namespace,
+ action_name=self.action_name,
+ **params
)
def test_exception(self):
with self.assertRaises(Exception):
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, self.action_name, params=params
+ "cluster", self.kdu_instance, self.action_name, params=params
)
)
self.assertIsNone(output)
+ self.k8s_juju_conn._obtain_namespace_from_db.assert_called_once_with(
+ kdu_instance=self.kdu_instance
+ )
self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
- self.application_name, self.model_name
+ application_name=self.application_name, model_name=self.default_namespace
)
self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
- self.application_name, self.model_name, self.action_name, **params
+ application_name=self.application_name,
+ model_name=self.default_namespace,
+ action_name=self.action_name,
+ **params
)
def test_missing_application_name_in_params(self):
with self.assertRaises(K8sException):
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, self.action_name, params=params
+ "cluster", self.kdu_instance, self.action_name, params=params
)
)
with self.assertRaises(K8sException):
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, self.action_name
+ "cluster", self.kdu_instance, self.action_name
)
)
with self.assertRaises(K8sException):
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, "non-existing-action", params=params
+ "cluster", self.kdu_instance, "non-existing-action", params=params
)
)
self.assertIsNone(output)
+ self.k8s_juju_conn._obtain_namespace_from_db.assert_called_once_with(
+ kdu_instance=self.kdu_instance
+ )
self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
- self.application_name, self.model_name
+ application_name=self.application_name, model_name=self.default_namespace
)
self.k8s_juju_conn.libjuju.execute_action.assert_not_called()
with self.assertRaises(K8sException):
output = self.loop.run_until_complete(
self.k8s_juju_conn.exec_primitive(
- "cluster", self.model_name, self.action_name, params=params
+ "cluster", self.kdu_instance, self.action_name, params=params
)
)
self.assertIsNone(output)
+ self.k8s_juju_conn._obtain_namespace_from_db.assert_called_once_with(
+ kdu_instance=self.kdu_instance
+ )
self.k8s_juju_conn.libjuju.get_actions.assert_called_once_with(
- self.application_name, self.model_name
+ application_name=self.application_name, model_name=self.default_namespace
)
self.k8s_juju_conn.libjuju.execute_action.assert_called_once_with(
- self.application_name, self.model_name, self.action_name, **params
+ application_name=self.application_name,
+ model_name=self.default_namespace,
+ action_name=self.action_name,
+ **params
)
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()
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.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()