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         for key, value in expect_resp_data.items():
 108             self.assertEqual(resp_data[key], value)
 109         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
 110
 111     def test_query_single_vnf(self):
 112         VnfPackageModel.objects.create(
 113             vnfPackageId="222",
 114             vnfdId="zte-hss-1.0",
 115             vnfVendor="zte",
 116             vnfdProductName="hss",
 117             vnfSoftwareVersion="1.0.0",
 118             vnfdVersion="1.0.0",
 119             checksum='{"algorithm":"111", "hash": "11"}',
 120             onboardingState="CREATED",
 121             operationalState="DISABLED",
 122             usageState="NOT_IN_USE",
 123             userDefinedData='{"a": "A"}'
 124         )
 125         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
 126         expect_data = {
 127             "id": "222",
 128             "vnfdId": "zte-hss-1.0",
 129             "vnfProductName": "hss",
 130             "vnfSoftwareVersion": "1.0.0",
 131             "vnfdVersion": "1.0.0",
 132             "checksum": {"algorithm": "111", "hash": "11"},
 133             "softwareImages": None,
 134             "additionalArtifacts": None,
 135             "onboardingState": "CREATED",
 136             "operationalState": "DISABLED",
 137             "usageState": "NOT_IN_USE",
 138             "userDefinedData": {"a": "A"},
 139             "_links": None
 140         }
 141         for key, value in expect_data.items():
 142             self.assertEqual(response.data[key], value)
 143         # self.assertEqual(response.data, expect_data)
 144         self.assertEqual(response.status_code, status.HTTP_200_OK)
 145
 146     def test_query_single_vnf_failed(self):
 147         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
 148         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 149
 150     def test_query_multiple_vnf(self):
 151         VnfPackageModel.objects.create(
 152             vnfPackageId="111",
 153             vnfdId="zte-hss-1.0",
 154             vnfVendor="zte",
 155             vnfdProductName="hss",
 156             vnfSoftwareVersion="1.0.0",
 157             vnfdVersion="1.0.0",
 158             checksum='{"algorithm":"111", "hash": "11"}',
 159             onboardingState="CREATED",
 160             operationalState="DISABLED",
 161             usageState="NOT_IN_USE",
 162             userDefinedData='{"a": "A"}'
 163         )
 164         VnfPackageModel.objects.create(
 165             vnfPackageId="222",
 166             vnfdId="zte-hss-1.0",
 167             vnfVendor="zte",
 168             vnfdProductName="hss",
 169             vnfSoftwareVersion="1.0.0",
 170             vnfdVersion="1.0.0",
 171             checksum='{"algorithm":"111", "hash": "11"}',
 172             onboardingState="CREATED",
 173             operationalState="DISABLED",
 174             usageState="NOT_IN_USE",
 175             userDefinedData='{"a": "A"}'
 176         )
 177         response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
 178         expect_data = [
 179             {
 180                 "id": "111",
 181                 "vnfdId": "zte-hss-1.0",
 182                 "vnfProductName": "hss",
 183                 "vnfSoftwareVersion": "1.0.0",
 184                 "vnfdVersion": "1.0.0",
 185                 "checksum": {"algorithm": "111", "hash": "11"},
 186                 "softwareImages": None,
 187                 "additionalArtifacts": None,
 188                 "onboardingState": "CREATED",
 189                 "operationalState": "DISABLED",
 190                 "usageState": "NOT_IN_USE",
 191                 "userDefinedData": {"a": "A"},
 192                 "_links": None
 193             },
 194             {
 195                 "id": "222",
 196                 "vnfdId": "zte-hss-1.0",
 197                 "vnfProductName": "hss",
 198                 "vnfSoftwareVersion": "1.0.0",
 199                 "vnfdVersion": "1.0.0",
 200                 "checksum": {"algorithm": "111", "hash": "11"},
 201                 "softwareImages": None,
 202                 "additionalArtifacts": None,
 203                 "onboardingState": "CREATED",
 204                 "operationalState": "DISABLED",
 205                 "usageState": "NOT_IN_USE",
 206                 "userDefinedData": {"a": "A"},
 207                 "_links": None
 208             }
 209         ]
 210         # self.assertEqual(response.data, expect_data)
 211         self.assertEqual(len(response.data), len(expect_data))
 212         for i in range(0, len(expect_data)):
 213             for key, value in expect_data[i].items():
 214                 self.assertEqual(response.data[i][key], value)
 215
 216         self.assertEqual(response.status_code, status.HTTP_200_OK)
 217
 218     def test_delete_single_vnf_pkg(self):
 219         VnfPackageModel.objects.create(
 220             vnfPackageId="222",
 221             vnfdId="zte-hss-1.0",
 222             vnfVendor="zte",
 223             vnfdProductName="hss",
 224             vnfSoftwareVersion="1.0.0",
 225             vnfdVersion="1.0.0",
 226             checksum='{"algorithm":"111", "hash": "11"}',
 227             onboardingState="CREATED",
 228             operationalState="DISABLED",
 229             usageState="NOT_IN_USE",
 230             userDefinedData='{"a": "A"}'
 231         )
 232         response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
 233         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 234         self.assertEqual(response.data, None)
 235
 236     def test_delete_when_vnf_pkg_not_exist(self):
 237         response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
 238         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 239         self.assertEqual(response.data, None)
 240
 241     def test_fetch_vnf_pkg(self):
 242         with open("vnfPackage.csar", "wb") as fp:
 243             fp.writelines("AAAABBBBCCCCDDDD")
 244         VnfPackageModel.objects.create(
 245             vnfPackageId="222",
 246             onboardingState="ONBOARDED",
 247             localFilePath="vnfPackage.csar"
 248         )
 249         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
 250         file_content = ''
 251         for data in response.streaming_content:
 252             file_content = file_content + data
 253         self.assertEqual(response.status_code, status.HTTP_200_OK)
 254         self.assertEqual('AAAABBBBCCCCDDDD', file_content)
 255         os.remove("vnfPackage.csar")
 256
 257     def test_fetch_partical_vnf_pkg(self):
 258         with open("vnfPackage.csar", "wb") as fp:
 259             fp.writelines("AAAABBBBCCCCDDDD")
 260         VnfPackageModel.objects.create(
 261             vnfPackageId="222",
 262             onboardingState="ONBOARDED",
 263             localFilePath="vnfPackage.csar"
 264         )
 265         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", HTTP_RANGE="4-7")
 266         partial_file_content = ''
 267         for data in response.streaming_content:
 268             partial_file_content = partial_file_content + data
 269         self.assertEqual(response.status_code, status.HTTP_200_OK)
 270         self.assertEqual('BBB', partial_file_content)
 271         os.remove("vnfPackage.csar")
 272
 273     def test_fetch_last_partical_vnf_pkg(self):
 274         with open("vnfPackage.csar", "wb") as fp:
 275             fp.writelines("AAAABBBBCCCCDDDD")
 276         VnfPackageModel.objects.create(
 277             vnfPackageId="222",
 278             onboardingState="ONBOARDED",
 279             localFilePath="vnfPackage.csar"
 280         )
 281         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content", HTTP_RANGE=" 4-")
 282         partial_file_content = ''
 283         for data in response.streaming_content:
 284             partial_file_content = partial_file_content + data
 285         self.assertEqual(response.status_code, status.HTTP_200_OK)
 286         self.assertEqual('BBBBCCCCDDDD', partial_file_content)
 287         os.remove("vnfPackage.csar")
 288
 289     def test_fetch_vnf_pkg_when_pkg_not_exist(self):
 290         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
 291         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 292
 293     def test_fetch_vnf_pkg_when_catch_cataloge_exception(self):
 294         VnfPackageModel.objects.create(
 295             vnfPackageId="222",
 296             onboardingState="CREATED",
 297             localFilePath="vnfPackage.csar"
 298         )
 299         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
 300         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 301
 302     @mock.patch.object(VnfPackage, "create_vnf_pkg")
 303     def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
 304         mock_create_vnf_pkg.side_effect = TypeError('integer type')
 305         req_data = {
 306             "userDefinedData": {"a": "A"}
 307         }
 308         response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
 309         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 310
 311     @mock.patch.object(VnfPackage, "delete_vnf_pkg")
 312     def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
 313         mock_delete_vnf_pkg.side_effect = TypeError("integer type")
 314         response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
 315         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 316
 317     @mock.patch.object(VnfPackage, "query_single")
 318     def test_query_single_when_catch_exception(self, mock_query_single):
 319         mock_query_single.side_effect = TypeError("integer type")
 320         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
 321         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 322
 323     @mock.patch.object(VnfPackage, "query_multiple")
 324     def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
 325         mock_query_muitiple.side_effect = TypeError("integer type")
 326         response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
 327         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 328
 329     @mock.patch.object(toscaparsers, 'parse_vnfd')
 330     def test_upload_when_catch_exception(self, mock_parse_vnfd):
 331         data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "empty.txt"), "rb")}
 332         VnfPackageModel.objects.create(
 333             vnfPackageId="222",
 334             onboardingState="CREATED"
 335         )
 336         mock_parse_vnfd.side_effect = TypeError("integer type")
 337         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
 338         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 339
 340     @mock.patch.object(VnfPkgUploadThread, 'start')
 341     def test_upload_from_uri_when_catch_exception(self, mock_start):
 342         req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
 343         mock_start.side_effect = TypeError("integer type")
 344         response = self.client.post("/api/vnfpkgm/v1/vnf_packages/111/package_content/upload_from_uri", data=req_data)
 345         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 346
 347     @mock.patch.object(VnfPackage, 'download')
 348     def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
 349         mock_download.side_effect = TypeError("integer type")
 350         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
 351         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 352
 353     @mock.patch.object(toscaparsers, 'parse_vnfd')
 354     def test_fetch_vnf_artifact(self, mock_parse_vnfd):
 355         data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
 356         VnfPackageModel.objects.create(
 357             vnfPackageId="222",
 358             onboardingState="CREATED"
 359         )
 360         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 361         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
 362         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 363         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image")
 364         self.assertEqual(response.status_code, status.HTTP_200_OK)
 365         self.assertEqual(response.getvalue(), "ubuntu_16.04\n")
 366
 367     @mock.patch.object(toscaparsers, 'parse_vnfd')
 368     def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
 369         data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
 370         VnfPackageModel.objects.create(
 371             vnfPackageId="222",
 372             onboardingState="CREATED"
 373         )
 374         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 375         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
 376         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 377         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/1451/artifacts/image")
 378         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 379
 380     @mock.patch.object(toscaparsers, 'parse_vnfd')
 381     def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
 382         data = {'file': open(os.path.join(GENERICPARSER_ROOT_PATH, "resource_test.csar"), "rb")}
 383         VnfPackageModel.objects.create(
 384             vnfPackageId="222",
 385             onboardingState="CREATED"
 386         )
 387         mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
 388         response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
 389         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
 390         response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image1")
 391         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)