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