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_execute_hooks_with_multiple_hook(self):
hook_1 = MagicMock(spec=Hook)
hook_2 = MagicMock(spec=Hook)
execute_hooks([hook_1, hook_2])
hook_1.run.called_once_with()
hook_2.run.called_once_with()
# -*- coding: utf-8 -*-
from mock import MagicMock
from sceptre.hooks import Hook, HookProperty, add_stack_hooks, execute_hooks
class MockHook(Hook):
def __init__(self, *args, **kwargs):
super(MockHook, self).__init__(*args, **kwargs)
def run(self):
pass
class TestHooksFunctions(object):
def setup_method(self, test_method):
pass
def test_add_stack_hooks(self):
mock_hook_before = MagicMock(spec=Hook)
mock_hook_after = MagicMock(spec=Hook)
mock_object = MagicMock()
def test_hook_inheritance(self):
assert isinstance(self.hook, Hook)
from shutil import rmtree
try:
from StringIO import StringIO as BufferIO
except ImportError:
from io import BytesIO as BufferIO
try:
import zlib
compression = zipfile.ZIP_DEFLATED
except ImportError:
compression = zipfile.ZIP_STORED
class S3Package(Hook):
NAME = "s3_package"
TARGET = "dist"
DELIMITER = "^^"
def __init__(self, *args, **kwargs):
super(S3Package, self).__init__(*args, **kwargs)
def run(self):
if self.DELIMITER in self.argument:
fn_root_dir, s3_object = self.argument.split(self.DELIMITER, 1)
s3_bucket, s3_key = s3_object.split("/", 1)
self.logger.debug(
"[{}] S3 bucket/key parsed from the argument".format(self.NAME)
)
elif "sceptre_user_data" in self.stack_config:
code = self.stack_config.get("sceptre_user_data").get("Code", {})
import subprocess
from sceptre.hooks import Hook
from sceptre.exceptions import InvalidHookArgumentTypeError
class Cmd(Hook):
"""
Cmd implements a Sceptre hook which can run arbitrary commands.
"""
def __init__(self, *args, **kwargs):
super(Cmd, self).__init__(*args, **kwargs)
def run(self):
"""
Runs the argument string in a subprocess.
:raises: sceptre.exceptions.InvalidTaskArgumentTypeException
:raises: subprocess.CalledProcessError
"""
try:
subprocess.check_call(self.argument, shell=True)
# -*- coding: utf-8 -*-
from six import string_types
from sceptre.hooks import Hook
from sceptre.exceptions import InvalidHookArgumentTypeError
from sceptre.exceptions import InvalidHookArgumentSyntaxError
from sceptre.exceptions import InvalidHookArgumentValueError
class ASGScalingProcesses(Hook):
"""
Resumes or suspends autoscaling group scaling processes. This is
useful as scheduled actions must be suspended when updating stacks with
autoscaling groups.
"""
def __init__(self, *args, **kwargs):
super(ASGScalingProcesses, self).__init__(*args, **kwargs)
def run(self):
"""
Either suspends or resumes any scaling processes on all autoscaling
groups within the current stack.
:raises: InvalidHookArgumentSyntaxError, when syntax is not using "::".
:raises: InvalidHookArgumentTypeError, if argument is not a string.
# -*- coding: utf-8 -*-
import warnings
from six import string_types
from colorama import Fore, Style
from sceptre.hooks import Hook
from sceptre.exceptions import InvalidHookArgumentValueError
from sceptre.exceptions import InvalidHookArgumentTypeError
class ASGScheduledActions(Hook):
"""
This hook has been deprecated in favor of the asg_scaling_processes hook.
A command to resume or suspend autoscaling group scheduled actions. This is
useful as schedule actions must be suspended when updating stacks with
on autoscaling groups.
"""
def __init__(self, *args, **kwargs):
super(ASGScheduledActions, self).__init__(*args, **kwargs)
def run(self):
"""
Either suspends or resumes any scheduled actions on all autoscaling
groups with in the current stack.
"""
import subprocess
import warnings
from six import string_types
from colorama import Fore, Style
from sceptre.hooks import Hook
from sceptre.exceptions import InvalidHookArgumentTypeError
class Bash(Hook):
"""
This hook has been deprecated in favor of cmd hook.
This command with execute the argument string with bash.
"""
ALLOW_COMMAND_ERROR = True
def __init__(self, *args, **kwargs):
super(Bash, self).__init__(*args, **kwargs)
def run(self):
"""
Runs argument string in child process with bash.
:raise: sceptre.exceptions.InvalidTaskArgumentTypeException
"""