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_sceptreise_path_with_trailing_slash(self):
with pytest.raises(PathConversionError):
sceptreise_path(
"this/path/is/invalid/"
)
def test_sceptreise_path_with_trailing_backslash(self):
with pytest.raises(PathConversionError):
sceptreise_path(
'this\path\is\invalid\\'
)
stack_group_configs = {}
while todo:
abs_path = todo.pop()
rel_path = path.relpath(
abs_path, start=self.context.full_config_path())
directory, filename = path.split(rel_path)
if directory in stack_group_configs:
stack_group_config = stack_group_configs[directory]
else:
stack_group_config = stack_group_configs[directory] = \
self.read(path.join(directory, self.context.config_file))
stack = self._construct_stack(rel_path, stack_group_config)
stack_map[sceptreise_path(rel_path)] = stack
if abs_path.startswith(self.context.full_command_path()):
command_stacks.add(stack)
stacks = self.resolve_stacks(stack_map)
return stacks, command_stacks
def _collect_s3_details(stack_name, config):
"""
Collects and constructs details for where to store the Template in S3.
:param stack_name: Stack name.
:type stack_name: str
:param config: Config with details.
:type config: dict
:returns: S3 details.
:rtype: dict
"""
s3_details = None
if "template_bucket_name" in config:
template_key = "/".join([
sceptreise_path(stack_name), "{time_stamp}.json".format(
time_stamp=datetime.datetime.utcnow().strftime(
"%Y-%m-%d-%H-%M-%S-%fZ"
)
)
])
bucket_region = config.get("region", None)
if "template_key_prefix" in config:
prefix = config["template_key_prefix"]
template_key = "/".join([prefix.strip("/"), template_key])
s3_details = {
"bucket_name": config["template_bucket_name"],
"bucket_key": template_key,
"bucket_region": bucket_region
def _valid_stack_paths(self):
return [
sceptreise_path(path.relpath(path.join(dirpath, f), self.context.config_path))
for dirpath, dirnames, files in walk(self.context.config_path)
for f in files
if not f.endswith(self.context.config_file)
]
def __init__(
self, name, project_code, template_path, region, template_bucket_name=None,
template_key_prefix=None, required_version=None, parameters=None,
sceptre_user_data=None, hooks=None, s3_details=None, iam_role=None,
dependencies=None, role_arn=None, protected=False, tags=None,
external_name=None, notifications=None, on_failure=None, profile=None,
stack_timeout=0, stack_group_config={}
):
self.logger = logging.getLogger(__name__)
self.name = sceptreise_path(name)
self.project_code = project_code
self.region = region
self.template_bucket_name = template_bucket_name
self.template_key_prefix = template_key_prefix
self.required_version = required_version
self.external_name = external_name or get_external_stack_name(self.project_code, self.name)
self.template_path = template_path
self.s3_details = s3_details
self._template = None
self._connection_manager = None
self.protected = protected
self.role_arn = role_arn
self.on_failure = on_failure
self.dependencies = dependencies or []
"""
Transforms map of Stacks into a set of Stacks, transforms dependencies
from a list of Strings (stack names) to a list of Stacks.
:param stack_map: Map of stacks, containing dependencies as list of Strings.
:type base_config: dict
:returns: Set of stacks, containing dependencies as list of Stacks.
:rtype: set
:raises: sceptre.exceptions.DependencyDoesNotExistError
"""
stacks = set()
for stack in stack_map.values():
if not self.context.ignore_dependencies:
for i, dep in enumerate(stack.dependencies):
try:
stack.dependencies[i] = stack_map[sceptreise_path(dep)]
except KeyError:
raise DependencyDoesNotExistError(
"{stackname}: Dependency {dep} not found. "
"Valid dependency names are: "
"{stackkeys}. "
"Please make sure that your dependencies stack_outputs "
"have their full path from `config` defined."
.format(stackname=stack.name, dep=dep,
stackkeys=", ".join(stack_map.keys())))
else:
stack.dependencies = []
stacks.add(stack)
return stacks
parsed_stack_group_config = self._parsed_stack_group_config(stack_group_config)
config = self.read(rel_path, stack_group_config)
stack_name = path.splitext(rel_path)[0]
# Check for missing mandatory attributes
for required_key in REQUIRED_KEYS:
if required_key not in config:
raise InvalidConfigFileError(
"Required attribute '{0}' not found in configuration of '{1}'.".format(
required_key, stack_name
)
)
abs_template_path = path.join(
self.context.project_path, self.context.templates_path,
sceptreise_path(config["template_path"])
)
s3_details = self._collect_s3_details(
stack_name, config
)
stack = Stack(
name=stack_name,
project_code=config["project_code"],
template_path=abs_template_path,
region=config["region"],
template_bucket_name=config.get("template_bucket_name"),
template_key_prefix=config.get("template_key_prefix"),
required_version=config.get("required_version"),
iam_role=config.get("iam_role"),
profile=config.get("profile"),
parameters=config.get("parameters", {}),