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_init(self, tmp_path):
"""Test init process."""
env_name = 'test'
env_region = 'us-east-1'
env_root = str(tmp_path)
config_file = tmp_path / 'config'
config_file.write_text(u'[profile test]\n'
'aws_access_key_id = bar\n'
'aws_secret_access_key = foo')
with environ(dict(AWS_PROFILE='test', **TEST_CREDENTIALS)):
env_vars = os.environ.copy()
env_vars['DEPLOY_ENVIRONMENT'] = env_name
ctx = Context(env_name, env_region, env_root)
assert not ctx.command
assert not ctx.debug
assert ctx.env_name == env_name
assert ctx.env_region == env_region
assert ctx.env_root == env_root
assert sorted(ctx.env_vars) == sorted(env_vars)
with environ({'AWS_PROFILE': 'test', 'AWS_CONFIG_FILE': str(config_file)}):
os.environ.pop('AWS_ACCESS_KEY_ID', None)
os.environ.pop('AWS_SECRET_ACCESS_KEY', None)
os.environ.pop('AWS_SESSION_TOKEN', None)
env_vars = os.environ.copy()
env_vars['DEPLOY_ENVIRONMENT'] = env_name
ctx = Context(env_name, env_region, env_root)
def test_persistent_graph_location_no_bucket(self):
"""Return an empty dict if key is set but no bucket name."""
cp_config = self.persist_graph_raw_config.copy()
cp_config['cfngin_bucket'] = ''
context = Context(config=Config(cp_config))
self.assertEqual({}, context.persistent_graph_location)
def test_persistent_graph_lock_code_none(self):
"""Return 'None' when the tag is not set."""
context = Context(config=self.persist_graph_config)
stubber = Stubber(context.s3_conn)
stubber.add_response('get_object_tagging', {'TagSet': []},
context.persistent_graph_location)
with stubber:
self.assertIsNone(context.persistent_graph_lock_code)
self.assertIsNone(context._persistent_graph_lock_code)
stubber.assert_no_pending_responses()
def test_put_persistent_graph_unlocked(self):
"""Error raised when trying to update an unlocked object."""
context = Context(config=self.persist_graph_config)
context._s3_bucket_verified = True
context._persistent_graph = Graph.from_dict({'stack1': []}, context)
stubber = Stubber(context.s3_conn)
stubber.add_response('get_object_tagging', {'TagSet': []},
context.persistent_graph_location)
with stubber:
with self.assertRaises(PersistentGraphUnlocked):
context.put_persistent_graph('')
stubber.assert_no_pending_responses()
def setUp(self):
"""Run before tests."""
self.context = Context(
config=Config({'namespace': 'test', 'stacker_bucket': 'test'}))
self.provider = mock_provider(region="us-east-1")
def test_unlock_persistent_graph_code_missmatch(self):
"""Error raised when local code does not match object."""
code = '0000'
context = Context(config=self.persist_graph_config)
context._s3_bucket_verified = True
context._persistent_graph = Graph.from_dict({'stack1': []}, context)
stubber = Stubber(context.s3_conn)
stubber.add_response('get_object_tagging',
{'TagSet': gen_tagset(
{context._persistent_graph_lock_tag: '1111'}
)},
context.persistent_graph_location)
with stubber:
with self.assertRaises(PersistentGraphCannotUnlock):
context.unlock_persistent_graph(code)
stubber.assert_no_pending_responses()
def test_context_default_bucket_name(self):
"""Test context default bucket name."""
context = Context(config=Config({"namespace": "test"}))
self.assertEqual(context.bucket_name, "stacker-test")
def test_persistent_graph_location_no_key(self):
"""Return an empty dict if key is not set."""
context = Context(config=self.config)
self.assertEqual({}, context.persistent_graph_location)
def test_context_get_fqn_replace_dot(self):
"""Test context get fqn replace dot."""
context = Context(config=Config({"namespace": "my.namespace"}))
fqn = context.get_fqn()
self.assertEqual(fqn, "my-namespace")
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