Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def create_zip_file_cli(source_path, base_dir, zip_file):
# Using the native zip command can be an order of magnitude faster on Travis-CI
source = '*' if source_path == base_dir else os.path.basename(source_path)
command = 'cd %s; zip -r %s %s' % (base_dir, zip_file, source)
run(command)
def copy_dir(source, target):
if is_alpine():
# Using the native command can be an order of magnitude faster on Travis-CI
return run('cp -r %s %s' % (source, target))
shutil.copytree(source, target)
def get_all_container_names(self):
"""
Returns a list of container names for lambda containers.
:return: A String[] localstack docker container names for each function.
"""
with self.docker_container_lock:
LOG.debug('Getting all lambda containers names.')
cmd = '%s ps -a --filter="name=localstack_lambda_*" --format "{{.Names}}"' % self._docker_cmd()
LOG.debug(cmd)
cmd_result = run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE).strip()
if len(cmd_result) > 0:
container_names = cmd_result.split('\n')
else:
container_names = []
return container_names
def ensure_webapp_installed():
web_dir = os.path.realpath(os.path.join(os.path.dirname(__file__), 'web'))
node_modules_dir = os.path.join(web_dir, 'node_modules', 'jquery')
if not os.path.exists(node_modules_dir):
print('Initializing installation of Web application (this could take a long time, please be patient)')
common.run('cd "%s"; npm install' % web_dir)
"""
with self.docker_container_lock:
status = self.get_docker_container_status(func_arn)
docker_cmd = self._docker_cmd()
# Get the container name and id.
container_name = self.get_container_name(func_arn)
if status == 1:
LOG.debug('Stopping container: %s' % container_name)
cmd = (
'%s stop -t0 %s'
) % (docker_cmd, container_name)
LOG.debug(cmd)
run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)
status = self.get_docker_container_status(func_arn)
if status == -1:
LOG.debug('Removing container: %s' % container_name)
cmd = (
'%s rm %s'
) % (docker_cmd, container_name)
LOG.debug(cmd)
run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)
if status == 0:
return ''
# Get the container name.
container_name = self.get_container_name(func_arn)
docker_cmd = self._docker_cmd()
# Get the container network
LOG.debug('Getting container network: %s' % container_name)
cmd = (
'%s inspect %s'
' --format "{{ .HostConfig.NetworkMode }}"'
) % (docker_cmd, container_name)
LOG.debug(cmd)
cmd_result = run(cmd, asynchronous=False, stderr=subprocess.PIPE, outfile=subprocess.PIPE)
container_network = cmd_result.strip()
return container_network
def use_docker():
global DO_USE_DOCKER
if DO_USE_DOCKER is None:
DO_USE_DOCKER = False
if 'docker' in config.LAMBDA_EXECUTOR:
try:
run('docker images', print_error=False)
DO_USE_DOCKER = True
except Exception:
pass
return DO_USE_DOCKER
def install_kinesalite():
target_dir = '%s/kinesalite' % INSTALL_DIR_NPM
if not os.path.exists(target_dir):
log_install_msg('Kinesis')
run('cd "%s" && npm install' % ROOT_PATH)