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_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_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 generate_tfstate_cfn_template():
"""Return rendered CFN template yaml."""
# pylint: disable=import-outside-toplevel
from runway.blueprints.tf_state import TfState
return to_yaml(TfState('test',
Context({"namespace": "test"}),
None).to_json())
def _get_context(self, config, config_path):
"""Initialize a CFNgin context object.
Args:
config (:class:`runway.cfngin.config.Config): CFNgin config object.
config_path (str): Path to the config file that was provided.
Returns:
:class:`runway.cfngin.context.Context`
"""
return CFNginContext(
boto3_credentials=self.__ctx.boto3_credentials,
config=config,
config_path=config_path,
environment=self.parameters,
force_stacks=[], # placeholder
region=self.region,
stack_names=[] # placeholder
)