operation=OPERATION_TYPE.INSTANTIATE,
task=OPERATION_TASK.INSTANTIATE
)
- self.pre_deal()
def run(self):
try:
+ self.pre_deal()
self.inst_pre()
self.lcm_op_occ.notify_lcm(OPERATION_STATE_TYPE.STARTING)
self.apply_grant()
)
except NFLCMException as e:
self.vnf_inst_failed_handle(e.args[0])
+ except NFLCMExceptionConflict as e:
+ self.vnf_inst_failed_handle(e.args[0])
except Exception as e:
logger.error(str(e))
logger.error(traceback.format_exc())
# See the License for the specific language governing permissions and
# limitations under the License.
-
+import mock
from django.test import TestCase, Client
from rest_framework import status
from lcm.pub.database.models import StorageInstModel
from lcm.pub.database.models import NfvoRegInfoModel
from lcm.pub.utils.timeutil import now_time
+from lcm.nf.biz.delete_vnf import DeleteVnf
+from lcm.pub.exceptions import NFLCMException
class TestNFTerminate(TestCase):
def test_delete_vnf_identifier_when_vnf_not_exist(self):
response = self.client.delete("/api/vnflcm/v1/vnf_instances/1111")
self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
+
+ @mock.patch.object(DeleteVnf, 'do_biz')
+ def test_delete_vnf_inner_error(self, mock_DeleteVnf_do_biz):
+ mock_DeleteVnf_do_biz.side_effect = NFLCMException('Boom!')
+ response = self.client.delete("/api/vnflcm/v1/vnf_instances/1234")
+ self.assertEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code)
from .const import vnfpackage_info\r
from .const import instantiate_grant_result\r
\r
-from lcm.pub.database.models import NfInstModel\r
+from lcm.pub.database.models import NfInstModel, VNFLcmOpOccModel\r
from lcm.pub.database.models import JobStatusModel\r
from lcm.pub.database.models import SubscriptionModel\r
from lcm.pub.utils import restcall\r
from lcm.pub.exceptions import NFLCMException\r
\r
from lcm.nf.biz.instantiate_vnf import InstantiateVnf\r
+from lcm.nf import const\r
\r
\r
class TestNFInstantiate(TestCase):\r
nf_inst_id=self.nf_inst_id,\r
job_id=self.job_id\r
).run()\r
+ self.assert_job_result(\r
+ self.job_id,\r
+ 100,\r
+ 'Instantiate Vnf success.'\r
+ )\r
\r
@mock.patch.object(JobUtil, 'create_job')\r
def test_instantiate_inner_error(self, mock_run):\r
)\r
NfInstModel.objects.filter(nfinstid='144').delete()\r
self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)\r
+\r
+ @mock.patch.object(restcall, 'call_req')\r
+ def test_instantiate_operating_fail(self, mock_call_req):\r
+ NfInstModel.objects.create(\r
+ nfinstid='1111',\r
+ nf_name='vFW_01',\r
+ package_id='222',\r
+ version='',\r
+ vendor='',\r
+ netype='',\r
+ vnfd_model='',\r
+ status='NOT_INSTANTIATED',\r
+ nf_desc='vFW in Nanjing TIC Edge',\r
+ vnfdid='111',\r
+ create_time=now_time()\r
+ )\r
+ r1_get_vnfpackage_by_vnfdid = [\r
+ 0,\r
+ json.JSONEncoder().encode(vnfpackage_info),\r
+ '200'\r
+ ]\r
+ mock_call_req.side_effect = [\r
+ r1_get_vnfpackage_by_vnfdid\r
+ ]\r
+ self.nf_inst_id = '1111'\r
+ self.job_id = JobUtil.create_job('NF', 'CREATE', self.nf_inst_id)\r
+ JobUtil.add_job_status(self.job_id, 0, 'INST_VNF_READY')\r
+ VNFLcmOpOccModel.objects.create(vnf_instance_id=self.nf_inst_id,\r
+ id=self.job_id,\r
+ operation=const.OPERATION_TYPE.INSTANTIATE,\r
+ operation_state=const.OPERATION_STATE_TYPE.PROCESSING)\r
+ InstantiateVnf(\r
+ inst_req_data,\r
+ nf_inst_id=self.nf_inst_id,\r
+ job_id=self.job_id\r
+ ).run()\r
+ self.assert_job_result(\r
+ self.job_id,\r
+ 255,\r
+ 'VNF(%s) %s in processing.' % (self.nf_inst_id, const.OPERATION_TYPE.INSTANTIATE)\r
+ )\r