Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_merge(self):
# Config files are loaded and merged
unlink(self.temp)
conf = ChainConfig([
('a', PathConfig(self.a)),
('b', PathConfig(self.b))])
eq_(+conf, PathConfig(self.final))
def setUpModule():
# Test gramex.services.cache() as a pure function
info.folder = os.path.dirname(os.path.abspath(__file__))
info.config = gramex.config.PathConfig(os.path.join(info.folder, 'gramex.yaml'))
gramex.services.cache(info.config.cache)
def test_variables(self):
# Templates interpolate string variables
# Create configuration with 2 layers and a subdirectory import
conf = +ChainConfig(
base=PathConfig(self.chain.base),
child=PathConfig(self.chain.child),
)
# Custom variables are deleted after use
ok_('variables' not in conf)
for key in ['base', 'child', 'subdir']:
# {.} maps to YAML file's directory
eq_(conf['%s_DOT' % key], str(self.chain[key].parent))
# $YAMLPATH maps to YAML file's directory
eq_(conf['%s_YAMLPATH' % key], str(self.chain[key].parent))
# $YAMLURL is the relative path to YAML file's directory
eq_(conf['%s_YAMLURL' % key], conf['%s_YAMLURL_EXPECTED' % key])
# Environment variables are present by default
eq_(conf['%s_HOME' % key], os.environ.get('HOME', ''))
# Non-existent variables map to ''
eq_(conf['%s_NONEXISTENT' % key], os.environ.get('NONEXISTENT', ''))
# Custom variables are applied
eq_(conf['%s_THIS' % key], key)
def test_merge(self):
# Config files are loaded and merged
unlink(self.temp)
conf = ChainConfig([
('a', PathConfig(self.a)),
('b', PathConfig(self.b))])
eq_(+conf, PathConfig(self.final))
def test_import(self):
# Check if config files are imported
conf_imp = ChainConfig(conf=PathConfig(self.imp))
conf_b = ChainConfig(conf=PathConfig(self.b))
# When temp is missing, config matches b
unlink(self.temp)
eq_(+conf_imp, +conf_b)
# Once temp file is created, it is automatically imported
data = AttrDict(a=1, b=2)
with self.temp.open('w') as out:
yaml.dump(data, out)
result = +conf_b
result.update(data)
eq_(+conf_imp, result)
# Once removed, it no longer used
unlink(self.temp)
eq_(+conf_imp, +conf_b)
def test_variables(self):
# Templates interpolate string variables
# Create configuration with 2 layers and a subdirectory import
conf = +ChainConfig(
base=PathConfig(self.chain.base),
child=PathConfig(self.chain.child),
)
# Custom variables are deleted after use
ok_('variables' not in conf)
for key in ['base', 'child', 'subdir']:
# {.} maps to YAML file's directory
eq_(conf['%s_DOT' % key], str(self.chain[key].parent))
# $YAMLPATH maps to YAML file's directory
eq_(conf['%s_YAMLPATH' % key], str(self.chain[key].parent))
# $YAMLURL is the relative path to YAML file's directory
eq_(conf['%s_YAMLURL' % key], conf['%s_YAMLURL_EXPECTED' % key])
# Environment variables are present by default
eq_(conf['%s_HOME' % key], os.environ.get('HOME', ''))
# Non-existent variables map to ''
eq_(conf['%s_NONEXISTENT' % key], os.environ.get('NONEXISTENT', ''))
# Custom variables are applied
def test_update(self):
# Config files are updated on change
conf = ChainConfig(temp=PathConfig(self.temp))
# When the file is missing, config is empty
unlink(self.temp)
eq_(+conf, {})
# When the file is blank, config is empty
with self.temp.open('w') as out:
out.write(six.text_type(''))
eq_(+conf, {})
# Once created, it is automatically reloaded
data = AttrDict(a=1, b=2)
with self.temp.open('w') as out:
yaml.dump(data, out)
eq_(+conf, data)
Services are re-initialised if their configurations have changed. Service
callbacks are always re-run (even if the configuration hasn't changed.)
'''
# Reset variables
variables.clear()
variables.update(setup_variables())
# Initialise configuration layers with provided configurations
# AttrDicts are updated as-is. Paths are converted to PathConfig
paths.update(kwargs)
for key, val in paths.items():
if isinstance(val, Path):
if val.is_dir():
val = val / 'gramex.yaml'
val = PathConfig(val)
config_layers[key] = val
# Locate all config files
config_files = set()
for path_config in config_layers.values():
if hasattr(path_config, '__info__'):
for pathinfo in path_config.__info__.imports:
config_files.add(pathinfo.path)
config_files = list(config_files)
# Add config file folders to sys.path
sys.path[:] = _sys_path + [str(path.absolute().parent) for path in config_files]
from . import services
globals()['service'] = services.info # gramex.service = gramex.services.info
cached = _cache.get(key, None)
fstat = stat(path)
if cached is None or fstat != cached.get('stat'):
reloaded = True
if callable(callback):
data = callback(path, **kwargs)
elif callback_is_str:
method = None
if callback in _OPEN_CALLBACKS:
method = _OPEN_CALLBACKS[callback]
elif callback in {'json'}:
import json
method = opener(json.load)
elif callback in {'config'}:
from gramex.config import PathConfig
method = PathConfig
elif callback in {'xml', 'svg', 'rss', 'atom'}:
from lxml import etree
method = etree.parse
if method is not None:
data = method(path, **kwargs)
elif original_callback is None:
raise TypeError('gramex.cache.open: path "%s" has unknown extension' % path)
else:
raise TypeError('gramex.cache.open(callback="%s") is not a known type' % callback)
else:
raise TypeError('gramex.cache.open(callback=) must be a function, not %r' % callback)
if callable(transform):
data = transform(data)
_cache[key] = {'data': data, 'stat': fstat}
paths['source'] = Path(__file__).absolute().parent # Where gramex source code is
paths['base'] = Path('.') # Where gramex is run from
callbacks = {} # Services callbacks
# Populate __version__ from release.json
with (paths['source'] / 'release.json').open() as _release_file:
release = json.load(_release_file, object_pairs_hook=AttrDict)
__version__ = release.version
_sys_path = list(sys.path) # Preserve original sys.path
# List of URLs to warn about in case of duplicates
PathConfig.duplicate_warn = [
'url.*',
'cache.*',
'schedule.*',
'watch.*',
'email.*',
'alert.*',
'sms.*',
'log.loggers.*', 'log.handlers.*', 'log.formatters.*',
]
def parse_command_line(commands):
'''
Parse command line arguments. For example:
gramex cmd1 cmd2 --a=1 2 -b x --c --p.q=4