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": 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     def test_download_vnfd_when_pkg_not_exist(self):
 324         response = self.client.get(VNF_BASE_URL + "/222/vnfd")
 325         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 326
 327     def test_download_vnfd_when_catch_cataloge_exception(self):
 328         VnfPackageModel.objects.create(
 329             vnfPackageId="222",
 330             onboardingState="CREATED",
 331             localFilePath="vnfPackage.csar"
 332         )
 333         response = self.client.get(VNF_BASE_URL + "/222/vnfd")
 334         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 335
 336     @mock.patch.object(VnfPackage, "create_vnf_pkg")
 337     def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
 338         mock_create_vnf_pkg.side_effect = TypeError('integer type')
 339         req_data = {
 340             "userDefinedData": {"a": "A"}
 341         }
 342         response = self.client.post(VNF_BASE_URL, data=req_data, format="json")
 343         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 344
 345     @mock.patch.object(VnfPackage, "delete_vnf_pkg")
 346     def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
 347         mock_delete_vnf_pkg.side_effect = TypeError("integer type")
 348         response = self.client.delete(VNF_BASE_URL + "/222")
 349         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 350
 351     @mock.patch.object(VnfPackage, "query_single")
 352     def test_query_single_when_catch_exception(self, mock_query_single):
 353         mock_query_single.side_effect = TypeError("integer type")
 354         response = self.client.get(VNF_BASE_URL + "/222")
 355         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 356
 357     @mock.patch.object(VnfPackage, "query_multiple")
 358     def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
 359         mock_query_muitiple.side_effect = TypeError("integer type")
 360         response = self.client.get(VNF_BASE_URL)
 361         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 362
 363     @mock.patch.object(toscaparser, 'parse_vnfd')
 364     def test_upload_when_catch_exception(self, mock_parse_vnfd):
 365         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
 366         VnfPackageModel.objects.create(
 367             vnfPackageId="222",
 368             onboardingState="CREATED"
 369         )
 370         mock_parse_vnfd.side_effect = TypeError("integer type")
 371         response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 372         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 373
 374     @mock.patch.object(VnfPkgUploadThread, 'start')
 375     def test_upload_from_uri_when_catch_exception(self, mock_start):
 376         req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
 377         mock_start.side_effect = TypeError("integer type")
 378         response = self.client.post(VNF_BASE_URL + "/111/package_content/upload_from_uri", data=req_data)
 379         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 380
 381     @mock.patch.object(VnfPackage, 'download')
 382     def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
 383         mock_download.side_effect = TypeError("integer type")
 384         response = self.client.get(VNF_BASE_URL + "/222/package_content")
 385         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 386
 387     @mock.patch.object(toscaparser, 'parse_vnfd')
 388     def test_fetch_vnf_artifact(self, mock_parse_vnfd):
 389         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
 390         VnfPackageModel.objects.create(
 391             vnfPackageId="222",
 392             onboardingState="CREATED"
 393         )
 394         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 395         response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 396         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 397         response = self.client.get(VNF_BASE_URL + "/222/artifacts/image")
 398         self.assertEqual(response.status_code, status.HTTP_200_OK)
 399         self.assertEqual(response.getvalue(), b"ubuntu_16.04\n")
 400
 401     @mock.patch.object(toscaparser, 'parse_vnfd')
 402     def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
 403         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
 404         VnfPackageModel.objects.create(
 405             vnfPackageId="222",
 406             onboardingState="CREATED"
 407         )
 408         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 409         response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 410         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 411         response = self.client.get(VNF_BASE_URL + "/1451/artifacts/image")
 412         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 413
 414     @mock.patch.object(toscaparser, 'parse_vnfd')
 415     def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
 416         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
 417         VnfPackageModel.objects.create(
 418             vnfPackageId="222",
 419             onboardingState="CREATED"
 420         )
 421         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 422         response = self.client.put(VNF_BASE_URL + "/222/package_content", data=data)
 423         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 424         response = self.client.get(VNF_BASE_URL + "/222/artifacts/image1")
 425         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 426
 427     def test_upload_vnf_pkg_with_artifacts(self):
 428         data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "vgw.csar"), "rb")}
 429         VnfPackageModel.objects.create(
 430             vnfPackageId="222",
 431             onboardingState="CREATED"
 432         )
 433         response = self.client.put("%s/222/package_content" % VNF_BASE_URL, data=data)
 434         vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId="222")
 435         self.assertEqual(PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
 436         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 437         response = self.client.get("%s/222" % VNF_BASE_URL)
 438         print(response.data)
 439         self.assertEqual(response.status_code, status.HTTP_200_OK)
 440         expact_response_data = {
 441             "id": "222",
 442             "vnfdId": "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
 443             "vnfProductName": "vcpe_vgw",
 444             "vnfSoftwareVersion": "1.0",
 445             "vnfdVersion": "1.0",
 446             "softwareImages": None,
 447             "additionalArtifacts": [
 448                 {
 449                     "artifactPath": "MainServiceTemplate.yaml",
 450                     "checksum": {
 451                         "algorithm": "Null",
 452                         "hash": "Null"
 453                     }
 454                 }
 455             ],
 456             "onboardingState": "ONBOARDED",
 457             "operationalState": "ENABLED",
 458             "usageState": "NOT_IN_USE",
 459             "_links": None
 460         }
 461         self.assertEqual(response.data, expact_response_data)