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("00342b18-a5c7-11e8-998c-bf1755941f12", 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("00342b18-a5c7-11e8-998c-bf1755941f12", 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": {'self': {'href': '/api/vnfpkgm/v1/vnf_packages/222'},
 154                        'vnfd': {
 155                            'href': '/api/vnfpkgm/v1/vnf_packages/222/vnfd'},
 156                        'packageContent': {
 157                            'href': '/api/vnfpkgm/v1/vnf_packages/222/package_content'}
 158                        }
 159         }
 160         self.assertEqual(response.data, expect_data)
 161         self.assertEqual(response.status_code, status.HTTP_200_OK)
 162
 163     def test_query_single_vnf_failed(self):
 164         response = self.client.get(VNF_BASE_URL + "/222")
 165         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 166
 167     def test_query_multiple_vnf(self):
 168         VnfPackageModel.objects.create(
 169             vnfPackageId="111",
 170             vnfdId="zte-hss-1.0",
 171             vnfVendor="zte",
 172             vnfdProductName="hss",
 173             vnfSoftwareVersion="1.0.0",
 174             vnfdVersion="1.0.0",
 175             checksum='{"algorithm":"111", "hash": "11"}',
 176             onboardingState="CREATED",
 177             operationalState="DISABLED",
 178             usageState="NOT_IN_USE",
 179             userDefinedData='{"a": "A"}'
 180         )
 181         VnfPackageModel.objects.create(
 182             vnfPackageId="222",
 183             vnfdId="zte-hss-1.0",
 184             vnfVendor="zte",
 185             vnfdProductName="hss",
 186             vnfSoftwareVersion="1.0.0",
 187             vnfdVersion="1.0.0",
 188             checksum='{"algorithm":"111", "hash": "11"}',
 189             onboardingState="CREATED",
 190             operationalState="DISABLED",
 191             usageState="NOT_IN_USE",
 192             userDefinedData='{"a": "A"}'
 193         )
 194         response = self.client.get(VNF_BASE_URL)
 195         expect_data = [
 196             {
 197                 "id": "111",
 198                 "vnfdId": "zte-hss-1.0",
 199                 "vnfProductName": "hss",
 200                 "vnfSoftwareVersion": "1.0.0",
 201                 "vnfdVersion": "1.0.0",
 202                 "checksum": {"algorithm": "111", "hash": "11"},
 203                 "softwareImages": None,
 204                 "additionalArtifacts": None,
 205                 "onboardingState": "CREATED",
 206                 "operationalState": "DISABLED",
 207                 "usageState": "NOT_IN_USE",
 208                 "userDefinedData": {"a": "A"},
 209                 "_links": {
 210                     "self": {
 211                         "href": "/api/vnfpkgm/v1/vnf_packages/111"
 212                     },
 213                     "vnfd": {
 214                         "href": "/api/vnfpkgm/v1/vnf_packages/111/vnfd"
 215                     },
 216                     "packageContent": {
 217                         "href": "/api/vnfpkgm/v1/vnf_packages/111/package_content"
 218                     }
 219                 }
 220             },
 221             {
 222                 "id": "222",
 223                 "vnfdId": "zte-hss-1.0",
 224                 "vnfProductName": "hss",
 225                 "vnfSoftwareVersion": "1.0.0",
 226                 "vnfdVersion": "1.0.0",
 227                 "checksum": {"algorithm": "111", "hash": "11"},
 228                 "softwareImages": None,
 229                 "additionalArtifacts": None,
 230                 "onboardingState": "CREATED",
 231                 "operationalState": "DISABLED",
 232                 "usageState": "NOT_IN_USE",
 233                 "userDefinedData": {"a": "A"},
 234                 "_links": {'self': {'href': '/api/vnfpkgm/v1/vnf_packages/222'},
 235                            'vnfd': {'href': '/api/vnfpkgm/v1/vnf_packages/222/vnfd'},
 236                            'packageContent': {
 237                                'href': '/api/vnfpkgm/v1/vnf_packages/222/package_content'}}
 238             }
 239         ]
 240         self.assertEqual(response.data, expect_data)
 241         self.assertEqual(response.status_code, status.HTTP_200_OK)
 242
 243     def test_delete_single_vnf_pkg(self):
 244         VnfPackageModel.objects.create(
 245             vnfPackageId="222",
 246             vnfdId="zte-hss-1.0",
 247             vnfVendor="zte",
 248             vnfdProductName="hss",
 249             vnfSoftwareVersion="1.0.0",
 250             vnfdVersion="1.0.0",
 251             checksum='{"algorithm":"111", "hash": "11"}',
 252             onboardingState="CREATED",
 253             operationalState="DISABLED",
 254             usageState="NOT_IN_USE",
 255             userDefinedData='{"a": "A"}'
 256         )
 257         response = self.client.delete(VNF_BASE_URL + "/222")
 258         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 259         self.assertEqual(response.data, None)
 260
 261     def test_delete_when_vnf_pkg_not_exist(self):
 262         response = self.client.delete(VNF_BASE_URL + "/222")
 263         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 264         self.assertEqual(response.data, None)
 265
 266     def test_fetch_vnf_pkg(self):
 267         with open("vnfPackage.csar", "wt") as fp:
 268             fp.writelines("AAAABBBBCCCCDDDD")
 269         VnfPackageModel.objects.create(
 270             vnfPackageId="222",
 271             onboardingState="ONBOARDED",
 272             localFilePath="vnfPackage.csar"
 273         )
 274         response = self.client.get(VNF_BASE_URL + "/222/package_content")
 275         file_content = ''
 276         for data in response.streaming_content:
 277             file_content = file_content + data.decode()
 278         self.assertEqual(response.status_code, status.HTTP_200_OK)
 279         self.assertEqual('AAAABBBBCCCCDDDD', file_content)
 280         os.remove("vnfPackage.csar")
 281
 282     def test_fetch_partical_vnf_pkg(self):
 283         with open("vnfPackage.csar", "wt") as fp:
 284             fp.writelines("AAAABBBBCCCCDDDD")
 285         VnfPackageModel.objects.create(
 286             vnfPackageId="222",
 287             onboardingState="ONBOARDED",
 288             localFilePath="vnfPackage.csar"
 289         )
 290         response = self.client.get("%s/222/package_content" % VNF_BASE_URL, HTTP_RANGE="4-7")
 291         partial_file_content = ''
 292         for data in response.streaming_content:
 293             partial_file_content = partial_file_content + data.decode()
 294         self.assertEqual(response.status_code, status.HTTP_200_OK)
 295         self.assertEqual('BBB', partial_file_content)
 296         os.remove("vnfPackage.csar")
 297
 298     def test_fetch_last_partical_vnf_pkg(self):
 299         with open("vnfPackage.csar", "wt") as fp:
 300             fp.writelines("AAAABBBBCCCCDDDD")
 301         VnfPackageModel.objects.create(
 302             vnfPackageId="222",
 303             onboardingState="ONBOARDED",
 304             localFilePath="vnfPackage.csar"
 305         )
 306         response = self.client.get(VNF_BASE_URL + "/222/package_content", HTTP_RANGE=" 4-")
 307         partial_file_content = ''
 308         for data in response.streaming_content:
 309             partial_file_content = partial_file_content + data.decode()
 310         self.assertEqual(response.status_code, status.HTTP_200_OK)
 311         self.assertEqual('BBBBCCCCDDDD', partial_file_content)
 312         os.remove("vnfPackage.csar")
 313
 314     def test_fetch_vnf_pkg_when_pkg_not_exist(self):
 315         response = self.client.get(VNF_BASE_URL + "/222/package_content")
 316         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 317
 318     def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
 319         VnfPackageModel.objects.create(
 320             vnfPackageId="222",
 321             onboardingState="CREATED",
 322             localFilePath="vnfPackage.csar"
 323         )
 324         response = self.client.get(VNF_BASE_URL + "/222/package_content")
 325         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 326
 327     def test_download_vnfd(self):
 328         VnfPackageModel.objects.create(
 329             vnfPackageId="222",
 330             onboardingState="ONBOARDED",
 331             localFilePath=os.path.join(CATALOG_ROOT_PATH, "resource_test.csar")
 332         )
 333         response = self.client.get(VNF_BASE_URL + "/222/vnfd")
 334         self.assertEqual(response.status_code, status.HTTP_200_OK)
 335         with open("vnfd.csar", 'wb') as local_file:
 336             for chunk in response.streaming_content:
 337                 local_file.write(chunk)
 338         self.assertTrue(zipfile.is_zipfile("vnfd.csar"))
 339         os.remove("vnfd.csar")
 340
 341     def test_download_vnfd_when_pkg_not_exist(self):
 342         response = self.client.get(VNF_BASE_URL + "/222/vnfd")
 343         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 344
 345     def test_download_vnfd_when_catch_cataloge_exception(self):
 346         VnfPackageModel.objects.create(
 347             vnfPackageId="222",
 348             onboardingState="CREATED",
 349             localFilePath="vnfPackage.csar"
 350         )
 351         response = self.client.get(VNF_BASE_URL + "/222/vnfd")
 352         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 353
 354     @mock.patch.object(VnfPackage, "create_vnf_pkg")
 355     def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
 356         mock_create_vnf_pkg.side_effect = TypeError('integer type')
 357         req_data = {
 358             "userDefinedData": {"a": "A"}
 359         }
 360         response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
 361         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 362
 363     @mock.patch.object(VnfPackage, "delete_vnf_pkg")
 364     def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
 365         mock_delete_vnf_pkg.side_effect = TypeError("integer type")
 366         response = self.client.delete(VNF_BASE_URL + "/222")
 367         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 368
 369     @mock.patch.object(VnfPackage, "query_single")
 370     def test_query_single_when_catch_exception(self, mock_query_single):
 371         mock_query_single.side_effect = TypeError("integer type")
 372         response = self.client.get(VNF_BASE_URL + "/222")
 373         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 374
 375     @mock.patch.object(VnfPackage, "query_multiple")
 376     def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
 377         mock_query_muitiple.side_effect = TypeError("integer type")
 378         response = self.client.get(VNF_BASE_URL)
 379         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 380
 381     @mock.patch.object(toscaparser, 'parse_vnfd')
 382     def test_upload_when_catch_exception(self, mock_parse_vnfd):
 383         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
 384         VnfPackageModel.objects.create(
 385             vnfPackageId="222",
 386             onboardingState="CREATED"
 387         )
 388         mock_parse_vnfd.side_effect = TypeError("integer type")
 389         response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 390         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 391
 392     @mock.patch.object(VnfPkgUploadThread, 'start')
 393     def test_upload_from_uri_when_catch_exception(self, mock_start):
 394         req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
 395         mock_start.side_effect = TypeError("integer type")
 396         response = self.client.post(VNF_BASE_URL + "/111/package_content/upload_from_uri", data=req_data)
 397         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 398
 399     @mock.patch.object(VnfPackage, 'download')
 400     def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
 401         mock_download.side_effect = TypeError("integer type")
 402         response = self.client.get(VNF_BASE_URL + "/222/package_content")
 403         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 404
 405     @mock.patch.object(toscaparser, 'parse_vnfd')
 406     def test_fetch_vnf_artifact(self, mock_parse_vnfd):
 407         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
 408         VnfPackageModel.objects.create(
 409             vnfPackageId="222",
 410             onboardingState="CREATED"
 411         )
 412         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 413         response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 414         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 415         response = self.client.get(VNF_BASE_URL + "/222/artifacts/image")
 416         self.assertEqual(response.status_code, status.HTTP_200_OK)
 417         self.assertEqual(response.getvalue(), b"ubuntu_16.04\n")
 418
 419     @mock.patch.object(toscaparser, 'parse_vnfd')
 420     def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
 421         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
 422         VnfPackageModel.objects.create(
 423             vnfPackageId="222",
 424             onboardingState="CREATED"
 425         )
 426         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 427         response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 428         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 429         response = self.client.get(VNF_BASE_URL + "/1451/artifacts/image")
 430         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 431
 432     @mock.patch.object(toscaparser, 'parse_vnfd')
 433     def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
 434         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
 435         VnfPackageModel.objects.create(
 436             vnfPackageId="222",
 437             onboardingState="CREATED"
 438         )
 439         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 440         response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 441         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 442         response = self.client.get(VNF_BASE_URL + "/222/artifacts/image1")
 443         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 444
 445     def test_upload_vnf_pkg_with_artifacts(self):
 446         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "vgw.csar"), "rb")}
 447         VnfPackageModel.objects.create(
 448             vnfPackageId="222",
 449             onboardingState="CREATED"
 450         )
 451         response = self.client.put("%s/222/package_content" % VNF_BASE_URL, data=data)
 452         vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId="222")
 453         self.assertEqual(PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
 454         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 455         response = self.client.get("%s/222" % VNF_BASE_URL)
 456         print(response.data)
 457         self.assertEqual(response.status_code, status.HTTP_200_OK)
 458         expact_response_data = {
 459             "id": "222",
 460             "vnfdId": "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
 461             "vnfProductName": "vcpe_vgw",
 462             "vnfSoftwareVersion": "1.0",
 463             "vnfdVersion": "1.0",
 464             "softwareImages": None,
 465             "additionalArtifacts": [
 466                 {
 467                     "artifactPath": "MainServiceTemplate.yaml",
 468                     "checksum": {
 469                         "algorithm": "Null",
 470                         "hash": "Null"
 471                     }
 472                 }
 473             ],
 474             "onboardingState": "ONBOARDED",
 475             "operationalState": "ENABLED",
 476             "usageState": "NOT_IN_USE",
 477             "_links": {
 478                 "self": {
 479                     "href": "/api/vnfpkgm/v1/vnf_packages/222"
 480                 },
 481                 "vnfd": {
 482                     "href": "/api/vnfpkgm/v1/vnf_packages/222/vnfd"
 483                 },
 484                 "packageContent": {
 485                     "href": "/api/vnfpkgm/v1/vnf_packages/222/package_content"
 486                 }
 487             }
 488         }
 489         self.assertEqual(response.data, expact_response_data)