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 query artifact(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(sdc, 'get_artifacts')
 123     def test_ns_when_not_distributed_by_sdc(self, mock_get_artifacts):
 124         mock_get_artifacts.return_value = [{
 125             "uuid": "1",
 126             "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb",
 127             "name": "underlayvpn",
 128             "version": "2.0",
 129             "toscaModelURL": "/sdc/v1/catalog/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel",
 130             "category": "Volte",
 131             "subCategory": "VolteVNF",
 132             "resourceType": "VF",
 133             "lifecycleState": "CERTIFIED",
 134             "distributionStatus": "DISTRIBUTION_APPROVED",
 135             "lastUpdaterUserId": "jh0003"
 136         }]
 137         resp = self.client.post(
 138             "/api/catalog/v1/nspackages", {"csarId": "1"}, format='json')
 139         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 140         self.assertEqual("failed", resp.data["status"])
 141         self.assertEqual(
 142             "The artifact (services,1) is not distributed from sdc.",
 143             resp.data["statusDescription"])
 144
 145     ##########################################################################
 146
 147     def test_ns_pkg_normal_delete(self):
 148         NSPackageModel(nsPackageId="8", nsdId="2").save()
 149         resp = self.client.delete("/api/catalog/v1/nspackages/8")
 150         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 151         self.assertEqual("success", resp.data["status"])
 152         self.assertEqual(
 153             "Delete CSAR(8) successfully.",
 154             resp.data["statusDescription"])
 155
 156     def test_ns_pkg_get_all(self):
 157         NSPackageModel(
 158             nsPackageId="13",
 159             nsdId="2",
 160             nsdDesginer="2",
 161             nsdVersion="2",
 162             nsPackageUri="13.csar",
 163             nsdModel="").save()
 164         NSPackageModel(
 165             nsPackageId="14",
 166             nsdId="3",
 167             nsdDesginer="3",
 168             nsdVersion="3",
 169             nsPackageUri="14.csar",
 170             nsdModel="").save()
 171         resp = self.client.get("/api/catalog/v1/nspackages")
 172         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 173         expect_data = [{"csarId": "13",
 174                         "packageInfo": {"csarName": "13.csar",
 175                                         "nsdProvider": "2",
 176                                         "nsdId": "2",
 177                                         "nsPackageId": "13",
 178                                         "downloadUrl": "http://127.0.0.1:8806/static/catalog/13/13.csar",
 179                                         "nsdModel": "",
 180                                         "nsdVersion": "2",
 181                                         "nsdInvariantId": None
 182                                         }},
 183                        {"csarId": "14",
 184                         "packageInfo": {"csarName": "14.csar",
 185                                         "nsdProvider": "3",
 186                                         "nsdId": "3",
 187                                         "nsPackageId": "14",
 188                                         "downloadUrl": "http://127.0.0.1:8806/static/catalog/14/14.csar",
 189                                         "nsdModel": "",
 190                                         "nsdVersion": "3",
 191                                         "nsdInvariantId": None}}]
 192         self.assertEqual(expect_data, resp.data)
 193
 194     def test_ns_pkg_get_one(self):
 195         NSPackageModel(
 196             nsPackageId="14",
 197             nsdId="2",
 198             nsdDesginer="3",
 199             nsdVersion="4",
 200             nsPackageUri="14.csar",
 201             nsdModel="").save()
 202         resp = self.client.get("/api/catalog/v1/nspackages/14")
 203         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 204         expect_data = {
 205             "csarId": "14",
 206             "packageInfo": {
 207                 "nsdId": "2",
 208                 "nsPackageId": "14",
 209                 "nsdProvider": "3",
 210                 "nsdVersion": "4",
 211                 "csarName": "14.csar",
 212                 "nsdModel": "",
 213                 "downloadUrl": "http://127.0.0.1:8806/static/catalog/14/14.csar",
 214                 "nsdInvariantId": None}}
 215         self.assertEqual(expect_data, resp.data)
 216
 217     def test_ns_pkg_get_one_not_found(self):
 218         resp = self.client.get("/api/catalog/v1/nspackages/22")
 219         self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 220         self.assertEqual(
 221             {"error": "Ns package[22] not Found."},
 222             resp.data)
 223
 224     ##########################################################################
 225
 226     @mock.patch.object(toscaparser, 'parse_nsd')
 227     def test_nsd_parse_normal(self, mock_parse_nsd):
 228         NSPackageModel(nsPackageId="18", nsdId="12").save()
 229         mock_parse_nsd.return_value = json.JSONEncoder().encode({"a": "b"})
 230         req_data = {"csarId": "18", "inputs": []}
 231         resp = self.client.post(
 232             "/api/parser/v1/parsernsd",
 233             req_data,
 234             format='json')
 235         self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
 236         self.assertEqual({"model": '{"a": "b"}'}, resp.data)
 237
 238     def test_nsd_parse_when_csar_not_exist(self):
 239         req_data = {"csarId": "1", "inputs": []}
 240         resp = self.client.post(
 241             "/api/parser/v1/parsernsd",
 242             req_data,
 243             format='json')
 244         self.assertEqual(
 245             resp.status_code,
 246             status.HTTP_500_INTERNAL_SERVER_ERROR)
 247         self.assertEqual(resp.data, {"error": "NS CSAR(1) does not exist."})