Source code for robot.variables.filesetter

#  Copyright 2008-2015 Nokia Networks
#  Copyright 2016-     Robot Framework Foundation
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.

import inspect
import io
import json
try:
    import yaml
except ImportError:
    yaml = None

from robot.errors import DataError
from robot.output import LOGGER
from robot.utils import (get_error_message, is_dict_like, is_list_like,
                         is_string, seq2str2, type_name, DotDict, Importer)


[docs]class VariableFileSetter: def __init__(self, store): self._store = store
[docs] def set(self, path_or_variables, args=None, overwrite=False): variables = self._import_if_needed(path_or_variables, args) self._set(variables, overwrite) return variables
def _import_if_needed(self, path_or_variables, args=None): if not is_string(path_or_variables): return path_or_variables LOGGER.info("Importing variable file '%s' with args %s" % (path_or_variables, args)) if path_or_variables.lower().endswith(('.yaml', '.yml')): importer = YamlImporter() elif path_or_variables.lower().endswith('.json'): importer = JsonImporter() else: importer = PythonImporter() try: return importer.import_variables(path_or_variables, args) except: args = 'with arguments %s ' % seq2str2(args) if args else '' raise DataError("Processing variable file '%s' %sfailed: %s" % (path_or_variables, args, get_error_message())) def _set(self, variables, overwrite=False): for name, value in variables: self._store.add(name, value, overwrite)
[docs]class YamlImporter:
[docs] def import_variables(self, path, args=None): if args: raise DataError('YAML variable files do not accept arguments.') variables = self._import(path) return [('${%s}' % name, self._dot_dict(value)) for name, value in variables]
def _import(self, path): with io.open(path, encoding='UTF-8') as stream: variables = self._load_yaml(stream) if not is_dict_like(variables): raise DataError('YAML variable file must be a mapping, got %s.' % type_name(variables)) return variables.items() def _load_yaml(self, stream): if not yaml: raise DataError('Using YAML variable files requires PyYAML module ' 'to be installed. Typically you can install it ' 'by running `pip install pyyaml`.') if yaml.__version__.split('.')[0] == '3': return yaml.load(stream) return yaml.full_load(stream) def _dot_dict(self, value): if is_dict_like(value): return DotDict((k, self._dot_dict(v)) for k, v in value.items()) if is_list_like(value): return [self._dot_dict(v) for v in value] return value
[docs]class PythonImporter:
[docs] def import_variables(self, path, args=None): importer = Importer('variable file', LOGGER).import_class_or_module var_file = importer(path, instantiate_with_args=()) return self._get_variables(var_file, args)
def _get_variables(self, var_file, args): if self._is_dynamic(var_file): variables = self._get_dynamic(var_file, args) else: variables = self._get_static(var_file) return list(self._decorate_and_validate(variables)) def _is_dynamic(self, var_file): return (hasattr(var_file, 'get_variables') or hasattr(var_file, 'getVariables')) def _get_dynamic(self, var_file, args): get_variables = (getattr(var_file, 'get_variables', None) or getattr(var_file, 'getVariables')) variables = get_variables(*args) if is_dict_like(variables): return variables.items() raise DataError("Expected '%s' to return dict-like value, got %s." % (get_variables.__name__, type_name(variables))) def _get_static(self, var_file): names = [attr for attr in dir(var_file) if not attr.startswith('_')] if hasattr(var_file, '__all__'): names = [name for name in names if name in var_file.__all__] variables = [(name, getattr(var_file, name)) for name in names] if not inspect.ismodule(var_file): variables = [(n, v) for n, v in variables if not callable(v)] return variables def _decorate_and_validate(self, variables): for name, value in variables: name = self._decorate(name) self._validate(name, value) yield name, value def _decorate(self, name): if name.startswith('LIST__'): return '@{%s}' % name[6:] if name.startswith('DICT__'): return '&{%s}' % name[6:] return '${%s}' % name def _validate(self, name, value): if name[0] == '@' and not is_list_like(value): raise DataError("Invalid variable '%s': Expected list-like value, " "got %s." % (name, type_name(value))) if name[0] == '&' and not is_dict_like(value): raise DataError("Invalid variable '%s': Expected dict-like value, " "got %s." % (name, type_name(value)))
[docs]class JsonImporter:
[docs] def import_variables(self, path, args=None): if args: raise DataError('JSON variable files do not accept arguments.') variables = self._import(path) return [('${%s}' % name, self._dot_dict(value)) for name, value in variables]
def _import(self, path): with io.open(path, encoding='UTF-8') as stream: variables = json.load(stream) if not is_dict_like(variables): raise DataError('JSON variable file must be a mapping, got %s.' % type_name(variables)) return variables.items() def _dot_dict(self, value): if is_dict_like(value): return DotDict((k, self._dot_dict(v)) for k, v in value.items()) if is_list_like(value): return [self._dot_dict(v) for v in value] return value