genericparser/packages/tests/test_vnfpackage.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 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
  25 from .const import vnfd_data
  26
  27
  28 class TestNfPackage(TestCase):
  29     def setUp(self):
  30         self.client = APIClient()
  31         VnfPackageModel.objects.filter().delete()
  32         JobModel.objects.filter().delete()
  33         JobStatusModel.objects.filter().delete()
  34         self.vnfd_data = vnfd_data
  35
  36     def tearDown(self):
  37         pass
  38
  39     def assert_job_result(self, job_id, job_progress, job_detail):
  40         jobs = JobStatusModel.objects.filter(
  41             jobid=job_id,
  42             progress=job_progress,
  43             descp=job_detail)
  44         self.assertEqual(1, len(jobs))
  45
  46     @mock.patch.object(NfDistributeThread, 'run')
  47     def test_nf_pkg_distribute_normal(self, mock_run):
  48         resp = self.client.post(
  49             "/api/parser/v1/vnfpackages",
  50             {
  51                 "csarId": "1",
  52                 "vimIds": ["1"]
  53             },
  54             format='json'
  55         )
  56         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
  57
  58     def test_nf_pkg_distribute_when_csar_already_exist(self):
  59         VnfPackageModel(
  60             vnfPackageId="1",
  61             vnfdId="vcpe_vfw_zte_1_0"
  62         ).save()
  63         NfDistributeThread(
  64             csar_id="1",
  65             vim_ids=["1"],
  66             lab_vim_id="",
  67             job_id="2"
  68         ).run()
  69         self.assert_job_result("2", 255, "NF CSAR(1) already exists.")
  70
  71     @mock.patch.object(restcall, 'call_req')
  72     @mock.patch.object(sdc, 'download_artifacts')
  73     @mock.patch.object(toscaparsers, 'parse_vnfd')
  74     def test_nf_pkg_distribute_when_vnfd_already_exist(self,
  75                                                        mock_parse_vnfd,
  76                                                        mock_download_artifacts,
  77                                                        mock_call_req):
  78         mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
  79         mock_download_artifacts.return_value = "/home/hss.csar"
  80         mock_call_req.return_value = [0, json.JSONEncoder().encode([{
  81             "uuid": "1",
  82             "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
  83         }]), '200']
  84         VnfPackageModel(vnfPackageId="2",
  85                         vnfdId="zte-hss-1.0").save()
  86         NfDistributeThread(csar_id="1",
  87                            vim_ids=["1"],
  88                            lab_vim_id="",
  89                            job_id="2").run()
  90         self.assert_job_result("2", 255, "VNF package(zte-hss-1.0) already exists.")
  91
  92     @mock.patch.object(restcall, 'call_req')
  93     @mock.patch.object(sdc, 'download_artifacts')
  94     @mock.patch.object(toscaparsers, 'parse_vnfd')
  95     def test_nf_pkg_distribute_successfully(self,
  96                                             mock_parse_vnfd, mock_download_artifacts, mock_call_req):
  97         mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
  98         mock_download_artifacts.return_value = "/home/hss.csar"
  99         mock_call_req.return_value = [0, json.JSONEncoder().encode([{
 100             "uuid": "1",
 101             "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
 102         }]), '200']
 103         NfDistributeThread(csar_id="1",
 104                            vim_ids=["1"],
 105                            lab_vim_id="",
 106                            job_id="4").run()
 107         self.assert_job_result("4", 100, "CSAR(1) distribute successfully.")
 108
 109     ###############################################################################################################
 110
 111     @mock.patch.object(NfPkgDeleteThread, 'run')
 112     def test_nf_pkg_delete_normal(self, mock_run):
 113         resp = self.client.delete("/api/parser/v1/vnfpackages/1")
 114         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 115
 116     def test_nf_pkg_normal_delete(self):
 117         VnfPackageModel(vnfPackageId="2", vnfdId="vcpe_vfw_zte_1_0").save()
 118         NfPkgDeleteThread(csar_id="2", job_id="2").run()
 119         self.assert_job_result("2", 100, "Delete CSAR(2) successfully.")
 120
 121     def test_nf_pkg_get_all(self):
 122         VnfPackageModel(
 123             vnfPackageId="3",
 124             vnfdId="3",
 125             vnfVendor='3',
 126             vnfdVersion='3',
 127             vnfSoftwareVersion='',
 128             vnfPackageUri='',
 129             vnfdModel=''
 130         ).save()
 131         VnfPackageModel(
 132             vnfPackageId="4",
 133             vnfdId="4",
 134             vnfVendor='4',
 135             vnfdVersion='4',
 136             vnfSoftwareVersion='',
 137             vnfPackageUri='',
 138             vnfdModel=''
 139         ).save()
 140         resp = self.client.get("/api/parser/v1/vnfpackages")
 141         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 142         expect_data = [
 143             {
 144                 "imageInfo": [],
 145                 "csarId": "3",
 146                 "packageInfo": {
 147                     "csarName": "",
 148                     "vnfdModel": "",
 149                     "vnfdProvider": "3",
 150                     "vnfdId": "3",
 151                     "downloadUrl": "http://127.0.0.1:8806/static/genericparser/3/",
 152                     "vnfVersion": "",
 153                     "vnfdVersion": "3",
 154                     "vnfPackageId": "3"
 155                 }
 156             },
 157             {
 158                 "imageInfo": [],
 159                 "csarId": "4",
 160                 "packageInfo": {
 161                     "csarName": "",
 162                     "vnfdModel": "",
 163                     "vnfdProvider": "4",
 164                     "vnfdId": "4",
 165                     "downloadUrl": "http://127.0.0.1:8806/static/genericparser/4/",
 166                     "vnfVersion": "",
 167                     "vnfdVersion": "4",
 168                     "vnfPackageId": "4"
 169                 }
 170             }
 171         ]
 172         self.assertEqual(expect_data, resp.data)
 173
 174     def test_nf_pkg_get_one(self):
 175         VnfPackageModel(
 176             vnfPackageId="4",
 177             vnfdId="4",
 178             vnfVendor='4',
 179             vnfdVersion='4',
 180             vnfSoftwareVersion='',
 181             vnfPackageUri='',
 182             vnfdModel=''
 183         ).save()
 184
 185         resp = self.client.get("/api/parser/v1/vnfpackages/4")
 186         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 187         expect_data = {
 188             "imageInfo": [],
 189             "csarId": "4",
 190             "packageInfo": {
 191                 "csarName": "",
 192                 "vnfdModel": "",
 193                 "vnfdProvider": "4",
 194                 "vnfdId": "4",
 195                 "downloadUrl": "http://127.0.0.1:8806/static/genericparser/4/",
 196                 "vnfVersion": "",
 197                 "vnfdVersion": "4",
 198                 "vnfPackageId": "4"
 199             }
 200         }
 201         self.assertEqual(expect_data, resp.data)
 202
 203     def test_nf_pkg_get_one_failed(self):
 204         VnfPackageModel(vnfPackageId="4",
 205                         vnfdId="4",
 206                         vnfVendor='4',
 207                         vnfdVersion='4',
 208                         vnfSoftwareVersion='',
 209                         vnfPackageUri='',
 210                         vnfdModel='').save()
 211
 212         resp = self.client.get("/api/parser/v1/vnfpackages/2")
 213         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 214         self.assertEqual({'error': 'Vnf package[2] not Found.'}, resp.data)
 215
 216     ###############################################################################################################
 217
 218     @mock.patch.object(toscaparsers, 'parse_vnfd')
 219     def test_vnfd_parse_normal(self, mock_parse_vnfd):
 220         VnfPackageModel(vnfPackageId="8", vnfdId="10").save()
 221         mock_parse_vnfd.return_value = json.JSONEncoder().encode({"c": "d"})
 222         req_data = {"csarId": "8", "inputs": []}
 223         resp = self.client.post("/api/parser/v1/parservnfd", req_data, format='json')
 224         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 225         self.assertEqual({"model": '{"c": "d"}'}, resp.data)
 226
 227     def test_vnfd_parse_when_csar_not_exist(self):
 228         req_data = {"csarId": "1", "inputs": []}
 229         resp = self.client.post("/api/parser/v1/parservnfd", req_data, format='json')
 230         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 231         self.assertEqual(resp.data, {"error": "VNF CSAR(1) does not exist."})