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()
def test_success_local(self, mock_chdir):
- expected_kdu_instance = "{}-{}".format(self.kdu_name, "id")
- kdu_instance = self.loop.run_until_complete(
+ self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
self.local_bundle,
+ self.kdu_instance,
atomic=True,
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
)
)
- self.assertEqual(kdu_instance, expected_kdu_instance)
self.assertEqual(mock_chdir.call_count, 2)
self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
cloud_name=self.cluster_uuid,
credential_name="cred-{}".format(self.cluster_uuid),
)
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
"local:{}".format(self.local_bundle),
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
wait=True,
timeout=1800,
)
def test_success_cs(self, mock_chdir):
- expected_kdu_instance = "{}-{}".format(self.kdu_name, "id")
- kdu_instance = self.loop.run_until_complete(
+ self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
self.cs_bundle,
+ self.kdu_instance,
atomic=True,
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
)
)
- self.assertEqual(kdu_instance, expected_kdu_instance)
self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
cloud_name=self.cluster_uuid,
credential_name="cred-{}".format(self.cluster_uuid),
)
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.cs_bundle,
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
wait=True,
timeout=1800,
)
def test_success_http(self, mock_chdir):
- expected_kdu_instance = "{}-{}".format(self.kdu_name, "id")
- kdu_instance = self.loop.run_until_complete(
+ self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
self.http_bundle,
+ self.kdu_instance,
atomic=True,
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
)
)
- self.assertEqual(kdu_instance, expected_kdu_instance)
self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
cloud_name=self.cluster_uuid,
credential_name="cred-{}".format(self.cluster_uuid),
)
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.http_bundle,
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
wait=True,
timeout=1800,
)
def test_success_not_kdu_name(self, mock_chdir):
- expected_kdu_instance = "id"
- kdu_instance = self.loop.run_until_complete(
+ self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
self.cs_bundle,
+ self.kdu_instance,
atomic=True,
db_dict=self.db_dict,
timeout=1800,
)
)
- self.assertEqual(kdu_instance, expected_kdu_instance)
self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
cloud_name=self.cluster_uuid,
credential_name="cred-{}".format(self.cluster_uuid),
)
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.cs_bundle,
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
wait=True,
timeout=1800,
)
def test_missing_db_dict(self, mock_chdir):
kdu_instance = None
with self.assertRaises(K8sException):
- kdu_instance = self.loop.run_until_complete(
+ self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
self.cs_bundle,
+ self.kdu_instance,
atomic=True,
kdu_name=self.kdu_name,
timeout=1800,
@asynctest.mock.patch("os.getcwd")
def test_getcwd_exception(self, mock_getcwd, mock_chdir):
mock_getcwd.side_effect = FileNotFoundError()
- expected_kdu_instance = "{}-{}".format(self.kdu_name, "id")
- kdu_instance = self.loop.run_until_complete(
+ self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
self.cs_bundle,
+ self.kdu_instance,
atomic=True,
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
)
)
- self.assertEqual(kdu_instance, expected_kdu_instance)
self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
cloud_name=self.cluster_uuid,
credential_name="cred-{}".format(self.cluster_uuid),
)
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
self.cs_bundle,
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
wait=True,
timeout=1800,
)
def test_missing_bundle(self, mock_chdir):
- kdu_instance = None
with self.assertRaises(K8sException):
- kdu_instance = self.loop.run_until_complete(
+ self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
"",
+ self.kdu_instance,
atomic=True,
kdu_name=self.kdu_name,
timeout=1800,
db_dict=self.db_dict,
)
)
- self.assertIsNone(kdu_instance)
self.k8s_juju_conn.libjuju.add_model.assert_not_called()
self.k8s_juju_conn.libjuju.deploy.assert_not_called()
def test_missing_exception(self, mock_chdir):
- expected_kdu_instance = "{}-{}".format(self.kdu_name, "id")
- kdu_instance = None
self.k8s_juju_conn.libjuju.deploy.side_effect = Exception()
with self.assertRaises(Exception):
- kdu_instance = self.loop.run_until_complete(
+ self.loop.run_until_complete(
self.k8s_juju_conn.install(
self.cluster_uuid,
self.local_bundle,
+ self.kdu_instance,
atomic=True,
kdu_name=self.kdu_name,
db_dict=self.db_dict,
timeout=1800,
)
)
- self.assertIsNone(kdu_instance)
self.k8s_juju_conn.libjuju.add_model.assert_called_once_with(
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
cloud_name=self.cluster_uuid,
credential_name="cred-{}".format(self.cluster_uuid),
)
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
"local:{}".format(self.local_bundle),
- model_name=expected_kdu_instance,
+ model_name=self.kdu_instance,
wait=True,
timeout=1800,
)