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