lcm/lcm/nf/tests/test_create_vnf.py

   1 # Copyright 2017 ZTE Corporation.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import json
  16 import uuid
  17 import mock
  18
  19 from django.test import TestCase
  20 from rest_framework import status
  21 from rest_framework.test import APIClient
  22
  23 from .const import vnfpackage_info
  24 from lcm.pub.database.models import NfInstModel
  25 from lcm.pub.database.models import JobStatusModel
  26 from lcm.pub.utils import restcall
  27 from lcm.pub.utils.timeutil import now_time
  28 from lcm.pub.exceptions import NFLCMException
  29 from lcm.nf.biz.create_vnf import CreateVnf
  30
  31
  32 class TestNFInstantiate(TestCase):
  33     def setUp(self):
  34         self.client = APIClient()
  35
  36         self.grant_result = {
  37             "vimid": 'vimid_1',
  38             "tenant": 'tenantname_1'
  39         }
  40
  41     def tearDown(self):
  42         pass
  43
  44     def assert_job_result(self, job_id, job_progress, job_detail):
  45         jobs = JobStatusModel.objects.filter(
  46             jobid=job_id,
  47             progress=job_progress,
  48             descp=job_detail
  49         )
  50         self.assertEqual(1, len(jobs))
  51
  52     def test_create_vnf_identifier_when_vnf_is_exist(self):
  53         NfInstModel.objects.create(
  54             nfinstid='1111',
  55             nf_name='vFW_01',
  56             package_id='222',
  57             version='',
  58             vendor='',
  59             netype='',
  60             vnfd_model='',
  61             status='NOT_INSTANTIATED',
  62             nf_desc='vFW in Nanjing TIC Edge',
  63             vnfdid='111',
  64             create_time=now_time()
  65         )
  66         data = {
  67             "vnfdId": "111",
  68             "vnfInstanceName": "vFW_01",
  69             "vnfInstanceDescription": "vFW in Nanjing TIC Edge"
  70         }
  71         response = self.client.post(
  72             "/api/vnflcm/v1/vnf_instances",
  73             data=data,
  74             format='json'
  75         )
  76         self.assertEqual(
  77             status.HTTP_500_INTERNAL_SERVER_ERROR,
  78             response.status_code
  79         )
  80         context = json.loads(response.content)
  81         self.assertEqual({
  82             'detail': 'VNF is already exist.',
  83             'status': 500
  84         }, context)
  85
  86     @mock.patch.object(restcall, 'call_req')
  87     @mock.patch.object(uuid, 'uuid4')
  88     def test_create_vnf_identifier(self, mock_uuid4, mock_call_req):
  89         r2_get_vnfpackage_from_catalog = [
  90             0,
  91             json.JSONEncoder().encode(vnfpackage_info),
  92             '200'
  93         ]
  94         mock_call_req.return_value = r2_get_vnfpackage_from_catalog
  95         mock_uuid4.return_value = "1"
  96         data = {
  97             "vnfdId": "111",
  98             "vnfInstanceName": "vFW_01",
  99             "vnfInstanceDescription": "vFW in Nanjing TIC Edge"
 100         }
 101         response = self.client.post(
 102             "/api/vnflcm/v1/vnf_instances",
 103             data=data,
 104             format='json'
 105         )
 106         expect_data = {
 107             "id": "1",
 108             "vnfProvider": "huawei",
 109             "vnfdVersion": "1.0",
 110             "vnfPkgId": "111",
 111             "instantiationState": "NOT_INSTANTIATED"
 112         }
 113         self.assertEqual(status.HTTP_201_CREATED, response.status_code)
 114         self.assertEqual(expect_data, response.data)
 115
 116     @mock.patch.object(restcall, 'call_req')
 117     @mock.patch.object(uuid, 'uuid4')
 118     def test_create_vnf_inner_error(self, mock_uuid4, mock_call_req):
 119         mock_call_req.side_effect = NFLCMException('Boom!')
 120         mock_uuid4.return_value = "1"
 121         data = {
 122             "vnfdId": "111",
 123             "vnfInstanceName": "vFW_01",
 124             "vnfInstanceDescription": "vFW in Nanjing TIC Edge"
 125         }
 126         response = self.client.post(
 127             "/api/vnflcm/v1/vnf_instances",
 128             data=data,
 129             format='json'
 130         )
 131         self.assertEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code)
 132
 133     def test_create_vnf_bad_req(self):
 134         data = {
 135             "vnfInstanceName": "vFW_01",
 136             "vnfInstanceDescription": "vFW in Nanjing TIC Edge"
 137         }
 138         response = self.client.post(
 139             "/api/vnflcm/v1/vnf_instances",
 140             data=data,
 141             format='json'
 142         )
 143         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 144
 145     @mock.patch.object(CreateVnf, 'do_biz')
 146     def test_create_vnf_bad_response(self, moc_CreateVnf_do_biz):
 147         moc_CreateVnf_do_biz.return_value = {
 148             # "id": "1",
 149             "vnfProvider": "huawei",
 150             "vnfdVersion": "1.0",
 151             "vnfPkgId": "111",
 152             "instantiationState": "NOT_INSTANTIATED"
 153         }
 154         data = {
 155             "vnfdId": "111",
 156             "vnfInstanceName": "vFW_01",
 157             "vnfInstanceDescription": "vFW in Nanjing TIC Edge"
 158         }
 159         response = self.client.post(
 160             "/api/vnflcm/v1/vnf_instances",
 161             data=data,
 162             format='json'
 163         )
 164         self.assertEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code)