1 # Copyright 2017 ZTE Corporation.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
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
23 class InterfacesTest(TestCase):
25 self.client = Client()
30 @mock.patch.object(restcall, 'call_req')
31 def test_instantiate_vnf(self, mock_call_req):
35 'name': 'ZTE_VNFM_237_62',
36 'vimId': '516cee95-e8ca-4d26-9268-38e343c2e31e',
37 'url': 'http://192.168.237.165:2324',
40 'vnfmId': 'b0797c9b-3da9-459c-b25c-3813e9d8fd70',
42 'type': 'ztevmanagerdriver',
43 'createTime': '2016-10-31 11:08:39',
48 "jobId":"NF-CREATE-8-b384535c-9f45-11e6-8749-fa163e91c2f9"
52 "vnfLcOpId":"NF-CREATE-8-b384535c-9f45-11e6-8749-fa163e91c2f9"
55 r1 = [0, json.JSONEncoder().encode(vnfm_info), "200"]
56 ret = [0, json.JSONEncoder().encode(job_info), '200']
57 ret2 = [0, json.JSONEncoder().encode(vnflcm_info), '200']
58 mock_call_req.side_effect = [r1, ret, r1, ret2]
60 'vnfInstanceName': 'VFW_f88c0cb7-512a-44c4-bd09-891663f19367',
61 'vnfPackageId': 'd852e1be-0aac-48f1-b1a4-cd825f6cdf9a',
62 'vnfDescriptorId': 'vcpe_vfw_zte_1_0',
64 'sdncontroller': 'e4d637f1-a4ec-4c59-8b20-4e8ab34daba9',
65 'NatIpRange': '192.167.0.10-192.168.0.20',
66 'm6000_mng_ip': '192.168.11.11',
67 'externalPluginManageNetworkName': 'plugin_net_2014',
68 'location': '516cee95-e8ca-4d26-9268-38e343c2e31e',
69 'externalManageNetworkName': 'mng_net_2017',
70 'sfc_data_network': 'sfc_data_net_2016',
71 'externalDataNetworkName': 'Flow_out_net',
75 response = self.client.post("/api/gvnfmdriver/v1/1/vnfs",
76 data=json.dumps(req_data), content_type="application/json")
77 self.assertEqual(status.HTTP_201_CREATED, response.status_code)
80 self.assertEqual(job_info, response.data)
83 @mock.patch.object(restcall, 'call_req')
84 def test_terminate_vnf(self, mock_call_req):
86 "vnfmId": "19ecbb3a-3242-4fa3-9926-8dfb7ddc29ee",
92 "description": "vnfm",
94 "url": "http://10.74.44.11",
97 "createTime": "2016-07-06 15:33:18"
99 job_info = {"vnfInstanceId": "1", "vnfLcOpId": "1"}
100 job_status_info = {"VnfLcOpResponseDescriptor":{"progress":"100"}}
101 r1 = [0, json.JSONEncoder().encode(vnfm_info), "200"]
102 r2 = [0, json.JSONEncoder().encode(job_info), "200"]
103 job_ret = [0, json.JSONEncoder().encode(job_status_info), "200"]
104 mock_call_req.side_effect = [r1, r2, r1, job_ret, r1, r2]
105 response = self.client.post("/api/gvnfmdriver/v1/ztevnfmid/vnfs/2/terminate")
106 self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
107 self.assertEqual(job_info, response.data)
110 @mock.patch.object(restcall, 'call_req')
111 def test_query_vnf(self, mock_call_req):
113 "vnfmId": "19ecbb3a-3242-4fa3-9926-8dfb7ddc29ee",
119 "description": "vnfm",
120 "certificateUrl": "",
121 "url": "http://10.74.44.11",
124 "createTime": "2016-07-06 15:33:18"
126 job_info = {"ResponseInfo": {"vnfInstanceId":"88","instantiationState":"INSTANTIATED","vnfSoftwareVersion":"v1.2.3"}}
127 r1 = [0, json.JSONEncoder().encode(vnfm_info), "200"]
128 r2 = [0, json.JSONEncoder().encode(job_info), "200"]
129 mock_call_req.side_effect = [r1, r2]
130 response = self.client.get("/api/gvnfmdriver/v1/19ecbb3a-3242-4fa3-9926-8dfb7ddc29ee/vnfs/88")
131 self.assertEqual(status.HTTP_200_OK, response.status_code)
132 expect_resp_data = {"vnfInfo": {"vnfInstanceId": "88", "vnfStatus": "ACTIVE","version":"v1.2.3"}}
133 self.assertEqual(expect_resp_data, response.data)
136 @mock.patch.object(restcall, 'call_req')
137 def test_operation_status(self, mock_call_req):
141 'name': 'ZTE_VNFM_237_62',
142 'vimId': '516cee95-e8ca-4d26-9268-38e343c2e31e',
143 'url': 'http://192.168.237.165:2324',
144 'certificateUrl': '',
146 'vnfmId': 'b0797c9b-3da9-459c-b25c-3813e9d8fd70',
148 'type': 'ztevmanagerdriver',
149 'createTime': '2016-10-31 11:08:39',
153 "jobId": "NF-CREATE-11-ec6c2f2a-9f48-11e6-9405-fa163e91c2f9",
154 "responseDescriptor":{
157 "status": "PROCESSING",
158 "statusDescription": "OMC VMs are decommissioned in VIM",
160 "responseHistoryList": [
166 "statusdescription": "'JsonParser' object has no attribute 'parser_info'"
173 "vnfLcOpId":"NF-CREATE-11-ec6c2f2a-9f48-11e6-9405-fa163e91c2f9",
174 "responseDescriptor":{
177 "lcmOperationStatus": "PROCESSING",
178 "statusDescription": "OMC VMs are decommissioned in VIM",
180 "responseHistoryList": [
185 "statusdescription": "'JsonParser' object has no attribute 'parser_info'"}]
189 r1 = [0, json.JSONEncoder().encode(vnfm_info), '200']
190 r2 = [0, json.JSONEncoder().encode(resp_body), '200']
191 mock_call_req.side_effect = [r1, r2]
192 response = self.client.get("/api/gvnfmdriver/v1/{vnfmid}/jobs/{jobid}?responseId={responseId}".
193 format(vnfmid=vnfm_info["vnfmId"],jobid=resp_body["ResponseInfo"]["vnfLcOpId"],
194 responseId=resp_body["ResponseInfo"]["responseDescriptor"]["responseId"]))
195 self.assertEqual(status.HTTP_200_OK, response.status_code)
198 self.assertDictEqual(expected_body, response.data)
201 @mock.patch.object(restcall, 'call_req')
202 def test_grantvnf(self, mock_call_req):
203 vim_info = {"vim":{"accessinfo":{"tenant":"admin"},"vimid":"516cee95-e8ca-4d26-9268-38e343c2e31e"}}
205 "vnfmid": "13232222",
206 "nfvoid": "03212234",
208 "exvimidlist ":["exvimid"],
209 "tenant": " tenant1",
210 "vnfistanceid": "1234",
211 "operationright": "0",
221 mock_call_req.return_value = [0, json.JSONEncoder().encode(vim_info), '201']
222 response = self.client.put("/api/gvnfmdriver/v1/resource/grant",
223 data=json.dumps(req_data), content_type='application/json')
224 self.assertEqual(str(status.HTTP_201_CREATED), response.status_code)
225 expect_resp_data = {"vimid": "516cee95-e8ca-4d26-9268-38e343c2e31e", "tenant": "admin"}
226 self.assertDictEqual(expect_resp_data, response.data)
228 @mock.patch.object(restcall, 'call_req')
229 def test_notify(self, mock_call_req):
232 "vimInfoId": "111111",
237 "protocolType": "None"},
241 "password": "password"},
242 "interfaceEndpoint": "http://127.0.0.1/api/v1"
246 "resourceDefinitionId": "xxxxx",
249 "removeResource": "",
251 "computeResourceFlavour": {
254 "vimFlavourId": "12"},
259 "additionalParam": ""
261 r2 = [0, json.JSONEncoder().encode(vim_info), "200"]
262 mock_call_req.side_effect = [r2]
265 "vnfmid": "876543211",
267 "timestamp": "1234567890",
268 "vnfinstanceid": "1",
275 "vmidlist ": ["vmuuid"]},
279 "vmidlist ": ["vmuuid"]}
282 response = self.client.post("/api/gvnfmdriver/v1/vnfs/lifecyclechangesnotification",
283 data=json.dumps(req_data), content_type='application/json')
284 self.assertEqual(str(status.HTTP_200_OK), response.status_code)
285 expect_resp_data = None
286 self.assertEqual(expect_resp_data, response.data)
288 @mock.patch.object(restcall, 'call_req')
289 def test_get_vnfpkgs(self, mock_call_req):
290 mock_call_req.return_value = [0, json.JSONEncoder().encode({
296 resp = self.client.get("/api/gvnfmdriver/v1/vnfpackages")
297 self.assertEqual(status.HTTP_200_OK, resp.status_code)
298 self.assertEqual(1, len(resp.data["csars"]))
299 self.assertEqual("1", resp.data["csars"][0]["csarId"])
300 self.assertEqual("2", resp.data["csars"][0]["vnfdId"])