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 rest_framework.test import APIClient
18 from django.test import TestCase
19 from rest_framework import status
20 from genericparser.packages.biz.sdc_vnf_package import NfDistributeThread, NfPkgDeleteThread
21 from genericparser.pub.database.models import JobStatusModel, JobModel
22 from genericparser.pub.database.models import VnfPackageModel
23 from genericparser.pub.msapi import sdc
24 from genericparser.pub.utils import restcall, toscaparsers
27 class TestNfPackage(TestCase):
29 self.client = APIClient()
30 VnfPackageModel.objects.filter().delete()
31 JobModel.objects.filter().delete()
32 JobStatusModel.objects.filter().delete()
41 "unit_size": 1000000000
43 "type_of_storage": "volume",
44 "rdma_enabled": False,
47 "volume_storage_id": "vNAT_Storage_6wdgwzedlb6sq18uzrr41sof7",
55 "vNAT_Storage_6wdgwzedlb6sq18uzrr41sof7"
62 "configurable_properties": {
64 "additional_vnfc_configurable_properties": {
71 "description": "the virtual machine of vNat",
82 "artifact_name": "vNatVNFImage",
83 "type": "tosca.artifacts.nfv.SwImage",
85 "operating_system": "linux",
86 "sw_image": "/swimages/vRouterVNF_ControlPlane.qcow2",
87 "name": "vNatVNFImage",
88 "container_format": "bare",
90 "disk_format": "qcow2",
91 "supported_virtualisation_environments": [
99 "file": "/swimages/vRouterVNF_ControlPlane.qcow2"
103 "flavor_extra_specs": {
104 "hw:cpu_sockets": "2",
105 "sw:ovs_dpdk": "true",
106 "hw:cpu_threads": "2",
107 "hw:numa_mem.1": "3072",
108 "hw:numa_mem.0": "1024",
109 "hw:numa_nodes": "2",
110 "hw:numa_cpus.0": "0,1",
111 "hw:numa_cpus.1": "2,3,4,5",
113 "hw:cpu_threads_policy": "isolate"
115 "cpu_frequency": "2.4 GHz",
119 "local_storages": [],
120 "image_file": "vNatVNFImage",
127 "operating_system": "linux",
128 "sw_image": "/swimages/vRouterVNF_ControlPlane.qcow2",
129 "name": "vNatVNFImage",
130 "container_format": "bare",
132 "disk_format": "qcow2",
133 "supported_virtualisation_environments": [
141 "image_file_id": "vNatVNFImage",
146 "local_storages": [],
150 "key_name": "sriov_plane",
151 "cp_id": "SRIOV_Port"
159 "vl_id": "sriov_link",
160 "route_external": False,
166 "connectivity_type": {
167 "layer_protocol": "ipv4",
168 "flow_pattern": "flat"
170 "description": "sriov_link",
179 "vl_id": "sriov_link",
180 "vdu_id": "vdu_vNat",
182 "cp_id": "SRIOV_Port",
186 "address_type": "ip_address",
188 "ip_address_type": "ipv4",
189 "floating_ip_activated": False,
190 "number_of_ip_address": 1,
191 "ip_address_assignment": True
195 "description": "sriov port",
196 "layer_protocol": "ipv4",
197 "virtual_network_interface_requirements": [
202 "support_mandatory": False,
204 "description": "sriov"
210 "support_mandatory": False,
212 "description": "normal"
216 "bitrate_requirement": 10
221 "vnfSoftwareVersion": "1.0.0",
222 "vnfProductName": "zte",
223 "localizationLanguage": [
227 "vnfProvider": "zte",
229 "defaultLocalizationLanguage": "english",
230 "vnfdId": "zte-hss-1.0",
232 "vnfProductInfoDescription": "hss",
233 "vnfdVersion": "1.0.0",
234 "vnfProductInfoName": "hss"
238 "descriptor_id": "zte-hss-1.0",
239 "descriptor_verison": "1.0.0",
240 "software_version": "1.0.0",
251 def assert_job_result(self, job_id, job_progress, job_detail):
252 jobs = JobStatusModel.objects.filter(
254 progress=job_progress,
256 self.assertEqual(1, len(jobs))
258 @mock.patch.object(NfDistributeThread, 'run')
259 def test_nf_pkg_distribute_normal(self, mock_run):
260 resp = self.client.post("/api/genericparser/v1/vnfpackages", {
264 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
266 def test_nf_pkg_distribute_when_csar_already_exist(self):
267 VnfPackageModel(vnfPackageId="1", vnfdId="vcpe_vfw_zte_1_0").save()
268 NfDistributeThread(csar_id="1",
272 self.assert_job_result("2", 255, "NF CSAR(1) already exists.")
274 @mock.patch.object(restcall, 'call_req')
275 @mock.patch.object(sdc, 'download_artifacts')
276 @mock.patch.object(toscaparsers, 'parse_vnfd')
277 def test_nf_pkg_distribute_when_vnfd_already_exist(self,
278 mock_parse_vnfd, mock_download_artifacts, mock_call_req):
279 mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
280 mock_download_artifacts.return_value = "/home/hss.csar"
281 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
283 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
285 VnfPackageModel(vnfPackageId="2", vnfdId="zte-hss-1.0").save()
286 NfDistributeThread(csar_id="1",
290 self.assert_job_result("2", 255, "VNF package(zte-hss-1.0) already exists.")
292 @mock.patch.object(restcall, 'call_req')
293 @mock.patch.object(sdc, 'download_artifacts')
294 @mock.patch.object(toscaparsers, 'parse_vnfd')
295 def test_nf_pkg_distribute_successfully(self,
296 mock_parse_vnfd, mock_download_artifacts, mock_call_req):
297 mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
298 mock_download_artifacts.return_value = "/home/hss.csar"
299 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
301 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
303 NfDistributeThread(csar_id="1",
307 self.assert_job_result("4", 100, "CSAR(1) distribute successfully.")
309 ###############################################################################################################
311 @mock.patch.object(NfPkgDeleteThread, 'run')
312 def test_nf_pkg_delete_normal(self, mock_run):
313 resp = self.client.delete("/api/genericparser/v1/vnfpackages/1")
314 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
316 def test_nf_pkg_normal_delete(self):
317 VnfPackageModel(vnfPackageId="2", vnfdId="vcpe_vfw_zte_1_0").save()
318 NfPkgDeleteThread(csar_id="2", job_id="2").run()
319 self.assert_job_result("2", 100, "Delete CSAR(2) successfully.")
321 def test_nf_pkg_get_all(self):
322 VnfPackageModel(vnfPackageId="3", vnfdId="3", vnfVendor='3', vnfdVersion='3',
323 vnfSoftwareVersion='', vnfPackageUri='', vnfdModel='').save()
324 VnfPackageModel(vnfPackageId="4", vnfdId="4", vnfVendor='4', vnfdVersion='4',
325 vnfSoftwareVersion='', vnfPackageUri='', vnfdModel='').save()
326 resp = self.client.get("/api/genericparser/v1/vnfpackages")
327 self.assertEqual(resp.status_code, status.HTTP_200_OK)
337 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/3/",
351 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/4/",
358 self.assertEqual(expect_data, resp.data)
360 def test_nf_pkg_get_one(self):
361 VnfPackageModel(vnfPackageId="4", vnfdId="4", vnfVendor='4', vnfdVersion='4',
362 vnfSoftwareVersion='', vnfPackageUri='', vnfdModel='').save()
364 resp = self.client.get("/api/genericparser/v1/vnfpackages/4")
365 self.assertEqual(resp.status_code, status.HTTP_200_OK)
374 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/4/",
380 self.assertEqual(expect_data, resp.data)
382 def test_nf_pkg_get_one_failed(self):
383 VnfPackageModel(vnfPackageId="4", vnfdId="4", vnfVendor='4', vnfdVersion='4',
384 vnfSoftwareVersion='', vnfPackageUri='', vnfdModel='').save()
386 resp = self.client.get("/api/genericparser/v1/vnfpackages/2")
387 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
388 self.assertEqual({'error': 'Vnf package[2] not Found.'}, resp.data)
390 ###############################################################################################################
392 @mock.patch.object(toscaparsers, 'parse_vnfd')
393 def test_vnfd_parse_normal(self, mock_parse_vnfd):
394 VnfPackageModel(vnfPackageId="8", vnfdId="10").save()
395 mock_parse_vnfd.return_value = json.JSONEncoder().encode({"c": "d"})
396 req_data = {"csarId": "8", "inputs": []}
397 resp = self.client.post("/api/genericparser/v1/parservnfd", req_data, format='json')
398 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
399 self.assertEqual({"model": '{"c": "d"}'}, resp.data)
401 def test_vnfd_parse_when_csar_not_exist(self):
402 req_data = {"csarId": "1", "inputs": []}
403 resp = self.client.post("/api/genericparser/v1/parservnfd", req_data, format='json')
404 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
405 self.assertEqual(resp.data, {"error": "VNF CSAR(1) does not exist."})