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_troposphere(self, cfngin_context):
"""Test with troposphere object like returned from lambda hook."""
bucket = 'test-bucket'
s3_key = 'lambda_functions/my_function'
cfngin_context.set_hook_data('lambda',
{'my_function': Code(
S3Bucket=bucket,
S3Key=s3_key
)})
var_bucket = Variable('test',
'${hook_data lambda.my_function::'
'load=troposphere,get=S3Bucket}',
variable_type='cfngin')
var_key = Variable('test',
'${hook_data lambda.my_function::get=S3Key}',
variable_type='cfngin')
var_bucket.resolve(cfngin_context)
var_key.resolve(cfngin_context)
assert var_bucket.value == bucket
assert var_key.value == s3_key
def test_resolve_variables_lookup_returns_troposphere_obj(self):
"""Test resolve variables lookup returns troposphere obj."""
class TestBlueprint(Blueprint):
"""Test blueprint."""
VARIABLES = {
"Param1": {"type": Base64},
}
def return_obj(*_args, **_kwargs):
"""Return object."""
return Base64("test")
register_lookup_handler("custom", return_obj)
blueprint = TestBlueprint(name="test", context=MagicMock())
variables = [Variable("Param1", "${custom non-string-return-val}",
'cfngin')]
for var in variables:
var._value.resolve({}, {})
blueprint.resolve_variables(variables)
self.assertEqual(blueprint.resolved_variables["Param1"].data,
Base64("test").data)
def test_variable_resolve_multiple_lookups_string(self):
"""Test variable resolve multiple lookups string."""
var = Variable(
"Param1",
"url://${output fakeStack::FakeOutput}@"
"${output fakeStack::FakeOutput2}",
)
stack = Stack(
definition=generate_definition("vpc", 1),
context=self.context)
stack.set_outputs({
"FakeOutput": "resolved",
"FakeOutput2": "resolved2",
})
self.context.get_stack.return_value = stack
var.resolve(self.context, self.provider)
self.assertTrue(var.resolved)
def test_handle(self, cfngin_context):
"""Test handle with simple usage."""
cfngin_context.set_hook_data('fake_hook',
{'nested': {'result': 'good'}})
var_top = Variable('test', '${hook_data fake_hook}',
variable_type='cfngin')
var_nested = Variable('test', '${hook_data fake_hook.nested.result}',
variable_type='cfngin')
var_top.resolve(cfngin_context)
var_nested.resolve(cfngin_context)
assert var_top.value == {'nested': {'result': 'good'}}
assert var_nested.value == 'good'
def test_variable_resolve_nested_lookup(self):
"""Test variable resolve nested lookup."""
stack = Stack(
definition=generate_definition("vpc", 1),
context=self.context)
stack.set_outputs({
"FakeOutput": "resolved",
"FakeOutput2": "resolved2",
})
def mock_handler(value, context, provider, **kwargs):
return "looked up: {}".format(value)
register_lookup_handler("lookup", mock_handler)
self.context.get_stack.return_value = stack
var = Variable(
"Param1",
"${lookup ${lookup ${output fakeStack::FakeOutput}}}",
)
var.resolve(self.context, self.provider)
self.assertTrue(var.resolved)
self.assertEqual(var.value, "looked up: looked up: resolved")
sort_keys=True,
indent=4
)
blueprint = RawTemplateBlueprint(
name="stack1",
context=mock_context(
extra_config_args={'stacks': [{'name': 'stack1',
'template_path': 'unused',
'variables': {
'Param1': 'param1val',
'bar': 'foo'}}]},
environment={'foo': 'bar'}),
raw_template_path=RAW_J2_TEMPLATE_PATH
)
blueprint.resolve_variables([Variable("Param1", "param1val", 'cfngin'),
Variable("bar", "foo", 'cfngin')])
self.assertEqual(
expected_json,
blueprint.to_json()
)
'account_id', deployment.pop('account-id', {})
), 'runway') # type: Variable
self._assume_role = Variable(
self.name + '.assume_role', deployment.pop(
'assume_role', deployment.pop('assume-role', {})
), 'runway'
) # type: Variable
self._environments = Variable(
self.name + '.environments', deployment.pop('environments', {}),
'runway'
) # type: Variable
self._parameters = Variable(
self.name + '.parameters', deployment.pop('parameters', {}),
'runway'
) # type: Variable
self._env_vars = Variable(self.name + '.env_vars', deployment.pop(
'env_vars', deployment.pop('env-vars', {})
), 'runway') # type: Variable
if deployment.pop('current_dir', False):
# Deprecated in 1.0 (late 2019). Retain for at least a major version.
LOGGER.warning('DEPRECATION WARNING: The "current_dir" option has '
'been deprecated in favor of a "./" module '
'definition. Please update your config.')
modules = ['.' + os.sep]
else:
if not deployment.get('modules'):
LOGGER.error('No modules have been defined in your Runway '
'deployment.')
sys.exit(1)
modules = deployment.pop('modules')
self.modules = ModuleDefinition.from_list(
modules
- `Troposphere`_
- `Terraform`_
- `Kubernetes`_
- :ref:`Static Site`
- :ref:`Module Configurations` -
detailed module ``options``
- :ref:`Repo Structure` - examples of
directory structure
- :ref:`command-deploy`
- :ref:`command-destroy`
- :ref:`command-plan`
"""
self.name = name
self._path = Variable(name + '.path', path, 'runway')
self._class_path = Variable(name + '.class_path', class_path, 'runway')
self.type = type_str
self._environments = Variable(name + '.environments',
environments or {}, 'runway')
self._parameters = Variable(name + '.parameters', parameters or {},
'runway')
self._env_vars = Variable(name + '.env_vars', env_vars or {}, 'runway')
self._options = Variable(name + '.options', options or {}, 'runway')
self.tags = tags or {}
self.child_modules = child_modules or []
def __getitem__(self, key):
# type: (str) -> Any
"""Implement evaluation of self[key]."""
result = getattr(self, key, getattr(self, key.replace('-', '_')))
if isinstance(result, Variable):
return result.value
return result
'account_alias', deployment.pop('account-alias', {})
), 'runway'
) # type: Variable
self._account_id = Variable(self.name + '.account_id', deployment.pop(
'account_id', deployment.pop('account-id', {})
), 'runway') # type: Variable
self._assume_role = Variable(
self.name + '.assume_role', deployment.pop(
'assume_role', deployment.pop('assume-role', {})
), 'runway'
) # type: Variable
self._environments = Variable(
self.name + '.environments', deployment.pop('environments', {}),
'runway'
) # type: Variable
self._parameters = Variable(
self.name + '.parameters', deployment.pop('parameters', {}),
'runway'
) # type: Variable
self._env_vars = Variable(self.name + '.env_vars', deployment.pop(
'env_vars', deployment.pop('env-vars', {})
), 'runway') # type: Variable
if deployment.pop('current_dir', False):
# Deprecated in 1.0 (late 2019). Retain for at least a major version.
LOGGER.warning('DEPRECATION WARNING: The "current_dir" option has '
'been deprecated in favor of a "./" module '
'definition. Please update your config.')
modules = ['.' + os.sep]
else:
if not deployment.get('modules'):
LOGGER.error('No modules have been defined in your Runway '
'deployment.')