# 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 from unittest import mock import ddt 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', 'Additional Artifacts']) 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', 'Additional Artifacts'] 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,additionalArtifacts', "--filter", '(eq,onboardingState,ONBOARDED)'], [('exclude_fields', 'softwareImages,checksum,' 'userDefinedData,additionalArtifacts'), ('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') vnf_pkg.pop('additionalArtifacts') 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,additionalArtifacts', 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', 'additionalArtifacts']), 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', 'Additional Artifacts']) 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.assertCountEqual(_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) @ddt.ddt class TestDownloadVnfPackageArtifact(TestVnfPackage): # The new vnf package created. _vnf_package = vnf_package_fakes.vnf_package_obj( attrs={'userDefinedData': {'Test_key': 'Test_value'}}) def setUp(self): super(TestDownloadVnfPackageArtifact, self).setUp() self.download_vnf_package_artifacts = vnf_package.\ DownloadVnfPackageArtifact( self.app, self.app_args, cmd_name='vnf package artifact download') def test_download_no_options(self): self.assertRaises(base.ParserException, self.check_parser, self.download_vnf_package_artifacts, [], []) def _mock_request_url_for_download_artifacts( self, artifact_path, artifact_data): self.header = {'content-type': 'text/plain'} url = os.path.join(self.url, 'vnfpkgm/v1/vnf_packages', self._vnf_package['id'], 'artifacts', artifact_path) self.requests_mock.register_uri('GET', url, headers=self.header, text=artifact_data) def _get_arglist_and_verifylist(self, localfile): arglist = [ self._vnf_package['id'], localfile.name[1:], '--file', localfile.name ] verifylist = [ ('vnf_package', self._vnf_package['id']), ('artifact_path', localfile.name[1:]), ('file', localfile.name) ] return arglist, verifylist def test_download_artifacts_from_vnf_package(self): test_file = ('./tackerclient/tests//unit/osc/v1/fixture_data/' 'sample_vnf_package_artifacts/Scripts/' 'install.sh') local_file = tempfile.NamedTemporaryFile(suffix='install.sh') artifact_data = open(test_file, 'r').read() arglist, verifylist = self._get_arglist_and_verifylist( local_file) parsed_args = self.check_parser( self.download_vnf_package_artifacts, arglist, verifylist) self._mock_request_url_for_download_artifacts( local_file.name[1:], artifact_data) self.download_vnf_package_artifacts.take_action(parsed_args) self.assertTrue(filecmp.cmp(test_file, local_file.name), "Downloaded contents don't match test file")