X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_lcm%2Ftests%2Ftest_vim_sdn.py;h=970dd00d477b0ae103ebfe4ec6de6094f45eb489;hb=8e53c6d0046a3015bf015b745aa3cf9a30a8fbe6;hp=f6b75e0f2b2052532f150248a85666336db0a84d;hpb=120695e9fe7ee315b83466d664a0f143c5f90044;p=osm%2FLCM.git diff --git a/osm_lcm/tests/test_vim_sdn.py b/osm_lcm/tests/test_vim_sdn.py index f6b75e0..970dd00 100644 --- a/osm_lcm/tests/test_vim_sdn.py +++ b/osm_lcm/tests/test_vim_sdn.py @@ -116,6 +116,164 @@ class TestVcaLcm(TestCase): 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"