Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def cli(ctx, path, **kwds):
"""Generate human readable tool test reports.
Creates reports in various formats (HTML, text, markdown)
from the structured test output (tool_test_output.json).
"""
if not os.path.exists(path):
io.error("Failed to tool test json file at %s" % path)
return 1
test_data = StructuredData(path)
test_data.calculate_summary_data_if_needed()
handle_reports(ctx, test_data.structured_data, kwds)
def test_shed_diff(self):
with self._isolate_repo("single_tool") as f:
upload_command = ["shed_upload", "--force_repository_creation"]
upload_command.extend(self._shed_args())
self._check_exit_code(upload_command)
io.write_file(
join(f, "related_file"),
"A related non-tool file (modified).\n",
)
self._check_diff(f, True)
self._check_diff(f, False)
def test_diff_recursive(self):
with self._isolate_repo("multi_repos_nested") as f:
self._shed_create(recursive=True)
diff_command = ["shed_diff", "-r"]
diff_command.extend(self._shed_args(read_only=True))
self._check_exit_code(diff_command, exit_code=0)
io.write_file(
join(f, "cat1", "related_file"),
"A related non-tool file (modified).\n",
)
self._check_exit_code(diff_command, exit_code=1)
def _copy_repo(self, name, dest):
repo = os.path.join(TEST_REPOS_DIR, name)
io.shell(['cp', '-r', "%s/." % repo, dest])
def append_macro_file(tool_files, kwds):
macro_contents = None
if kwds["macros"]:
macro_contents = _render(kwds, MACROS_TEMPLATE)
macros_file = "macros.xml"
if not os.path.exists(macros_file):
tool_files.append(ToolFile(macros_file, macro_contents, "macros"))
io.info(REUSING_MACROS_MESSAGE)
def not_exclusive(x, y):
if kwds.get(x) and kwds.get(y):
io.error("Can only specify one of --%s and --%s" % (x, y))
return True
if port is None:
port = network_util.get_free_port()
kwds["port"] = port
with galaxy_config(ctx, runnables, **kwds) as config:
cmd = config.startup_command(ctx, **kwds)
action = "Starting galaxy"
exit_code = run_galaxy_command(
ctx,
cmd,
config.env,
action,
)
if exit_code:
message = "Problem running Galaxy command [%s]." % config.log_contents
io.warn(message)
raise Exception(message)
host = kwds.get("host", "127.0.0.1")
timeout = 500
galaxy_url = "http://%s:%s" % (host, port)
galaxy_alive = sleep(galaxy_url, verbose=ctx.verbose, timeout=timeout)
if not galaxy_alive:
raise Exception("Attempted to serve Galaxy at %s, but it failed to start in %d seconds." % (galaxy_url, timeout))
config.install_workflows()
if kwds.get("pid_file"):
real_pid_file = config.pid_file
if os.path.exists(config.pid_file):
os.symlink(real_pid_file, kwds["pid_file"])
else:
io.warn("Can't find Galaxy pid file [%s] to link" % real_pid_file)
return config
def not_exclusive(x, y):
if kwds.get(x) and kwds.get(y):
io.error("Can only specify one of --%s and --%s" % (x, y))
return True
def cli(ctx, paths, **kwds):
"""Launch Galaxy with Tool Shed dependencies.
This command will start a Galaxy instance configured to target the
specified shed, find published artifacts (tools and dependencies)
corresponding to command-line arguments and ``.shed.yml`` file(s),
install these artifacts, and serve a Galaxy instances that can be
logged into and explored interactively.
"""
kwds['galaxy_skip_client_build'] = False
install_args_list = shed.install_arg_lists(ctx, paths, **kwds)
with shed_serve(ctx, install_args_list, **kwds) as config:
io.info("Galaxy running with tools installed at %s" % config.galaxy_url)
sleep_for_serve()
def print_path_list(paths, **kwds):
with io.open_file_or_standard_output(kwds["output"], "w") as f:
for path in paths:
print(path, file=f)