Args = collections.namedtuple('Args',
['source', 'entry', 'manifest', 'history', 'tests',
- 'licenses', 'digest', 'certificate', 'privkey'])
+ 'licenses', 'digest', 'certificate', 'privkey', 'sol241'])
ARGS_MANIFEST = {
'digest': None,
'certificate': None,
'privkey': None,
+ 'sol241': False,
}
ARGS_MANIFEST_DIGEST = {
'digest': 'sha-256',
'certificate': None,
'privkey': None,
+ 'sol241': False,
}
ARGS_MANIFEST_DIGEST_CERT = {
'licenses': 'Licenses',
'digest': 'sha-256',
'certificate': 'test.crt',
- 'privkey': os.path.join(ROOT_DIR, 'tests', 'resources', 'signature', 'test.key')
+ 'privkey': os.path.join(ROOT_DIR, 'tests', 'resources', 'signature', 'test.key'),
+ 'sol241': False,
}
ARGS_NO_MANIFEST = {
'digest': None,
'certificate': None,
'privkey': None,
+ 'sol241': True,
}
INVALID_ARGS_NO_MANIFEST = {
'digest': 'sha-256',
'certificate': None,
'privkey': None,
+ 'sol241': True,
}
INVALID_ARGS_NO_PRIVKEY = {
'digest': None,
'certificate': 'test.crt',
'privkey': None,
+ 'sol241': True,
}
--- /dev/null
+#
+# Copyright (c) 2017 GigaSpaces Technologies Ltd. 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 copy
+import os
+import pytest
+import shutil
+import tempfile
+
+from vnfsdk_pkgtools.packager import toscameta
+from vnfsdk_pkgtools import util
+
+
+ROOT_DIR = util.get_project_root()
+
+CSAR_RESOURCE_DIR = os.path.join(ROOT_DIR, 'tests', 'resources', 'csar')
+CSAR_ENTRY_FILE = 'test_entry.yaml'
+CSAR_OUTPUT_FILE = 'output.csar'
+
+ARGS_MANIFEST = {
+ 'base_dir': CSAR_RESOURCE_DIR,
+ 'entry': CSAR_ENTRY_FILE,
+ 'manifest': 'test_entry.mf',
+ 'changelog': 'ChangeLog.txt',
+ 'licenses': 'Licenses',
+ 'tests': 'Tests',
+ 'certificate': None,
+ }
+
+ARGS_MANIFEST_CERTIFICATE = {
+ 'base_dir': CSAR_RESOURCE_DIR,
+ 'entry': CSAR_ENTRY_FILE,
+ 'manifest': 'test_entry.mf',
+ 'changelog': 'ChangeLog.txt',
+ 'licenses': 'Licenses',
+ 'tests': 'Tests',
+ 'certificate': 'test.crt',
+ }
+
+ARGS_NO_MANIFEST = {
+ 'base_dir': CSAR_RESOURCE_DIR,
+ 'entry': CSAR_ENTRY_FILE,
+ 'manifest': None,
+ 'changelog': None,
+ 'licenses': None,
+ 'tests': None,
+ 'certificate': None,
+ }
+
+
+def _validate_metadata(cls, expected):
+ metadata = cls(**expected)
+ assert toscameta.META_CREATED_BY_VALUE == metadata.created_by
+ assert toscameta.META_CSAR_VERSION_VALUE == metadata.csar_version
+ assert toscameta.META_FILE_VERSION_VALUE == metadata.meta_file_version
+ assert expected['entry'] == metadata.entry_definitions
+ assert expected['manifest'] == metadata.entry_manifest_file
+ assert expected['changelog'] == metadata.entry_history_file
+ assert expected['licenses'] == metadata.entry_licenses_dir
+ assert expected['tests'] == metadata.entry_tests_dir
+ assert expected['certificate'] == metadata.entry_certificate_file
+ return metadata
+
+
+def test_261():
+ metadata = _validate_metadata(toscameta.ToscaMeta261, ARGS_MANIFEST)
+ assert "ETSI-Entry-Change-Log: ChangeLog.txt\n" in metadata.dump_as_string()
+
+
+def test_241():
+ metadata = _validate_metadata(toscameta.ToscaMeta241, ARGS_MANIFEST_CERTIFICATE)
+ assert "Entry-Certificate: test.crt\n" in metadata.dump_as_string()
+
+
+def test_261_no_manifest():
+ with pytest.raises(ValueError):
+ toscameta.ToscaMeta261(**ARGS_NO_MANIFEST)
+
+
+def test_241_no_manifest():
+ metadata = _validate_metadata(toscameta.ToscaMeta241, ARGS_NO_MANIFEST)
+ assert "Entry-Definitions: test_entry.yaml\n" in metadata.dump_as_string()
+
+
+def test_invalid_entry():
+ args = copy.copy(ARGS_MANIFEST)
+ args['entry'] = 'test_entry.mf'
+ with pytest.raises(ValueError):
+ toscameta.ToscaMeta261(**args)
+
+
+def test_invalid_csar_version():
+ args = copy.copy(ARGS_MANIFEST)
+ args['meta_csar_version'] = '1.2'
+ with pytest.raises(ValueError):
+ toscameta.ToscaMeta241(**args)
+
+
+FROM_FILE_CASES = ['TOSCA.meta.sol261', 'TOSCA.meta.sol241']
+
+def _prepare(target, metafile_path):
+ shutil.copytree(CSAR_RESOURCE_DIR, target)
+ os.mkdir(os.path.join(target, 'TOSCA-Metadata'))
+ shutil.copy(metafile_path, os.path.join(target,
+ 'TOSCA-Metadata',
+ 'TOSCA.meta'))
+
+def test_create_from_file():
+ for case in FROM_FILE_CASES:
+ target = tempfile.mkdtemp()
+ base_dir = os.path.join(target, 'mytest')
+ _prepare(base_dir,
+ os.path.join(ROOT_DIR,
+ 'tests',
+ 'resources',
+ case))
+ try:
+ toscameta.create_from_file(base_dir)
+ finally:
+ shutil.rmtree(target, ignore_errors=True)
with pytest.raises(subprocess.CalledProcessError):
utils.verify(str(p), CERT_FILE, cms, no_verify_cert=True)
-
+
+
+CHECK_FILE_CASES = [
+ {
+ 'negative': False,
+ 'params': {'root': RESOURCES_DIR,
+ 'entry': 'test.key',
+ 'msg': '',
+ 'check_for_non': False,
+ 'check_dir': False,
+ }
+ },
+ {
+ 'negative': False,
+ 'params': {'root': RESOURCES_DIR,
+ 'entry': 'non-existing-file',
+ 'msg': '',
+ 'check_for_non': True,
+ 'check_dir': False,
+ }
+ },
+ {
+ 'negative': True,
+ 'params': {'root': RESOURCES_DIR,
+ 'entry': 'non-existing-file',
+ 'msg': '',
+ 'check_for_non': False,
+ 'check_dir': False,
+ }
+ },
+ {
+ 'negative': False,
+ 'params': {'root': ROOT_DIR,
+ 'entry': 'tests',
+ 'msg': '',
+ 'check_for_non': False,
+ 'check_dir': True,
+ }
+ },
+ {
+ 'negative': False,
+ 'params': {'root': ROOT_DIR,
+ 'entry': 'non-existing-dir',
+ 'msg': '',
+ 'check_for_non': True,
+ 'check_dir': True,
+ }
+ },
+ {
+ 'negative': True,
+ 'params': {'root': ROOT_DIR,
+ 'entry': 'non-existing-dir',
+ 'msg': '',
+ 'check_for_non': False,
+ 'check_dir': True,
+ }
+ },
+ ]
+
+
+def test_check_file_dir():
+ for case in CHECK_FILE_CASES:
+ if case['negative']:
+ with pytest.raises(ValueError):
+ utils.check_file_dir(**case['params'])
+ else:
+ utils.check_file_dir(**case['params'])
--- /dev/null
+TOSCA-Meta-File-Version: 1.0
+CSAR-Version: 1.1
+Created-By: ONAP VNFSDK pkgtools
+Entry-Definitions: test_entry.yaml
+Entry-Manifest: test_entry.mf
+Entry-Change-Log: ChangeLog.txt
+Entry-Licenses: Licenses
+Entry-Tests: Tests
+Entry-Certificate: test.crt
--- /dev/null
+TOSCA-Meta-File-Version: 1.0
+CSAR-Version: 1.1
+Created-By: ONAP VNFSDK pkgtools
+Entry-Definitions: test_entry.yaml
+ETSI-Entry-Manifest: test_entry.mf
+ETSI-Entry-Change-Log: ChangeLog.txt
+ETSI-Entry-Licenses: Licenses
+ETSI-Entry-Tests: Tests
+ETSI-Entry-Certificate: test.crt
required=True)
csar_create.add_argument(
'--manifest',
- help='Manifest file relative to service template directory')
+ help='Manifest file relative to service template directory',
+ required=True)
csar_create.add_argument(
'--history',
- help='Change history file relative to service template directory')
+ help='Change history file relative to service template directory',
+ required=True)
csar_create.add_argument(
'--tests',
help='Directory containing test information, relative to service template directory')
csar_create.add_argument(
'--licenses',
- help='Directory containing license information, relative to service template directory')
+ help='Directory containing license information, relative to service template directory',
+ required=True)
csar_create.add_argument(
'--digest',
choices=manifest.SUPPORTED_HASH_ALGO,
csar_create.add_argument(
'--privkey',
help='Private key file for certification, absoluate or relative path')
+ csar_create.add_argument(
+ '--sol241',
+ action='store_true',
+ help='Generate SOL004 v2.4.1 csar for backward compatilibity')
csar_open = subparsers.add_parser('csar-open')
import zipfile
import requests
-from ruamel import yaml # @UnresolvedImport
from vnfsdk_pkgtools.packager import manifest
+from vnfsdk_pkgtools.packager import toscameta
from vnfsdk_pkgtools.packager import utils
LOG = logging.getLogger(__name__)
-META_FILE = 'TOSCA-Metadata/TOSCA.meta'
-META_FILE_VERSION_KEY = 'TOSCA-Meta-File-Version'
-META_FILE_VERSION_VALUE = '1.0'
-META_CSAR_VERSION_KEY = 'CSAR-Version'
-META_CSAR_VERSION_VALUE = '1.1'
-META_CREATED_BY_KEY = 'Created-By'
-META_CREATED_BY_VALUE = 'ONAP'
-META_ENTRY_DEFINITIONS_KEY = 'Entry-Definitions'
-META_ENTRY_MANIFEST_FILE_KEY = 'Entry-Manifest'
-META_ENTRY_HISTORY_FILE_KEY = 'Entry-Change-Log'
-META_ENTRY_TESTS_DIR_KEY = 'Entry-Tests'
-META_ENTRY_LICENSES_DIR_KEY = 'Entry-Licenses'
-META_ENTRY_CERT_FILE_KEY = 'Entry-Certificate'
-
-BASE_METADATA = {
- META_FILE_VERSION_KEY: META_FILE_VERSION_VALUE,
- META_CSAR_VERSION_KEY: META_CSAR_VERSION_VALUE,
- META_CREATED_BY_KEY: META_CREATED_BY_VALUE,
-}
-
-
-def check_file_dir(root, entry, msg, check_for_non=False, check_dir=False):
- path = os.path.join(root, entry)
- if check_for_non:
- ret = not os.path.exists(path)
- error_msg = '{0} already exists. ' + msg
- elif check_dir:
- ret = os.path.isdir(path)
- error_msg = '{0} is not an existing directory. ' + msg
- else:
- ret = os.path.isfile(path)
- error_msg = '{0} is not an existing file. ' + msg
- if not ret:
- raise ValueError(error_msg.format(path))
-
def write(source, entry, destination, args):
source = os.path.expanduser(source)
destination = os.path.expanduser(destination)
- metadata = BASE_METADATA.copy()
-
- check_file_dir(root=source,
- entry='',
- msg='Please specify the service template directory.',
- check_dir=True)
-
- check_file_dir(root=source,
- entry=entry,
- msg='Please specify a valid entry point.',
- check_dir=False)
- metadata[META_ENTRY_DEFINITIONS_KEY] = entry
-
- check_file_dir(root='',
- entry=destination,
- msg='Please provide a path to where the CSAR should be created.',
- check_for_non=True)
-
- check_file_dir(root=source,
- entry=META_FILE,
- msg='This commands generates a meta file for you. Please '
- 'remove the existing metafile.',
- check_for_non=True)
-
- if(args.manifest):
- check_file_dir(root=source,
- entry=args.manifest,
- msg='Please specify a valid manifest file.',
- check_dir=False)
- metadata[META_ENTRY_MANIFEST_FILE_KEY] = args.manifest
- manifest_file = manifest.Manifest(source, args.manifest)
+
+ utils.check_file_dir(root=source,
+ entry='',
+ msg='Please specify the service template directory.',
+ check_dir=True)
+
+ utils.check_file_dir(root='',
+ entry=destination,
+ msg='Please provide a path to where the CSAR should be created.',
+ check_for_non=True)
+
+ utils.check_file_dir(root=source,
+ entry=toscameta.META_FILE,
+ msg='This commands generates a meta file for you.'
+ 'Please remove the existing metafile.',
+ check_for_non=True)
+ if args.sol241:
+ metadatacls = toscameta.ToscaMeta241
+ else:
+ metadatacls = toscameta.ToscaMeta261
+ metadata = metadatacls(source, args.entry, args.manifest,
+ args.history, args.licenses,
+ args.tests, args.certificate)
+
+ if args.manifest:
+ manifest_file = manifest.Manifest(source, args.manifest)
manifest_file_full_path = os.path.join(source, args.manifest)
elif args.certificate or args.digest:
raise ValueError("Must specify manifest file if certificate or digest is specified")
manifest_file = None
manifest_file_full_path = None
-
- if(args.history):
- check_file_dir(root=source,
- entry=args.history,
- msg='Please specify a valid change history file.',
- check_dir=False)
- metadata[META_ENTRY_HISTORY_FILE_KEY] = args.history
-
if args.certificate:
- check_file_dir(root=source,
- entry=args.certificate,
- msg='Please specify a valid certificate file.',
- check_dir=False)
- metadata[META_ENTRY_CERT_FILE_KEY] = args.certificate
if not args.privkey:
raise ValueError('Need private key file for signing')
- check_file_dir(root='',
- entry=args.privkey,
- msg='Please specify a valid private key file.',
- check_dir=False)
-
- if(args.tests):
- check_file_dir(root=source,
- entry=args.tests,
- msg='Please specify a valid test directory.',
- check_dir=True)
- metadata[META_ENTRY_TESTS_DIR_KEY] = args.tests
-
- if(args.licenses):
- check_file_dir(root=source,
- entry=args.licenses,
- msg='Please specify a valid license directory.',
- check_dir=True)
- metadata[META_ENTRY_LICENSES_DIR_KEY] = args.licenses
+ utils.check_file_dir(root='',
+ entry=args.privkey,
+ msg='Please specify a valid private key file.',
+ check_dir=False)
LOG.debug('Compressing root directory to ZIP')
with zipfile.ZipFile(destination, 'w', zipfile.ZIP_DEFLATED) as f:
for file in files:
file_full_path = os.path.join(root, file)
# skip manifest file here in case we need to generate digest
- if file_full_path!=manifest_file_full_path:
+ if file_full_path != manifest_file_full_path:
file_relative_path = os.path.relpath(file_full_path, source)
LOG.debug('Writing to archive: {0}'.format(file_relative_path))
f.write(file_full_path, file_relative_path)
if manifest_file and args.digest:
LOG.debug('Update file digest: {0}'.format(file_relative_path))
manifest_file.add_file(file_relative_path, args.digest)
-
if manifest_file:
LOG.debug('Update manifest file to temporary file')
manifest_file_full_path = manifest_file.update_to_file(True)
LOG.debug('Writing to archive: {0}'.format(args.manifest))
f.write(manifest_file_full_path, args.manifest)
- LOG.debug('Writing new metadata file to {0}'.format(META_FILE))
- f.writestr(META_FILE, yaml.dump(metadata, default_flow_style=False))
+ LOG.debug('Writing new metadata file to {0}'.format(toscameta.META_FILE))
+ f.writestr(toscameta.META_FILE, metadata.dump_as_string())
class _CSARReader(object):
source = download_target
self.source = os.path.expanduser(source)
self.destination = os.path.expanduser(destination)
- self.metadata = {}
+ self.metadata = None
self.manifest = None
try:
if not os.path.exists(self.source):
@property
def created_by(self):
- return self.metadata.get(META_CREATED_BY_KEY)
+ return self.metadata.created_by
@property
def csar_version(self):
- return self.metadata.get(META_CSAR_VERSION_KEY)
+ return self.metadata.csar_version
@property
def meta_file_version(self):
- return self.metadata.get(META_FILE_VERSION_KEY)
+ return self.metadata.meta_file_version
@property
def entry_definitions(self):
- return self.metadata.get(META_ENTRY_DEFINITIONS_KEY)
+ return self.metadata.entry_definitions
@property
def entry_definitions_yaml(self):
@property
def entry_manifest_file(self):
- return self.metadata.get(META_ENTRY_MANIFEST_FILE_KEY)
+ return self.metadata.entry_manifest_file
@property
def entry_history_file(self):
- return self.metadata.get(META_ENTRY_HISTORY_FILE_KEY)
+ return self.metadata.entry_history_file
@property
def entry_tests_dir(self):
- return self.metadata.get(META_ENTRY_TESTS_DIR_KEY)
+ return self.metadata.entry_tests_dir
@property
def entry_licenses_dir(self):
- return self.metadata.get(META_ENTRY_LICENSES_DIR_KEY)
+ return self.metadata.entry_licenses_dir
@property
def entry_certificate_file(self):
- return self.metadata.get(META_ENTRY_CERT_FILE_KEY)
+ return self.metadata.entry_certificate_file
def _extract(self):
LOG.debug('Extracting CSAR contents')
LOG.debug('CSAR contents successfully extracted')
def _read_metadata(self):
- csar_metafile = os.path.join(self.destination, META_FILE)
- if not os.path.exists(csar_metafile):
- raise ValueError('Metadata file {0} is missing from the CSAR'.format(csar_metafile))
- LOG.debug('CSAR metadata file: {0}'.format(csar_metafile))
- LOG.debug('Attempting to parse CSAR metadata YAML')
- with open(csar_metafile) as f:
- self.metadata.update(yaml.safe_load(f))
- LOG.debug('CSAR metadata:\n{0}'.format(pprint.pformat(self.metadata)))
+ self.metadata = toscameta.create_from_file(self.destination)
def _validate(self, no_verify_cert):
- def validate_key(key, expected=None):
- if not self.metadata.get(key):
- raise ValueError('{0} is missing from the metadata file.'.format(key))
- actual = str(self.metadata[key])
- if expected and actual != expected:
- raise ValueError('{0} is expected to be {1} in the metadata file while it is in '
- 'fact {2}.'.format(key, expected, actual))
- validate_key(META_FILE_VERSION_KEY, expected=META_FILE_VERSION_VALUE)
- validate_key(META_CSAR_VERSION_KEY, expected=META_CSAR_VERSION_VALUE)
- validate_key(META_CREATED_BY_KEY)
- validate_key(META_ENTRY_DEFINITIONS_KEY)
LOG.debug('CSAR entry definitions: {0}'.format(self.entry_definitions))
LOG.debug('CSAR manifest file: {0}'.format(self.entry_manifest_file))
LOG.debug('CSAR change history file: {0}'.format(self.entry_history_file))
LOG.debug('CSAR licenses directory: {0}'.format(self.entry_licenses_dir))
LOG.debug('CSAR certificate file: {0}'.format(self.entry_certificate_file))
- check_file_dir(self.destination,
- self.entry_definitions,
- 'The entry definitions {0} referenced by the metadata '
- 'file does not exist.'.format(self.entry_definitions),
- check_dir=False)
-
- if(self.entry_manifest_file):
- check_file_dir(self.destination,
- self.entry_manifest_file,
- 'The manifest file {0} referenced by the metadata '
- 'file does not exist.'.format(self.entry_manifest_file),
- check_dir=False)
- self.manifest = manifest.Manifest(self.destination,
- self.entry_manifest_file)
-
-
- if(self.entry_history_file):
- check_file_dir(self.destination,
- self.entry_history_file,
- 'The change history file {0} referenced by the metadata '
- 'file does not exist.'.format(self.entry_history_file),
- check_dir=False)
-
- if(self.entry_tests_dir):
- check_file_dir(self.destination,
- self.entry_tests_dir,
- 'The test directory {0} referenced by the metadata '
- 'file does not exist.'.format(self.entry_tests_dir),
- check_dir=True)
-
- if(self.entry_licenses_dir):
- check_file_dir(self.destination,
- self.entry_licenses_dir,
- 'The license directory {0} referenced by the metadata '
- 'file does not exist.'.format(self.entry_licenses_dir),
- check_dir=True)
+ if self.entry_manifest_file:
+ self.manifest = manifest.Manifest(self.destination,
+ self.entry_manifest_file)
if(self.entry_certificate_file):
- # check certificate
- check_file_dir(self.destination,
- self.entry_certificate_file,
- 'The certificate file {0} referenced by the metadata '
- 'file does not exist.'.format(self.entry_certificate_file),
- check_dir=False)
tmp_manifest = self.manifest.save_to_temp_without_cms()
utils.verify(tmp_manifest,
os.path.join(self.destination, self.entry_certificate_file),
--- /dev/null
+# Copyright (c) 2019 Intel Corp. All rights reserved.
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 logging
+import os
+import pprint
+
+from ruamel import yaml # @UnresolvedImport
+import six
+
+from vnfsdk_pkgtools.packager import utils
+
+LOG = logging.getLogger(__name__)
+
+META_FILE = 'TOSCA-Metadata/TOSCA.meta'
+
+META_FILE_VERSION_KEY = 'TOSCA-Meta-File-Version'
+META_FILE_VERSION_VALUE = '1.0'
+META_CSAR_VERSION_KEY = 'CSAR-Version'
+META_CSAR_VERSION_VALUE = '1.1'
+META_CREATED_BY_KEY = 'Created-By'
+META_CREATED_BY_VALUE = 'ONAP VNFSDK pkgtools'
+
+META_ENTRY_DEFINITIONS_KEY = 'Entry-Definitions'
+
+BASE_META = {
+ META_FILE_VERSION_KEY: META_FILE_VERSION_VALUE,
+ META_CSAR_VERSION_KEY: META_CSAR_VERSION_VALUE,
+}
+
+
+class ToscaMeta(object):
+ META_ENTRY_MANIFEST_FILE_KEY = 'ETSI-Entry-Manifest'
+ META_ENTRY_HISTORY_FILE_KEY = 'ETSI-Entry-Change-Log'
+ META_ENTRY_TESTS_DIR_KEY = 'ETSI-Entry-Tests'
+ META_ENTRY_LICENSES_DIR_KEY = 'ETSI-Entry-Licenses'
+ META_ENTRY_CERT_FILE_KEY = 'ETSI-Entry-Certificate'
+ REQUIRED_KEYS = [ META_FILE_VERSION_KEY, META_CSAR_VERSION_KEY,
+ META_CREATED_BY_KEY, META_ENTRY_DEFINITIONS_KEY,
+ META_ENTRY_MANIFEST_FILE_KEY, META_ENTRY_HISTORY_FILE_KEY,
+ META_ENTRY_LICENSES_DIR_KEY,
+ ]
+ OPTIONAL_KEYS = [META_ENTRY_TESTS_DIR_KEY, META_ENTRY_CERT_FILE_KEY]
+
+ def __init__(self, base_dir, entry, manifest=None, changelog=None,
+ licenses=None, tests=None, certificate=None,
+ meta_file_version=META_FILE_VERSION_VALUE,
+ meta_csar_version=META_CSAR_VERSION_VALUE,
+ meta_created_by=META_CREATED_BY_VALUE):
+
+ self.base_dir = base_dir
+
+ metadata = {}
+ metadata[META_FILE_VERSION_KEY] = str(meta_file_version)
+ metadata[META_CSAR_VERSION_KEY] = str(meta_csar_version)
+ metadata[META_CREATED_BY_KEY] = meta_created_by
+ metadata[META_ENTRY_DEFINITIONS_KEY] = entry
+ if manifest:
+ metadata[self.META_ENTRY_MANIFEST_FILE_KEY] = manifest
+ if changelog:
+ metadata[self.META_ENTRY_HISTORY_FILE_KEY] = changelog
+ if licenses:
+ metadata[self.META_ENTRY_LICENSES_DIR_KEY] = licenses
+ if tests:
+ metadata[self.META_ENTRY_TESTS_DIR_KEY] = tests
+ if certificate:
+ metadata[self.META_ENTRY_CERT_FILE_KEY] = certificate
+
+ self.metadata = self._validate(metadata)
+
+ def _validate(self, metadata):
+ for (key, value) in six.iteritems(BASE_META):
+ if metadata.get(key) != value:
+ raise ValueError('TOSCA.meta: {} must be {}'.format(key, value))
+
+ utils.check_file_dir(root=self.base_dir,
+ entry=metadata.get(META_ENTRY_DEFINITIONS_KEY),
+ msg='Please specify a valid entry point.',
+ check_dir=False)
+ entry_file = os.path.join(self.base_dir,
+ metadata.get(META_ENTRY_DEFINITIONS_KEY))
+ try:
+ with open(entry_file) as f:
+ v = yaml.safe_load(f)['tosca_definitions_version']
+ except:
+ raise ValueError('Entry file {} is not a valid tosca simple yaml file'.format(entry_file))
+
+ if metadata.get(self.META_ENTRY_MANIFEST_FILE_KEY):
+ utils.check_file_dir(root=self.base_dir,
+ entry=metadata[self.META_ENTRY_MANIFEST_FILE_KEY],
+ msg='Please specify a valid manifest file.',
+ check_dir=False)
+ if metadata.get(self.META_ENTRY_HISTORY_FILE_KEY):
+ utils.check_file_dir(root=self.base_dir,
+ entry=metadata[self.META_ENTRY_HISTORY_FILE_KEY],
+ msg='Please specify a valid change history file.',
+ check_dir=False)
+ if metadata.get(self.META_ENTRY_LICENSES_DIR_KEY):
+ utils.check_file_dir(root=self.base_dir,
+ entry=metadata[self.META_ENTRY_LICENSES_DIR_KEY],
+ msg='Please specify a valid license directory.',
+ check_dir=True)
+ if metadata.get(self.META_ENTRY_TESTS_DIR_KEY):
+ utils.check_file_dir(root=self.base_dir,
+ entry=metadata[self.META_ENTRY_TESTS_DIR_KEY],
+ msg='Please specify a valid test directory.',
+ check_dir=True)
+ if metadata.get(self.META_ENTRY_CERT_FILE_KEY):
+ utils.check_file_dir(root=self.base_dir,
+ entry=metadata[self.META_ENTRY_CERT_FILE_KEY],
+ msg='Please specify a valid certificate file.',
+ check_dir=False)
+ missing_keys = [key for key in self.REQUIRED_KEYS if key not in metadata]
+ if missing_keys:
+ raise ValueError('TOSCA.meta: missing keys: {}'.format(','.join(missing_keys)))
+ return metadata
+
+ def dump_as_string(self):
+ s = ""
+ for key in self.REQUIRED_KEYS + self.OPTIONAL_KEYS:
+ if self.metadata.get(key):
+ s += "{}: {}\n".format(key, self.metadata.get(key))
+ return s
+
+ @property
+ def created_by(self):
+ return self.metadata.get(META_CREATED_BY_KEY)
+
+ @property
+ def csar_version(self):
+ return self.metadata.get(META_CSAR_VERSION_KEY)
+
+ @property
+ def meta_file_version(self):
+ return self.metadata.get(META_FILE_VERSION_KEY)
+
+ @property
+ def entry_definitions(self):
+ return self.metadata.get(META_ENTRY_DEFINITIONS_KEY)
+
+ @property
+ def entry_manifest_file(self):
+ return self.metadata.get(self.META_ENTRY_MANIFEST_FILE_KEY)
+
+ @property
+ def entry_history_file(self):
+ return self.metadata.get(self.META_ENTRY_HISTORY_FILE_KEY)
+
+ @property
+ def entry_tests_dir(self):
+ return self.metadata.get(self.META_ENTRY_TESTS_DIR_KEY)
+
+ @property
+ def entry_licenses_dir(self):
+ return self.metadata.get(self.META_ENTRY_LICENSES_DIR_KEY)
+
+ @property
+ def entry_certificate_file(self):
+ return self.metadata.get(self.META_ENTRY_CERT_FILE_KEY)
+
+
+class ToscaMeta241(ToscaMeta):
+ # SOL004 v2.4.1
+ META_ENTRY_MANIFEST_FILE_KEY = 'Entry-Manifest'
+ META_ENTRY_HISTORY_FILE_KEY = 'Entry-Change-Log'
+ META_ENTRY_TESTS_DIR_KEY = 'Entry-Tests'
+ META_ENTRY_LICENSES_DIR_KEY = 'Entry-Licenses'
+ META_ENTRY_CERT_FILE_KEY = 'Entry-Certificate'
+ REQUIRED_KEYS = [ META_FILE_VERSION_KEY, META_CSAR_VERSION_KEY,
+ META_CREATED_BY_KEY, META_ENTRY_DEFINITIONS_KEY,
+ ]
+ OPTIONAL_KEYS = [ META_ENTRY_MANIFEST_FILE_KEY, META_ENTRY_HISTORY_FILE_KEY,
+ META_ENTRY_LICENSES_DIR_KEY, META_ENTRY_TESTS_DIR_KEY,
+ META_ENTRY_CERT_FILE_KEY,
+ ]
+
+
+class ToscaMeta261(ToscaMeta):
+ # SOL004 v2.6.1
+ pass
+
+
+def create_from_file(base_dir):
+ csar_metafile = os.path.join(base_dir, META_FILE)
+ if not os.path.exists(csar_metafile):
+ raise ValueError('Metadata file {0} is missing from the CSAR'.format(csar_metafile))
+ LOG.debug('CSAR metadata file: {0}'.format(csar_metafile))
+ LOG.debug('Attempting to parse CSAR metadata YAML')
+ with open(csar_metafile) as f:
+ metadata = yaml.safe_load(f)
+ LOG.debug('CSAR metadata:\n{0}'.format(pprint.pformat(metadata)))
+ # By default we assume it's SOL004 2.4.1
+ cls = ToscaMeta241
+ for key in metadata.keys():
+ if key.startswith('ETSI-'):
+ cls = ToscaMeta261
+ break
+ return cls(base_dir,
+ entry=metadata.get(META_ENTRY_DEFINITIONS_KEY),
+ manifest=metadata.get(cls.META_ENTRY_MANIFEST_FILE_KEY),
+ changelog=metadata.get(cls.META_ENTRY_HISTORY_FILE_KEY),
+ licenses=metadata.get(cls.META_ENTRY_LICENSES_DIR_KEY),
+ tests=metadata.get(cls.META_ENTRY_TESTS_DIR_KEY),
+ certificate=metadata.get(cls.META_ENTRY_CERT_FILE_KEY),
+ meta_file_version=metadata.get(META_FILE_VERSION_KEY),
+ meta_csar_version=metadata.get(META_CSAR_VERSION_KEY),
+ meta_created_by=metadata.get(META_CREATED_BY_KEY))
+
LOG = logging.getLogger(__name__)
+def check_file_dir(root, entry, msg, check_for_non=False, check_dir=False):
+ path = os.path.join(root, entry)
+ if check_for_non:
+ ret = not os.path.exists(path)
+ error_msg = '{0} already exists. ' + msg
+ elif check_dir:
+ ret = os.path.isdir(path)
+ error_msg = '{0} is not an existing directory. ' + msg
+ else:
+ ret = os.path.isfile(path)
+ error_msg = '{0} is not an existing file. ' + msg
+ if not ret:
+ raise ValueError(error_msg.format(path))
+
+
def _hash_value_for_file(f, hash_function, block_size=2**20):
while True:
data = f.read(block_size)
global __version__
-__version__='1.3.0'
+__version__='1.4.0pre'
import six
from stevedore import driver
-from vnfsdk_pkgtools.packager import csar
+from vnfsdk_pkgtools.packager import toscameta
from vnfsdk_pkgtools.validator import toscaparser_validator as tv
from vnfsdk_pkgtools import vnfreq
for file in files:
full_path = os.path.join(root, file)
rel_path = os.path.relpath(full_path, reader.destination)
- if rel_path not in (reader.entry_manifest_file, csar.META_FILE):
+ if rel_path not in (reader.entry_manifest_file, toscameta.META_FILE):
if rel_path not in reader.manifest.digests:
raise vnfreq.VnfRequirementError("Package component %s not found in manifest file" % rel_path)
return 0
Code Review / vnfsdk / pkgtools.git / commitdiff