+
+
+class TestK8SClusterLcm(TestCase):
+ @patch("osm_lcm.vim_sdn.K8sHelmConnector")
+ @patch("osm_lcm.vim_sdn.K8sHelm3Connector")
+ @patch("osm_lcm.vim_sdn.K8sJujuConnector")
+ @patch("osm_lcm.lcm_utils.Database")
+ @patch("osm_lcm.lcm_utils.Filesystem")
+ def setUp(
+ self,
+ mock_filesystem,
+ mock_database,
+ juju_connector,
+ helm3_connector,
+ helm_connector,
+ ):
+ self.loop = asyncio.get_event_loop()
+ self.msg = Mock(msgbase.MsgBase())
+ self.lcm_tasks = Mock()
+ self.config = {"database": {"driver": "mongo"}}
+ self.vca_config = {
+ "VCA": {
+ "helmpath": "/usr/local/bin/helm",
+ "helm3path": "/usr/local/bin/helm3",
+ "kubectlpath": "/usr/bin/kubectl",
+ }
+ }
+ self.k8scluster_lcm = K8sClusterLcm(
+ self.msg, self.lcm_tasks, self.vca_config, self.loop
+ )
+ self.k8scluster_lcm.db = Mock()
+ self.k8scluster_lcm.fs = Mock()
+
+ def test_k8scluster_edit(self):
+ k8scluster_content = {"op_id": "op-id", "_id": "id"}
+ order_id = "order-id"
+ self.lcm_tasks.lock_HA.return_value = True
+ self.loop.run_until_complete(
+ self.k8scluster_lcm.edit(k8scluster_content, order_id)
+ )
+ self.lcm_tasks.unlock_HA.assert_called_with(
+ "k8scluster",
+ "edit",
+ "op-id",
+ operationState="COMPLETED",
+ detailed_status="Not implemented",
+ )
+ self.lcm_tasks.remove.assert_called_with("k8scluster", "id", order_id)
+
+ def test_k8scluster_edit_lock_false(self):
+ k8scluster_content = {"op_id": "op-id", "_id": "id"}
+ order_id = "order-id"
+ self.lcm_tasks.lock_HA.return_value = False
+ self.loop.run_until_complete(
+ self.k8scluster_lcm.edit(k8scluster_content, order_id)
+ )
+ self.lcm_tasks.unlock_HA.assert_not_called()
+ self.lcm_tasks.remove.assert_not_called()
+
+ def test_k8scluster_edit_no_opid(self):
+ k8scluster_content = {"_id": "id"}
+ order_id = "order-id"
+ self.lcm_tasks.lock_HA.return_value = True
+ self.loop.run_until_complete(
+ self.k8scluster_lcm.edit(k8scluster_content, order_id)
+ )
+ self.lcm_tasks.unlock_HA.assert_called_with(
+ "k8scluster",
+ "edit",
+ None,
+ operationState="COMPLETED",
+ detailed_status="Not implemented",
+ )
+ self.lcm_tasks.remove.assert_called_with("k8scluster", "id", order_id)
+
+ def test_k8scluster_edit_no_orderid(self):
+ k8scluster_content = {"op_id": "op-id", "_id": "id"}
+ order_id = None
+ self.lcm_tasks.lock_HA.return_value = True
+ self.loop.run_until_complete(
+ self.k8scluster_lcm.edit(k8scluster_content, order_id)
+ )
+ self.lcm_tasks.unlock_HA.assert_called_with(
+ "k8scluster",
+ "edit",
+ "op-id",
+ operationState="COMPLETED",
+ detailed_status="Not implemented",
+ )
+ self.lcm_tasks.remove.assert_called_with("k8scluster", "id", order_id)