genericparser/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 json
  16 import os
  17 import urllib2
  18 import mock
  19
  20 from django.test import TestCase
  21 from rest_framework import status
  22 from rest_framework.test import APIClient
  23
  24 from genericparser.packages.biz.vnf_package import VnfPackage, VnfPkgUploadThread
  25 from genericparser.packages.const import PKG_STATUS
  26 from genericparser.packages.tests.const import vnfd_data
  27 from genericparser.pub.config.config import GENERICPARSER_ROOT_PATH
  28 from genericparser.pub.database.models import VnfPackageModel
  29 from genericparser.pub.utils import toscaparsers
  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(toscaparsers, 'parse_vnfd')
  48     def test_upload_vnf_pkg(self, mock_parse_vnfd):
  49         data = {'file': open(os.path.join(GENERICPARSER_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     def test_upload_vnf_pkg_failed(self):
  62         data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "empty.txt"), "rb")}
  63         VnfPackageModel.objects.create(
  64             vnfPackageId="222",
  65         )
  66         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
  67         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
  68
  69     @mock.patch.object(toscaparsers, 'parse_vnfd')
  70     @mock.patch.object(urllib2, 'urlopen')
  71     def test_upload_nf_pkg_from_uri(self, mock_urlopen, mock_parse_vnfd):
  72         vnf_pkg = VnfPackageModel.objects.create(
  73             vnfPackageId="222",
  74             onboardingState="CREATED"
  75         )
  76         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
  77         req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
  78         mock_urlopen.return_value = MockReq()
  79         vnf_pkg_id = vnf_pkg.vnfPackageId
  80         VnfPkgUploadThread(req_data, vnf_pkg_id).run()
  81         vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="222")
  82         self.assertEqual("zte-hss-1.0", vnf_pkg1[0].vnfdId)
  83
  84     def test_upload_from_uri_failed(self):
  85         req_data = {"username": "123"}
  86         response = self.client.post("/api/vnfpkgm/v1/vnf_packages/111/package_content/upload_from_uri", data=req_data)
  87         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
  88
  89     def test_create_vnf_pkg(self):
  90         req_data = {
  91             "userDefinedData": {"a": "A"}
  92         }
  93         response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
  94         resp_data = json.loads(response.content)
  95         expect_resp_data = {
  96             "id": resp_data.get("id"),
  97             "onboardingState": "CREATED",
  98             "operationalState": "DISABLED",
  99             "usageState": "NOT_IN_USE",
 100             "userDefinedData": {"a": "A"},
 101             "_links": None  # TODO
 102         }
 103         self.assertEqual(expect_resp_data, resp_data)
 104         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
 105
 106     def test_query_single_vnf(self):
 107         VnfPackageModel.objects.create(
 108             vnfPackageId="222",
 109             vnfdId="zte-hss-1.0",
 110             vnfVendor="zte",
 111             vnfdProductName="hss",
 112             vnfSoftwareVersion="1.0.0",
 113             vnfdVersion="1.0.0",
 114             checksum='{"algorithm":"111", "hash": "11"}',
 115             onboardingState="CREATED",
 116             operationalState="DISABLED",
 117             usageState="NOT_IN_USE",
 118             userDefinedData='{"a": "A"}'
 119         )
 120         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
 121         expect_data = {
 122             "id": "222",
 123             "vnfdId": "zte-hss-1.0",
 124             "vnfProductName": "hss",
 125             "vnfSoftwareVersion": "1.0.0",
 126             "vnfdVersion": "1.0.0",
 127             "checksum": {"algorithm": "111", "hash": "11"},
 128             "softwareImages": None,
 129             "additionalArtifacts": None,
 130             "onboardingState": "CREATED",
 131             "operationalState": "DISABLED",
 132             "usageState": "NOT_IN_USE",
 133             "userDefinedData": {"a": "A"},
 134             "_links": None
 135         }
 136         self.assertEqual(response.data, expect_data)
 137         self.assertEqual(response.status_code, status.HTTP_200_OK)
 138
 139     def test_query_single_vnf_failed(self):
 140         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
 141         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 142
 143     def test_query_multiple_vnf(self):
 144         VnfPackageModel.objects.create(
 145             vnfPackageId="111",
 146             vnfdId="zte-hss-1.0",
 147             vnfVendor="zte",
 148             vnfdProductName="hss",
 149             vnfSoftwareVersion="1.0.0",
 150             vnfdVersion="1.0.0",
 151             checksum='{"algorithm":"111", "hash": "11"}',
 152             onboardingState="CREATED",
 153             operationalState="DISABLED",
 154             usageState="NOT_IN_USE",
 155             userDefinedData='{"a": "A"}'
 156         )
 157         VnfPackageModel.objects.create(
 158             vnfPackageId="222",
 159             vnfdId="zte-hss-1.0",
 160             vnfVendor="zte",
 161             vnfdProductName="hss",
 162             vnfSoftwareVersion="1.0.0",
 163             vnfdVersion="1.0.0",
 164             checksum='{"algorithm":"111", "hash": "11"}',
 165             onboardingState="CREATED",
 166             operationalState="DISABLED",
 167             usageState="NOT_IN_USE",
 168             userDefinedData='{"a": "A"}'
 169         )
 170         response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
 171         expect_data = [
 172             {
 173                 "id": "111",
 174                 "vnfdId": "zte-hss-1.0",
 175                 "vnfProductName": "hss",
 176                 "vnfSoftwareVersion": "1.0.0",
 177                 "vnfdVersion": "1.0.0",
 178                 "checksum": {"algorithm": "111", "hash": "11"},
 179                 "softwareImages": None,
 180                 "additionalArtifacts": None,
 181                 "onboardingState": "CREATED",
 182                 "operationalState": "DISABLED",
 183                 "usageState": "NOT_IN_USE",
 184                 "userDefinedData": {"a": "A"},
 185                 "_links": None
 186             },
 187             {
 188                 "id": "222",
 189                 "vnfdId": "zte-hss-1.0",
 190                 "vnfProductName": "hss",
 191                 "vnfSoftwareVersion": "1.0.0",
 192                 "vnfdVersion": "1.0.0",
 193                 "checksum": {"algorithm": "111", "hash": "11"},
 194                 "softwareImages": None,
 195                 "additionalArtifacts": None,
 196                 "onboardingState": "CREATED",
 197                 "operationalState": "DISABLED",
 198                 "usageState": "NOT_IN_USE",
 199                 "userDefinedData": {"a": "A"},
 200                 "_links": None
 201             }
 202         ]
 203         self.assertEqual(response.data, expect_data)
 204         self.assertEqual(response.status_code, status.HTTP_200_OK)
 205
 206     def test_delete_single_vnf_pkg(self):
 207         VnfPackageModel.objects.create(
 208             vnfPackageId="222",
 209             vnfdId="zte-hss-1.0",
 210             vnfVendor="zte",
 211             vnfdProductName="hss",
 212             vnfSoftwareVersion="1.0.0",
 213             vnfdVersion="1.0.0",
 214             checksum='{"algorithm":"111", "hash": "11"}',
 215             onboardingState="CREATED",
 216             operationalState="DISABLED",
 217             usageState="NOT_IN_USE",
 218             userDefinedData='{"a": "A"}'
 219         )
 220         response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
 221         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 222         self.assertEqual(response.data, None)
 223
 224     def test_delete_when_vnf_pkg_not_exist(self):
 225         response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
 226         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 227         self.assertEqual(response.data, None)
 228
 229     def test_fetch_vnf_pkg(self):
 230         with open("vnfPackage.csar", "wb") as fp:
 231             fp.writelines("AAAABBBBCCCCDDDD")
 232         VnfPackageModel.objects.create(
 233             vnfPackageId="222",
 234             onboardingState="ONBOARDED",
 235             localFilePath="vnfPackage.csar"
 236         )
 237         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
 238         file_content = ''
 239         for data in response.streaming_content:
 240             file_content = file_content + data
 241         self.assertEqual(response.status_code, status.HTTP_200_OK)
 242         self.assertEqual('AAAABBBBCCCCDDDD', file_content)
 243         os.remove("vnfPackage.csar")
 244
 245     def test_fetch_partical_vnf_pkg(self):
 246         with open("vnfPackage.csar", "wb") as fp:
 247             fp.writelines("AAAABBBBCCCCDDDD")
 248         VnfPackageModel.objects.create(
 249             vnfPackageId="222",
 250             onboardingState="ONBOARDED",
 251             localFilePath="vnfPackage.csar"
 252         )
 253         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", RANGE="4-7")
 254         partial_file_content = ''
 255         for data in response.streaming_content:
 256             partial_file_content = partial_file_content + data
 257         self.assertEqual(response.status_code, status.HTTP_200_OK)
 258         self.assertEqual('BBB', partial_file_content)
 259         os.remove("vnfPackage.csar")
 260
 261     def test_fetch_vnf_pkg_when_pkg_not_exist(self):
 262         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
 263         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 264
 265     def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
 266         VnfPackageModel.objects.create(
 267             vnfPackageId="222",
 268             onboardingState="CREATED",
 269             localFilePath="vnfPackage.csar"
 270         )
 271         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
 272         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 273
 274     @mock.patch.object(VnfPackage, "create_vnf_pkg")
 275     def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
 276         mock_create_vnf_pkg.side_effect = TypeError('integer type')
 277         req_data = {
 278             "userDefinedData": {"a": "A"}
 279         }
 280         response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
 281         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 282
 283     @mock.patch.object(VnfPackage, "delete_vnf_pkg")
 284     def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
 285         mock_delete_vnf_pkg.side_effect = TypeError("integer type")
 286         response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
 287         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 288
 289     @mock.patch.object(VnfPackage, "query_single")
 290     def test_query_single_when_catch_exception(self, mock_query_single):
 291         mock_query_single.side_effect = TypeError("integer type")
 292         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
 293         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 294
 295     @mock.patch.object(VnfPackage, "query_multiple")
 296     def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
 297         mock_query_muitiple.side_effect = TypeError("integer type")
 298         response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
 299         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 300
 301     @mock.patch.object(toscaparsers, 'parse_vnfd')
 302     def test_upload_when_catch_exception(self, mock_parse_vnfd):
 303         data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "empty.txt"), "rb")}
 304         VnfPackageModel.objects.create(
 305             vnfPackageId="222",
 306             onboardingState="CREATED"
 307         )
 308         mock_parse_vnfd.side_effect = TypeError("integer type")
 309         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
 310         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 311
 312     @mock.patch.object(VnfPkgUploadThread, 'start')
 313     def test_upload_from_uri_when_catch_exception(self, mock_start):
 314         req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
 315         mock_start.side_effect = TypeError("integer type")
 316         response = self.client.post("/api/vnfpkgm/v1/vnf_packages/111/package_content/upload_from_uri", data=req_data)
 317         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 318
 319     @mock.patch.object(VnfPackage, 'download')
 320     def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
 321         mock_download.side_effect = TypeError("integer type")
 322         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
 323         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 324
 325     @mock.patch.object(toscaparsers, 'parse_vnfd')
 326     def test_fetch_vnf_artifact(self, mock_parse_vnfd):
 327         data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
 328         VnfPackageModel.objects.create(
 329             vnfPackageId="222",
 330             onboardingState="CREATED"
 331         )
 332         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 333         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
 334         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 335         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image")
 336         self.assertEqual(response.status_code, status.HTTP_200_OK)
 337         self.assertEqual(response.getvalue(), "ubuntu_16.04\n")
 338
 339     @mock.patch.object(toscaparsers, 'parse_vnfd')
 340     def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
 341         data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
 342         VnfPackageModel.objects.create(
 343             vnfPackageId="222",
 344             onboardingState="CREATED"
 345         )
 346         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 347         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
 348         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 349         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/1451/artifacts/image")
 350         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 351
 352     @mock.patch.object(toscaparsers, 'parse_vnfd')
 353     def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
 354         data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
 355         VnfPackageModel.objects.create(
 356             vnfPackageId="222",
 357             onboardingState="CREATED"
 358         )
 359         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 360         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
 361         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 362         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image1")
 363         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)