catalog/packages/tests/test_vnf_package.py

   1 # Copyright 2018 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 os
  16 import json
  17 import mock
  18 import urllib2
  19
  20 from rest_framework.test import APIClient
  21 from django.test import TestCase
  22 from rest_framework import status
  23 from catalog.pub.config.config import CATALOG_ROOT_PATH
  24 from catalog.packages.biz.vnf_package import VnfPkgUploadThread
  25 from catalog.pub.database.models import VnfPackageModel
  26 from catalog.pub.utils import toscaparser
  27 from catalog.packages.const import PKG_STATUS
  28 from catalog.packages.tests.const import vnfd_data
  29 from catalog.packages.biz.vnf_package import VnfPackage
  30
  31
  32 class MockReq():
  33     def read(self):
  34         return "1"
  35
  36     def close(self):
  37         pass
  38
  39
  40 class TestVnfPackage(TestCase):
  41     def setUp(self):
  42         self.client = APIClient()
  43
  44     def tearDown(self):
  45         pass
  46
  47     @mock.patch.object(toscaparser, 'parse_vnfd')
  48     def test_upload_vnf_pkg(self, mock_parse_vnfd):
  49         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
  50         VnfPackageModel.objects.create(
  51             vnfPackageId="222",
  52             onboardingState="CREATED"
  53         )
  54         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
  55         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
  56         vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId="222")
  57         self.assertEqual("zte-hss-1.0", vnf_pkg[0].vnfdId)
  58         self.assertEqual(PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
  59         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
  60
  61         os.remove(vnf_pkg[0].localFilePath)
  62         os.removedirs(os.path.join(CATALOG_ROOT_PATH, vnf_pkg[0].vnfPackageId))
  63
  64     def test_upload_vnf_pkg_failed(self):
  65         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
  66         VnfPackageModel.objects.create(
  67             vnfPackageId="222",
  68         )
  69         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
  70         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
  71
  72     @mock.patch.object(toscaparser, 'parse_vnfd')
  73     @mock.patch.object(urllib2, 'urlopen')
  74     def test_upload_nf_pkg_from_uri(self, mock_urlopen, mock_parse_vnfd):
  75         vnf_pkg = VnfPackageModel.objects.create(
  76             vnfPackageId="222",
  77             onboardingState="CREATED"
  78         )
  79         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
  80         req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
  81         mock_urlopen.return_value = MockReq()
  82         vnf_pkg_id = vnf_pkg.vnfPackageId
  83         VnfPkgUploadThread(req_data, vnf_pkg_id).run()
  84         vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="222")
  85         self.assertEqual("zte-hss-1.0", vnf_pkg1[0].vnfdId)
  86
  87         os.remove(vnf_pkg1[0].localFilePath)
  88         os.removedirs(os.path.join(CATALOG_ROOT_PATH, vnf_pkg1[0].vnfPackageId))
  89
  90     def test_create_vnf_pkg(self):
  91         req_data = {
  92             "userDefinedData": {"a": "A"}
  93         }
  94         response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
  95         resp_data = json.loads(response.content)
  96         expect_resp_data = {
  97             "id": resp_data.get("id"),
  98             "onboardingState": "CREATED",
  99             "operationalState": "DISABLED",
 100             "usageState": "NOT_IN_USE",
 101             "userDefinedData": {"a": "A"},
 102             "_links": None  # TODO
 103         }
 104         self.assertEqual(expect_resp_data, resp_data)
 105         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
 106
 107     def test_query_single_vnf(self):
 108         VnfPackageModel.objects.create(
 109             vnfPackageId="222",
 110             vnfdId="zte-hss-1.0",
 111             vnfVendor="zte",
 112             vnfdProductName="hss",
 113             vnfSoftwareVersion="1.0.0",
 114             vnfdVersion="1.0.0",
 115             checksum='{"algorithm":"111", "hash": "11"}',
 116             onboardingState="CREATED",
 117             operationalState="DISABLED",
 118             usageState="NOT_IN_USE",
 119             userDefinedData='{"a": "A"}'
 120         )
 121         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
 122         expect_data = {
 123             "id": "222",
 124             "vnfdId": "zte-hss-1.0",
 125             "vnfProductName": "hss",
 126             "vnfSoftwareVersion": "1.0.0",
 127             "vnfdVersion": "1.0.0",
 128             "checksum": {"algorithm": "111", "hash": "11"},
 129             "softwareImages": None,
 130             "additionalArtifacts": None,
 131             "onboardingState": "CREATED",
 132             "operationalState": "DISABLED",
 133             "usageState": "NOT_IN_USE",
 134             "userDefinedData": {"a": "A"},
 135             "_links": None
 136         }
 137         self.assertEqual(response.data, expect_data)
 138         self.assertEqual(response.status_code, status.HTTP_200_OK)
 139
 140     def test_query_single_vnf_failed(self):
 141         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
 142         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 143
 144     def test_query_multiple_vnf(self):
 145         VnfPackageModel.objects.create(
 146             vnfPackageId="111",
 147             vnfdId="zte-hss-1.0",
 148             vnfVendor="zte",
 149             vnfdProductName="hss",
 150             vnfSoftwareVersion="1.0.0",
 151             vnfdVersion="1.0.0",
 152             checksum='{"algorithm":"111", "hash": "11"}',
 153             onboardingState="CREATED",
 154             operationalState="DISABLED",
 155             usageState="NOT_IN_USE",
 156             userDefinedData='{"a": "A"}'
 157         )
 158         VnfPackageModel.objects.create(
 159             vnfPackageId="222",
 160             vnfdId="zte-hss-1.0",
 161             vnfVendor="zte",
 162             vnfdProductName="hss",
 163             vnfSoftwareVersion="1.0.0",
 164             vnfdVersion="1.0.0",
 165             checksum='{"algorithm":"111", "hash": "11"}',
 166             onboardingState="CREATED",
 167             operationalState="DISABLED",
 168             usageState="NOT_IN_USE",
 169             userDefinedData='{"a": "A"}'
 170         )
 171         response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
 172         expect_data = [
 173             {
 174                 "id": "111",
 175                 "vnfdId": "zte-hss-1.0",
 176                 "vnfProductName": "hss",
 177                 "vnfSoftwareVersion": "1.0.0",
 178                 "vnfdVersion": "1.0.0",
 179                 "checksum": {"algorithm": "111", "hash": "11"},
 180                 "softwareImages": None,
 181                 "additionalArtifacts": None,
 182                 "onboardingState": "CREATED",
 183                 "operationalState": "DISABLED",
 184                 "usageState": "NOT_IN_USE",
 185                 "userDefinedData": {"a": "A"},
 186                 "_links": None
 187             },
 188             {
 189                 "id": "222",
 190                 "vnfdId": "zte-hss-1.0",
 191                 "vnfProductName": "hss",
 192                 "vnfSoftwareVersion": "1.0.0",
 193                 "vnfdVersion": "1.0.0",
 194                 "checksum": {"algorithm": "111", "hash": "11"},
 195                 "softwareImages": None,
 196                 "additionalArtifacts": None,
 197                 "onboardingState": "CREATED",
 198                 "operationalState": "DISABLED",
 199                 "usageState": "NOT_IN_USE",
 200                 "userDefinedData": {"a": "A"},
 201                 "_links": None
 202             }
 203         ]
 204         self.assertEqual(response.data, expect_data)
 205         self.assertEqual(response.status_code, status.HTTP_200_OK)
 206
 207     def test_delete_single_vnf_pkg(self):
 208         VnfPackageModel.objects.create(
 209             vnfPackageId="222",
 210             vnfdId="zte-hss-1.0",
 211             vnfVendor="zte",
 212             vnfdProductName="hss",
 213             vnfSoftwareVersion="1.0.0",
 214             vnfdVersion="1.0.0",
 215             checksum='{"algorithm":"111", "hash": "11"}',
 216             onboardingState="CREATED",
 217             operationalState="DISABLED",
 218             usageState="NOT_IN_USE",
 219             userDefinedData='{"a": "A"}'
 220         )
 221         response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
 222         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 223         self.assertEqual(response.data, None)
 224
 225     def test_delete_when_vnf_pkg_not_exist(self):
 226         response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
 227         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 228         self.assertEqual(response.data, None)
 229
 230     def test_fetch_vnf_pkg(self):
 231         with open("vnfPackage.csar", "wb") as fp:
 232             fp.writelines("AAAABBBBCCCCDDDD")
 233         VnfPackageModel.objects.create(
 234             vnfPackageId="222",
 235             onboardingState="ONBOARDED",
 236             localFilePath="vnfPackage.csar"
 237         )
 238         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
 239         file_content = ''
 240         for data in response.streaming_content:
 241             file_content = file_content + data
 242         self.assertEqual(response.status_code, status.HTTP_200_OK)
 243         self.assertEqual('AAAABBBBCCCCDDDD', file_content)
 244         os.remove("vnfPackage.csar")
 245
 246     def test_fetch_partical_vnf_pkg(self):
 247         with open("vnfPackage.csar", "wb") as fp:
 248             fp.writelines("AAAABBBBCCCCDDDD")
 249         VnfPackageModel.objects.create(
 250             vnfPackageId="222",
 251             onboardingState="ONBOARDED",
 252             localFilePath="vnfPackage.csar"
 253         )
 254         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", RANGE="4-7")
 255         partial_file_content = ''
 256         for data in response.streaming_content:
 257             partial_file_content = partial_file_content + data
 258         self.assertEqual(response.status_code, status.HTTP_200_OK)
 259         self.assertEqual('BBBB', partial_file_content)
 260         os.remove("vnfPackage.csar")
 261
 262     def test_fetch_vnf_pkg_when_pkg_not_exist(self):
 263         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
 264         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 265
 266     @mock.patch.object(VnfPackage, "create_vnf_pkg")
 267     def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
 268         mock_create_vnf_pkg.side_effect = TypeError('integer type')
 269         req_data = {
 270             "userDefinedData": {"a": "A"}
 271         }
 272         response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
 273         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 274
 275     @mock.patch.object(VnfPackage, "delete_vnf_pkg")
 276     def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
 277         mock_delete_vnf_pkg.side_effect = TypeError("integer type")
 278         response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
 279         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 280
 281     @mock.patch.object(VnfPackage, "query_single")
 282     def test_query_single_when_catch_exception(self, mock_query_single):
 283         mock_query_single.side_effect = TypeError("integer type")
 284         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
 285         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 286
 287     @mock.patch.object(VnfPackage, "query_multiple")
 288     def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
 289         mock_query_muitiple.side_effect = TypeError("integer type")
 290         response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
 291         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 292
 293     @mock.patch.object(toscaparser, 'parse_vnfd')
 294     def test_upload_when_catch_exception(self, mock_parse_vnfd):
 295         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
 296         VnfPackageModel.objects.create(
 297             vnfPackageId="222",
 298             onboardingState="CREATED"
 299         )
 300         mock_parse_vnfd.side_effect = TypeError("integer type")
 301         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
 302         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)