catalog/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 catalog.packages.biz.sdc_vnf_package import NfDistributeThread, NfPkgDeleteThread
  21 from catalog.pub.database.models import JobStatusModel, JobModel
  22 from catalog.pub.database.models import VnfPackageModel
  23 from catalog.pub.msapi import sdc
  24 from catalog.pub.utils import restcall, toscaparser
  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/catalog/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(toscaparser, '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", vnfdId="zte-hss-1.0").save()
  85         NfDistributeThread(
  86             csar_id="1",
  87             vim_ids=["1"],
  88             lab_vim_id="",
  89             job_id="2"
  90         ).run()
  91         self.assert_job_result("2", 255, "VNF package(zte-hss-1.0) already exists.")
  92
  93     @mock.patch.object(restcall, 'call_req')
  94     @mock.patch.object(sdc, 'download_artifacts')
  95     @mock.patch.object(toscaparser, 'parse_vnfd')
  96     def test_nf_pkg_distribute_successfully(self,
  97                                             mock_parse_vnfd,
  98                                             mock_download_artifacts,
  99                                             mock_call_req):
 100         mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
 101         mock_download_artifacts.return_value = "/home/hss.csar"
 102         mock_call_req.return_value = [0, json.JSONEncoder().encode([{
 103             "uuid": "1",
 104             "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
 105         }]), '200']
 106         NfDistributeThread(
 107             csar_id="1",
 108             vim_ids=["1"],
 109             lab_vim_id="",
 110             job_id="4"
 111         ).run()
 112         self.assert_job_result("4", 100, "CSAR(1) distribute successfully.")
 113
 114     ###############################################################################################################
 115
 116     @mock.patch.object(NfPkgDeleteThread, 'run')
 117     def test_nf_pkg_delete_normal(self, mock_run):
 118         resp = self.client.delete("/api/catalog/v1/vnfpackages/1")
 119         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 120
 121     def test_nf_pkg_normal_delete(self):
 122         VnfPackageModel(
 123             vnfPackageId="2",
 124             vnfdId="vcpe_vfw_zte_1_0"
 125         ).save()
 126         NfPkgDeleteThread(
 127             csar_id="2",
 128             job_id="2"
 129         ).run()
 130         self.assert_job_result("2", 100, "Delete CSAR(2) successfully.")
 131
 132     def test_nf_pkg_get_all(self):
 133         VnfPackageModel(
 134             vnfPackageId="3",
 135             vnfdId="3",
 136             vnfVendor='3',
 137             vnfdVersion='3',
 138             vnfSoftwareVersion='',
 139             vnfPackageUri='',
 140             vnfdModel=''
 141         ).save()
 142         VnfPackageModel(
 143             vnfPackageId="4",
 144             vnfdId="4",
 145             vnfVendor='4',
 146             vnfdVersion='4',
 147             vnfSoftwareVersion='',
 148             vnfPackageUri='',
 149             vnfdModel=''
 150         ).save()
 151         resp = self.client.get("/api/catalog/v1/vnfpackages")
 152         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 153         expect_data = [
 154             {
 155                 "imageInfo": [],
 156                 "csarId": "3",
 157                 "packageInfo": {
 158                     "csarName": "",
 159                     "vnfdModel": "",
 160                     "vnfdProvider": "3",
 161                     "vnfdId": "3",
 162                     "downloadUrl": "http://127.0.0.1:8806/static/catalog/3/",
 163                     "vnfVersion": "",
 164                     "vnfdVersion": "3",
 165                     "vnfPackageId": "3"
 166                 }
 167             },
 168             {
 169                 "imageInfo": [],
 170                 "csarId": "4",
 171                 "packageInfo": {
 172                     "csarName": "",
 173                     "vnfdModel": "",
 174                     "vnfdProvider": "4",
 175                     "vnfdId": "4",
 176                     "downloadUrl": "http://127.0.0.1:8806/static/catalog/4/",
 177                     "vnfVersion": "",
 178                     "vnfdVersion": "4",
 179                     "vnfPackageId": "4"
 180                 }
 181             }
 182         ]
 183         self.assertEqual(expect_data, resp.data)
 184
 185     def test_nf_pkg_get_one(self):
 186         VnfPackageModel(
 187             vnfPackageId="4",
 188             vnfdId="4",
 189             vnfVendor='4',
 190             vnfdVersion='4',
 191             vnfSoftwareVersion='',
 192             vnfPackageUri='',
 193             vnfdModel=''
 194         ).save()
 195
 196         resp = self.client.get("/api/catalog/v1/vnfpackages/4")
 197         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 198         expect_data = {
 199             "imageInfo": [],
 200             "csarId": "4",
 201             "packageInfo": {
 202                 "csarName": "",
 203                 "vnfdModel": "",
 204                 "vnfdProvider": "4",
 205                 "vnfdId": "4",
 206                 "downloadUrl": "http://127.0.0.1:8806/static/catalog/4/",
 207                 "vnfVersion": "",
 208                 "vnfdVersion": "4",
 209                 "vnfPackageId": "4"
 210             }
 211         }
 212         self.assertEqual(expect_data, resp.data)
 213
 214     def test_nf_pkg_get_one_failed(self):
 215         VnfPackageModel(
 216             vnfPackageId="4",
 217             vnfdId="4",
 218             vnfVendor='4',
 219             vnfdVersion='4',
 220             vnfSoftwareVersion='',
 221             vnfPackageUri='',
 222             vnfdModel=''
 223         ).save()
 224
 225         resp = self.client.get("/api/catalog/v1/vnfpackages/2")
 226         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 227         self.assertEqual({'error': 'Vnf package[2] not Found.'}, resp.data)
 228
 229     ###############################################################################################################
 230
 231     @mock.patch.object(toscaparser, 'parse_vnfd')
 232     def test_vnfd_parse_normal(self, mock_parse_vnfd):
 233         VnfPackageModel(
 234             vnfPackageId="8",
 235             vnfdId="10"
 236         ).save()
 237         mock_parse_vnfd.return_value = json.JSONEncoder().encode({"c": "d"})
 238         req_data = {
 239             "csarId": "8",
 240             "inputs": []
 241         }
 242         resp = self.client.post(
 243             "/api/catalog/v1/parservnfd",
 244             req_data,
 245             format='json'
 246         )
 247         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 248         self.assertEqual({"model": '{"c": "d"}'}, resp.data)
 249
 250     def test_vnfd_parse_when_csar_not_exist(self):
 251         req_data = {"csarId": "1", "inputs": []}
 252         resp = self.client.post(
 253             "/api/catalog/v1/parservnfd",
 254             req_data,
 255             format='json'
 256         )
 257         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 258         self.assertEqual(resp.data, {"error": "VNF CSAR(1) does not exist."})