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