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_bad_req(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_400_BAD_REQUEST)
  94
  95     @mock.patch.object(urllib.request, 'urlopen')
  96     def test_upload_from_uri_failed(self, mock_urlopen):
  97         vnf_pkg = VnfPackageModel.objects.create(
  98             vnfPackageId="333",
  99             onboardingState="CREATED"
 100         )
 101         req_data = {"addressInformation": "error"}
 102         mock_urlopen.return_value = Exception('Boom!')
 103         vnf_pkg_id = vnf_pkg.vnfPackageId
 104         VnfPkgUploadThread(req_data, vnf_pkg_id).run()
 105         vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="333")
 106         self.assertEqual("CREATED", vnf_pkg1[0].onboardingState)
 107
 108     def test_create_vnf_pkg(self):
 109         req_data = {
 110             "userDefinedData": {"a": "A"}
 111         }
 112         response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
 113         resp_data = json.loads(response.content)
 114         expect_resp_data = {
 115             "id": resp_data.get("id"),
 116             "onboardingState": "CREATED",
 117             "operationalState": "DISABLED",
 118             "usageState": "NOT_IN_USE",
 119             "userDefinedData": {"a": "A"},
 120             "_links": None  # TODO
 121         }
 122         self.assertEqual(expect_resp_data, resp_data)
 123         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
 124
 125     def test_query_single_vnf(self):
 126         VnfPackageModel.objects.create(
 127             vnfPackageId="222",
 128             vnfdId="zte-hss-1.0",
 129             vnfVendor="zte",
 130             vnfdProductName="hss",
 131             vnfSoftwareVersion="1.0.0",
 132             vnfdVersion="1.0.0",
 133             checksum='{"algorithm":"111", "hash": "11"}',
 134             onboardingState="CREATED",
 135             operationalState="DISABLED",
 136             usageState="NOT_IN_USE",
 137             userDefinedData='{"a": "A"}'
 138         )
 139         response = self.client.get("%s/222" % VNF_BASE_URL)
 140         expect_data = {
 141             "id": "222",
 142             "vnfdId": "zte-hss-1.0",
 143             "vnfProductName": "hss",
 144             "vnfSoftwareVersion": "1.0.0",
 145             "vnfdVersion": "1.0.0",
 146             "checksum": {"algorithm": "111", "hash": "11"},
 147             "softwareImages": None,
 148             "additionalArtifacts": None,
 149             "onboardingState": "CREATED",
 150             "operationalState": "DISABLED",
 151             "usageState": "NOT_IN_USE",
 152             "userDefinedData": {"a": "A"},
 153             "_links": None
 154         }
 155         self.assertEqual(response.data, expect_data)
 156         self.assertEqual(response.status_code, status.HTTP_200_OK)
 157
 158     def test_query_single_vnf_failed(self):
 159         response = self.client.get(VNF_BASE_URL + "/222")
 160         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 161
 162     def test_query_multiple_vnf(self):
 163         VnfPackageModel.objects.create(
 164             vnfPackageId="111",
 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         VnfPackageModel.objects.create(
 177             vnfPackageId="222",
 178             vnfdId="zte-hss-1.0",
 179             vnfVendor="zte",
 180             vnfdProductName="hss",
 181             vnfSoftwareVersion="1.0.0",
 182             vnfdVersion="1.0.0",
 183             checksum='{"algorithm":"111", "hash": "11"}',
 184             onboardingState="CREATED",
 185             operationalState="DISABLED",
 186             usageState="NOT_IN_USE",
 187             userDefinedData='{"a": "A"}'
 188         )
 189         response = self.client.get(VNF_BASE_URL)
 190         expect_data = [
 191             {
 192                 "id": "111",
 193                 "vnfdId": "zte-hss-1.0",
 194                 "vnfProductName": "hss",
 195                 "vnfSoftwareVersion": "1.0.0",
 196                 "vnfdVersion": "1.0.0",
 197                 "checksum": {"algorithm": "111", "hash": "11"},
 198                 "softwareImages": None,
 199                 "additionalArtifacts": None,
 200                 "onboardingState": "CREATED",
 201                 "operationalState": "DISABLED",
 202                 "usageState": "NOT_IN_USE",
 203                 "userDefinedData": {"a": "A"},
 204                 "_links": None
 205             },
 206             {
 207                 "id": "222",
 208                 "vnfdId": "zte-hss-1.0",
 209                 "vnfProductName": "hss",
 210                 "vnfSoftwareVersion": "1.0.0",
 211                 "vnfdVersion": "1.0.0",
 212                 "checksum": {"algorithm": "111", "hash": "11"},
 213                 "softwareImages": None,
 214                 "additionalArtifacts": None,
 215                 "onboardingState": "CREATED",
 216                 "operationalState": "DISABLED",
 217                 "usageState": "NOT_IN_USE",
 218                 "userDefinedData": {"a": "A"},
 219                 "_links": None
 220             }
 221         ]
 222         self.assertEqual(response.data, expect_data)
 223         self.assertEqual(response.status_code, status.HTTP_200_OK)
 224
 225     def test_delete_single_vnf_pkg(self):
 226         VnfPackageModel.objects.create(
 227             vnfPackageId="222",
 228             vnfdId="zte-hss-1.0",
 229             vnfVendor="zte",
 230             vnfdProductName="hss",
 231             vnfSoftwareVersion="1.0.0",
 232             vnfdVersion="1.0.0",
 233             checksum='{"algorithm":"111", "hash": "11"}',
 234             onboardingState="CREATED",
 235             operationalState="DISABLED",
 236             usageState="NOT_IN_USE",
 237             userDefinedData='{"a": "A"}'
 238         )
 239         response = self.client.delete(VNF_BASE_URL + "/222")
 240         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 241         self.assertEqual(response.data, None)
 242
 243     def test_delete_when_vnf_pkg_not_exist(self):
 244         response = self.client.delete(VNF_BASE_URL + "/222")
 245         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 246         self.assertEqual(response.data, None)
 247
 248     def test_fetch_vnf_pkg(self):
 249         with open("vnfPackage.csar", "wt") 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(VNF_BASE_URL + "/222/package_content")
 257         file_content = ''
 258         for data in response.streaming_content:
 259             file_content = file_content + data.decode()
 260         self.assertEqual(response.status_code, status.HTTP_200_OK)
 261         self.assertEqual('AAAABBBBCCCCDDDD', file_content)
 262         os.remove("vnfPackage.csar")
 263
 264     def test_fetch_partical_vnf_pkg(self):
 265         with open("vnfPackage.csar", "wt") 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("%s/222/package_content" % VNF_BASE_URL, HTTP_RANGE="4-7")
 273         partial_file_content = ''
 274         for data in response.streaming_content:
 275             partial_file_content = partial_file_content + data.decode()
 276         self.assertEqual(response.status_code, status.HTTP_200_OK)
 277         self.assertEqual('BBB', partial_file_content)
 278         os.remove("vnfPackage.csar")
 279
 280     def test_fetch_last_partical_vnf_pkg(self):
 281         with open("vnfPackage.csar", "wt") as fp:
 282             fp.writelines("AAAABBBBCCCCDDDD")
 283         VnfPackageModel.objects.create(
 284             vnfPackageId="222",
 285             onboardingState="ONBOARDED",
 286             localFilePath="vnfPackage.csar"
 287         )
 288         response = self.client.get(VNF_BASE_URL + "/222/package_content", HTTP_RANGE=" 4-")
 289         partial_file_content = ''
 290         for data in response.streaming_content:
 291             partial_file_content = partial_file_content + data.decode()
 292         self.assertEqual(response.status_code, status.HTTP_200_OK)
 293         self.assertEqual('BBBBCCCCDDDD', partial_file_content)
 294         os.remove("vnfPackage.csar")
 295
 296     def test_fetch_vnf_pkg_when_pkg_not_exist(self):
 297         response = self.client.get(VNF_BASE_URL + "/222/package_content")
 298         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 299
 300     def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
 301         VnfPackageModel.objects.create(
 302             vnfPackageId="222",
 303             onboardingState="CREATED",
 304             localFilePath="vnfPackage.csar"
 305         )
 306         response = self.client.get(VNF_BASE_URL + "/222/package_content")
 307         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 308
 309     def test_download_vnfd(self):
 310         VnfPackageModel.objects.create(
 311             vnfPackageId="222",
 312             onboardingState="ONBOARDED",
 313             localFilePath=os.path.join(CATALOG_ROOT_PATH, "resource_test.csar")
 314         )
 315         response = self.client.get(VNF_BASE_URL + "/222/vnfd")
 316         self.assertEqual(response.status_code, status.HTTP_200_OK)
 317         with open("vnfd.csar", 'wb') as local_file:
 318             for chunk in response.streaming_content:
 319                 local_file.write(chunk)
 320         self.assertTrue(zipfile.is_zipfile("vnfd.csar"))
 321         os.remove("vnfd.csar")
 322
 323
 324 def test_download_vnfd_when_pkg_not_exist(self):
 325     response = self.client.get(VNF_BASE_URL + "/222/vnfd")
 326     self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 327
 328
 329 def test_download_vnfd_when_catch_cataloge_exception(self):
 330     VnfPackageModel.objects.create(
 331         vnfPackageId="222",
 332         onboardingState="CREATED",
 333         localFilePath="vnfPackage.csar"
 334     )
 335     response = self.client.get(VNF_BASE_URL + "/222/vnfd")
 336     self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 337
 338
 339 @mock.patch.object(VnfPackage, "create_vnf_pkg")
 340 def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
 341     mock_create_vnf_pkg.side_effect = TypeError('integer type')
 342     req_data = {
 343         "userDefinedData": {"a": "A"}
 344     }
 345     response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
 346     self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 347
 348
 349 @mock.patch.object(VnfPackage, "delete_vnf_pkg")
 350 def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
 351     mock_delete_vnf_pkg.side_effect = TypeError("integer type")
 352     response = self.client.delete(VNF_BASE_URL + "/222")
 353     self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 354
 355
 356 @mock.patch.object(VnfPackage, "query_single")
 357 def test_query_single_when_catch_exception(self, mock_query_single):
 358     mock_query_single.side_effect = TypeError("integer type")
 359     response = self.client.get(VNF_BASE_URL + "/222")
 360     self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 361
 362
 363 @mock.patch.object(VnfPackage, "query_multiple")
 364 def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
 365     mock_query_muitiple.side_effect = TypeError("integer type")
 366     response = self.client.get(VNF_BASE_URL)
 367     self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 368
 369
 370 @mock.patch.object(toscaparser, 'parse_vnfd')
 371 def test_upload_when_catch_exception(self, mock_parse_vnfd):
 372     data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
 373     VnfPackageModel.objects.create(
 374         vnfPackageId="222",
 375         onboardingState="CREATED"
 376     )
 377     mock_parse_vnfd.side_effect = TypeError("integer type")
 378     response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 379     self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 380
 381
 382 @mock.patch.object(VnfPkgUploadThread, 'start')
 383 def test_upload_from_uri_when_catch_exception(self, mock_start):
 384     req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
 385     mock_start.side_effect = TypeError("integer type")
 386     response = self.client.post(VNF_BASE_URL + "/111/package_content/upload_from_uri", data=req_data)
 387     self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 388
 389
 390 @mock.patch.object(VnfPackage, 'download')
 391 def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
 392     mock_download.side_effect = TypeError("integer type")
 393     response = self.client.get(VNF_BASE_URL + "/222/package_content")
 394     self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 395
 396
 397 @mock.patch.object(toscaparser, 'parse_vnfd')
 398 def test_fetch_vnf_artifact(self, mock_parse_vnfd):
 399     data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
 400     VnfPackageModel.objects.create(
 401         vnfPackageId="222",
 402         onboardingState="CREATED"
 403     )
 404     mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 405     response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 406     self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 407     response = self.client.get(VNF_BASE_URL + "/222/artifacts/image")
 408     self.assertEqual(response.status_code, status.HTTP_200_OK)
 409     self.assertEqual(response.getvalue(), b"ubuntu_16.04\n")
 410
 411
 412 @mock.patch.object(toscaparser, 'parse_vnfd')
 413 def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
 414     data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
 415     VnfPackageModel.objects.create(
 416         vnfPackageId="222",
 417         onboardingState="CREATED"
 418     )
 419     mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 420     response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 421     self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 422     response = self.client.get(VNF_BASE_URL + "/1451/artifacts/image")
 423     self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 424
 425
 426 @mock.patch.object(toscaparser, 'parse_vnfd')
 427 def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
 428     data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
 429     VnfPackageModel.objects.create(
 430         vnfPackageId="222",
 431         onboardingState="CREATED"
 432     )
 433     mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 434     response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 435     self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 436     response = self.client.get(VNF_BASE_URL + "/222/artifacts/image1")
 437     self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)