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 import os
  18 import catalog.pub.utils.timeutil
  19 from requests.auth import HTTPBasicAuth
  20
  21 from django.test import TestCase
  22 from rest_framework import status
  23 from rest_framework.test import APIClient
  24
  25 from catalog.packages.biz.sdc_vnf_package import NfDistributeThread, NfPkgDeleteThread
  26 from catalog.pub.database.models import JobStatusModel, JobModel
  27 from catalog.pub.database.models import VnfPackageModel
  28 from catalog.pub.msapi import sdc
  29 from catalog.pub.utils import restcall, toscaparser
  30 from .const import vnfd_data
  31 from catalog.pub.config.config import CATALOG_ROOT_PATH
  32 from catalog.packages import const
  33 from catalog.pub.config import config as pub_config
  34
  35
  36 class TestNfPackage(TestCase):
  37     def setUp(self):
  38         self.client = APIClient()
  39         VnfPackageModel.objects.filter().delete()
  40         JobModel.objects.filter().delete()
  41         JobStatusModel.objects.filter().delete()
  42         self.vnfd_data = vnfd_data
  43
  44     def tearDown(self):
  45         pass
  46
  47     def assert_job_result(self, job_id, job_progress, job_detail):
  48         jobs = JobStatusModel.objects.filter(
  49             jobid=job_id,
  50             progress=job_progress,
  51             descp=job_detail)
  52         self.assertEqual(1, len(jobs))
  53
  54     @mock.patch.object(NfDistributeThread, 'run')
  55     def test_nf_pkg_distribute_normal(self, mock_run):
  56         resp = self.client.post(
  57             "/api/catalog/v1/vnfpackages",
  58             {
  59                 "csarId": "1",
  60                 "vimIds": ["1"]
  61             },
  62             format='json'
  63         )
  64         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
  65
  66     def test_nf_pkg_distribute_when_csar_already_exist(self):
  67         VnfPackageModel(
  68             vnfPackageId="1",
  69             vnfdId="vcpe_vfw_zte_1_0"
  70         ).save()
  71         NfDistributeThread(
  72             csar_id="1",
  73             vim_ids=["1"],
  74             lab_vim_id="",
  75             job_id="2"
  76         ).run()
  77         self.assert_job_result("2", 255, "NF CSAR(1) already exists.")
  78
  79     @mock.patch.object(restcall, 'call_req')
  80     @mock.patch.object(sdc, 'download_artifacts')
  81     @mock.patch.object(toscaparser, 'parse_vnfd')
  82     def test_nf_pkg_distribute_when_vnfd_already_exist(self,
  83                                                        mock_parse_vnfd,
  84                                                        mock_download_artifacts,
  85                                                        mock_call_req):
  86         mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
  87         mock_download_artifacts.return_value = "/home/hss.csar"
  88         mock_call_req.return_value = [0, json.JSONEncoder().encode([{
  89             "uuid": "1",
  90             "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
  91         }]), '200']
  92         VnfPackageModel(vnfPackageId="2", vnfdId="00342b18-a5c7-11e8-998c-bf1755941f12").save()
  93         NfDistributeThread(
  94             csar_id="1",
  95             vim_ids=["1"],
  96             lab_vim_id="",
  97             job_id="2"
  98         ).run()
  99         self.assert_job_result("2", 255, "VNF package(00342b18-a5c7-11e8-998c-bf1755941f12) already exists.")
 100
 101     @mock.patch.object(restcall, 'call_req')
 102     @mock.patch.object(sdc, 'download_artifacts')
 103     @mock.patch.object(toscaparser, 'parse_vnfd')
 104     def test_nf_pkg_distribute_successfully(self,
 105                                             mock_parse_vnfd,
 106                                             mock_download_artifacts,
 107                                             mock_call_req):
 108         mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
 109         mock_download_artifacts.return_value = "/home/hss.csar"
 110         mock_call_req.return_value = [0, json.JSONEncoder().encode([{
 111             "uuid": "1",
 112             "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
 113         }]), '200']
 114         NfDistributeThread(
 115             csar_id="1",
 116             vim_ids=["1"],
 117             lab_vim_id="",
 118             job_id="4"
 119         ).run()
 120         self.assert_job_result("4", 100, "CSAR(1) distribute successfully.")
 121
 122     ###############################################################################################################
 123
 124     @mock.patch.object(NfPkgDeleteThread, 'run')
 125     def test_nf_pkg_delete_normal(self, mock_run):
 126         resp = self.client.delete("/api/catalog/v1/vnfpackages/1")
 127         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 128
 129     def test_nf_pkg_normal_delete(self):
 130         VnfPackageModel(
 131             vnfPackageId="2",
 132             vnfdId="vcpe_vfw_zte_1_0"
 133         ).save()
 134         NfPkgDeleteThread(
 135             csar_id="2",
 136             job_id="2"
 137         ).run()
 138         self.assert_job_result("2", 100, "Delete CSAR(2) successfully.")
 139
 140     def test_nf_pkg_get_all(self):
 141         VnfPackageModel(
 142             vnfPackageId="3",
 143             vnfdId="3",
 144             vnfVendor='3',
 145             vnfdVersion='3',
 146             vnfSoftwareVersion='',
 147             vnfPackageUri='',
 148             vnfdModel=''
 149         ).save()
 150         VnfPackageModel(
 151             vnfPackageId="4",
 152             vnfdId="4",
 153             vnfVendor='4',
 154             vnfdVersion='4',
 155             vnfSoftwareVersion='',
 156             vnfPackageUri='',
 157             vnfdModel=''
 158         ).save()
 159         resp = self.client.get("/api/catalog/v1/vnfpackages")
 160         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 161         expect_data = [
 162             {
 163                 "imageInfo": [],
 164                 "csarId": "3",
 165                 "packageInfo": {
 166                     "csarName": "",
 167                     "vnfdModel": "",
 168                     "vnfdProvider": "3",
 169                     "vnfdId": "3",
 170                     "downloadUrl": "http://127.0.0.1:8806/static/catalog/3/",
 171                     "vnfVersion": "",
 172                     "vnfdVersion": "3",
 173                     "vnfPackageId": "3"
 174                 }
 175             },
 176             {
 177                 "imageInfo": [],
 178                 "csarId": "4",
 179                 "packageInfo": {
 180                     "csarName": "",
 181                     "vnfdModel": "",
 182                     "vnfdProvider": "4",
 183                     "vnfdId": "4",
 184                     "downloadUrl": "http://127.0.0.1:8806/static/catalog/4/",
 185                     "vnfVersion": "",
 186                     "vnfdVersion": "4",
 187                     "vnfPackageId": "4"
 188                 }
 189             }
 190         ]
 191         self.assertEqual(expect_data, resp.data)
 192
 193     def test_nf_pkg_get_one(self):
 194         VnfPackageModel(
 195             vnfPackageId="4",
 196             vnfdId="4",
 197             vnfVendor='4',
 198             vnfdVersion='4',
 199             vnfSoftwareVersion='',
 200             vnfPackageUri='',
 201             vnfdModel=''
 202         ).save()
 203
 204         resp = self.client.get("/api/catalog/v1/vnfpackages/4")
 205         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 206         expect_data = {
 207             "imageInfo": [],
 208             "csarId": "4",
 209             "packageInfo": {
 210                 "csarName": "",
 211                 "vnfdModel": "",
 212                 "vnfdProvider": "4",
 213                 "vnfdId": "4",
 214                 "downloadUrl": "http://127.0.0.1:8806/static/catalog/4/",
 215                 "vnfVersion": "",
 216                 "vnfdVersion": "4",
 217                 "vnfPackageId": "4"
 218             }
 219         }
 220         self.assertEqual(expect_data, resp.data)
 221
 222     def test_nf_pkg_get_one_failed(self):
 223         VnfPackageModel(
 224             vnfPackageId="4",
 225             vnfdId="4",
 226             vnfVendor='4',
 227             vnfdVersion='4',
 228             vnfSoftwareVersion='',
 229             vnfPackageUri='',
 230             vnfdModel=''
 231         ).save()
 232
 233         resp = self.client.get("/api/catalog/v1/vnfpackages/2")
 234         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 235         self.assertEqual({'error': 'Vnf package[2] not Found.'}, resp.data)
 236
 237     ###############################################################################################################
 238
 239     @mock.patch.object(toscaparser, 'parse_vnfd')
 240     def test_vnfd_parse_normal(self, mock_parse_vnfd):
 241         VnfPackageModel(
 242             vnfPackageId="8",
 243             vnfdId="10"
 244         ).save()
 245         mock_parse_vnfd.return_value = json.JSONEncoder().encode({"c": "d"})
 246         req_data = {
 247             "csarId": "8",
 248             "inputs": []
 249         }
 250         resp = self.client.post(
 251             "/api/parser/v1/parservnfd",
 252             req_data,
 253             format='json'
 254         )
 255         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 256         self.assertEqual({"model": '{"c": "d"}'}, resp.data)
 257
 258     def test_vnfd_parse_when_csar_not_exist(self):
 259         req_data = {"csarId": "1", "inputs": []}
 260         resp = self.client.post(
 261             "/api/parser/v1/parservnfd",
 262             req_data,
 263             format='json'
 264         )
 265         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 266         self.assertEqual(resp.data, {"error": "VNF CSAR(1) does not exist."})
 267
 268     @mock.patch("requests.post")
 269     @mock.patch.object(sdc, 'download_artifacts')
 270     @mock.patch.object(sdc, 'get_artifact')
 271     @mock.patch("requests.get")
 272     @mock.patch("uuid.uuid4")
 273     @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
 274     def test_service_pkg_distribute_and_notify(self, mock_nowtime, mock_uuid, mock_requests_get, mock_get_artifact,
 275                                                mock_download_artifacts, mock_requests_post):
 276         mock_nowtime.return_value = "2019-02-16 14:41:16"
 277         uuid_subscriptid = "99442b18-a5c7-11e8-998c-bf1755941f13"
 278         uuid_csarid = "1234"
 279         mock_uuid.side_effect = [uuid_subscriptid, "1111", "2222"]
 280         mock_requests_get.return_value.status_code = 204
 281         vnf_subscription_data = {
 282             "filter": {
 283                 "notificationTypes": [
 284                     "VnfPackageOnboardingNotification",
 285                     "VnfPackageChangeNotification"
 286                 ],
 287                 "vnfPkgId": [uuid_csarid],
 288                 "operationalState": ["ENABLED", "DISABLED"]
 289             },
 290             "callbackUri": "https://so-vnfm-simulator.onap:9093/vnfpkgm/v1/notification",
 291             "authentication": {
 292                 "authType": [
 293                     "BASIC"
 294                 ],
 295                 "paramsBasic": {
 296                     "userName": "vnfm",
 297                     "password": "password1$"
 298                 }
 299             }
 300         }
 301         response = self.client.post(
 302             "/api/vnfpkgm/v1/subscriptions",
 303             data=vnf_subscription_data,
 304             format='json')
 305         self.assertEqual(201, response.status_code)
 306         mock_download_artifacts.return_value = os.path.join(CATALOG_ROOT_PATH, "Sol004VnfCSAR.csar")
 307         mock_get_artifact.return_value = {
 308             "uuid": "c94490a0-f7ef-48be-b3f8-8d8662a37236",
 309             "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb",
 310             "name": "Sol004VnfCSAR",
 311             "version": "2.0",
 312             "toscaModelURL": "/sdc/v1/catalog/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel",
 313             "category": "Volte",
 314             "subCategory": "VolteVF",
 315             "resourceType": "VF",
 316             "lifecycleState": "CERTIFIED",
 317             "lastUpdaterUserId": "jh0003"
 318         }
 319         NfDistributeThread(csar_id=uuid_csarid, vim_ids=["1"], lab_vim_id="", job_id="4").on_distribute()
 320         print(VnfPackageModel.objects.all().values())
 321         expect_onboarding_notification = {
 322             'id': "1111",
 323             'notificationType': const.PKG_NOTIFICATION_TYPE.ONBOARDING,
 324             'timeStamp': "2019-02-16 14:41:16",
 325             'vnfPkgId': "1234",
 326             'vnfdId': "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
 327             '_links': {
 328                 'vnfPackage': {
 329                     'href': '%s/%s/vnf_packages/%s' % (pub_config.MSB_BASE_URL, const.PKG_URL_PREFIX, uuid_csarid)},
 330                 'subscription': {
 331                     'href': '%s/%s%s' % (pub_config.MSB_BASE_URL, const.VNFPKG_SUBSCRIPTION_ROOT_URI, uuid_subscriptid)}
 332
 333             },
 334             "subscriptionId": uuid_subscriptid
 335         }
 336         mock_requests_post.return_value.status_code = 204
 337         mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"],
 338                                               data=json.dumps(expect_onboarding_notification),
 339                                               headers={'Connection': 'close',
 340                                                        'content-type': 'application/json',
 341                                                        'accept': 'application/json'},
 342                                               auth=HTTPBasicAuth("vnfm", "password1$"),
 343                                               verify=False)
 344         mock_requests_post.return_value.status_code = 204
 345         expect_deleted_notification = {
 346             'id': "2222",
 347             'notificationType': const.PKG_NOTIFICATION_TYPE.CHANGE,
 348             'timeStamp': "2019-02-16 14:41:16",
 349             'vnfPkgId': "1234",
 350             'vnfdId': "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
 351             '_links': {
 352                 'vnfPackage': {
 353                     'href': '%s/%s/vnf_packages/%s' % (pub_config.MSB_BASE_URL, const.PKG_URL_PREFIX,
 354                                                        uuid_csarid)},
 355                     'subscription': {
 356                         'href': '%s/%s%s' % (pub_config.MSB_BASE_URL, const.VNFPKG_SUBSCRIPTION_ROOT_URI,
 357                                              uuid_subscriptid)}
 358
 359             },
 360             'changeType': const.PKG_CHANGE_TYPE.PKG_DELETE,
 361             'operationalState': None,
 362             "subscriptionId": uuid_subscriptid
 363         }
 364         NfPkgDeleteThread(csar_id=uuid_csarid, job_id="5").delete_csar()
 365         mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"],
 366                                               data=json.dumps(expect_deleted_notification),
 367                                               headers={'Connection': 'close',
 368                                                        'content-type': 'application/json',
 369                                                        'accept': 'application/json'},
 370                                               auth=HTTPBasicAuth("vnfm", "password1$"),
 371                                               verify=False)