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 InvalidPropertyValueError
18 from toscaparser.common.exception import MissingRequiredFieldError
19 from toscaparser.common.exception import UnknownFieldError
20 from toscaparser.common.exception import ValidationError
21 from toscaparser.elements.tosca_type_validation import TypeValidation
22 from toscaparser.utils.gettextutils import _
23 import toscaparser.utils.urlutils
24 import toscaparser.utils.yamlparser
26 YAML_LOADER = toscaparser.utils.yamlparser.load_yaml
27 log = logging.getLogger("tosca")
30 class ImportsLoader(object):
32 IMPORTS_SECTION = (FILE, REPOSITORY, NAMESPACE_URI, NAMESPACE_PREFIX) = \
33 ('file', 'repository', 'namespace_uri',
36 def __init__(self, importslist, path, type_definition_list=None,
38 self.importslist = importslist
40 self.nested_tosca_tpls = []
41 if not path and not tpl:
42 msg = _('Input tosca template is not provided.')
44 ExceptionCollector.appendException(ValidationError(message=msg))
46 self.repositories = {}
47 if tpl and tpl.get('repositories'):
48 self.repositories = tpl.get('repositories')
49 self.type_definition_list = []
50 if type_definition_list:
51 if isinstance(type_definition_list, list):
52 self.type_definition_list = type_definition_list
54 self.type_definition_list.append(type_definition_list)
55 self._validate_and_load_imports()
57 def get_custom_defs(self):
58 return self.custom_defs
60 def get_nested_tosca_tpls(self):
61 return self.nested_tosca_tpls
63 def _validate_and_load_imports(self):
66 if not self.importslist:
67 msg = _('"imports" keyname is defined without including '
70 ExceptionCollector.appendException(ValidationError(message=msg))
73 for import_def in self.importslist:
74 if isinstance(import_def, dict):
75 for import_name, import_uri in import_def.items():
76 if import_name in imports_names:
77 msg = (_('Duplicate import name "%s" was found.') %
80 ExceptionCollector.appendException(
81 ValidationError(message=msg))
82 imports_names.add(import_name)
84 full_file_name, custom_type = self._load_import_template(
85 import_name, import_uri)
86 namespace_prefix = None
87 if isinstance(import_uri, dict):
88 namespace_prefix = import_uri.get(
89 self.NAMESPACE_PREFIX)
91 TypeValidation(custom_type, import_def)
92 self._update_custom_def(custom_type, namespace_prefix)
93 else: # old style of imports
94 full_file_name, custom_type = self._load_import_template(
98 custom_type, import_def)
99 self._update_custom_def(custom_type, None)
101 self._update_nested_tosca_tpls(full_file_name, custom_type)
103 def _update_custom_def(self, custom_type, namespace_prefix):
104 outer_custom_types = {}
105 for type_def in self.type_definition_list:
106 outer_custom_types = custom_type.get(type_def)
107 if outer_custom_types:
108 if type_def == "imports":
109 for i in self.custom_defs.get('imports', []):
110 if i not in outer_custom_types:
111 outer_custom_types.append(i)
112 self.custom_defs.update({'imports': outer_custom_types})
115 prefix_custom_types = {}
116 for type_def_key in outer_custom_types.keys():
117 namespace_prefix_to_key = (namespace_prefix +
119 prefix_custom_types[namespace_prefix_to_key] = \
120 outer_custom_types[type_def_key]
121 self.custom_defs.update(prefix_custom_types)
123 self.custom_defs.update(outer_custom_types)
125 def _update_nested_tosca_tpls(self, full_file_name, custom_tpl):
126 if full_file_name and custom_tpl:
127 topo_tpl = {full_file_name: custom_tpl}
128 self.nested_tosca_tpls.append(topo_tpl)
130 def _validate_import_keys(self, import_name, import_uri_def):
131 if self.FILE not in import_uri_def.keys():
132 log.warning(_('Missing keyname "file" in import "%(name)s".')
133 % {'name': import_name})
134 ExceptionCollector.appendException(
135 MissingRequiredFieldError(
136 what='Import of template "%s"' % import_name,
138 for key in import_uri_def.keys():
139 if key not in self.IMPORTS_SECTION:
140 log.warning(_('Unknown keyname "%(key)s" error in '
141 'imported definition "%(def)s".')
142 % {'key': key, 'def': import_name})
143 ExceptionCollector.appendException(
145 what='Import of template "%s"' % import_name,
148 def _load_import_template(self, import_name, import_uri_def):
149 """Handle custom types defined in imported template files
151 This method loads the custom type definitions referenced in "imports"
152 section of the TOSCA YAML template by determining whether each import
153 is specified via a file reference (by relative or absolute path) or a
157 +----------+--------+------------------------------+
158 | template | import | comment |
159 +----------+--------+------------------------------+
162 | preparsed| file | file must be a full path |
163 | preparsed| URL | OK |
164 | URL | file | file must be a relative path |
166 +----------+--------+------------------------------+
168 short_import_notation = False
169 if isinstance(import_uri_def, dict):
170 self._validate_import_keys(import_name, import_uri_def)
171 file_name = import_uri_def.get(self.FILE)
172 repository = import_uri_def.get(self.REPOSITORY)
173 repos = self.repositories.keys()
174 if repository is not None:
175 if repository not in repos:
176 ExceptionCollector.appendException(
177 InvalidPropertyValueError(
178 what=_('Repository is not found in "%s"') % repos))
180 file_name = import_uri_def
182 short_import_notation = True
185 msg = (_('A template file name is not provided with import '
186 'definition "%(import_name)s".')
187 % {'import_name': import_name})
189 ExceptionCollector.appendException(ValidationError(message=msg))
192 if toscaparser.utils.urlutils.UrlUtils.validate_url(file_name):
193 return file_name, YAML_LOADER(file_name, False)
195 import_template = None
197 if toscaparser.utils.urlutils.UrlUtils.validate_url(self.path):
198 if os.path.isabs(file_name):
199 msg = (_('Absolute file name "%(name)s" cannot be '
200 'used in a URL-based input template '
202 % {'name': file_name, 'template': self.path})
204 ExceptionCollector.appendException(ImportError(msg))
206 import_template = toscaparser.utils.urlutils.UrlUtils.\
207 join_url(self.path, file_name)
211 main_a_file = os.path.isfile(self.path)
214 if os.path.isfile(file_name):
215 import_template = file_name
217 full_path = os.path.join(
218 os.path.dirname(os.path.abspath(self.path)),
220 if os.path.isfile(full_path):
221 import_template = full_path
223 file_path = file_name.rpartition("/")
224 dir_path = os.path.dirname(os.path.abspath(
226 if file_path[0] != '' and dir_path.endswith(
228 import_template = dir_path + "/" +\
230 if not os.path.isfile(import_template):
231 msg = (_('"%(import_template)s" is'
233 % {'import_template':
236 ExceptionCollector.appendException
238 else: # template is pre-parsed
239 if os.path.isabs(file_name) and os.path.isfile(file_name):
241 import_template = file_name
243 msg = (_('Relative file name "%(name)s" cannot be used '
244 'in a pre-parsed input template.')
245 % {'name': file_name})
247 ExceptionCollector.appendException(ImportError(msg))
250 if not import_template:
251 log.error(_('Import "%(name)s" is not valid.') %
252 {'name': import_uri_def})
253 ExceptionCollector.appendException(
254 ImportError(_('Import "%s" is not valid.') %
257 return import_template, YAML_LOADER(import_template, a_file)
259 if short_import_notation:
260 log.error(_('Import "%(name)s" is not valid.') % import_uri_def)
261 ExceptionCollector.appendException(
262 ImportError(_('Import "%s" is not valid.') % import_uri_def))
267 if self.repositories:
268 for repo_name, repo_def in self.repositories.items():
269 if repo_name == repository:
270 # Remove leading, ending spaces and strip
271 # the last character if "/"
272 repo_url = ((repo_def['url']).strip()).rstrip("//")
273 full_url = repo_url + "/" + file_name
276 msg = (_('referenced repository "%(n_uri)s" in import '
277 'definition "%(tpl)s" not found.')
278 % {'n_uri': repository, 'tpl': import_name})
280 ExceptionCollector.appendException(ImportError(msg))
283 if toscaparser.utils.urlutils.UrlUtils.validate_url(full_url):
284 return full_url, YAML_LOADER(full_url, False)
286 msg = (_('repository url "%(n_uri)s" is not valid in import '
287 'definition "%(tpl)s".')
288 % {'n_uri': repo_url, 'tpl': import_name})
290 ExceptionCollector.appendException(ImportError(msg))