gvnfmadapter/driver/interfaces/tests.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 mock
  17 from django.test import Client
  18 from django.test import TestCase
  19 from rest_framework import status
  20 from driver.pub.utils import restcall
  21
  22
  23 class InterfacesTest(TestCase):
  24     def setUp(self):
  25         self.client = Client()
  26
  27     def tearDown(self):
  28         pass
  29
  30     @mock.patch.object(restcall, 'call_req')
  31     def test_instantiate_vnf(self, mock_call_req):
  32         vnfm_info = {
  33             'userName': 'admin',
  34             'vendor': 'ZTE',
  35             'name': 'ZTE_VNFM_237_62',
  36             'vimId': '516cee95-e8ca-4d26-9268-38e343c2e31e',
  37             'url': 'http: //192.168.237.165: 2324',
  38             'certificateUrl': '',
  39             'version': 'V1.0',
  40             'vnfmId': 'b0797c9b-3da9-459c-b25c-3813e9d8fd70',
  41             'password': 'admin',
  42             'type': 'gvnfmdriver',
  43             'createTime': '2016-10-3111: 08: 39',
  44             'description': ''
  45         }
  46         job_info = {
  47             "vnfInstanceId": "8",
  48             "jobId": "NF-CREATE-8-b384535c-9f45-11e6-8749-fa163e91c2f9"
  49         }
  50         vnflcm_info = {
  51             "vnfInstanceId": "8",
  52             "vnfLcOpId": "NF-INST-8-6ffa8083-6705-49b3-ae54-cbd6265fbe7a"
  53         }
  54         r1 = [0, json.JSONEncoder().encode(vnfm_info), "200"]
  55         ret = [0, json.JSONEncoder().encode(job_info), '200']
  56         ret2 = [0, json.JSONEncoder().encode(vnflcm_info), '200']
  57         mock_call_req.side_effect = [r1, ret, r1, ret2]
  58         req_data = {
  59             'vnfInstanceName': 'VFW_f88c0cb7-512a-44c4-bd09-891663f19367',
  60             'vnfPackageId': 'd852e1be-0aac-48f1-b1a4-cd825f6cdf9a',
  61             'vnfDescriptorId': 'vcpe_vfw_zte_1_0',
  62             'additionalParam': {
  63                 'sdncontroller': 'e4d637f1-a4ec-4c59-8b20-4e8ab34daba9',
  64                 'NatIpRange': '192.167.0.10-192.168.0.20',
  65                 'm6000_mng_ip': '192.168.11.11',
  66                 'externalPluginManageNetworkName': 'plugin_net_2014',
  67                 'location': '516cee95-e8ca-4d26-9268-38e343c2e31e',
  68                 'externalManageNetworkName': 'mng_net_2017',
  69                 'sfc_data_network': 'sfc_data_net_2016',
  70                 'externalDataNetworkName': 'Flow_out_net',
  71                 'inputs': {}
  72             }
  73         }
  74         response = self.client.post("/api/gvnfmdriver/v1/1/vnfs",
  75                                     data=json.dumps(req_data), content_type="application/json")
  76         self.assertEqual(status.HTTP_201_CREATED, response.status_code)
  77         expect_data = {
  78             "vnfInstanceId": "8",
  79             "jobId": "NF-INST-8-6ffa8083-6705-49b3-ae54-cbd6265fbe7a"
  80         }
  81         self.assertEqual(expect_data, response.data)
  82
  83     @mock.patch.object(restcall, 'call_req')
  84     def test_terminate_vnf(self, mock_call_req):
  85         vnfm_info = {
  86             "vnfmId": "19ecbb3a-3242-4fa3-9926-8dfb7ddc29ee",
  87             "name": "g_vnfm",
  88             "type": "gvnfmdriver",
  89             "vimId": "",
  90             "vendor": "ZTE",
  91             "version": "v1.0",
  92             "description": "vnfm",
  93             "certificateUrl": "",
  94             "url": "http://10.74.44.11",
  95             "userName": "admin",
  96             "password": "admin",
  97             "createTime": "2016-07-06 15:33:18"
  98         }
  99         job_info = {"vnfInstanceId": "1", "vnfLcOpId": "1"}
 100         job_status_info = {
 101             "jobId": "1",
 102             "responseDescriptor": {
 103                 "status": "",
 104                 "progress": 100,
 105                 "statusDescription": "",
 106                 "errorCode": "",
 107                 "responseId": "2",
 108                 "responseHistoryList": [
 109                     {
 110                         "status": "",
 111                         "progress": "",
 112                         "statusDescription": "",
 113                         "errorCode": "",
 114                         "responseId": ""
 115                     }
 116                 ]
 117             }
 118         }
 119         r1 = [0, json.JSONEncoder().encode(vnfm_info), "200"]
 120         r2 = [0, json.JSONEncoder().encode(job_info), "200"]
 121         job_ret = [0, json.JSONEncoder().encode(job_status_info), "200"]
 122         r3 = [0, json.JSONEncoder().encode(None), "200"]
 123         mock_call_req.side_effect = [r1, r2, r1, job_ret, r1, r3]
 124         response = self.client.post("/api/gvnfmdriver/v1/ztevnfmid/vnfs/2/terminate")
 125         self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 126         self.assertEqual(None, response.data)
 127
 128     @mock.patch.object(restcall, 'call_req')
 129     def test_query_vnf(self, mock_call_req):
 130         vnfm_info = {
 131             "vnfmId": "19ecbb3a-3242-4fa3-9926-8dfb7ddc29ee",
 132             "name": "g_vnfm",
 133             "type": "gvnfmdriver",
 134             "vimId": "",
 135             "vendor": "ZTE",
 136             "version": "v1.0",
 137             "description": "vnfm",
 138             "certificateUrl": "",
 139             "url": "http://10.74.44.11",
 140             "userName": "admin",
 141             "password": "admin",
 142             "createTime": "2016-07-06 15:33:18"
 143         }
 144         job_info = {
 145             "ResponseInfo": {
 146                 "id": "88",
 147                 "instantiationState": "INSTANTIATED",
 148                 "vnfSoftwareVersion": "v1.2.3"
 149             }
 150         }
 151         r1 = [0, json.JSONEncoder().encode(vnfm_info), "200"]
 152         r2 = [0, json.JSONEncoder().encode(job_info), "200"]
 153         mock_call_req.side_effect = [r1, r2]
 154         response = self.client.get("/api/gvnfmdriver/v1/19ecbb3a-3242-4fa3-9926-8dfb7ddc29ee/vnfs/88")
 155         self.assertEqual(status.HTTP_200_OK, response.status_code)
 156         expect_resp_data = {
 157             "vnfInfo": {
 158                 "vnfInstanceId": "88",
 159                 "vnfStatus": "ACTIVE",
 160                 "version": "v1.2.3"
 161             }
 162         }
 163         self.assertEqual(expect_resp_data, response.data)
 164
 165     @mock.patch.object(restcall, 'call_req')
 166     def test_operation_status(self, mock_call_req):
 167         vnfm_info = {
 168             'userName': 'admin',
 169             'vendor': 'ZTE',
 170             'name': 'ZTE_VNFM_237_62',
 171             'vimId': '516cee95-e8ca-4d26-9268-38e343c2e31e',
 172             'url': 'http: //192.168.237.165: 2324',
 173             'certificateUrl': '',
 174             'version': 'V1.0',
 175             'vnfmId': 'b0797c9b-3da9-459c-b25c-3813e9d8fd70',
 176             'password': 'admin',
 177             'type': 'gvnfmdriver',
 178             'createTime': '2016-10-3111: 08: 39',
 179             'description': ''
 180         }
 181         expected_body = {
 182             "jobId": "NF-CREATE-11-ec6c2f2a-9f48-11e6-9405-fa163e91c2f9",
 183             "responseDescriptor": {
 184                 "responseId": 3,
 185                 "progress": 40,
 186                 "status": "PROCESSING",
 187                 "statusDescription": "OMC VMs are decommissioned in VIM",
 188                 "errorCode": "null",
 189                 "responseHistoryList": [
 190                     {
 191                         "status": "error",
 192                         "progress": 255,
 193                         "errorcode": "",
 194                         "responseid": 20,
 195                         "statusdescription": "'JsonParser' object has no attribute 'parser_info'"
 196                     }
 197                 ]
 198             }
 199         }
 200         resp_body = {
 201             "ResponseInfo": {
 202                 "vnfLcOpId": "NF-CREATE-11-ec6c2f2a-9f48-11e6-9405-fa163e91c2f9",
 203                 "responseDescriptor": {
 204                     "responseId": 3,
 205                     "progress": 40,
 206                     "lcmOperationStatus": "PROCESSING",
 207                     "statusDescription": "OMC VMs are decommissioned in VIM",
 208                     "errorCode": "null",
 209                     "responseHistoryList": [
 210                         {
 211                             "status": "error",
 212                             "progress": 255,
 213                             "errorcode": "",
 214                             "responseid": 20,
 215                             "statusdescription": "'JsonParser' object has no attribute 'parser_info'"
 216                         }
 217                     ]
 218                 }
 219             }
 220         }
 221         r1 = [0, json.JSONEncoder().encode(vnfm_info), '200']
 222         r2 = [0, json.JSONEncoder().encode(resp_body), '200']
 223         mock_call_req.side_effect = [r1, r2]
 224         response = self.client.get("/api/gvnfmdriver/v1/%s/jobs/%s?responseId=0"
 225                                    % (vnfm_info["vnfmId"], expected_body["jobId"]))
 226         self.assertEqual(status.HTTP_200_OK, response.status_code)
 227         self.assertDictEqual(expected_body, response.data)
 228
 229     @mock.patch.object(restcall, 'call_req')
 230     def test_grantvnf(self, mock_call_req):
 231         vim_info = {
 232             "vim": {
 233                 "accessinfo": {
 234                     "tenant": "admin"
 235                 },
 236                 "vimid": "516cee95-e8ca-4d26-9268-38e343c2e31e"
 237             }
 238         }
 239         req_data = {
 240             "vnfmid": "13232222",
 241             "nfvoid": "03212234",
 242             "vimid": "12345678",
 243             "exvimidlist ": "exvimid",
 244             "tenant": " tenant1",
 245             "vnfinstanceid": "1234",
 246             "operationright": "0",
 247             "vmlist": [
 248                 {
 249                     "vmflavor": "SMP",
 250                     "vmnumber": "3"
 251                 },
 252                 {
 253                     "vmflavor": "CMP",
 254                     "vmnumber": "3"
 255                 }
 256             ]
 257         }
 258         mock_call_req.return_value = [0, json.JSONEncoder().encode(vim_info), '201']
 259         response = self.client.put("/api/gvnfmdriver/v1/resource/grant",
 260                                    data=json.dumps(req_data), content_type='application/json')
 261         self.assertEqual(status.HTTP_201_CREATED, response.status_code)
 262         expect_resp_data = {
 263             "vimid": "516cee95-e8ca-4d26-9268-38e343c2e31e",
 264             "tenant": "admin"
 265         }
 266         self.assertDictEqual(expect_resp_data, response.data)
 267
 268     @mock.patch.object(restcall, 'call_req')
 269     def test_notify(self, mock_call_req):
 270         vim_info = {
 271             "vim": {
 272                 "vimInfoId": "111111",
 273                 "vimId": "12345678",
 274                 "interfaceInfo": {
 275                     "vimType": "vnf",
 276                     "apiVersion": "v1",
 277                     "protocolType": "None"
 278                 },
 279                 "accessInfo": {
 280                     "tenant": "tenant1",
 281                     "username": "admin",
 282                     "password": "password"
 283                 },
 284                 "interfaceEndpoint": "http://127.0.0.1/api/v1"
 285             },
 286             "zone": "",
 287             "addResource": {
 288                 "resourceDefinitionId": "xxxxx",
 289                 "vimId": "12345678",
 290                 "zoneId": "000"
 291             },
 292             "removeResource": "",
 293             "vimAssets": {
 294                 "computeResourceFlavour": {
 295                     "vimId": "12345678",
 296                     "vduId": "sdfasdf",
 297                     "vimFlavourId": "12"
 298                 },
 299                 "softwareImage": {
 300                     "vimId": "12345678",
 301                     "imageName": "AAA",
 302                     "vimImageId": ""
 303                 }
 304             },
 305             "additionalParam": ""
 306         }
 307         r2 = [0, json.JSONEncoder().encode(vim_info), "200"]
 308         mock_call_req.side_effect = [r2]
 309         req_data = {
 310             "nfvoid": "1",
 311             "vnfmid": "876543211",
 312             "vimid": "6543211",
 313             "timestamp": "1234567890",
 314             "vnfinstanceid": "1",
 315             "eventtype": "0",
 316             "vmlist": [
 317                 {
 318                     "vmflavor": "SMP",
 319                     "vmnumber": "3",
 320                     "vmidlist ": [
 321                         "vmuuid"
 322                     ]
 323                 },
 324                 {
 325                     "vmflavor": "CMP",
 326                     "vmnumber": "3",
 327                     "vmidlist ": [
 328                         "vmuuid"
 329                     ]
 330                 }
 331             ]
 332         }
 333         response = self.client.post("/api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification",
 334                                     data=json.dumps(req_data),
 335                                     content_type='application/json')
 336         self.assertEqual(status.HTTP_200_OK, response.status_code)
 337         expect_resp_data = None
 338         self.assertEqual(expect_resp_data, response.data)
 339
 340     @mock.patch.object(restcall, 'call_req')
 341     def test_get_vnfpkgs(self, mock_call_req):
 342         vnfpkgs_info = {
 343             "csars": [{
 344                 "csarId": "1",
 345                 "vnfdId": "2"
 346             }]
 347         }
 348         mock_call_req.return_value = [0, json.JSONEncoder().encode(vnfpkgs_info), '200']
 349         resp = self.client.get("/api/gvnfmdriver/v1/vnfpackages")
 350         self.assertEqual(status.HTTP_200_OK, resp.status_code)
 351         self.assertEqual(1, len(resp.data["csars"]))
 352         self.assertEqual("1", resp.data["csars"][0]["csarId"])
 353         self.assertEqual("2", resp.data["csars"][0]["vnfdId"])