catalog/packages/tests/test_nspackage.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 Client
  19 from django.test import TestCase
  20 from rest_framework import status
  21
  22 from catalog.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
  23 from catalog.pub.msapi import sdc
  24 from catalog.pub.utils import restcall, toscaparser
  25 from .const import nsd_data
  26
  27
  28 class TestNsPackage(TestCase):
  29     def setUp(self):
  30         self.client = Client()
  31         NSPackageModel.objects.filter().delete()
  32         VnfPackageModel.objects.filter().delete()
  33         self.nsd_data = nsd_data
  34
  35     def tearDown(self):
  36         pass
  37
  38     def test_ns_pkg_distribute_when_ns_exists(self):
  39         NSPackageModel(nsPackageId="1", nsdId="2").save()
  40         resp = self.client.post(
  41             "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
  42         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
  43         self.assertEqual("failed", resp.data["status"])
  44         self.assertEqual(
  45             "NS CSAR(1) already exists.",
  46             resp.data["statusDescription"])
  47
  48     @mock.patch.object(restcall, 'call_req')
  49     def test_ns_pkg_distribute_when_csar_not_exist(self, mock_call_req):
  50         mock_call_req.return_value = [0, "{}", '200']
  51         resp = self.client.post(
  52             "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
  53         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
  54         self.assertEqual("failed", resp.data["status"])
  55         self.assertEqual(
  56             "Failed to get asset(services, 1) from sdc.",
  57             resp.data["statusDescription"])
  58
  59     @mock.patch.object(restcall, 'call_req')
  60     @mock.patch.object(sdc, 'download_artifacts')
  61     @mock.patch.object(toscaparser, 'parse_nsd')
  62     def test_ns_pkg_distribute_when_nsd_already_exists(
  63             self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
  64         mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
  65         mock_download_artifacts.return_value = "/home/vcpe.csar"
  66         mock_call_req.return_value = [0, json.JSONEncoder().encode({
  67             "uuid": "1",
  68             "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
  69             "distributionStatus": "DISTRIBUTED"
  70         }), '200']
  71         NSPackageModel(nsPackageId="2", nsdId="VCPE_NS").save()
  72         resp = self.client.post(
  73             "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
  74         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
  75         self.assertEqual("failed", resp.data["status"])
  76         self.assertEqual(
  77             "NSD(VCPE_NS) already exists.",
  78             resp.data["statusDescription"])
  79
  80     @mock.patch.object(restcall, 'call_req')
  81     @mock.patch.object(sdc, 'download_artifacts')
  82     @mock.patch.object(toscaparser, 'parse_nsd')
  83     def test_ns_pkg_distribute_when_nf_not_distributed(
  84             self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
  85         mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
  86         mock_download_artifacts.return_value = "/home/vcpe.csar"
  87         mock_call_req.return_value = [0, json.JSONEncoder().encode({
  88             "uuid": "1",
  89             "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
  90             "distributionStatus": "DISTRIBUTED",
  91         }), '200']
  92         resp = self.client.post(
  93             "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
  94         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
  95         self.assertEqual("failed", resp.data["status"])
  96         self.assertEqual(
  97             "VNF package(vcpe_vfw_zte_1_0) is not distributed.",
  98             resp.data["statusDescription"])
  99
 100     @mock.patch.object(restcall, 'call_req')
 101     @mock.patch.object(sdc, 'download_artifacts')
 102     @mock.patch.object(toscaparser, 'parse_nsd')
 103     def test_ns_pkg_distribute_when_successfully(
 104             self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
 105         mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
 106         mock_download_artifacts.return_value = "/home/vcpe.csar"
 107         mock_call_req.return_value = [0, json.JSONEncoder().encode({
 108             "uuid": "1",
 109             "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
 110             "distributionStatus": "DISTRIBUTED"
 111         }), '200']
 112         VnfPackageModel(vnfPackageId="1", vnfdId="vcpe_vfw_zte_1_0").save()
 113         PnfPackageModel(pnfPackageId="1", pnfdId="m6000_s").save()
 114         resp = self.client.post(
 115             "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
 116         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 117         self.assertEqual("success", resp.data["status"])
 118         self.assertEqual(
 119             "CSAR(1) distributed successfully.",
 120             resp.data["statusDescription"])
 121
 122     @mock.patch.object(restcall, 'call_req')
 123     def test_ns_when_not_distributed_by_sdc(self, mock_call_req):
 124         mock_call_req.return_value = [0, json.JSONEncoder().encode({
 125             "uuid": "1",
 126             "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
 127             "distributionStatus": "DISTRIBUTION_APPROVED"
 128         }), '200']
 129         resp = self.client.post(
 130             "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
 131         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 132         self.assertEqual("failed", resp.data["status"])
 133         self.assertEqual(
 134             "The asset (services,1) is not distributed from sdc.",
 135             resp.data["statusDescription"])
 136
 137     ##########################################################################
 138
 139     def test_ns_pkg_normal_delete(self):
 140         NSPackageModel(nsPackageId="8", nsdId="2").save()
 141         resp = self.client.delete("/api/catalog/v1/nspackages/8")
 142         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 143         self.assertEqual("success", resp.data["status"])
 144         self.assertEqual(
 145             "Delete CSAR(8) successfully.",
 146             resp.data["statusDescription"])
 147
 148     def test_ns_pkg_get_all(self):
 149         NSPackageModel(
 150             nsPackageId="13",
 151             nsdId="2",
 152             nsdDesginer="2",
 153             nsdVersion="2",
 154             nsPackageUri="13.csar",
 155             nsdModel="").save()
 156         NSPackageModel(
 157             nsPackageId="14",
 158             nsdId="3",
 159             nsdDesginer="3",
 160             nsdVersion="3",
 161             nsPackageUri="14.csar",
 162             nsdModel="").save()
 163         resp = self.client.get("/api/catalog/v1/nspackages")
 164         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 165         expect_data = [{"csarId": "13",
 166                         "packageInfo": {"csarName": "13.csar",
 167                                         "nsdProvider": "2",
 168                                         "nsdId": "2",
 169                                         "nsPackageId": "13",
 170                                         "downloadUrl": "http://127.0.0.1:8806/static/catalog/13/13.csar",
 171                                         "nsdModel": "",
 172                                         "nsdVersion": "2",
 173                                         "nsdInvariantId": None
 174                                         }},
 175                        {"csarId": "14",
 176                         "packageInfo": {"csarName": "14.csar",
 177                                         "nsdProvider": "3",
 178                                         "nsdId": "3",
 179                                         "nsPackageId": "14",
 180                                         "downloadUrl": "http://127.0.0.1:8806/static/catalog/14/14.csar",
 181                                         "nsdModel": "",
 182                                         "nsdVersion": "3",
 183                                         "nsdInvariantId": None}}]
 184         self.assertEqual(expect_data, resp.data)
 185
 186     def test_ns_pkg_get_one(self):
 187         NSPackageModel(
 188             nsPackageId="14",
 189             nsdId="2",
 190             nsdDesginer="3",
 191             nsdVersion="4",
 192             nsPackageUri="14.csar",
 193             nsdModel="").save()
 194         resp = self.client.get("/api/catalog/v1/nspackages/14")
 195         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 196         expect_data = {
 197             "csarId": "14",
 198             "packageInfo": {
 199                 "nsdId": "2",
 200                 "nsPackageId": "14",
 201                 "nsdProvider": "3",
 202                 "nsdVersion": "4",
 203                 "csarName": "14.csar",
 204                 "nsdModel": "",
 205                 "downloadUrl": "http://127.0.0.1:8806/static/catalog/14/14.csar",
 206                 "nsdInvariantId": None}}
 207         self.assertEqual(expect_data, resp.data)
 208
 209     def test_ns_pkg_get_one_not_found(self):
 210         resp = self.client.get("/api/catalog/v1/nspackages/22")
 211         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 212         self.assertEqual(
 213             {"error": "Ns package[22] not Found."},
 214             resp.data)
 215
 216     ##########################################################################
 217
 218     @mock.patch.object(toscaparser, 'parse_nsd')
 219     def test_nsd_parse_normal(self, mock_parse_nsd):
 220         NSPackageModel(nsPackageId="18", nsdId="12").save()
 221         mock_parse_nsd.return_value = json.JSONEncoder().encode({"a": "b"})
 222         req_data = {"csarId": "18", "inputs": []}
 223         resp = self.client.post(
 224             "/api/parser/v1/parsernsd",
 225             req_data,
 226             format='json')
 227         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 228         self.assertEqual({"model": '{"a": "b"}'}, resp.data)
 229
 230     def test_nsd_parse_when_csar_not_exist(self):
 231         req_data = {"csarId": "1", "inputs": []}
 232         resp = self.client.post(
 233             "/api/parser/v1/parsernsd",
 234             req_data,
 235             format='json')
 236         self.assertEqual(
 237             resp.status_code,
 238             status.HTTP_500_INTERNAL_SERVER_ERROR)
 239         self.assertEqual(resp.data, {"error": "NS CSAR(1) does not exist."})