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_format_results(self):
"""Test format_results."""
test_dict = {
'nested': {
'bool': True,
'nested_key': 'nested_value'
},
'test_key': 'test_value'
}
mute_map = MutableMap(**test_dict.copy())
assert LookupHandler.format_results(test_dict) == test_dict
assert LookupHandler.format_results(mute_map) == test_dict
assert LookupHandler.format_results(
test_dict, get='test_key'
) == test_dict['test_key']
assert LookupHandler.format_results(
mute_map, get='test_key'
) == mute_map.test_key
assert LookupHandler.format_results(
mute_map, get='nested'
) == mute_map.nested.data
assert LookupHandler.format_results(
mute_map, get='nested.nested_key'
) == mute_map.nested.nested_key
def test_bool(self):
"""Validates the bool value.
Also tests setting an attr using bracket notation.
"""
mute_map = MutableMap()
assert not mute_map
mute_map['str_val'] = 'test'
assert mute_map
def test_pre_process_resolve(self, yaml_fixtures):
"""Test that pre-process resolution only resolves specific vars."""
raw_config = deepcopy(yaml_fixtures['config.runway.yml']['deployments'])
raw_vars = deepcopy(yaml_fixtures['config.runway.variables.yml'])
deployment = DeploymentDefinition.from_list(raw_config)[0]
raw_context = {'env_vars': os.environ.copy()}
raw_context['env_vars'].update(ENV_VARS)
deployment.resolve(MutableMap(**raw_context),
variables=MutableMap(**raw_vars),
pre_process=True)
# check resolved variables for pre_process
assert not deployment.account_id == 123456789101
assert deployment.assume_role['arn'] == 'arn:aws:iam::role/some-role'
assert deployment.env_vars == {'MY_USERNAME': 'test'}
assert deployment.regions == ['us-east-1']
assert not deployment.parallel_regions, 'not set in test config, should be None'
# these should be unresolved at this point
for attr in ['environments', 'module_options']:
with pytest.raises(UnresolvedVariable):
getattr(deployment, attr)
def test_update_termination_protection(self):
"""Test update_termination_protection."""
stack_name = 'fake-stack'
test_cases = [
MutableMap(aws=False, defined=True, expected=True),
MutableMap(aws=True, defined=False, expected=False),
MutableMap(aws=True, defined=True, expected=None),
MutableMap(aws=False, defined=False, expected=None)
]
for test in test_cases:
self.stubber.add_response(
'describe_stacks',
{'Stacks': [
generate_describe_stacks_stack(
stack_name, termination_protection=test.aws
)
]},
{'StackName': stack_name}
)
if isinstance(test.expected, bool):
type_str=mod.pop('type', None),
environments=mod.pop('environments', {}),
env_vars=mod.pop('env_vars', {}),
options=mod.pop('options', {}),
parameters=mod.pop('parameters', {}),
tags=mod.pop('tags', {}),
child_modules=child_modules))
if mod:
LOGGER.warning(
'Invalid keys found in module %s have been ignored: %s',
name, ', '.join(mod.keys())
)
return results
class FutureDefinition(MutableMap):
"""Opt-in to future functionality before the next major release.
Availability of these toggles will be removed at each major release as
the functionality will then be considered current.
Lookups are not supported as these values should be static.
"""
def __init__(self, strict_environments=False, **kwargs):
# type: (bool, bool) -> None
"""Instantiate class.
Keyword Args:
strict_environments (bool): Wether to enable strict environments.
from six import string_types
import yaml
from .util import MutableMap, cached_property
from .variables import Variable
# python2 supported pylint sees this is cyclic even though its only for type checking
# pylint: disable=cyclic-import
if TYPE_CHECKING:
from .context import Context # noqa: F401 pylint: disable=unused-import
LOGGER = logging.getLogger('runway')
NoneType = type(None)
class ConfigComponent(MutableMap):
"""Base class for Runway config components.
Attributes:
SUPPORTS_VARIABLES: A list of directives that support the use of
variable.
PRE_PROCESS_VARIABLES: A list of directives that support the use of
variables and needs to be resolved before the component is
processed.
"""
SUPPORTS_VARIABLES = ['env_vars', 'environments', 'parameters'] # type: List[str]
PRE_PROCESS_VARIABLES = [] # type: List[str]
@property
def data(self):
for index, test in enumerate(tests):
name = test.pop('name', 'test_{}'.format(index + 1))
results.append(cls(name, test.pop('type'),
test.pop('args', {}),
test.pop('required', False)))
if test:
LOGGER.warning(
'Invalid keys found in test %s have been ignored: %s',
name, ', '.join(test.keys())
)
return results
class VariablesDefinition(MutableMap):
"""A variable definitions for the Runway config file.
Runway variables are used to fill values that could change based on any
number of circumstances. They can also be used to simplify the Runway config
file by pulling lengthy definitions into another file. Variables can be used
in the config file by providing the `var lookup`_ to any keyword/directive
that supports :ref:`Lookups `.
By default, Runway will look for and load a ``runway.variables.yml`` or
``runway.variables.yaml`` file that is in the same directory as the
Runway config file. The file path and name of the file can optionally be
defined in the config file. If the file path is explicitly provided and
the file can't be found, an error will be raised.
Variables can also be defined in the Runway config file directly. This can
either be in place of a dedicated variables file, extend an existing file,
Returns:
MutableMap
"""
result = {}
supported_names = ['{}.env'.format(self.__ctx.env_name),
'{}-{}.env'.format(self.__ctx.env_name,
self.region)]
for _, file_name in enumerate(supported_names):
file_path = os.path.join(self.sys_path, file_name)
if os.path.isfile(file_path):
LOGGER.info('Found environment file: %s', file_path)
self._env_file_name = file_path
with open(file_path, 'r') as file_:
result.update(parse_environment(file_.read()))
return MutableMap(**result)
def _transform_to_string(cls, value, delimiter=None, **kwargs):
# type: (Any, str, Any) -> str
"""Transform anything into a string.
If the datatype of ``value`` is a list or similar to a list, ``join()``
is used to construct the list using a given delimiter or ``,``.
Args:
value: The value to be transformed into a string.
delimited: Used when transforming a list like object into a string
to join each element together.
"""
if isinstance(value, (list, set, tuple)):
return '{}'.format(delimiter or ',').join(value)
if isinstance(value, MutableMap):
# convert into a dict with protected attrs removed
value = value.data
if isinstance(value, dict):
# dumped twice for an escaped json dict
return json.dumps(json.dumps(value,
indent=int(kwargs.get('indent', 0))))
if isinstance(value, bool):
return json.dumps(str(value))
return str(value)
def data(self):
# type: () -> Dict[str, Any]
"""Sanitized output of __dict__.
Removes anything that starts with ``_``.
"""
result = {}
for key, val in self.__dict__.items():
if key.startswith('_'):
continue
if isinstance(val, MutableMap):
result[key] = val.data
else:
result[key] = val
return result