openstack/deb-python-tosca-parser
Code Issues Proposed changes

Merge "Fix a bug where the temp folder is not always removed"

Browse Source
This commit is contained in:
Jenkins
2015-11-09 15:52:43 +00:00
committed by Gerrit Code Review
parent 6e55a1b905 d83ce965f0
commit fe83d5977f
3 changed files with 96 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
toscaparser/prereq/csar.py
View File

@@ -12,6 +12,7 @@
import os.path import os.path
import requests import requests
import shutil
import six import six
import tempfile import tempfile
import yaml import yaml
@@ -36,6 +37,7 @@ class CSAR(object):
self.a_file = a_file self.a_file = a_file
self.is_validated = False self.is_validated = False
self.csar = None self.csar = None
self.temp_dir = None
def validate(self): def validate(self):
"""Validate the provided CSAR file.""" """Validate the provided CSAR file."""
@@ -152,10 +154,9 @@ class CSAR(object):
def decompress(self): def decompress(self):
if not self.is_validated: if not self.is_validated:
self.validate() self.validate()
folder = tempfile.NamedTemporaryFile().name self.temp_dir = tempfile.NamedTemporaryFile().name
with zipfile.ZipFile(self.csar, "r") as zf: with zipfile.ZipFile(self.csar, "r") as zf:
zf.extractall(folder) zf.extractall(self.temp_dir)
return folder
def _validate_external_references(self): def _validate_external_references(self):
"""Extracts files referenced in the main template """Extracts files referenced in the main template
@@ -165,64 +166,67 @@ class CSAR(object):
* interface implementations * interface implementations
* artifacts * artifacts
""" """
temp_dir = self.decompress() try:
main_tpl_file = self.get_main_template() self.decompress()
main_tpl = self.get_main_template_yaml() main_tpl_file = self.get_main_template()
main_tpl = self.get_main_template_yaml()
if 'imports' in main_tpl: if 'imports' in main_tpl:
ImportsLoader(main_tpl['imports'], ImportsLoader(main_tpl['imports'],
os.path.join(temp_dir, main_tpl_file)) os.path.join(self.temp_dir, main_tpl_file))
if 'topology_template' not in main_tpl: if 'topology_template' in main_tpl:
return topology_template = main_tpl['topology_template']
topology_template = main_tpl['topology_template']
if 'node_templates' not in topology_template: if 'node_templates' in topology_template:
return node_templates = topology_template['node_templates']
node_templates = topology_template['node_templates']
for node_template_key in node_templates: for node_template_key in node_templates:
node_template = node_templates[node_template_key] node_template = node_templates[node_template_key]
if 'artifacts' in node_template: if 'artifacts' in node_template:
artifacts = node_template['artifacts'] artifacts = node_template['artifacts']
for artifact_key in artifacts: for artifact_key in artifacts:
artifact = artifacts[artifact_key] artifact = artifacts[artifact_key]
if isinstance(artifact, six.string_types): if isinstance(artifact, six.string_types):
self._validate_external_reference(temp_dir, self._validate_external_reference(
main_tpl_file, main_tpl_file,
artifact) artifact)
elif isinstance(artifact, dict): elif isinstance(artifact, dict):
if 'file' in artifact: if 'file' in artifact:
self._validate_external_reference(temp_dir, self._validate_external_reference(
main_tpl_file, main_tpl_file,
artifact['file']) artifact['file'])
else: else:
raise ValueError(_('Unexpected artifact definition ' raise ValueError(
'for %s.') % artifact_key) _('Unexpected artifact definition '
if 'interfaces' in node_template: 'for %s.') % artifact_key)
interfaces = node_template['interfaces'] if 'interfaces' in node_template:
for interface_key in interfaces: interfaces = node_template['interfaces']
interface = interfaces[interface_key] for interface_key in interfaces:
for opertation_key in interface: interface = interfaces[interface_key]
operation = interface[opertation_key] for opertation_key in interface:
if isinstance(operation, six.string_types): operation = interface[opertation_key]
self._validate_external_reference(temp_dir, if isinstance(operation, six.string_types):
main_tpl_file, self._validate_external_reference(
operation, main_tpl_file,
False) operation,
elif isinstance(operation, dict): False)
if 'implementation' in operation: elif isinstance(operation, dict):
self._validate_external_reference( if 'implementation' in operation:
temp_dir, main_tpl_file, self._validate_external_reference(
operation['implementation']) main_tpl_file,
operation['implementation'])
finally:
if self.temp_dir:
shutil.rmtree(self.temp_dir)
def _validate_external_reference(self, base_dir, tpl_file, resource_file, def _validate_external_reference(self, tpl_file, resource_file,
raise_exc=True): raise_exc=True):
"""Verify that the external resource exists """Verify that the external resource exists
If resource_file is a URL verify that the URL is valid. If resource_file is a URL verify that the URL is valid.
If resource_file is a relative path verify that the path is valid If resource_file is a relative path verify that the path is valid
considering base_dir and tpl_file. considering base folder (self.temp_dir) and tpl_file.
Note that in a CSAR resource_file cannot be an absolute path. Note that in a CSAR resource_file cannot be an absolute path.
""" """
if UrlUtils.validate_url(resource_file): if UrlUtils.validate_url(resource_file):
@@ -235,7 +239,7 @@ class CSAR(object):
except Exception: except Exception:
raise URLException(what=msg) raise URLException(what=msg)
if os.path.isfile(os.path.join(base_dir, if os.path.isfile(os.path.join(self.temp_dir,
os.path.dirname(tpl_file), os.path.dirname(tpl_file),
resource_file)): resource_file)):
return return

40
toscaparser/tests/test_prereq.py
View File

@@ -11,6 +11,7 @@
# under the License. # under the License.
import os import os
import shutil
import zipfile import zipfile
from toscaparser.common.exception import URLException from toscaparser.common.exception import URLException
@@ -30,6 +31,8 @@ class CSARPrereqTest(TestCase):
csar = CSAR(path) csar = CSAR(path)
error = self.assertRaises(ValidationError, csar.validate) error = self.assertRaises(ValidationError, csar.validate)
self.assertEqual(_('%s does not exist.') % path, str(error)) self.assertEqual(_('%s does not exist.') % path, str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_file_is_zip(self): def test_file_is_zip(self):
path = os.path.join(self.base_path, "data/CSAR/csar_not_zip.zip") path = os.path.join(self.base_path, "data/CSAR/csar_not_zip.zip")
@@ -43,6 +46,8 @@ class CSARPrereqTest(TestCase):
csar = CSAR(path, False) csar = CSAR(path, False)
error = self.assertRaises(ValidationError, csar.validate) error = self.assertRaises(ValidationError, csar.validate)
self.assertEqual(_('%s is not a valid zip file.') % path, str(error)) self.assertEqual(_('%s is not a valid zip file.') % path, str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_metadata_file_exists(self): def test_metadata_file_exists(self):
path = os.path.join(self.base_path, path = os.path.join(self.base_path,
@@ -52,6 +57,8 @@ class CSARPrereqTest(TestCase):
self.assertEqual(_('%s is not a valid CSAR as it does not contain the ' self.assertEqual(_('%s is not a valid CSAR as it does not contain the '
'required file "TOSCA.meta" in the folder ' 'required file "TOSCA.meta" in the folder '
'"TOSCA-Metadata".') % path, str(error)) '"TOSCA-Metadata".') % path, str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_valid_metadata_file_exists(self): def test_valid_metadata_file_exists(self):
path = os.path.join(self.base_path, path = os.path.join(self.base_path,
@@ -61,6 +68,8 @@ class CSARPrereqTest(TestCase):
self.assertEqual(_('%s is not a valid CSAR as it does not contain the ' self.assertEqual(_('%s is not a valid CSAR as it does not contain the '
'required file "TOSCA.meta" in the folder ' 'required file "TOSCA.meta" in the folder '
'"TOSCA-Metadata".') % path, str(error)) '"TOSCA-Metadata".') % path, str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_metadata_is_yaml(self): def test_metadata_is_yaml(self):
path = os.path.join(self.base_path, path = os.path.join(self.base_path,
@@ -70,6 +79,8 @@ class CSARPrereqTest(TestCase):
self.assertEqual(_('The file "TOSCA-Metadata/TOSCA.meta" in the CSAR ' self.assertEqual(_('The file "TOSCA-Metadata/TOSCA.meta" in the CSAR '
'%s does not contain valid YAML content.') % path, '%s does not contain valid YAML content.') % path,
str(error)) str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_metadata_exists(self): def test_metadata_exists(self):
path = os.path.join(self.base_path, path = os.path.join(self.base_path,
@@ -79,6 +90,8 @@ class CSARPrereqTest(TestCase):
self.assertEqual(_('The CSAR %s is missing the required metadata ' self.assertEqual(_('The CSAR %s is missing the required metadata '
'"Entry-Definitions" in ' '"Entry-Definitions" in '
'"TOSCA-Metadata/TOSCA.meta".') % path, str(error)) '"TOSCA-Metadata/TOSCA.meta".') % path, str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_entry_def_exists(self): def test_entry_def_exists(self):
path = os.path.join(self.base_path, path = os.path.join(self.base_path,
@@ -87,6 +100,8 @@ class CSARPrereqTest(TestCase):
error = self.assertRaises(ValidationError, csar.validate) error = self.assertRaises(ValidationError, csar.validate)
self.assertEqual(_('The "Entry-Definitions" file defined in the CSAR ' self.assertEqual(_('The "Entry-Definitions" file defined in the CSAR '
'%s does not exist.') % path, str(error)) '%s does not exist.') % path, str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_csar_invalid_import_path(self): def test_csar_invalid_import_path(self):
path = os.path.join(self.base_path, path = os.path.join(self.base_path,
@@ -95,6 +110,8 @@ class CSARPrereqTest(TestCase):
error = self.assertRaises(ImportError, csar.validate) error = self.assertRaises(ImportError, csar.validate)
self.assertEqual(_('Import Definitions/wordpress.yaml is not valid'), self.assertEqual(_('Import Definitions/wordpress.yaml is not valid'),
str(error)) str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_csar_invalid_import_url(self): def test_csar_invalid_import_url(self):
path = os.path.join(self.base_path, path = os.path.join(self.base_path,
@@ -107,6 +124,8 @@ class CSARPrereqTest(TestCase):
'tosca_single_instance_wordpress/Definitions/' 'tosca_single_instance_wordpress/Definitions/'
'wordpress1.yaml. Reason is : Not Found".'), 'wordpress1.yaml. Reason is : Not Found".'),
str(error)) str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_csar_invalid_script_path(self): def test_csar_invalid_script_path(self):
path = os.path.join(self.base_path, path = os.path.join(self.base_path,
@@ -118,6 +137,8 @@ class CSARPrereqTest(TestCase):
'not exist.') or 'not exist.') or
str(error) == _('The resource Scripts/WordPress/configure.sh does ' str(error) == _('The resource Scripts/WordPress/configure.sh does '
'not exist.')) 'not exist.'))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_csar_invalid_script_url(self): def test_csar_invalid_script_url(self):
path = os.path.join(self.base_path, path = os.path.join(self.base_path,
@@ -130,17 +151,23 @@ class CSARPrereqTest(TestCase):
'tosca_single_instance_wordpress/Scripts/WordPress/' 'tosca_single_instance_wordpress/Scripts/WordPress/'
'install1.sh cannot be accessed".'), 'install1.sh cannot be accessed".'),
str(error)) str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_valid_csar(self): def test_valid_csar(self):
path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip") path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip")
csar = CSAR(path) csar = CSAR(path)
self.assertIsNone(csar.validate()) self.assertIsNone(csar.validate())
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_valid_csar_with_url_import_and_script(self): def test_valid_csar_with_url_import_and_script(self):
path = os.path.join(self.base_path, "data/CSAR/csar_wordpress_with_url" path = os.path.join(self.base_path, "data/CSAR/csar_wordpress_with_url"
"_import_and_script.zip") "_import_and_script.zip")
csar = CSAR(path) csar = CSAR(path)
self.assertIsNone(csar.validate()) self.assertIsNone(csar.validate())
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_metadata_invalid_csar(self): def test_metadata_invalid_csar(self):
path = os.path.join(self.base_path, path = os.path.join(self.base_path,
@@ -150,6 +177,8 @@ class CSARPrereqTest(TestCase):
self.assertEqual(_('The file "TOSCA-Metadata/TOSCA.meta" in the CSAR ' self.assertEqual(_('The file "TOSCA-Metadata/TOSCA.meta" in the CSAR '
'%s does not contain valid YAML content.') % path, '%s does not contain valid YAML content.') % path,
str(error)) str(error))
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_metadata_valid_csar(self): def test_metadata_valid_csar(self):
path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip") path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip")
@@ -167,6 +196,8 @@ class CSARPrereqTest(TestCase):
self.assertEqual('tosca_helloworld.yaml', csar.get_main_template()) self.assertEqual('tosca_helloworld.yaml', csar.get_main_template())
self.assertEqual('Template for deploying a single server with ' self.assertEqual('Template for deploying a single server with '
'predefined properties.', csar.get_description()) 'predefined properties.', csar.get_description())
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_csar_main_template(self): def test_csar_main_template(self):
path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip") path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip")
@@ -175,13 +206,18 @@ class CSARPrereqTest(TestCase):
"data/tosca_helloworld.yaml") "data/tosca_helloworld.yaml")
expected_yaml = toscaparser.utils.yamlparser.load_yaml(yaml_file) expected_yaml = toscaparser.utils.yamlparser.load_yaml(yaml_file)
self.assertEqual(expected_yaml, csar.get_main_template_yaml()) self.assertEqual(expected_yaml, csar.get_main_template_yaml())
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))
def test_decompress(self): def test_decompress(self):
path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip") path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip")
csar = CSAR(path) csar = CSAR(path)
tmp_dir = csar.decompress() csar.decompress()
zf = zipfile.ZipFile(path) zf = zipfile.ZipFile(path)
for name in zf.namelist(): for name in zf.namelist():
tmp_path = os.path.join(tmp_dir, name) tmp_path = os.path.join(csar.temp_dir, name)
self.assertTrue(os.path.isdir(tmp_path) or self.assertTrue(os.path.isdir(tmp_path) or
os.path.isfile(tmp_path)) os.path.isfile(tmp_path))
shutil.rmtree(csar.temp_dir)
self.assertTrue(csar.temp_dir is None or
not os.path.exists(csar.temp_dir))

4
toscaparser/tosca_template.py
View File

@@ -174,9 +174,9 @@ class ToscaTemplate(object):
# a CSAR archive # a CSAR archive
csar = CSAR(path, self.a_file) csar = CSAR(path, self.a_file)
csar.validate() csar.validate()
folder = csar.decompress() csar.decompress()
self.a_file = True # the file has been decompressed locally self.a_file = True # the file has been decompressed locally
return os.path.join(folder, csar.get_main_template()) return os.path.join(csar.temp_dir, csar.get_main_template())
else: else:
raise ValueError(_("%(path)s is not a valid file.") raise ValueError(_("%(path)s is not a valid file.")
% {'path': path}) % {'path': path})