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_get_java_lib_folder_classpath(self):
jar_file = os.path.join(new_tmp_dir(), 'foo.jar')
save_file(jar_file, '')
self.assertEquals('.:foo.jar', lambda_executors.Util.get_java_classpath(jar_file))
def get_or_create_file(config_file):
if os.path.exists(config_file):
return config_file
try:
save_file(config_file, '{}')
return config_file
except Exception:
pass
dynamodb_url = DYNAMODB_JAR_URL_ALPINE if in_docker() else DYNAMODB_JAR_URL
download_and_extract_with_retry(dynamodb_url, tmp_archive, INSTALL_DIR_DDB)
# fix for Alpine, otherwise DynamoDBLocal fails with:
# DynamoDBLocal_lib/libsqlite4java-linux-amd64.so: __memcpy_chk: symbol not found
if is_alpine():
ddb_libs_dir = '%s/DynamoDBLocal_lib' % INSTALL_DIR_DDB
patched_marker = '%s/alpine_fix_applied' % ddb_libs_dir
if APPLY_DDB_ALPINE_FIX and not os.path.exists(patched_marker):
patched_lib = ('https://rawgit.com/bhuisgen/docker-alpine/master/alpine-dynamodb/' +
'rootfs/usr/local/dynamodb/DynamoDBLocal_lib/libsqlite4java-linux-amd64.so')
patched_jar = ('https://rawgit.com/bhuisgen/docker-alpine/master/alpine-dynamodb/' +
'rootfs/usr/local/dynamodb/DynamoDBLocal_lib/sqlite4java.jar')
run("curl -L -o %s/libsqlite4java-linux-amd64.so '%s'" % (ddb_libs_dir, patched_lib))
run("curl -L -o %s/sqlite4java.jar '%s'" % (ddb_libs_dir, patched_jar))
save_file(patched_marker, '')
# fix logging configuration for DynamoDBLocal
log4j2_config = """
"""
log4j2_file = os.path.join(INSTALL_DIR_DDB, 'log4j2.xml')
save_file(log4j2_file, log4j2_config)
run('cd "%s" && zip -u DynamoDBLocal.jar log4j2.xml || true' % INSTALL_DIR_DDB)
def log(self, s):
s = '%s\n' % s
if self.log_file:
save_file(self.log_file, s, append=True)
run("curl -L -o %s/sqlite4java.jar '%s'" % (ddb_libs_dir, patched_jar))
save_file(patched_marker, '')
# fix logging configuration for DynamoDBLocal
log4j2_config = """
"""
log4j2_file = os.path.join(INSTALL_DIR_DDB, 'log4j2.xml')
save_file(log4j2_file, log4j2_config)
run('cd "%s" && zip -u DynamoDBLocal.jar log4j2.xml || true' % INSTALL_DIR_DDB)
node-address {
protocol = http
host = "%s"
port = %s
context-path = ""
}
rest-sqs {
enabled = true
bind-port = %s
bind-hostname = "0.0.0.0"
sqs-limits = strict
}
""" % (LOCALSTACK_HOSTNAME, port, backend_port)
config_file = os.path.join(TMP_FOLDER, 'sqs.%s.conf' % short_uid())
TMP_FILES.append(config_file)
save_file(config_file, config_params)
# start process
cmd = ('java -Dconfig.file=%s -Xmx%s -jar %s/elasticmq-server.jar' % (
config_file, MAX_HEAP_SIZE, INSTALL_DIR_ELASTICMQ))
print('Starting mock SQS (%s port %s)...' % (get_service_protocol(), port))
start_proxy_for_service('sqs', port, backend_port, update_listener)
return do_run(cmd, asynchronous)
def exec_lambda_code(script, handler_function='handler', lambda_cwd=None, lambda_env=None):
if lambda_cwd or lambda_env:
exec_mutex.acquire()
if lambda_cwd:
previous_cwd = os.getcwd()
os.chdir(lambda_cwd)
sys.path = [lambda_cwd] + sys.path
if lambda_env:
previous_env = dict(os.environ)
os.environ.update(lambda_env)
# generate lambda file name
lambda_id = 'l_%s' % short_uid()
lambda_file = LAMBDA_SCRIPT_PATTERN.replace('*', lambda_id)
save_file(lambda_file, script)
# delete temporary .py and .pyc files on exit
TMP_FILES.append(lambda_file)
TMP_FILES.append('%sc' % lambda_file)
try:
handler_module = imp.load_source(lambda_id, lambda_file)
module_vars = handler_module.__dict__
except Exception as e:
LOG.error('Unable to exec: %s %s' % (script, traceback.format_exc()))
raise e
finally:
if lambda_cwd or lambda_env:
if lambda_cwd:
os.chdir(previous_cwd)
sys.path.pop(0)
if lambda_env:
os.environ = previous_env
# WARNING: this means we're pointing lambda_cwd to a local path in the user's
# file system! We must ensure that there is no data loss (i.e., we must *not* add
# this folder to TMP_FILES or similar).
return code['S3Key']
# get file content
zip_file_content = zip_file_content or get_zip_bytes(code)
# Save the zip file to a temporary file that the lambda executors can reference
code_sha_256 = base64.standard_b64encode(hashlib.sha256(zip_file_content).digest())
lambda_details.get_version('$LATEST')['CodeSize'] = len(zip_file_content)
lambda_details.get_version('$LATEST')['CodeSha256'] = code_sha_256.decode('utf-8')
tmp_dir = '%s/zipfile.%s' % (config.TMP_FOLDER, short_uid())
mkdir(tmp_dir)
tmp_file = '%s/%s' % (tmp_dir, LAMBDA_ZIP_FILE_NAME)
save_file(tmp_file, zip_file_content)
TMP_FILES.append(tmp_dir)
lambda_details.cwd = tmp_dir
return tmp_dir