-#\r
-# Copyright (c) 2017 GigaSpaces Technologies Ltd. All rights reserved.\r
-#\r
-# Licensed under the Apache License, Version 2.0 (the "License"); you may\r
-# not use this file except in compliance with the License. You may obtain\r
-# a copy of the License at\r
-#\r
-# http://www.apache.org/licenses/LICENSE-2.0\r
-#\r
-# Unless required by applicable law or agreed to in writing, software\r
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\r
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\r
-# License for the specific language governing permissions and limitations\r
-# under the License.\r
-#\r
-\r
-import pytest\r
-from vnfsdk_pkgtools.cli import __main__\r
-\r
-def test_main(capsys):\r
- with pytest.raises(SystemExit):\r
- args = __main__.parse_args(['csar-create', '-h'])\r
- args.func(args)\r
- out, err = capsys.readouterr()\r
- assert out.startswith('usage:')\r
-\r
+#
+# 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 pytest
+
+from vnfsdk_pkgtools.cli import __main__
+
+
+def test_main(capsys):
+ with pytest.raises(SystemExit):
+ args = __main__.parse_args(['csar-create', '-h'])
+ args.func(args)
+ out, err = capsys.readouterr()
+ assert out.startswith('usage:')
CSAR_OUTPUT_FILE = 'output.csar'
Args = collections.namedtuple('Args',
- ['source', 'entry', 'manifest', 'history', 'tests',
- 'licenses', 'digest', 'certificate', 'privkey', 'sol241'])
-
-
-ARGS_MANIFEST = {
- 'source': CSAR_RESOURCE_DIR,
- 'entry': CSAR_ENTRY_FILE,
- 'manifest': 'test_entry.mf',
- 'history': 'ChangeLog.txt',
- 'tests': 'Tests',
- 'licenses': 'Licenses',
- 'digest': None,
- 'certificate': None,
- 'privkey': None,
- 'sol241': False,
- }
-
-ARGS_MANIFEST_DIGEST = {
- 'source': CSAR_RESOURCE_DIR,
- 'entry': CSAR_ENTRY_FILE,
- 'manifest': 'test_entry.mf',
- 'history': 'ChangeLog.txt',
- 'tests': 'Tests',
- 'licenses': 'Licenses',
- 'digest': 'sha-256',
- 'certificate': None,
- 'privkey': None,
- 'sol241': False,
- }
-
-ARGS_MANIFEST_DIGEST_CERT = {
- 'source': CSAR_RESOURCE_DIR,
- 'entry': CSAR_ENTRY_FILE,
- 'manifest': 'test_entry.mf',
- 'history': 'ChangeLog.txt',
- 'tests': 'Tests',
- 'licenses': 'Licenses',
- 'digest': 'sha-256',
- 'certificate': 'test.crt',
- 'privkey': os.path.join(ROOT_DIR, 'tests', 'resources', 'signature', 'test.key'),
- 'sol241': False,
- }
-
-ARGS_NO_MANIFEST = {
- 'source': CSAR_RESOURCE_DIR,
- 'entry': CSAR_ENTRY_FILE,
- 'manifest': None,
- 'history': None,
- 'tests': None,
- 'licenses': None,
- 'digest': None,
- 'certificate': None,
- 'privkey': None,
- 'sol241': True,
- }
-
-INVALID_ARGS_NO_MANIFEST = {
- 'source': CSAR_RESOURCE_DIR,
- 'entry': CSAR_ENTRY_FILE,
- 'manifest': None,
- 'history': None,
- 'tests': None,
- 'licenses': None,
- 'digest': 'sha-256',
- 'certificate': None,
- 'privkey': None,
- 'sol241': True,
- }
-
-INVALID_ARGS_NO_PRIVKEY = {
- 'source': CSAR_RESOURCE_DIR,
- 'entry': CSAR_ENTRY_FILE,
- 'manifest': 'test_entry.mf',
- 'history': None,
- 'tests': None,
- 'licenses': None,
- 'digest': None,
- 'certificate': 'test.crt',
- 'privkey': None,
- 'sol241': True,
- }
+ ['source', 'entry', 'manifest',
+ 'history', 'tests', 'licenses',
+ 'digest', 'certificate', 'privkey',
+ 'sol241'])
+
+
+ARGS_MANIFEST = {'source': CSAR_RESOURCE_DIR,
+ 'entry': CSAR_ENTRY_FILE,
+ 'manifest': 'test_entry.mf',
+ 'history': 'ChangeLog.txt',
+ 'tests': 'Tests',
+ 'licenses': 'Licenses',
+ 'digest': None,
+ 'certificate': None,
+ 'privkey': None,
+ 'sol241': False,
+ }
+
+ARGS_MANIFEST_DIGEST = {'source': CSAR_RESOURCE_DIR,
+ 'entry': CSAR_ENTRY_FILE,
+ 'manifest': 'test_entry.mf',
+ 'history': 'ChangeLog.txt',
+ 'tests': 'Tests',
+ 'licenses': 'Licenses',
+ 'digest': 'sha-256',
+ 'certificate': None,
+ 'privkey': None,
+ 'sol241': False,
+ }
+
+ARGS_MANIFEST_DIGEST_CERT = {'source': CSAR_RESOURCE_DIR,
+ 'entry': CSAR_ENTRY_FILE,
+ 'manifest': 'test_entry.mf',
+ 'history': 'ChangeLog.txt',
+ 'tests': 'Tests',
+ 'licenses': 'Licenses',
+ 'digest': 'sha-256',
+ 'certificate': 'test.crt',
+ 'privkey': os.path.join(ROOT_DIR, 'tests',
+ 'resources',
+ 'signature',
+ 'test.key'),
+ 'sol241': False,
+ }
+
+ARGS_NO_MANIFEST = {'source': CSAR_RESOURCE_DIR,
+ 'entry': CSAR_ENTRY_FILE,
+ 'manifest': None,
+ 'history': None,
+ 'tests': None,
+ 'licenses': None,
+ 'digest': None,
+ 'certificate': None,
+ 'privkey': None,
+ 'sol241': True,
+ }
+
+INVALID_ARGS_NO_MANIFEST = {'source': CSAR_RESOURCE_DIR,
+ 'entry': CSAR_ENTRY_FILE,
+ 'manifest': None,
+ 'history': None,
+ 'tests': None,
+ 'licenses': None,
+ 'digest': 'sha-256',
+ 'certificate': None,
+ 'privkey': None,
+ 'sol241': True,
+ }
+
+INVALID_ARGS_NO_PRIVKEY = {'source': CSAR_RESOURCE_DIR,
+ 'entry': CSAR_ENTRY_FILE,
+ 'manifest': 'test_entry.mf',
+ 'history': None,
+ 'tests': None,
+ 'licenses': None,
+ 'digest': None,
+ 'certificate': 'test.crt',
+ 'privkey': None,
+ 'sol241': True,
+ }
def csar_write_test(args):
"vnf_provider_id: test",
"vnf_package_version: 1.0",
"vnf_release_data_time: 2017-09-15T15:00:10+08:00",
- ])
+ ])
METADATA_MISSING_KEY = '\n'.join(["metadata:",
- "vnf_product_name: test",
- "vnf_provider_id: test",
- "vnf_package_version:1.0",
- ])
+ "vnf_product_name: test",
+ "vnf_provider_id: test",
+ "vnf_package_version:1.0",
+ ])
METADATA_MISSING = "vnf_product_name: test"
FILE_DIGEST = '\n'.join(['Source: digest',
'Algorithm: SHA-256',
'Hash: 20a480339aa4371099f9503511dcc5a8051ce3884846678ced5611ec64bbfc9c',
- ])
+ ])
CMS = '\n'.join(['-----BEGIN CMS-----',
'MIICmAYJKoZIhvcNAQcCoIICiTCCAoUCAQExDTALBglghkgBZQMEAgEwCwYJKoZI',
'pqE+DUlSFyilc9CQWnSLubkHmM4dZnU7qnNoTBqplDYpOYH3WSNN9Cv322JusAzt',
'SzFEv182phI2C5pmjUnf7VG1WMKCH2WNtkYwMUCDcGvbHrh8n+kR8hL/BAs=',
'-----END CMS-----',
- ])
+ ])
FILE_SOURCE_ONLY = '\n'.join(['Source: source1',
'Source: source2',
- ])
+ ])
NON_MANO_ARTIFACTS = '\n'.join(['non_mano_artifact_sets:',
'foo_bar:',
'prv.happy-nfv.cool:',
'Source: happy/cool/123.html',
'Source: happy/cool/cool.json',
- ])
+ ])
def test_metadata(tmpdir):
manifest.Manifest(p.dirname, 'test.mf')
excinfo.match(r"Unrecognized file digest line vnf_product_name: test:")
+
def test_digest(tmpdir):
root = tmpdir.mkdir('csar')
mf = root.join('test.mf')
assert m.digests['digest'][0] == "SHA-256"
assert m.digests['digest'][1] == "20a480339aa4371099f9503511dcc5a8051ce3884846678ced5611ec64bbfc9c"
+
def test_add_file(tmpdir):
root = tmpdir.mkdir('csar')
mf = root.join('test.mf')
assert m.digests['digest'][0] == "SHA-256"
assert m.digests['digest'][1] == "20a480339aa4371099f9503511dcc5a8051ce3884846678ced5611ec64bbfc9c"
+
def test_update_to_file(tmpdir):
root = tmpdir.mkdir('csar')
mf = root.join('test.mf')
assert len(list(m2.digests.keys())) == 2
assert m2.signature == CMS
+
def test_signature(tmpdir):
p = tmpdir.mkdir('csar').join('test.mf')
p.write(METADATA + "\n\n" + CMS)
m = manifest.Manifest(p.dirname, 'test.mf')
assert m.signature == CMS
+
def test_illegal_signature(tmpdir):
p = tmpdir.mkdir('csar').join('test.mf')
p.write(METADATA + "\n\n" + CMS[:-17])
manifest.Manifest(p.dirname, 'test.mf')
excinfo.match(r"Can NOT find end of sigature block")
+
def test_signature_strip(tmpdir):
p = tmpdir.mkdir('csar').join('test.mf')
p.write(METADATA + "\n\n" + CMS)
assert m2.signature is None
os.unlink(newfile)
+
def test_source_only(tmpdir):
p = tmpdir.mkdir('csar').join('test.mf')
p.write(METADATA + "\n\n" + FILE_SOURCE_ONLY)
assert 'source1' in m.digests.keys()
assert 'source2' in m.digests.keys()
+
def test_non_mano_artifacts(tmpdir, mocker):
mocker.patch('vnfsdk_pkgtools.packager.utils.check_file_dir')
p = tmpdir.mkdir('csar').join('test.mf')
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,
- }
+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):
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'))
'TOSCA-Metadata',
'TOSCA.meta'))
+
def test_create_from_file():
for case in FROM_FILE_CASES:
target = tempfile.mkdtemp()
from vnfsdk_pkgtools import util
ROOT_DIR = util.get_project_root()
-RESOURCES_DIR= os.path.join(ROOT_DIR, 'tests', 'resources', 'signature')
+RESOURCES_DIR = os.path.join(ROOT_DIR, 'tests', 'resources', 'signature')
-MSG_FILE = os.path.join(RESOURCES_DIR, 'manifest.mf')
+MSG_FILE = os.path.join(RESOURCES_DIR, 'manifest.mf')
CERT_FILE = os.path.join(RESOURCES_DIR, 'test.crt')
-KEY_FILE = os.path.join(RESOURCES_DIR, 'test.key')
+KEY_FILE = os.path.join(RESOURCES_DIR, 'test.key')
CONTENT = b"needToBeHashed"
SHA256 = "20a480339aa4371099f9503511dcc5a8051ce3884846678ced5611ec64bbfc9c"
SHA512 = "dbed8672e752d51d0c7ca42050f67faf1534e58470bba96e787df5c4cf6a4f8ecf7ad45fb9307adbc5b9dec8432627d86b3eb1d3d43ee9c5e93f754ff2825320"
+
def test_cal_file_hash(tmpdir):
p = tmpdir.join("file_to_hash.txt")
p.write(CONTENT)
assert SHA512 == utils.cal_file_hash("", str(p), 'SHA512')
assert SHA256 == utils.cal_file_hash(p.dirname, p.basename, 'sha256')
+
def test_cal_file_hash_remote(mocker):
class FakeRequest(object):
def __init__(self, *args):
mocker.patch('requests.get', new=FakeRequest)
assert SHA256 == utils.cal_file_hash("", "http://fake", 'sha256')
+
def test_sign_verify_pairwise():
cms = utils.sign(MSG_FILE, CERT_FILE, KEY_FILE)
# We can't examine the exact content of cms because it contains timestamp
def test_verify_bad(tmpdir):
cms = utils.sign(MSG_FILE, CERT_FILE, KEY_FILE)
-
p = tmpdir.join("file_msg.txt")
p.write("BAD")
-
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,
- }
- },
- ]
+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():
HPA_PATH = os.path.join(RESOURCES_DIR, 'hpa.csar')
BAD_HPA_PATH = os.path.join(RESOURCES_DIR, 'hpa_bad.csar')
+
def test_validate(tmpdir):
reader = csar._CSARReader(CSAR_PATH, str(tmpdir.mkdir('validate')))
validator = toscaparser_validator.ToscaparserValidator()
validator.validate(reader)
assert hasattr(validator, 'tosca')
+
def test_validate_hpa(tmpdir):
reader = csar._CSARReader(HPA_PATH, str(tmpdir.mkdir('validate')))
validator = toscaparser_validator.ToscaparserValidator()
validator.validate(reader)
assert hasattr(validator, 'tosca')
+
def test_validate_hpa_bad(tmpdir):
reader = csar._CSARReader(BAD_HPA_PATH, str(tmpdir.mkdir('validate')))
validator = toscaparser_validator.ToscaparserValidator()
# under the License.
#
-import os
-
from ruamel import yaml
import pytest
BAD_CONTENT = "---\na: b\n- d"
+
def test_load_definition(tmpdir):
p = tmpdir.join("test_definition.yaml")
p.write(CONTENT)
# under the License.
#
-import pytest
-
from vnfsdk_pkgtools import vnfreq
check_result('R-04298', reader, None,
'No testing scripts found')
+
def test_R26881(mocker, tmpdir):
p1 = tmpdir.join('entry.yaml')
p1.write("")
validator.tosca.nodetemplates = [node]
node.entity_tpl = {'artifacts': {'sw_image': {'file': 'image',
'type': 'tosca.artifacts.nfv.SwImage',
- }
- }
- }
+ }
+ }
+ }
check_result('R-26881', reader, validator, None)
node = mocker.Mock()
validator.tosca.nodetemplates = [node]
check_result('R-35851', None, validator, None)
-
# under the License.
#
-import pytest
-
from vnfsdk_pkgtools import vnfreq
+
class FakeTester(vnfreq.TesterBase):
ID = 'fake'
DESC = 'fake'
#
[tox]
-envlist=py27,py3
+envlist=py27,py3,pep8
[testenv]
+basepython = python3
passenv =
TRAVIS
PYTHON
[testenv:py3]
basepython = python3
+
+[testenv:pep8]
+deps = flake8<3.8,>=3.7.0
+commands = flake8 {toxinidir}/vnfsdk_pkgtools {toxinidir}/tests
+
+[flake8]
+ignore = E501
import sys
import logging
import argparse
-import os
import shutil
import tempfile
def csar_create_func(namespace):
csar.write(namespace.source,
- namespace.entry,
- namespace.destination,
- args=namespace)
+ namespace.entry,
+ namespace.destination,
+ args=namespace)
+
def csar_open_func(namespace):
csar.read(namespace.source,
namespace.destination,
namespace.no_verify_cert)
+
def csar_validate_func(namespace):
workdir = tempfile.mkdtemp()
try:
"""
parser = argparse.ArgumentParser(description='VNF SDK CSAR manipulation tool')
parser.add_argument('-v', '--verbose',
- dest='verbosity',
- action='count',
- default=0,
- help='Set verbosity level (can be passed multiple times)')
+ dest='verbosity',
+ action='count',
+ default=0,
+ help='Set verbosity level (can be passed multiple times)')
subparsers = parser.add_subparsers(help='csar-create')
csar_create = subparsers.add_parser('csar-create')
action='store_true',
help='Generate SOL004 v2.4.1 csar for backward compatilibity')
-
csar_open = subparsers.add_parser('csar-open')
csar_open.set_defaults(func=csar_open_func)
csar_open.add_argument(
action='store_true',
help="Do NOT verify the signer's certificate")
-
csar_validate = subparsers.add_parser('csar-validate')
csar_validate.set_defaults(func=csar_validate_func)
csar_validate.add_argument(
import logging
import os
-import pprint
import tempfile
import zipfile
import requests
+from ruamel import yaml
from vnfsdk_pkgtools.packager import manifest
from vnfsdk_pkgtools.packager import toscameta
from vnfsdk_pkgtools.packager import utils
-METADATA_KEYS = [ 'vnf_provider_id',
- 'vnf_product_name',
- 'vnf_release_date_time',
- 'vnf_package_version']
-DIGEST_KEYS = [ 'Source', 'Algorithm', 'Hash' ]
+METADATA_KEYS = ['vnf_provider_id',
+ 'vnf_product_name',
+ 'vnf_release_date_time',
+ 'vnf_package_version']
+DIGEST_KEYS = ['Source', 'Algorithm', 'Hash']
SUPPORTED_HASH_ALGO = ['SHA-256', 'SHA-512']
NON_MANO_ARTIFACT_RE = re.compile(r'^[0-9a-z_-]+(\.[0-9a-z_-]+)*:$')
+
class ManifestException(Exception):
pass
+
class Manifest(object):
' Manifest file in CSAR package'
def __init__(self, root_path, manifest_path, sol241=False):
# non_mano_artifact dict
# :key = set identifier
# :value = list of files
- self.sol241=sol241
+ self.sol241 = sol241
self.non_mano_artifacts = {}
- self.blocks = [ ]
+ self.blocks = []
self._split_blocks()
self._parse_all_blocks()
@staticmethod
def __split_line(s):
- remain=s
+ remain = s
try:
- (key, value)=s.split(':', 1)
+ (key, value) = s.split(':', 1)
value = value.strip()
remain = None
except ValueError:
Split manifest file into blocks, each block is seperated by a empty
line or a line with only spaces and tabs.
'''
- block_content = [ ]
+ block_content = []
with open(os.path.join(self.root, self.path), 'rU') as fp:
for line in fp:
line = line.strip(' \t\n')
if key in METADATA_KEYS:
self.metadata[key] = value
elif key == 'vnf_release_data_time':
- #sol004 v2.4.1 compatibility
+ # sol004 v2.4.1 compatibility
self.metadata['vnf_release_date_time'] = value
else:
raise ManifestException("Unrecognized metadata %s:" % line)
- #validate metadata keys
+ # validate metadata keys
missing_keys = set(METADATA_KEYS) - set(self.metadata.keys())
if missing_keys:
raise ManifestException("Missing metadata keys: %s" % ','.join(missing_keys))
udatetime.from_string(self.metadata['vnf_release_date_time'])
except ValueError:
raise ManifestException("Incorrect IETF RFC 3339 vnf_release_date_time: %s"
- % self.metadata['vnf_release_date_time'])
+ % self.metadata['vnf_release_date_time'])
def parse_cms(self, lines):
if '--END CMS--' not in lines[-1]:
if key == 'Source':
self.digests[value] = (None, None)
elif key == 'Algorithm':
- #validate algorithm
+ # validate algorithm
desc['Algorithm'] = desc['Algorithm'].upper()
if desc['Algorithm'] not in SUPPORTED_HASH_ALGO:
raise ManifestException("Unsupported hash algorithm: %s" % desc['Algorithm'])
- #validate hash
+ # validate hash
if desc.get('Algorithm') and desc.get('Hash') and desc.get('Source'):
hash = utils.cal_file_hash(self.root, desc['Source'], desc['Algorithm'])
if hash != desc['Hash']:
# empty line between digest and signature section
ret += "\n"
# signature
- if self.signature:
+ if self.signature:
ret += self.signature
return ret
elif not skip:
lines.append(line)
content = ''.join(lines)
- tmpfile = tempfile.NamedTemporaryFile(mode='w',delete=False)
+ tmpfile = tempfile.NamedTemporaryFile(mode='w', delete=False)
tmpfile.write(content)
tmpfile.close()
return tmpfile.name
import os
import pprint
-from ruamel import yaml # @UnresolvedImport
+from ruamel import yaml
import six
from vnfsdk_pkgtools.packager import utils
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,
- ]
+ 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,
metadata.get(META_ENTRY_DEFINITIONS_KEY))
try:
with open(entry_file) as f:
- v = yaml.safe_load(f)['tosca_definitions_version']
- except:
+ yaml.safe_load(f)['tosca_definitions_version']
+ except Exception:
raise ValueError('Entry file {} is not a valid tosca simple yaml file'.format(entry_file))
if metadata.get(self.META_ENTRY_MANIFEST_FILE_KEY):
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,
- ]
+ 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):
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))
-
def _run_cmd(cmd, **kwargs):
if isinstance(cmd, list):
args = cmd
- elif isinstance(cmd, string):
+ elif isinstance(cmd, str):
args = [cmd]
else:
raise RuntimeError("cmd must be string or list")
def sign(msg_file, cert_file, key_file):
args = ["openssl", "cms", "-sign", "-binary"]
kwargs = {
- '-in': os.path.abspath(msg_file),
- '-signer': os.path.abspath(cert_file),
- '-inkey': os.path.abspath(key_file),
- '-outform': 'PEM',
- }
+ '-in': os.path.abspath(msg_file),
+ '-signer': os.path.abspath(cert_file),
+ '-inkey': os.path.abspath(key_file),
+ '-outform': 'PEM', }
return _run_cmd(args, **kwargs)
f.write(cms)
f.flush()
kwargs = {
- '-in': f.name,
- '-inform': 'PEM',
- '-content': os.path.abspath(msg_file),
- '-certfile': os.path.abspath(cert_file),
- }
+ '-in': f.name,
+ '-inform': 'PEM',
+ '-content': os.path.abspath(msg_file),
+ '-certfile': os.path.abspath(cert_file), }
return _run_cmd(args, **kwargs)
import six
from stevedore import driver
-
NS = "vnfsdk.pkgtools.validator"
+
def get_validator(params):
"""Get validate driver and load it.
def __init__(self):
self.tosca = None
-
@abc.abstractmethod
def validate(self, reader):
"""Validate the csar package.
self.hpa_schema_version = defs.get('metadata', {}).get('version')
self.hpa_schemas = defs.get('schemas', {})
self.hpa_mappings = defs.get('mappings', [])
- #validate schema defined in hpa.yaml is correct
+ # validate schema defined in hpa.yaml is correct
if not self.hpa_schema_version:
msg = "No defined version in metadata"
raise HpaSchemaDefError(msg)
def validate(self, reader):
entry_path = os.path.join(reader.destination,
- reader.entry_definitions)
+ reader.entry_definitions)
try:
- #TODO set debug_mode due to upstream bug
+ # TODO set debug_mode due to upstream bug
# https://jira.opnfv.org/browse/PARSER-181
self.tosca = ToscaTemplate(path=entry_path,
no_required_paras_check=True,
for (key, hpa_value) in six.iteritems(value):
if key not in hpa_schema:
msg = "node %s: %s is NOT a valid HPA key"
- raise HpaValueError(msg % (refkey, key))
+ raise HpaValueError(msg % (refkey, key))
try:
hpa_dict = json.loads(hpa_value)
- except:
+ except Exception:
msg = "node %s, HPA key %s: %s is NOT a valid json encoded string"
raise HpaValueError(msg % (refkey, key, hpa_value.encode('ascii', 'replace')))
if not isinstance(hpa_dict, dict):
attr_schema = hpa_schema[key][attr]
if not re.match(attr_schema, str(val)):
msg = ("node %s, HPA key %s, attr %s: %s is not a valid HPA "
- "attr value, expected re pattern is %s")
+ "attr value, expected re pattern is %s")
raise HpaValueError(msg % (refkey, key, attr, val.encode('ascii', 'replace'), attr_schema))
def validate_hpa_value(self, refkey, hpa_schema, values):
self.validate_hpa_value(refkey,
self.hpa_schemas[mapping['schema']],
value)
-
-
if hasattr(err, 'problem_mark'):
mark = err.problem_mark
errmsg = ("Invalid YAML syntax in Definitions file %(file)s "
- "at line: %(line)s, column: %(column)s." %
- dict(file=config_file,
- line=mark.line + 1,
- column=mark.column + 1))
+ "at line: %(line)s, column: %(column)s." %
+ dict(file=config_file,
+ line=mark.line + 1,
+ column=mark.column + 1))
else:
errmsg = ("YAML error reading Definitions file %s" % config_file)
LOG.error(errmsg)
global __version__
-__version__='1.4.0pre'
+__version__ = '1.4.0pre'
import six
from stevedore import driver
-
NS = "vnfsdk.pkgtools.vnfreq"
+
def get_vnfreq_tester(name):
"""Get vnf requirement tester.
self.err = self._do_check(reader, tosca)
except Exception as e:
self.err = e
-
# under the License.
#
-import abc
import os
import six
-from stevedore import driver
from vnfsdk_pkgtools.packager import toscameta
-from vnfsdk_pkgtools.validator import toscaparser_validator as tv
from vnfsdk_pkgtools import vnfreq
if file and \
os.path.isfile(os.path.join(entry_path, file)) or \
os.path.isfile(os.path.join(reader.destination, file)):
- valid_artifacts.append(file)
+ valid_artifacts.append(file)
if not valid_artifacts:
raise vnfreq.VnfRequirementError("No valid binaries or images for VNF instantion found")
return 0
for node in getattr(tosca.tosca, 'nodetemplates', []):
if tosca.is_type(node, 'tosca.nodes.nfv.VduCp') or \
tosca.is_type(node, 'tosca.nodes.nfv.VnfVirtualLink'):
- found = True
- break
+ found = True
+ break
if not found:
raise vnfreq.VnfRequirementError("No basic network or application connectivity found")
return 0
-
Code Review / vnfsdk / pkgtools.git / commitdiff