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 self.nested_topo_tpls = []
40 if not path and not tpl:
41 msg = _('Input tosca template is not provided.')
43 ExceptionCollector.appendException(ValidationError(message=msg))
45 self.repositories = {}
46 if tpl and tpl.get('repositories'):
47 self.repositories = tpl.get('repositories')
48 self.type_definition_list = []
49 if type_definition_list:
50 if isinstance(type_definition_list, list):
51 self.type_definition_list = type_definition_list
53 self.type_definition_list.append(type_definition_list)
54 self._validate_and_load_imports()
56 def get_custom_defs(self):
57 return self.custom_defs
59 def get_nested_topo_tpls(self):
60 return self.nested_topo_tpls
62 def _validate_and_load_imports(self):
65 if not self.importslist:
66 msg = _('"imports" keyname is defined without including '
69 ExceptionCollector.appendException(ValidationError(message=msg))
72 for import_def in self.importslist:
73 if isinstance(import_def, dict):
74 for import_name, import_uri in import_def.items():
75 if import_name in imports_names:
76 msg = (_('Duplicate import name "%s" was found.') %
79 ExceptionCollector.appendException(
80 ValidationError(message=msg))
81 imports_names.add(import_name)
83 full_file_name, custom_type = self._load_import_template(
84 import_name, import_uri)
85 namespace_prefix = None
86 if isinstance(import_uri, dict):
87 namespace_prefix = import_uri.get(
88 self.NAMESPACE_PREFIX)
90 TypeValidation(custom_type, import_def)
91 self._update_custom_def(custom_type, namespace_prefix)
92 else: # old style of imports
93 full_file_name, custom_type = self._load_import_template(
97 custom_type, import_def)
98 self._update_custom_def(custom_type, None)
100 self._update_nested_topo_tpls(full_file_name, custom_type)
102 def _update_custom_def(self, custom_type, namespace_prefix):
103 outer_custom_types = {}
104 for type_def in self.type_definition_list:
105 outer_custom_types = custom_type.get(type_def)
106 if outer_custom_types:
107 if type_def == "imports":
108 self.custom_defs.update({'imports': outer_custom_types})
111 prefix_custom_types = {}
112 for type_def_key in outer_custom_types.keys():
113 namespace_prefix_to_key = (namespace_prefix +
115 prefix_custom_types[namespace_prefix_to_key] = \
116 outer_custom_types[type_def_key]
117 self.custom_defs.update(prefix_custom_types)
119 self.custom_defs.update(outer_custom_types)
121 def _update_nested_topo_tpls(self, full_file_name, custom_tpl):
122 if full_file_name and custom_tpl:
123 topo_tpl = {full_file_name: custom_tpl}
124 self.nested_topo_tpls.append(topo_tpl)
126 def _validate_import_keys(self, import_name, import_uri_def):
127 if self.FILE not in import_uri_def.keys():
128 log.warning(_('Missing keyname "file" in import "%(name)s".')
129 % {'name': import_name})
130 ExceptionCollector.appendException(
131 MissingRequiredFieldError(
132 what='Import of template "%s"' % import_name,
134 for key in import_uri_def.keys():
135 if key not in self.IMPORTS_SECTION:
136 log.warning(_('Unknown keyname "%(key)s" error in '
137 'imported definition "%(def)s".')
138 % {'key': key, 'def': import_name})
139 ExceptionCollector.appendException(
141 what='Import of template "%s"' % import_name,
144 def _load_import_template(self, import_name, import_uri_def):
145 """Handle custom types defined in imported template files
147 This method loads the custom type definitions referenced in "imports"
148 section of the TOSCA YAML template by determining whether each import
149 is specified via a file reference (by relative or absolute path) or a
153 +----------+--------+------------------------------+
154 | template | import | comment |
155 +----------+--------+------------------------------+
158 | preparsed| file | file must be a full path |
159 | preparsed| URL | OK |
160 | URL | file | file must be a relative path |
162 +----------+--------+------------------------------+
165 short_import_notation = False
166 if isinstance(import_uri_def, dict):
167 self._validate_import_keys(import_name, import_uri_def)
168 file_name = import_uri_def.get(self.FILE)
169 repository = import_uri_def.get(self.REPOSITORY)
171 file_name = import_uri_def
173 short_import_notation = True
176 msg = (_('A template file name is not provided with import '
177 'definition "%(import_name)s".')
178 % {'import_name': import_name})
180 ExceptionCollector.appendException(ValidationError(message=msg))
183 if toscaparser.utils.urlutils.UrlUtils.validate_url(file_name):
184 return file_name, YAML_LOADER(file_name, False)
186 import_template = None
188 if toscaparser.utils.urlutils.UrlUtils.validate_url(self.path):
189 if os.path.isabs(file_name):
190 msg = (_('Absolute file name "%(name)s" cannot be '
191 'used in a URL-based input template '
193 % {'name': file_name, 'template': self.path})
195 ExceptionCollector.appendException(ImportError(msg))
197 import_template = toscaparser.utils.urlutils.UrlUtils.\
198 join_url(self.path, file_name)
202 main_a_file = os.path.isfile(self.path)
205 if os.path.isfile(file_name):
206 import_template = file_name
208 full_path = os.path.join(
209 os.path.dirname(os.path.abspath(self.path)),
211 if os.path.isfile(full_path):
212 import_template = full_path
214 file_path = file_name.rpartition("/")
215 dir_path = os.path.dirname(os.path.abspath(
217 if file_path[0] != '' and dir_path.endswith(
219 import_template = dir_path + "/" +\
221 if not os.path.isfile(import_template):
222 msg = (_('"%(import_template)s" is'
224 % {'import_template':
227 ExceptionCollector.appendException
229 else: # template is pre-parsed
230 if os.path.isabs(file_name) and os.path.isfile(file_name):
232 import_template = file_name
234 msg = (_('Relative file name "%(name)s" cannot be used '
235 'in a pre-parsed input template.')
236 % {'name': file_name})
238 ExceptionCollector.appendException(ImportError(msg))
241 if not import_template:
242 log.error(_('Import "%(name)s" is not valid.') %
243 {'name': import_uri_def})
244 ExceptionCollector.appendException(
245 ImportError(_('Import "%s" is not valid.') %
248 return import_template, YAML_LOADER(import_template, a_file)
250 if short_import_notation:
251 log.error(_('Import "%(name)s" is not valid.') % import_uri_def)
252 ExceptionCollector.appendException(
253 ImportError(_('Import "%s" is not valid.') % import_uri_def))
258 if self.repositories:
259 for repo_name, repo_def in self.repositories.items():
260 if repo_name == repository:
261 # Remove leading, ending spaces and strip
262 # the last character if "/"
263 repo_url = ((repo_def['url']).strip()).rstrip("//")
264 full_url = repo_url + "/" + file_name
267 msg = (_('referenced repository "%(n_uri)s" in import '
268 'definition "%(tpl)s" not found.')
269 % {'n_uri': repository, 'tpl': import_name})
271 ExceptionCollector.appendException(ImportError(msg))
274 if toscaparser.utils.urlutils.UrlUtils.validate_url(full_url):
275 return full_url, YAML_LOADER(full_url, False)
277 msg = (_('repository url "%(n_uri)s" is not valid in import '
278 'definition "%(tpl)s".')
279 % {'n_uri': repo_url, 'tpl': import_name})
281 ExceptionCollector.appendException(ImportError(msg))