self.lcm_tasks.unlock_HA.assert_not_called()
self.lcm_tasks.remove.assert_called_with("vca", "id", "order-id")
+ def test_vca_lcm_edit_success_no_config(self):
+ vca_content = {
+ "op_id": "order-id",
+ "_id": "id",
+ "description": "test-description",
+ }
+ db_vca = {
+ "_id": "vca-id",
+ "secret": "secret",
+ "cacert": "cacert",
+ "schema_version": "1.11",
+ }
+ order_id = "order-id"
+ self.lcm_tasks.lock_HA.return_value = True
+ self.vca_lcm.db.get_one.return_value = db_vca
+ self.vca_lcm.n2vc.validate_vca = AsyncMock()
+ self.vca_lcm.update_db_2 = Mock()
+ self.loop.run_until_complete(self.vca_lcm.edit(vca_content, order_id))
+ self.vca_lcm.n2vc.validate_vca.assert_not_called()
+ self.lcm_tasks.unlock_HA.assert_called_with(
+ "vca",
+ "edit",
+ "order-id",
+ operationState="COMPLETED",
+ detailed_status="Edited",
+ )
+ self.vca_lcm.update_db_2.assert_called_with(
+ "vca",
+ "id",
+ {},
+ )
+ self.lcm_tasks.remove.assert_called_with("vca", "id", "order-id")
+
+ def test_vca_lcm_edit_success_config(self):
+ vca_content = {"op_id": "order-id", "_id": "id", "cacert": "editcacert"}
+ db_vca = {
+ "_id": "vca-id",
+ "secret": "secret",
+ "cacert": "cacert",
+ "schema_version": "1.11",
+ }
+ order_id = "order-id"
+ self.lcm_tasks.lock_HA.return_value = True
+ self.vca_lcm.db.get_one.return_value = db_vca
+ self.vca_lcm.n2vc.validate_vca = AsyncMock()
+ self.vca_lcm.update_db_2 = Mock()
+ self.loop.run_until_complete(self.vca_lcm.edit(vca_content, order_id))
+ self.vca_lcm.n2vc.validate_vca.assert_called()
+ self.lcm_tasks.unlock_HA.assert_called_with(
+ "vca",
+ "edit",
+ "order-id",
+ operationState="COMPLETED",
+ detailed_status="Edited",
+ )
+ self.vca_lcm.update_db_2.assert_called_with(
+ "vca",
+ "id",
+ {
+ "_admin.operationalState": "ENABLED",
+ "_admin.detailed-status": "Connectivity: ok",
+ },
+ )
+ self.lcm_tasks.remove.assert_called_with("vca", "id", "order-id")
+
+ def test_vca_lcm_edit_exception_no_config(self):
+ vca_content = {
+ "op_id": "order-id",
+ "_id": "id",
+ "description": "new-description",
+ }
+ db_vca = {
+ "_id": "vca-id",
+ "secret": "secret",
+ "cacert": "cacert",
+ "schema_version": "1.11",
+ }
+ order_id = "order-id"
+ self.lcm_tasks.lock_HA.return_value = True
+ self.vca_lcm.db.get_one.return_value = db_vca
+ self.vca_lcm.n2vc.validate_vca = AsyncMock()
+ # validate_vca should not be called in this case
+ self.vca_lcm.n2vc.validate_vca.side_effect = Exception("failed")
+ self.vca_lcm.update_db_2 = Mock()
+ self.loop.run_until_complete(self.vca_lcm.edit(vca_content, order_id))
+ self.lcm_tasks.lock_HA.assert_called_with("vca", "edit", "order-id")
+ self.lcm_tasks.unlock_HA.assert_called_with(
+ "vca",
+ "edit",
+ "order-id",
+ operationState="COMPLETED",
+ detailed_status="Edited",
+ )
+ self.lcm_tasks.remove.assert_called_with("vca", "id", "order-id")
+
+ def test_vca_lcm_edit_exception_config(self):
+ vca_content = {"op_id": "order-id", "_id": "id", "user": "new-user"}
+ db_vca = {
+ "_id": "vca-id",
+ "secret": "secret",
+ "cacert": "cacert",
+ "schema_version": "1.11",
+ }
+ order_id = "order-id"
+ self.lcm_tasks.lock_HA.return_value = True
+ self.vca_lcm.db.get_one.return_value = db_vca
+ self.vca_lcm.n2vc.validate_vca = AsyncMock()
+ # validate_vca should be called in this case
+ self.vca_lcm.n2vc.validate_vca.side_effect = Exception("failed")
+ self.vca_lcm.update_db_2 = Mock()
+ self.loop.run_until_complete(self.vca_lcm.edit(vca_content, order_id))
+ self.lcm_tasks.lock_HA.assert_called_with("vca", "edit", "order-id")
+ self.lcm_tasks.unlock_HA.assert_called_with(
+ "vca",
+ "edit",
+ "order-id",
+ operationState="FAILED",
+ detailed_status="Failed with exception: failed",
+ )
+ self.vca_lcm.update_db_2.assert_called_with(
+ "vca",
+ "id",
+ {
+ "_admin.operationalState": "ERROR",
+ "_admin.detailed-status": "Failed with exception: failed",
+ },
+ )
+ self.lcm_tasks.remove.assert_called_with("vca", "id", "order-id")
+
+ def test_vca_lcm_edit_db_exception(self):
+ vca_content = {
+ "op_id": "order-id",
+ "_id": "id",
+ "description": "new-description",
+ }
+ db_vca = {
+ "_id": "vca-id",
+ "secret": "secret",
+ "cacert": "cacert",
+ "schema_version": "1.11",
+ }
+ order_id = "order-id"
+ self.lcm_tasks.lock_HA.return_value = True
+ self.vca_lcm.db.get_one.return_value = db_vca
+ self.vca_lcm.n2vc.validate_vca = AsyncMock()
+ self.vca_lcm.update_db_2 = Mock()
+ self.vca_lcm.update_db_2.side_effect = DbException("failed")
+ self.loop.run_until_complete(self.vca_lcm.edit(vca_content, order_id))
+ self.vca_lcm.n2vc.validate_vca.assert_not_called()
+ self.lcm_tasks.lock_HA.assert_called_with("vca", "edit", "order-id")
+ self.vca_lcm.update_db_2.assert_called_with(
+ "vca",
+ "id",
+ {},
+ )
+ self.lcm_tasks.unlock_HA.assert_not_called()
+ self.lcm_tasks.remove.assert_called_with("vca", "id", "order-id")
+
def test_vca_lcm_delete(self):
vca_content = {"op_id": "order-id", "_id": "id"}
order_id = "order-id"