Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _run_checks(self):
"""basic sanity checks for the file name (and others if needed) before
attempting parsing.
"""
if self.filename is not None:
# Does the recipe provided exist?
if not os.path.exists(self.filename):
bot.exit("Cannot find %s, is the path correct?" % self.filename)
# Ensure we carry fullpath
self.filename = os.path.abspath(self.filename)
"""
from spython.utils import run_command, check_install
check_install()
# If name provided, over write robot (default)
if name is not None:
self.name = name
# If an image isn't provided, we have an initialized instance
if image is None:
# Not having this means it was called as a command, without an image
if not hasattr(self, "_image"):
bot.exit("Please provide an image, or create an Instance first.")
image = self._image
# Derive subgroup command based on singularity version
subgroup = "instance.start"
if "version 3" in self.version():
subgroup = ["instance", "start"]
cmd = self._init_command(subgroup, singularity_options)
# Add options, if they are provided
if not isinstance(options, list):
options = [] if options is None else options.split(" ")
# Assemble the command!
cmd = cmd + options + [image, self.name]
ext = "sif"
# Force the build if the image / sandbox exists
if force:
cmd.append("--force")
# No image provided, default to use the client's loaded image
if recipe is None:
recipe = self._get_uri()
# If it's still None, try default build recipe
if recipe is None:
recipe = "Singularity"
if not os.path.exists(recipe):
bot.exit("Cannot find %s, exiting." % image)
if image is None:
if re.search("(docker|shub|library)://", recipe) and not robot_name:
image = self._get_filename(recipe, ext)
else:
image = "%s.%s" % (self.RobotNamer.generate(), ext)
# Does the user want a custom build folder?
if build_folder is not None:
if not os.path.exists(build_folder):
bot.exit("%s does not exist!" % build_folder)
image = os.path.join(build_folder, image)
# The user wants to run an isolated build
if isolated:
cmd.append("--isolated")
def decompress(self, image_path, quiet=True):
'''decompress will (properly) decompress an image'''
if not os.path.exists(image_path):
bot.exit("Cannot find image %s" %image_path)
extracted_file = image_path.replace('.gz', '')
cmd = ['gzip', '-d', '-f', image_path]
self.run_command(cmd, quiet=quiet) # exits if return code != 0
return extracted_file
def export(self, image_path, tmptar=None):
'''export will export an image, sudo must be used.
Parameters
==========
image_path: full path to image
tmptar: if defined, use custom temporary path for tar export
'''
from spython.utils import check_install
check_install()
if 'version 3' in self.version():
bot.exit('export is deprecated after Singularity 2.*')
if tmptar is None:
tmptar = "/%s/tmptar.tar" %(tempfile.mkdtemp())
cmd = ['singularity', 'image.export', '-f', tmptar, image_path]
self.run_command(cmd, sudo=False)
return tmptar
log_path: the path to store the log.
pid_file: specify the pid file path to use
sync_socket: the path to the unix socket for state synchronization.
command: the command (run or create) to use (default is run)
log_format: defaults to kubernetes. Can also be "basic" or "json"
singularity_options: a list of options to provide to the singularity client
"""
container_id = self.get_container_id(container_id)
# singularity oci create
cmd = self._init_command(command, singularity_options)
# Check that the bundle exists
if not os.path.exists(bundle):
bot.exit("Bundle not found at %s" % bundle)
# Add the bundle
cmd = cmd + ["--bundle", bundle]
# Additional Logging Files
cmd = cmd + ["--log-format", log_format]
if log_path is not None:
cmd = cmd + ["--log-path", log_path]
if pid_file is not None:
cmd = cmd + ["--pid-file", pid_file]
if sync_socket is not None:
cmd = cmd + ["--sync-socket", sync_socket]
if empty_process:
cmd.append("--empty-process")
def validate(self):
"""validate that all (required) fields are included for the Docker
recipe. We minimimally just need a FROM image, and must ensure
it's in a valid format. If anything is missing, we exit with error.
"""
if self.recipe is None:
bot.exit("Please provide a Recipe() to the writer first.")
# Write single recipe that includes all layer
recipe = []
# Number of layers
num_layers = len(self.recipe)
count = 0
# Write each layer to new file
for stage, parser in self.recipe.items():
# Set the first and active stage
self.stage = stage
# From header is required
if parser.fromHeader is None:
bot.exit("Singularity recipe requires a from header.")
recipe += ["\n\n\nBootstrap: docker"]
recipe += ["From: %s" % parser.fromHeader]
recipe += ["Stage: %s\n\n\n" % stage]
# TODO: stopped here - bug with files being found
# Add global files, and then layer files
recipe += self._create_section("files")
for layer, files in parser.layer_files.items():
recipe += create_keyval_section(files, "files", layer)
# Sections with key value pairs
recipe += self._create_section("labels")
recipe += self._create_section("install", "post")
recipe += self._create_section("environ", "environment")
if args.entrypoint is not None:
entrypoint = args.entrypoint
# This is only done if the user intended to print json here
recipeParser.entrypoint = args.entrypoint
recipeParser.cmd = None
force = True
if args.json:
if outfile is not None:
if not os.path.exists(outfile):
if force:
write_json(outfile, recipeParser.recipe.json())
else:
bot.exit('%s exists, set --force to overwrite.' % outfile)
else:
print(json.dumps(recipeParser.recipe.json(), indent=4))
else:
# Do the conversion
recipeWriter = writer(recipeParser.recipe)
result = recipeWriter.convert(runscript=entrypoint, force=force)
# If the user specifies an output file, save to it
if outfile is not None:
write_file(outfile, result)
# Otherwise, convert and print to screen
else:
print(result)