lcm/ns_vnfs/tests/vnf_tests.py

   1 # Copyright 2018 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 unittest
  16 import json
  17 import mock
  18 from rest_framework.test import APIClient
  19 from rest_framework import status
  20 from lcm.ns_vnfs.tests.const import GRANT_DATA, VNF_LCM_OP_OCC_NOTIFICATION_DATA, \
  21     VNF_IDENTIFIER_CREATION_NOTIFICATION_DATA, VNF_IDENTIFIER_DELETION_NOTIFICATION_DATA
  22 from lcm.pub.database.models import NfInstModel
  23 from lcm.pub.utils import restcall
  24
  25
  26 class VnfGrantViewTest(unittest.TestCase):
  27     def setUp(self):
  28         self.client = APIClient()
  29
  30     def tearDown(self):
  31         pass
  32
  33     @mock.patch.object(restcall, 'call_req')
  34     def test_grant_vnf_normal(self, mock_call_req):
  35         vim_connections = {
  36             "id": "1",
  37             "vimId": "1",
  38         }
  39         mock_call_req.return_value = [0, json.JSONEncoder().encode(vim_connections), '200']
  40         response = self.client.post("/api/nslcm/v2/grants", data=GRANT_DATA, format='json')
  41         self.assertEqual(status.HTTP_201_CREATED, response.status_code, response.content)
  42         resp_data = json.loads(response.content)
  43         expect_resp_data = {
  44             "id": resp_data.get("id"),
  45             "vnfInstanceId": "1",
  46             "vnfLcmOpOccId": "2",
  47             "vimConnections": [
  48                 {
  49                     "id": "1",
  50                     "vimId": "1"
  51                 }
  52             ]
  53         }
  54         self.assertEqual(expect_resp_data, resp_data)
  55
  56     def test_grant_vnf_when_vnfinst_not_exist(self):
  57         response = self.client.post("/api/nslcm/v2/grants", data=GRANT_DATA, format='json')
  58         self.failUnlessEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code)
  59
  60     @mock.patch.object(restcall, 'call_req')
  61     def test_grant_vnf(self, mock_call_req):
  62         data = {
  63             "vnfInstanceId": "1",
  64             "vnfLcmOpOccId": "2",
  65             "vnfdId": "3",
  66             "flavourId": "4",
  67             "operation": "INSTANTIATE",
  68             "isAutomaticInvocation": True,
  69             "instantiationLevelId": "5",
  70             "addResources": [
  71                 {
  72                     "id": "1",
  73                     "type": "COMPUTE",
  74                     "vduId": "2",
  75                     "resourceTemplateId": "3",
  76                     "vdu": "1"
  77                 }
  78             ],
  79             "additionalParams": {"vnfmid": "3"},
  80             "_links": {
  81                 "vnfLcmOpOcc": {
  82                     "href": "1"
  83                 },
  84                 "vnfInstance": {
  85                     "href": "2"
  86                 }
  87             }
  88         }
  89         vnfdModel = {
  90             "volume_storages": [],
  91             "vdus": [{
  92                 "vdu_id": "1",
  93                 "properties": {
  94                     "name": "1"
  95                 },
  96                 "local_storages": "2",
  97                 "virtual_compute": {
  98                     "virtual_cpu": {
  99                         "num_virtual_cpu": "111"
 100                     },
 101                     "virtual_memory": {
 102                         "virtual_mem_size": "3 B"
 103                     }
 104                 },
 105             }],
 106             "image_files": [],
 107             "routers": [],
 108             "local_storages": [{"local_storage_id": "1"}],
 109             "vnf_exposed": {
 110                 "external_cps": [],
 111                 "forward_cps": []
 112             },
 113             "vls": [],
 114             "cps": [],
 115             "metadata": {
 116                 "designer": "sdno",
 117                 "name": "underlayervpn",
 118                 "csarVersion": "1.0",
 119                 "csarType": "SSAR",
 120                 "csarProvider": "huawei",
 121                 "version": "1.0",
 122                 "type": "SSAR",
 123                 "id": "ns_underlayervpn_1_0"
 124             }
 125         }
 126
 127         vnfpackage_info = {
 128             "imageInfo": [],
 129             "csarId": "vOpenNAT",
 130             "packageInfo": {
 131                 "csarName": "vOpenNAT.csar",
 132                 "vnfdModel": json.dumps(vnfdModel),
 133                 "vnfdProvider": "Intel",
 134                 "vnfdId": "openNAT_1.0",
 135                 "downloadUrl": "http://10.96.33.39:8806/static/catalog/vOpenNAT/vOpenNAT.csar",
 136                 "vnfVersion": "v1.0",
 137                 "vnfdVersion": "v1.0",
 138                 "vnfPackageId": "vOpenNAT"
 139             }
 140         }
 141         vimConnections = {
 142             "id": "1",
 143             "vimId": "1",
 144         }
 145         NfInstModel.objects.create(nfinstid='1',
 146                                    package_id="2",
 147                                    vnfm_inst_id='3')
 148         get_vnfpackage = [0, json.JSONEncoder().encode(vnfpackage_info), '200']
 149         get_vimConnections = [0, json.JSONEncoder().encode(vimConnections), '200']
 150         mock_call_req.side_effect = [get_vnfpackage, get_vimConnections]
 151         response = self.client.post("/api/nslcm/v2/grants", data=data, format='json')
 152         self.assertEqual(status.HTTP_201_CREATED, response.status_code, response.content)
 153         resp_data = json.loads(response.content)
 154         expect_resp_data = {
 155             "id": resp_data.get("id"),
 156             "vnfInstanceId": "1",
 157             "vnfLcmOpOccId": "2",
 158             "vimConnections": [
 159                 {
 160                     "id": "1",
 161                     "vimId": "1"
 162                 }
 163             ]
 164         }
 165         self.assertEqual(expect_resp_data, resp_data)
 166
 167     def test_get_notify_vnf_normal(self):
 168         response = self.client.get("/api/nslcm/v2/ns/1/vnfs/1/Notify")
 169         self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code, response.content)
 170
 171     def test_notify_vnflcmopooc_normal(self):
 172         NfInstModel.objects.create(nfinstid='22',
 173                                    mnfinstid='2',
 174                                    vnfm_inst_id='1')
 175         response = self.client.post("/api/nslcm/v2/ns/1/vnfs/2/Notify",
 176                                     data=VNF_LCM_OP_OCC_NOTIFICATION_DATA,
 177                                     format='json')
 178         self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 179
 180     def test_notify_vnf_identifier_creation_normal(self):
 181         response = self.client.post("/api/nslcm/v2/ns/1/vnfs/2/Notify",
 182                                     data=VNF_IDENTIFIER_CREATION_NOTIFICATION_DATA,
 183                                     format='json')
 184         self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 185
 186     def test_notify_vnf_identifier_deletion_normal(self):
 187         NfInstModel.objects.create(nfinstid='22',
 188                                    mnfinstid='2',
 189                                    vnfm_inst_id='1')
 190         response = self.client.post("/api/nslcm/v2/ns/1/vnfs/2/Notify",
 191                                     data=VNF_IDENTIFIER_DELETION_NOTIFICATION_DATA,
 192                                     format='json')
 193         self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)