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