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