diff --git a/toscaparser/prereq/csar.py b/toscaparser/prereq/csar.py index 19479f0f..b1aa89ef 100644 --- a/toscaparser/prereq/csar.py +++ b/toscaparser/prereq/csar.py @@ -12,6 +12,7 @@ import os.path import requests +import shutil import six import tempfile import yaml @@ -36,6 +37,7 @@ class CSAR(object): self.a_file = a_file self.is_validated = False self.csar = None + self.temp_dir = None def validate(self): """Validate the provided CSAR file.""" @@ -152,10 +154,9 @@ class CSAR(object): def decompress(self): if not self.is_validated: self.validate() - folder = tempfile.NamedTemporaryFile().name + self.temp_dir = tempfile.NamedTemporaryFile().name with zipfile.ZipFile(self.csar, "r") as zf: - zf.extractall(folder) - return folder + zf.extractall(self.temp_dir) def _validate_external_references(self): """Extracts files referenced in the main template @@ -165,64 +166,67 @@ class CSAR(object): * interface implementations * artifacts """ - temp_dir = self.decompress() - main_tpl_file = self.get_main_template() - main_tpl = self.get_main_template_yaml() + try: + self.decompress() + main_tpl_file = self.get_main_template() + main_tpl = self.get_main_template_yaml() - if 'imports' in main_tpl: - ImportsLoader(main_tpl['imports'], - os.path.join(temp_dir, main_tpl_file)) + if 'imports' in main_tpl: + ImportsLoader(main_tpl['imports'], + os.path.join(self.temp_dir, main_tpl_file)) - if 'topology_template' not in main_tpl: - return - topology_template = main_tpl['topology_template'] + if 'topology_template' in main_tpl: + topology_template = main_tpl['topology_template'] - if 'node_templates' not in topology_template: - return - node_templates = topology_template['node_templates'] + if 'node_templates' in topology_template: + node_templates = topology_template['node_templates'] - for node_template_key in node_templates: - node_template = node_templates[node_template_key] - if 'artifacts' in node_template: - artifacts = node_template['artifacts'] - for artifact_key in artifacts: - artifact = artifacts[artifact_key] - if isinstance(artifact, six.string_types): - self._validate_external_reference(temp_dir, - main_tpl_file, - artifact) - elif isinstance(artifact, dict): - if 'file' in artifact: - self._validate_external_reference(temp_dir, - main_tpl_file, - artifact['file']) - else: - raise ValueError(_('Unexpected artifact definition ' - 'for %s.') % artifact_key) - if 'interfaces' in node_template: - interfaces = node_template['interfaces'] - for interface_key in interfaces: - interface = interfaces[interface_key] - for opertation_key in interface: - operation = interface[opertation_key] - if isinstance(operation, six.string_types): - self._validate_external_reference(temp_dir, - main_tpl_file, - operation, - False) - elif isinstance(operation, dict): - if 'implementation' in operation: - self._validate_external_reference( - temp_dir, main_tpl_file, - operation['implementation']) + for node_template_key in node_templates: + node_template = node_templates[node_template_key] + if 'artifacts' in node_template: + artifacts = node_template['artifacts'] + for artifact_key in artifacts: + artifact = artifacts[artifact_key] + if isinstance(artifact, six.string_types): + self._validate_external_reference( + main_tpl_file, + artifact) + elif isinstance(artifact, dict): + if 'file' in artifact: + self._validate_external_reference( + main_tpl_file, + artifact['file']) + else: + raise ValueError( + _('Unexpected artifact definition ' + 'for %s.') % artifact_key) + if 'interfaces' in node_template: + interfaces = node_template['interfaces'] + for interface_key in interfaces: + interface = interfaces[interface_key] + for opertation_key in interface: + operation = interface[opertation_key] + if isinstance(operation, six.string_types): + self._validate_external_reference( + main_tpl_file, + operation, + False) + elif isinstance(operation, dict): + if 'implementation' in operation: + self._validate_external_reference( + 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): """Verify that the external resource exists 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 - 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. """ if UrlUtils.validate_url(resource_file): @@ -235,7 +239,7 @@ class CSAR(object): except Exception: 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), resource_file)): return diff --git a/toscaparser/tests/test_prereq.py b/toscaparser/tests/test_prereq.py index fecaa5a6..75a5ef42 100644 --- a/toscaparser/tests/test_prereq.py +++ b/toscaparser/tests/test_prereq.py @@ -11,6 +11,7 @@ # under the License. import os +import shutil import zipfile from toscaparser.common.exception import URLException @@ -30,6 +31,8 @@ class CSARPrereqTest(TestCase): csar = CSAR(path) error = self.assertRaises(ValidationError, csar.validate) 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): path = os.path.join(self.base_path, "data/CSAR/csar_not_zip.zip") @@ -43,6 +46,8 @@ class CSARPrereqTest(TestCase): csar = CSAR(path, False) error = self.assertRaises(ValidationError, csar.validate) 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): 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 ' 'required file "TOSCA.meta" in the folder ' '"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): 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 ' 'required file "TOSCA.meta" in the folder ' '"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): 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 ' '%s does not contain valid YAML content.') % path, str(error)) + self.assertTrue(csar.temp_dir is None or + not os.path.exists(csar.temp_dir)) def test_metadata_exists(self): path = os.path.join(self.base_path, @@ -79,6 +90,8 @@ class CSARPrereqTest(TestCase): self.assertEqual(_('The CSAR %s is missing the required metadata ' '"Entry-Definitions" in ' '"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): path = os.path.join(self.base_path, @@ -87,6 +100,8 @@ class CSARPrereqTest(TestCase): error = self.assertRaises(ValidationError, csar.validate) self.assertEqual(_('The "Entry-Definitions" file defined in the CSAR ' '%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): path = os.path.join(self.base_path, @@ -95,6 +110,8 @@ class CSARPrereqTest(TestCase): error = self.assertRaises(ImportError, csar.validate) self.assertEqual(_('Import Definitions/wordpress.yaml is not valid'), str(error)) + self.assertTrue(csar.temp_dir is None or + not os.path.exists(csar.temp_dir)) def test_csar_invalid_import_url(self): path = os.path.join(self.base_path, @@ -107,6 +124,8 @@ class CSARPrereqTest(TestCase): 'tosca_single_instance_wordpress/Definitions/' 'wordpress1.yaml. Reason is : Not Found".'), str(error)) + self.assertTrue(csar.temp_dir is None or + not os.path.exists(csar.temp_dir)) def test_csar_invalid_script_path(self): path = os.path.join(self.base_path, @@ -118,6 +137,8 @@ class CSARPrereqTest(TestCase): 'not exist.') or str(error) == _('The resource Scripts/WordPress/configure.sh does ' 'not exist.')) + self.assertTrue(csar.temp_dir is None or + not os.path.exists(csar.temp_dir)) def test_csar_invalid_script_url(self): path = os.path.join(self.base_path, @@ -130,17 +151,23 @@ class CSARPrereqTest(TestCase): 'tosca_single_instance_wordpress/Scripts/WordPress/' 'install1.sh cannot be accessed".'), str(error)) + self.assertTrue(csar.temp_dir is None or + not os.path.exists(csar.temp_dir)) def test_valid_csar(self): path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip") csar = CSAR(path) 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): path = os.path.join(self.base_path, "data/CSAR/csar_wordpress_with_url" "_import_and_script.zip") csar = CSAR(path) 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): 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 ' '%s does not contain valid YAML content.') % path, str(error)) + self.assertTrue(csar.temp_dir is None or + not os.path.exists(csar.temp_dir)) def test_metadata_valid_csar(self): 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('Template for deploying a single server with ' '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): path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip") @@ -175,13 +206,18 @@ class CSARPrereqTest(TestCase): "data/tosca_helloworld.yaml") expected_yaml = toscaparser.utils.yamlparser.load_yaml(yaml_file) 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): path = os.path.join(self.base_path, "data/CSAR/csar_hello_world.zip") csar = CSAR(path) - tmp_dir = csar.decompress() + csar.decompress() zf = zipfile.ZipFile(path) 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 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)) diff --git a/toscaparser/tosca_template.py b/toscaparser/tosca_template.py index eeb380ca..d89b9650 100644 --- a/toscaparser/tosca_template.py +++ b/toscaparser/tosca_template.py @@ -174,9 +174,9 @@ class ToscaTemplate(object): # a CSAR archive csar = CSAR(path, self.a_file) csar.validate() - folder = csar.decompress() + csar.decompress() 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: raise ValueError(_("%(path)s is not a valid file.") % {'path': path})