1 # Licensed under the Apache License, Version 2.0 (the "License"); you may
2 # not use this file except in compliance with the License. You may obtain
3 # a copy of the License at
5 # http://www.apache.org/licenses/LICENSE-2.0
7 # Unless required by applicable law or agreed to in writing, software
8 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
9 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
10 # License for the specific language governing permissions and limitations
16 from toscaparser.common.exception import ExceptionCollector
17 from toscaparser.common.exception import MissingRequiredFieldError
18 from toscaparser.common.exception import UnknownFieldError
19 from toscaparser.common.exception import ValidationError
20 from toscaparser.elements.tosca_type_validation import TypeValidation
21 from toscaparser.utils.gettextutils import _
22 import toscaparser.utils.urlutils
23 import toscaparser.utils.yamlparser
25 YAML_LOADER = toscaparser.utils.yamlparser.load_yaml
26 log = logging.getLogger("tosca")
29 class ImportsLoader(object):
31 IMPORTS_SECTION = (FILE, REPOSITORY, NAMESPACE_URI, NAMESPACE_PREFIX) = \
32 ('file', 'repository', 'namespace_uri',
35 def __init__(self, importslist, path, type_definition_list=None,
37 self.importslist = importslist
39 if not path and not tpl:
40 msg = _('Input tosca template is not provided.')
42 ExceptionCollector.appendException(ValidationError(message=msg))
44 self.repositories = {}
45 if tpl and tpl.get('repositories'):
46 self.repositories = tpl.get('repositories')
47 self.type_definition_list = []
48 if type_definition_list:
49 if isinstance(type_definition_list, list):
50 self.type_definition_list = type_definition_list
52 self.type_definition_list.append(type_definition_list)
53 self._validate_and_load_imports()
55 def get_custom_defs(self):
56 return self.custom_defs
58 def _validate_and_load_imports(self):
61 if not self.importslist:
62 msg = _('"imports" keyname is defined without including '
65 ExceptionCollector.appendException(ValidationError(message=msg))
68 for import_def in self.importslist:
69 if isinstance(import_def, dict):
70 for import_name, import_uri in import_def.items():
71 if import_name in imports_names:
72 msg = (_('Duplicate import name "%s" was found.') %
75 ExceptionCollector.appendException(
76 ValidationError(message=msg))
77 imports_names.add(import_name)
79 custom_type = self._load_import_template(import_name,
81 namespace_prefix = None
82 if isinstance(import_uri, dict):
83 namespace_prefix = import_uri.get(
84 self.NAMESPACE_PREFIX)
86 TypeValidation(custom_type, import_def)
87 self._update_custom_def(custom_type, namespace_prefix)
88 else: # old style of imports
89 custom_type = self._load_import_template(None,
93 custom_type, import_def)
94 self._update_custom_def(custom_type, None)
96 def _update_custom_def(self, custom_type, namespace_prefix):
97 outer_custom_types = {}
98 for type_def in self.type_definition_list:
99 outer_custom_types = custom_type.get(type_def)
100 if outer_custom_types:
101 if type_def == "imports":
102 self.custom_defs.update({'imports': outer_custom_types})
105 prefix_custom_types = {}
106 for type_def_key in outer_custom_types.keys():
107 namespace_prefix_to_key = (namespace_prefix +
109 prefix_custom_types[namespace_prefix_to_key] = \
110 outer_custom_types[type_def_key]
111 self.custom_defs.update(prefix_custom_types)
113 self.custom_defs.update(outer_custom_types)
115 def _validate_import_keys(self, import_name, import_uri_def):
116 if self.FILE not in import_uri_def.keys():
117 log.warning(_('Missing keyname "file" in import "%(name)s".')
118 % {'name': import_name})
119 ExceptionCollector.appendException(
120 MissingRequiredFieldError(
121 what='Import of template "%s"' % import_name,
123 for key in import_uri_def.keys():
124 if key not in self.IMPORTS_SECTION:
125 log.warning(_('Unknown keyname "%(key)s" error in '
126 'imported definition "%(def)s".')
127 % {'key': key, 'def': import_name})
128 ExceptionCollector.appendException(
130 what='Import of template "%s"' % import_name,
133 def _load_import_template(self, import_name, import_uri_def):
134 """Handle custom types defined in imported template files
136 This method loads the custom type definitions referenced in "imports"
137 section of the TOSCA YAML template by determining whether each import
138 is specified via a file reference (by relative or absolute path) or a
142 +----------+--------+------------------------------+
143 | template | import | comment |
144 +----------+--------+------------------------------+
147 | preparsed| file | file must be a full path |
148 | preparsed| URL | OK |
149 | URL | file | file must be a relative path |
151 +----------+--------+------------------------------+
154 short_import_notation = False
155 if isinstance(import_uri_def, dict):
156 self._validate_import_keys(import_name, import_uri_def)
157 file_name = import_uri_def.get(self.FILE)
158 repository = import_uri_def.get(self.REPOSITORY)
160 file_name = import_uri_def
162 short_import_notation = True
165 msg = (_('A template file name is not provided with import '
166 'definition "%(import_name)s".')
167 % {'import_name': import_name})
169 ExceptionCollector.appendException(ValidationError(message=msg))
172 if toscaparser.utils.urlutils.UrlUtils.validate_url(file_name):
173 return YAML_LOADER(file_name, False)
175 import_template = None
177 if toscaparser.utils.urlutils.UrlUtils.validate_url(self.path):
178 if os.path.isabs(file_name):
179 msg = (_('Absolute file name "%(name)s" cannot be '
180 'used in a URL-based input template '
182 % {'name': file_name, 'template': self.path})
184 ExceptionCollector.appendException(ImportError(msg))
186 import_template = toscaparser.utils.urlutils.UrlUtils.\
187 join_url(self.path, file_name)
191 main_a_file = os.path.isfile(self.path)
194 if os.path.isfile(file_name):
195 import_template = file_name
197 full_path = os.path.join(
198 os.path.dirname(os.path.abspath(self.path)),
200 if os.path.isfile(full_path):
201 import_template = full_path
203 file_path = file_name.rpartition("/")
204 dir_path = os.path.dirname(os.path.abspath(
206 if file_path[0] != '' and dir_path.endswith(
208 import_template = dir_path + "/" +\
210 if not os.path.isfile(import_template):
211 msg = (_('"%(import_template)s" is'
213 % {'import_template':
216 ExceptionCollector.appendException
218 else: # template is pre-parsed
219 if os.path.isabs(file_name) and os.path.isfile(file_name):
221 import_template = file_name
223 msg = (_('Relative file name "%(name)s" cannot be used '
224 'in a pre-parsed input template.')
225 % {'name': file_name})
227 ExceptionCollector.appendException(ImportError(msg))
230 if not import_template:
231 log.error(_('Import "%(name)s" is not valid.') %
232 {'name': import_uri_def})
233 ExceptionCollector.appendException(
234 ImportError(_('Import "%s" is not valid.') %
237 return YAML_LOADER(import_template, a_file)
239 if short_import_notation:
240 log.error(_('Import "%(name)s" is not valid.') % import_uri_def)
241 ExceptionCollector.appendException(
242 ImportError(_('Import "%s" is not valid.') % import_uri_def))
247 if self.repositories:
248 for repo_name, repo_def in self.repositories.items():
249 if repo_name == repository:
250 # Remove leading, ending spaces and strip
251 # the last character if "/"
252 repo_url = ((repo_def['url']).strip()).rstrip("//")
253 full_url = repo_url + "/" + file_name
256 msg = (_('referenced repository "%(n_uri)s" in import '
257 'definition "%(tpl)s" not found.')
258 % {'n_uri': repository, 'tpl': import_name})
260 ExceptionCollector.appendException(ImportError(msg))
263 if toscaparser.utils.urlutils.UrlUtils.validate_url(full_url):
264 return YAML_LOADER(full_url, False)
266 msg = (_('repository url "%(n_uri)s" is not valid in import '
267 'definition "%(tpl)s".')
268 % {'n_uri': repo_url, 'tpl': import_name})
270 ExceptionCollector.appendException(ImportError(msg))