python-tackerclient/tackerclient/tests/unit/osc/v1/test_vnf_package.py

# Copyright (C) 2019 NTT DATA
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import filecmp
import os
import shutil
import sys
import tempfile

import ddt
import mock
import zipfile

from tackerclient.common import exceptions
from tackerclient.osc import utils as tacker_osc_utils
from tackerclient.osc.v1.vnfpkgm import vnf_package
from tackerclient.tests.unit.osc import base
from tackerclient.tests.unit.osc.v1.fixture_data import client
from tackerclient.tests.unit.osc.v1 import vnf_package_fakes
from tackerclient.v1_0 import client as proxy_client


def _create_zip():
    vnf_package_path = ('./tackerclient/tests//unit/osc/v1/fixture_data/'
                        'sample_vnf_package')
    tmpdir = tempfile.mkdtemp()
    tmparchive = os.path.join(tmpdir, 'sample_vnf_package')
    zip_file = shutil.make_archive(tmparchive, 'zip', vnf_package_path)
    return zip_file, tmpdir


def _get_columns_vnf_package(action='list', vnf_package_obj=None):
    columns = []
    if action == 'update':
        if vnf_package_obj.get('userDefinedData'):
            columns.extend(['User Defined Data'])
        if vnf_package_obj.get('operationalState'):
            columns.extend(['Operational State'])
        return columns

    columns.extend(['ID', 'Onboarding State', 'Operational State',
                    'Usage State', 'User Defined Data', 'Links'])

    if action in ['show', 'create']:
        if vnf_package_obj and vnf_package_obj[
            'onboardingState'] == 'ONBOARDED':
            columns.extend(['VNFD ID', 'VNF Provider', 'VNF Software Version',
                            'VNFD Version', 'Software Images',
                            'VNF Product Name', 'Checksum'])

    return columns


class TestVnfPackage(base.FixturedTestCase):
    client_fixture_class = client.ClientFixture

    def setUp(self):
        super(TestVnfPackage, self).setUp()
        self.url = client.TACKER_URL
        self.header = {'content-type': 'application/json'}
        self.app = mock.Mock()
        self.app_args = mock.Mock()
        self.client_manager = self.cs
        self.app.client_manager.tackerclient = self.client_manager


@ddt.ddt
class TestCreateVnfPackage(TestVnfPackage):

    def setUp(self):
        super(TestCreateVnfPackage, self).setUp()
        self.create_vnf_package = vnf_package.CreateVnfPackage(
            self.app, self.app_args, cmd_name='vnf package create')

    @ddt.data((["--user-data", 'Test_key=Test_value'],
               [('user_data', {'Test_key': 'Test_value'})]),
              ([], []))
    @ddt.unpack
    def test_take_action(self, arglist, verifylist):
        # command param
        parsed_args = self.check_parser(self.create_vnf_package, arglist,
                                        verifylist)

        if arglist:
            json = vnf_package_fakes.vnf_package_obj(
                attrs={'userDefinedData': {'Test_key': 'Test_value'}})
        else:
            json = vnf_package_fakes.vnf_package_obj()
        self.requests_mock.register_uri(
            'POST', self.url + '/vnfpkgm/v1/vnf_packages',
            json=json, headers=self.header)

        columns, data = (self.create_vnf_package.take_action(parsed_args))
        self.assertCountEqual(_get_columns_vnf_package(), columns)
        headers, attributes = vnf_package._get_columns(json)
        self.assertListItemsEqual(vnf_package_fakes.get_vnf_package_data(
            json, columns=attributes), data)


@ddt.ddt
class TestListVnfPackage(TestVnfPackage):

    def setUp(self):
        super(TestListVnfPackage, self).setUp()
        self.list_vnf_package = vnf_package.ListVnfPackage(
            self.app, self.app_args, cmd_name='vnf package list')
        self._vnf_packages = self._get_vnf_packages()

    def _get_vnf_packages(self, onboarded_vnf_package=False):
        return vnf_package_fakes.create_vnf_packages(
            count=3, onboarded_vnf_package=onboarded_vnf_package)

    def get_list_columns(self, all_fields=False, exclude_fields=None,
                         extra_fields=None, exclude_default=False):

        columns = ['Id', 'Vnf Product Name', 'Onboarding State', 'Usage State',
                   'Operational State', 'Links']
        complex_columns = ['Checksum', 'Software Images', 'User Defined Data']
        simple_columns = ['Vnfd Version', 'Vnf Provider', 'Vnfd Id',
                          'Vnf Software Version']

        if extra_fields:
            columns.extend(extra_fields)

        if exclude_fields:
            columns.extend([field for field in complex_columns
                           if field not in exclude_fields])
        if all_fields:
            columns.extend(complex_columns)
            columns.extend(simple_columns)

        if exclude_default:
            columns.extend(simple_columns)

        return columns

    def _get_mock_response_for_list_vnf_packages(
            self, filter_attribute, json=None):
        self.requests_mock.register_uri(
            'GET', self.url + '/vnfpkgm/v1/vnf_packages?' + filter_attribute,
            json=json if json else self._get_vnf_packages(),
            headers=self.header)

    def test_take_action_default_fields(self):
        parsed_args = self.check_parser(self.list_vnf_package, [], [])
        self.requests_mock.register_uri(
            'GET', self.url + '/vnfpkgm/v1/vnf_packages',
            json=self._vnf_packages, headers=self.header)
        actual_columns, data = self.list_vnf_package.take_action(parsed_args)
        expected_data = []
        headers, columns = tacker_osc_utils.get_column_definitions(
            self.list_vnf_package.get_attributes(), long_listing=True)

        for vnf_package_obj in self._vnf_packages['vnf_packages']:
            expected_data.append(vnf_package_fakes.get_vnf_package_data(
                vnf_package_obj, columns=columns, list_action=True))
        self.assertCountEqual(self.get_list_columns(), actual_columns)
        self.assertListItemsEqual(expected_data, list(data))

    @ddt.data('all_fields', 'exclude_default')
    def test_take_action(self, arg):
        parsed_args = self.check_parser(
            self.list_vnf_package,
            ["--" + arg, "--filter", '(eq,onboardingState,ONBOARDED)'],
            [(arg, True), ('filter', '(eq,onboardingState,ONBOARDED)')])
        vnf_packages = self._get_vnf_packages(onboarded_vnf_package=True)
        self._get_mock_response_for_list_vnf_packages(
            'filter=(eq,onboardingState,ONBOARDED)&' + arg, json=vnf_packages)

        actual_columns, data = self.list_vnf_package.take_action(parsed_args)
        expected_data = []
        kwargs = {arg: True}
        headers, columns = tacker_osc_utils.get_column_definitions(
            self.list_vnf_package.get_attributes(**kwargs), long_listing=True)

        for vnf_package_obj in vnf_packages['vnf_packages']:
            expected_data.append(vnf_package_fakes.get_vnf_package_data(
                vnf_package_obj, columns=columns, list_action=True, **kwargs))

        self.assertCountEqual(self.get_list_columns(**kwargs), actual_columns)
        self.assertListItemsEqual(expected_data, list(data))

    def test_take_action_with_exclude_fields(self):
        parsed_args = self.check_parser(
            self.list_vnf_package,
            ["--exclude_fields", 'softwareImages,checksum,userDefinedData',
             "--filter", '(eq,onboardingState,ONBOARDED)'],
            [('exclude_fields', 'softwareImages,checksum,userDefinedData'),
             ('filter', '(eq,onboardingState,ONBOARDED)')])
        vnf_packages = self._get_vnf_packages(onboarded_vnf_package=True)
        updated_vnf_packages = {'vnf_packages': []}
        for vnf_pkg in vnf_packages['vnf_packages']:
            vnf_pkg.pop('softwareImages')
            vnf_pkg.pop('checksum')
            vnf_pkg.pop('userDefinedData')
            updated_vnf_packages['vnf_packages'].append(vnf_pkg)
        self._get_mock_response_for_list_vnf_packages(
            'filter=(eq,onboardingState,ONBOARDED)&'
            'exclude_fields=softwareImages,checksum,userDefinedData',
            json=updated_vnf_packages)

        actual_columns, data = self.list_vnf_package.take_action(parsed_args)
        expected_data = []
        headers, columns = tacker_osc_utils.get_column_definitions(
            self.list_vnf_package.get_attributes(
                exclude_fields=['softwareImages', 'checksum',
                                'userDefinedData']),
            long_listing=True)

        for vnf_package_obj in updated_vnf_packages['vnf_packages']:
            expected_data.append(vnf_package_fakes.get_vnf_package_data(
                vnf_package_obj, columns=columns, list_action=True))
        expected_columns = self.get_list_columns(
            exclude_fields=['Software Images', 'Checksum',
                            'User Defined Data'])
        self.assertCountEqual(expected_columns, actual_columns)
        self.assertListItemsEqual(expected_data, list(data))

    @ddt.data((['--all_fields', '--fields', 'softwareImages'],
               [('all_fields', True), ('fields', 'softwareImages')]),
              (['--all_fields', '--exclude_fields', 'checksum'],
               [('all_fields', True), ('exclude_fields', 'checksum')]),
              (['--fields', 'softwareImages', '--exclude_fields', 'checksum'],
               [('fields', 'softwareImages'), ('exclude_fields', 'checksum')]))
    @ddt.unpack
    def test_take_action_with_invalid_combination(self, arglist, verifylist):
        self.assertRaises(base.ParserException, self.check_parser,
                          self.list_vnf_package, arglist, verifylist)

    def test_take_action_with_valid_combination(self):
        parsed_args = self.check_parser(
            self.list_vnf_package,
            ["--fields", 'softwareImages,checksum', "--exclude_default"],
            [('fields', 'softwareImages,checksum'), ('exclude_default', True)])
        vnf_packages = self._get_vnf_packages(onboarded_vnf_package=True)
        updated_vnf_packages = {'vnf_packages': []}
        for vnf_pkg in vnf_packages['vnf_packages']:
            vnf_pkg.pop('userDefinedData')
            updated_vnf_packages['vnf_packages'].append(vnf_pkg)

        self._get_mock_response_for_list_vnf_packages(
            'exclude_default&fields=softwareImages,checksum',
            json=updated_vnf_packages)

        actual_columns, data = self.list_vnf_package.take_action(parsed_args)
        expected_data = []
        headers, columns = tacker_osc_utils.get_column_definitions(
            self.list_vnf_package.get_attributes(
                extra_fields=['softwareImages', 'checksum'],
                exclude_default=True),
            long_listing=True)

        for vnf_package_obj in updated_vnf_packages['vnf_packages']:
            expected_data.append(vnf_package_fakes.get_vnf_package_data(
                vnf_package_obj, columns=columns, list_action=True,
                exclude_default=True))

        self.assertCountEqual(self.get_list_columns(
            extra_fields=['Software Images', 'Checksum'],
            exclude_default=True),
            actual_columns)
        self.assertListItemsEqual(expected_data, list(data))


@ddt.ddt
class TestShowVnfPackage(TestVnfPackage):

    def setUp(self):
        super(TestShowVnfPackage, self).setUp()
        self.show_vnf_package = vnf_package.ShowVnfPackage(
            self.app, self.app_args, cmd_name='vnf package show')

    @ddt.data(True, False)
    def test_take_action(self, onboarded):
        vnf_package_obj = vnf_package_fakes.vnf_package_obj(
            onboarded_state=onboarded)
        arglist = [vnf_package_obj['id']]
        verifylist = [('vnf_package', vnf_package_obj['id'])]
        parsed_args = self.check_parser(self.show_vnf_package, arglist,
                                        verifylist)
        url = self.url + '/vnfpkgm/v1/vnf_packages/' + vnf_package_obj['id']
        self.requests_mock.register_uri('GET', url, json=vnf_package_obj,
                                        headers=self.header)
        columns, data = (self.show_vnf_package.take_action(parsed_args))
        self.assertCountEqual(_get_columns_vnf_package(
            vnf_package_obj=vnf_package_obj, action='show'), columns)

        headers, attributes = vnf_package._get_columns(vnf_package_obj)
        self.assertListItemsEqual(
            vnf_package_fakes.get_vnf_package_data(vnf_package_obj,
                                                   columns=attributes), data)

    def test_show_no_options(self):
        self.assertRaises(base.ParserException, self.check_parser,
                          self.show_vnf_package, [], [])


class TestDeleteVnfPackage(TestVnfPackage):

    def setUp(self):
        super(TestDeleteVnfPackage, self).setUp()
        self.delete_vnf_package = vnf_package.DeleteVnfPackage(
            self.app, self.app_args, cmd_name='vnf package delete')

        # The Vnf Package to delete
        self._vnf_package = vnf_package_fakes.create_vnf_packages(count=3)

    def _mock_request_url_for_delete(self, vnf_pkg_index):
        url = (self.url + '/vnfpkgm/v1/vnf_packages/' +
               self._vnf_package['vnf_packages'][vnf_pkg_index]['id'])

        json = self._vnf_package['vnf_packages'][vnf_pkg_index]

        self.requests_mock.register_uri('GET', url, json=json,
                                        headers=self.header)
        self.requests_mock.register_uri('DELETE', url,
                                        headers=self.header, json={})

    def test_delete_one_vnf_package(self):
        arglist = [self._vnf_package['vnf_packages'][0]['id']]
        verifylist = [('vnf-package', [self._vnf_package['vnf_packages']
                                       [0]['id']])]

        parsed_args = self.check_parser(self.delete_vnf_package, arglist,
                                        verifylist)

        self._mock_request_url_for_delete(0)
        result = self.delete_vnf_package.take_action(parsed_args)
        self.assertIsNone(result)

    def test_delete_multiple_vnf_package(self):
        arglist = []
        for vnf_pkg in self._vnf_package['vnf_packages']:
            arglist.append(vnf_pkg['id'])
        verifylist = [('vnf-package', arglist)]
        parsed_args = self.check_parser(self.delete_vnf_package, arglist,
                                        verifylist)
        for i in range(0, 3):
            self._mock_request_url_for_delete(i)

        result = self.delete_vnf_package.take_action(parsed_args)
        self.assertIsNone(result)

    def test_delete_multiple_vnf_package_exception(self):
        arglist = [
            self._vnf_package['vnf_packages'][0]['id'],
            'xxxx-yyyy-zzzz',
            self._vnf_package['vnf_packages'][1]['id'],
        ]
        verifylist = [
            ('vnf-package', arglist),
        ]
        parsed_args = self.check_parser(self.delete_vnf_package,
                                        arglist, verifylist)

        self._mock_request_url_for_delete(0)

        url = (self.url + '/vnfpkgm/v1/vnf_packages/' + 'xxxx-yyyy-zzzz')
        body = {"error": exceptions.NotFound('404')}
        self.requests_mock.register_uri('GET', url, body=body,
                                        status_code=404, headers=self.header)
        self._mock_request_url_for_delete(1)
        self.assertRaises(exceptions.CommandError,
                          self.delete_vnf_package.take_action,
                          parsed_args)


@ddt.ddt
class TestUploadVnfPackage(TestVnfPackage):

    # The new vnf package created.
    _vnf_package = vnf_package_fakes.vnf_package_obj(
        attrs={'userDefinedData': {'Test_key': 'Test_value'}})

    def setUp(self):
        super(TestUploadVnfPackage, self).setUp()
        self.upload_vnf_package = vnf_package.UploadVnfPackage(
            self.app, self.app_args, cmd_name='vnf package upload')

    def test_upload_no_options(self):
        self.assertRaises(base.ParserException, self.check_parser,
                          self.upload_vnf_package, [], [])

    def _mock_request_url_for_upload(self, method, status_code=202, body={}):
        if method == 'PUT':
            self.header = {'content-type': 'application/zip'}
            url = (self.url + '/vnfpkgm/v1/vnf_packages/' +
                   self._vnf_package['id'] + '/package_content')
        else:
            url = (self.url + '/vnfpkgm/v1/vnf_packages/' +
                   self._vnf_package['id'] + '/package_content/'
                                             'upload_from_uri')

        self.requests_mock.register_uri(method, url, json=body,
                                        headers=self.header,
                                        status_code=status_code)

    def _get_arglist_and_verifylist(self, method, path):
        if method == 'path':
            arglist = [
                self._vnf_package['id'],
                "--path", path
            ]
            verifylist = [
                ('path', path),
                ('vnf_package', self._vnf_package['id'])
            ]
        else:
            arglist = [
                self._vnf_package['id'],
                "--url", "http://uri:8000/test.zip",
                "--user-name", "Test_username",
                "--password", "Test_passoword",
            ]
            verifylist = [
                ('url', "http://uri:8000/test.zip"),
                ('user_name', 'Test_username'),
                ('password', 'Test_passoword'),
                ('vnf_package', self._vnf_package['id'])
            ]
        return arglist, verifylist

    @ddt.data('path', 'url')
    def test_upload_vnf_package(self, method):
        path = None
        if method == 'path':
            zip_file, temp_dir = _create_zip()
            path = zip_file

        arglist, verifylist = self._get_arglist_and_verifylist(method,
                                                               path)
        parsed_args = self.check_parser(self.upload_vnf_package, arglist,
                                        verifylist)
        with mock.patch.object(proxy_client.ClientBase,
                               '_handle_fault_response') as m:
            if method == 'url':
                self._mock_request_url_for_upload('POST')
                self.upload_vnf_package.take_action(parsed_args)
            else:
                self._mock_request_url_for_upload('PUT')
                self.upload_vnf_package.take_action(parsed_args)
                # Delete temporary folder
                shutil.rmtree(temp_dir)
        # check no fault response is received
        self.assertNotCalled(m)

    def test_upload_vnf_package_with_conflict_error(self):
        # Scenario in which vnf package is already in on-boarded state
        zip_file, temp_dir = _create_zip()
        arglist, verifylist = self._get_arglist_and_verifylist('path',
                                                               zip_file)

        parsed_args = self.check_parser(self.upload_vnf_package, arglist,
                                        verifylist)

        body = {"conflictingRequest": {
            "message": "VNF Package " + self._vnf_package['id'] +
                       " onboarding state is not CREATED", "code": 409}}
        self._mock_request_url_for_upload('PUT', status_code=409, body=body)

        self.assertRaises(exceptions.TackerClientException,
                          self.upload_vnf_package.take_action, parsed_args)
        # Delete temporary folder
        shutil.rmtree(temp_dir)

    def test_upload_vnf_package_failed_with_404_not_found(self):
        # Scenario in which vnf package is not found
        zip_file, temp_dir = _create_zip()
        arglist = [
            'dumy-id',
            "--path", zip_file
        ]
        verifylist = [
            ('path', zip_file),
            ('vnf_package', 'dumy-id')
        ]

        parsed_args = self.check_parser(self.upload_vnf_package, arglist,
                                        verifylist)

        error_message = "Can not find requested vnf package: dummy-id"
        body = {"itemNotFound": {"message": error_message, "code": 404}}
        url = self.url + '/vnfpkgm/v1/vnf_packages/dumy-id/package_content'

        self.requests_mock.register_uri(
            'PUT', url, json=body,
            status_code=404)

        exception = self.assertRaises(
            exceptions.TackerClientException,
            self.upload_vnf_package.take_action, parsed_args)

        self.assertEqual(error_message, exception.message)
        # Delete temporary folder
        shutil.rmtree(temp_dir)


@ddt.ddt
class TestUpdateVnfPackage(TestVnfPackage):

    def setUp(self):
        super(TestUpdateVnfPackage, self).setUp()
        self.update_vnf_package = vnf_package.UpdateVnfPackage(
            self.app, self.app_args, cmd_name='vnf package update')

    @ddt.data((["--user-data", 'Test_key=Test_value',
                "--operational-state", 'DISABLED'],
               [('user_data', {'Test_key': 'Test_value'}),
                ('operational_state', 'DISABLED')]),
              (["--user-data", 'Test_key=Test_value'],
               [('user_data', {'Test_key': 'Test_value'})]),
              (["--operational-state", 'DISABLED'],
               [('operational_state', 'DISABLED')]))
    @ddt.unpack
    def test_take_action(self, arglist, verifylist):
        vnf_package_obj = vnf_package_fakes.vnf_package_obj(
            onboarded_state=True)
        arglist.append(vnf_package_obj['id'])
        verifylist.append(('vnf_package', vnf_package_obj['id']))

        parsed_args = self.check_parser(self.update_vnf_package, arglist,
                                        verifylist)
        url = os.path.join(self.url, 'vnfpkgm/v1/vnf_packages',
                           vnf_package_obj['id'])
        fake_response = vnf_package_fakes.get_fake_update_vnf_package_obj(
            arglist)
        self.requests_mock.register_uri('PATCH', url,
                                        json=fake_response,
                                        headers=self.header)

        columns, data = self.update_vnf_package.take_action(parsed_args)
        self.assertItemsEqual(_get_columns_vnf_package(
            vnf_package_obj=fake_response, action='update'), columns)
        self.assertListItemsEqual(
            vnf_package_fakes.get_vnf_package_data(fake_response), data)

    def test_update_no_options(self):
        self.assertRaises(base.ParserException, self.check_parser,
                          self.update_vnf_package, [], [])

    def test_update_without_user_data_and_operational_state(self):
        vnf_package_obj = vnf_package_fakes.vnf_package_obj(
            onboarded_state=True)
        arglist = [vnf_package_obj['id']]

        verifylist = [('vnf_package', vnf_package_obj['id'])]

        parsed_args = self.check_parser(self.update_vnf_package, arglist,
                                        verifylist)
        self.assertRaises(SystemExit, self.update_vnf_package.take_action,
                          parsed_args)


@ddt.ddt
class TestDownloadVnfPackage(TestVnfPackage):

    # The new vnf package created.
    _vnf_package = vnf_package_fakes.vnf_package_obj(
        attrs={'userDefinedData': {'Test_key': 'Test_value'}})

    def setUp(self):
        super(TestDownloadVnfPackage, self).setUp()
        self.download_vnf_package = vnf_package.DownloadVnfPackage(
            self.app, self.app_args, cmd_name='vnf package download')

    def test_download_no_options(self):
        self.assertRaises(base.ParserException, self.check_parser,
                          self.download_vnf_package, [], [])

    def _mock_request_url_for_download_vnfd(self, content_type, vnfd_data):
        self.header = {'content-type': content_type}
        url = os.path.join(self.url, 'vnfpkgm/v1/vnf_packages',
                           self._vnf_package['id'], 'vnfd')

        if content_type == 'text/plain':
            self.requests_mock.register_uri('GET', url,
                                            headers=self.header,
                                            text=vnfd_data)
        else:
            self.requests_mock.register_uri('GET', url,
                                            headers=self.header,
                                            content=vnfd_data)

    def _get_arglist_and_verifylist(self, accept_type, file_name):
        arglist = [
            self._vnf_package['id'],
            '--vnfd',
            '--type', accept_type,
            '--file', file_name
        ]
        verifylist = [
            ('type', accept_type),
            ('vnfd', True),
            ('vnf_package', self._vnf_package['id']),
            ('file', file_name)
        ]
        return arglist, verifylist

    def test_download_vnfd_from_vnf_package_for_type_text_plain(self):
        test_file = ('./tackerclient/tests//unit/osc/v1/fixture_data/'
                     'sample_vnf_package/Definitions/'
                     'etsi_nfv_sol001_common_types.yaml')

        local_file = tempfile.NamedTemporaryFile(suffix='vnfd_data.yaml')
        vnfd_data = open(test_file, 'r').read()
        arglist, verifylist = self._get_arglist_and_verifylist(
            'text/plain', local_file.name)
        parsed_args = self.check_parser(self.download_vnf_package, arglist,
                                        verifylist)
        self._mock_request_url_for_download_vnfd('text/plain', vnfd_data)
        self.download_vnf_package.take_action(parsed_args)
        self.assertTrue(filecmp.cmp(test_file, local_file.name),
                        "Downloaded contents don't match test file")

    @ddt.data('application/zip', 'both')
    def test_download_vnfd_from_vnf_package(self, accept_type):
        test_file, temp_dir = _create_zip()
        # File in which VNFD data will be stored.
        # For testing purpose we are creating temporary zip file.
        local_file = tempfile.NamedTemporaryFile(suffix='vnfd_data.zip')
        vnfd_data = open(test_file, 'rb').read()
        arglist, verifylist = self._get_arglist_and_verifylist(
            accept_type, local_file.name)
        parsed_args = self.check_parser(self.download_vnf_package, arglist,
                                        verifylist)
        # When --type argument is set to 'both', then 'Accept' header in
        # request is set to 'text/plain,application/zip' now it is up to the
        # NFVO to choose the format to return for a single-file VNFD and for
        # a multi-file VNFD, a ZIP file shall be returned. Here we have taken
        # the example of multi-file vnfd hence its retuning zip file and
        # setting the 'Content-Type' as 'application/zip' in response header.
        self._mock_request_url_for_download_vnfd('application/zip', vnfd_data)
        self.download_vnf_package.take_action(parsed_args)
        self.assertTrue(filecmp.cmp(test_file, local_file.name),
                        "Downloaded contents don't match test file")
        self.assertTrue(self._check_valid_zip_file(local_file.name))
        shutil.rmtree(temp_dir)

    def _check_valid_zip_file(self, zip_file):
        with zipfile.ZipFile(zip_file) as zf:
            ret = zf.testzip()
        return False if ret else True

    @mock.patch('builtins.print')
    def test_download_vnfd_from_vnf_package_without_file_arg(self, mock_print):
        # --file argument is optional when --type is set to 'text/plain'.
        arglist = [
            self._vnf_package['id'],
            '--vnfd',
            '--type', 'text/plain',
        ]
        verifylist = [
            ('type', 'text/plain'),
            ('vnfd', True),
            ('vnf_package', self._vnf_package['id']),
        ]
        parsed_args = self.check_parser(self.download_vnf_package, arglist,
                                        verifylist)
        test_file = ('./tackerclient/tests//unit/osc/v1/fixture_data/'
                     'sample_vnf_package/Definitions/'
                     'etsi_nfv_sol001_common_types.yaml')

        vnfd_data = open(test_file, 'r').read()
        self._mock_request_url_for_download_vnfd('text/plain', vnfd_data)
        self.download_vnf_package.take_action(parsed_args)
        mock_print.assert_called_once_with(vnfd_data)

    @ddt.data('application/zip', 'both')
    def test_download_vnfd_from_vnf_package_failed_with_no_file_arg(
            self, accept_type):
        arglist = [
            self._vnf_package['id'],
            '--vnfd',
            '--type', accept_type,
        ]
        verifylist = [
            ('type', accept_type),
            ('vnfd', True),
            ('vnf_package', self._vnf_package['id']),
        ]
        parsed_args = self.check_parser(self.download_vnf_package, arglist,
                                        verifylist)
        with mock.patch.object(sys.stdout, "isatty") as mock_isatty:
            mock_isatty.return_value = True
            self.assertRaises(SystemExit,
                              self.download_vnf_package.take_action,
                              parsed_args)

    def test_download_vnf_package(self):
        file_name = 'vnf_package_data.zip'
        test_file, temp_dir = _create_zip()

        # file in which VNF Package data will be stored.
        # for testing purpose we are creating temporary zip file.
        local_file = tempfile.NamedTemporaryFile(suffix=file_name)
        vnf_package_data = open(test_file, 'rb').read()

        arglist = [
            self._vnf_package['id'],
            '--file', local_file.name
        ]
        verifylist = [
            ('vnf_package', self._vnf_package['id']),
            ('file', local_file.name)
        ]

        parsed_args = self.check_parser(self.download_vnf_package, arglist,
                                        verifylist)

        url = os.path.join(self.url, '/vnfpkgm/v1/vnf_packages',
                           self._vnf_package['id'], 'package_content')

        self.requests_mock.register_uri(
            'GET', url, headers={'content-type': 'application/zip'},
            content=vnf_package_data)

        self.download_vnf_package.take_action(parsed_args)
        self.assertTrue(filecmp.cmp(test_file, local_file.name),
                        "Downloaded contents don't match test file")
        self.assertTrue(self._check_valid_zip_file(local_file.name))
        shutil.rmtree(temp_dir)