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.
16 from rest_framework import status
17 from django.test import TestCase
18 from django.test import Client
20 from lcm.pub.utils import restcall, toscaparser
21 from lcm.pub.utils import fileutil
22 from lcm.pub.database.models import NfPackageModel, NfInstModel
23 from lcm.pub.database.models import JobStatusModel, JobModel
24 from lcm.packages.sdc_nf_package import SdcNfDistributeThread, SdcNfPkgDeleteThread
25 from lcm.packages import sdc_nf_package
26 from lcm.pub.msapi import sdc
28 class TestNfPackage(TestCase):
30 self.client = Client()
31 NfPackageModel.objects.filter().delete()
32 NfInstModel.objects.filter().delete()
33 JobModel.objects.filter().delete()
34 JobStatusModel.objects.filter().delete()
43 "unit_size": 1000000000
45 "type_of_storage": "volume",
46 "rdma_enabled": False,
49 "volume_storage_id": "vNAT_Storage_6wdgwzedlb6sq18uzrr41sof7",
57 "vNAT_Storage_6wdgwzedlb6sq18uzrr41sof7"
64 "configurable_properties": {
66 "additional_vnfc_configurable_properties": {
73 "description": "the virtual machine of vNat",
84 "artifact_name": "vNatVNFImage",
85 "type": "tosca.artifacts.nfv.SwImage",
87 "operating_system": "linux",
88 "sw_image": "/swimages/vRouterVNF_ControlPlane.qcow2",
89 "name": "vNatVNFImage",
90 "container_format": "bare",
92 "disk_format": "qcow2",
93 "supported_virtualisation_environments": [
101 "file": "/swimages/vRouterVNF_ControlPlane.qcow2"
105 "flavor_extra_specs": {
106 "hw:cpu_sockets": "2",
107 "sw:ovs_dpdk": "true",
108 "hw:cpu_threads": "2",
109 "hw:numa_mem.1": "3072",
110 "hw:numa_mem.0": "1024",
111 "hw:numa_nodes": "2",
112 "hw:numa_cpus.0": "0,1",
113 "hw:numa_cpus.1": "2,3,4,5",
115 "hw:cpu_threads_policy": "isolate"
117 "cpu_frequency": "2.4 GHz",
121 "local_storages": [],
122 "image_file": "vNatVNFImage",
129 "operating_system": "linux",
130 "sw_image": "/swimages/vRouterVNF_ControlPlane.qcow2",
131 "name": "vNatVNFImage",
132 "container_format": "bare",
134 "disk_format": "qcow2",
135 "supported_virtualisation_environments": [
143 "image_file_id": "vNatVNFImage",
148 "local_storages": [],
152 "key_name": "sriov_plane",
153 "cp_id": "SRIOV_Port"
161 "vl_id": "sriov_link",
162 "route_external": False,
168 "connectivity_type": {
169 "layer_protocol": "ipv4",
170 "flow_pattern": "flat"
172 "description": "sriov_link",
181 "vl_id": "sriov_link",
182 "vdu_id": "vdu_vNat",
184 "cp_id": "SRIOV_Port",
188 "address_type": "ip_address",
190 "ip_address_type": "ipv4",
191 "floating_ip_activated": False,
192 "number_of_ip_address": 1,
193 "ip_address_assignment": True
197 "description": "sriov port",
198 "layer_protocol": "ipv4",
199 "virtual_network_interface_requirements": [
204 "support_mandatory": False,
206 "description": "sriov"
212 "support_mandatory": False,
214 "description": "normal"
218 "bitrate_requirement": 10
223 "vnfSoftwareVersion": "1.0.0",
224 "vnfProductName": "zte",
225 "localizationLanguage": [
229 "vnfProvider": "zte",
231 "defaultLocalizationLanguage": "english",
232 "vnfdId": "zte-hss-1.0",
234 "vnfProductInfoDescription": "hss",
235 "vnfdVersion": "1.0.0",
236 "vnfProductInfoName": "hss"
243 def assert_job_result(self, job_id, job_progress, job_detail):
244 jobs = JobStatusModel.objects.filter(
246 progress=job_progress,
248 self.assertEqual(1, len(jobs))
250 @mock.patch.object(SdcNfDistributeThread, 'run')
251 def test_nf_pkg_distribute_normal(self, mock_run):
252 resp = self.client.post("/api/nslcm/v1/vnfpackage", {
256 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
258 def test_nf_pkg_distribute_when_csar_already_exist(self):
259 NfPackageModel(uuid="1", nfpackageid="1", vnfdid="vcpe_vfw_zte_1_0").save()
260 SdcNfDistributeThread(csar_id="1",
264 self.assert_job_result("2", 255, "NF CSAR(1) already exists.")
266 @mock.patch.object(restcall, 'call_req')
267 @mock.patch.object(sdc, 'download_artifacts')
268 @mock.patch.object(toscaparser, 'parse_vnfd')
269 def test_nf_pkg_distribute_when_vnfd_already_exist(self,
270 mock_parse_vnfd, mock_download_artifacts, mock_call_req):
271 mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
272 mock_download_artifacts.return_value = "/home/hss.csar"
273 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
275 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
277 NfPackageModel(uuid="2", nfpackageid="2", vnfdid="zte-hss-1.0").save()
278 SdcNfDistributeThread(csar_id="1",
282 self.assert_job_result("2", 255, "NFD(zte-hss-1.0) already exists.")
284 @mock.patch.object(restcall, 'call_req')
285 @mock.patch.object(sdc, 'download_artifacts')
286 @mock.patch.object(toscaparser, 'parse_vnfd')
287 def test_nf_pkg_distribute_successfully(self,
288 mock_parse_vnfd, mock_download_artifacts, mock_call_req):
289 mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
290 mock_download_artifacts.return_value = "/home/hss.csar"
291 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
293 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
295 SdcNfDistributeThread(csar_id="1",
299 self.assert_job_result("4", 100, "CSAR(1) distribute successfully.")
302 ###############################################################################################################
304 @mock.patch.object(SdcNfPkgDeleteThread, 'run')
305 def test_nf_pkg_delete_normal(self, mock_run):
306 resp = self.client.delete("/api/nslcm/v1/vnfpackage/1")
307 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
309 def test_nf_pkg_normal_delete(self):
310 NfPackageModel(uuid="2", nfpackageid="2", vnfdid="vcpe_vfw_zte_1_0").save()
311 SdcNfPkgDeleteThread(csar_id="2", job_id="2", force_delete=False).run()
312 self.assert_job_result("2", 100, "Delete CSAR(2) successfully.")
314 def test_nf_pkg_force_delete(self):
315 NfPackageModel(uuid="1", nfpackageid="1", vnfdid="vcpe_vfw_zte_1_0").save()
316 NfInstModel(nfinstid="1", package_id="1").save()
317 SdcNfPkgDeleteThread(csar_id="1", job_id="2", force_delete=True).run()
318 self.assert_job_result("2", 100, "Delete CSAR(1) successfully.")
320 def test_nf_pkg_delete_when_pkg_in_using(self):
321 NfPackageModel(uuid="3", nfpackageid="3", vnfdid="vcpe_vfw_zte_1_0").save()
322 NfInstModel(nfinstid="3", package_id="3").save()
323 SdcNfPkgDeleteThread(csar_id="3", job_id="2", force_delete=False).run()
324 self.assert_job_result("2", 255, "NfInst by csar(3) exists, cannot delete.")