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