Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setup_mock_shed():
port = network_util.get_free_port()
directory = mkdtemp()
model = mock_model(directory)
def run():
app.debug = True
app.config["model"] = model
run_simple(
'localhost',
port,
app,
use_reloader=False,
use_debugger=True
)
t = threading.Thread(target=run)
t.start()
def test_serve_workflow(self):
"""Test serving a galaxy workflow via a daemon Galaxy process."""
port = network_util.get_free_port()
random_lines = os.path.join(PROJECT_TEMPLATES_DIR, "demo", "randomlines.xml")
cat = os.path.join(PROJECT_TEMPLATES_DIR, "demo", "cat.xml")
worklfow = os.path.join(TEST_DATA_DIR, "wf1.gxwf.yml")
extra_tools = [random_lines, cat]
config = galaxy_serve(
self.test_context,
[for_path(worklfow)],
install_galaxy=True,
port=port,
daemon=True,
extra_tools=extra_tools,
no_dependency_resolution=True,
)
user_gi = config.user_gi
assert user_gi.tools.get_tools(tool_id="random_lines1")
assert len(user_gi.workflows.get_workflows()) == 1
def test_shed_serve_daemon(self):
"""Test serving FASTQC from the tool shed via a daemon Galaxy process."""
port = network_util.get_free_port()
fastqc_path = os.path.join(TEST_REPOS_DIR, "fastqc")
ctx = self.test_context
install_args_list = shed.install_arg_lists(
ctx, [fastqc_path],
shed_target="toolshed",
)
with shed_serve(
ctx, install_args_list,
port=port,
skip_dependencies=True,
install_galaxy=True,
) as config:
assert network_util.wait_net_service(
"localhost",
config.port,
timeout=.1,
def setUp(self):
super(ServeTestCase, self).setUp()
self._port = network_util.get_free_port()
self._pid_file = os.path.join(self._home, "test.pid")
self._serve_artifact = os.path.join(TEST_REPOS_DIR, "single_tool", "cat.xml")
def test_serve(self):
port = network_util.get_free_port()
serve = functools.partial(self._run, port)
t = threading.Thread(target=serve)
t.daemon = True
t.start()
time.sleep(15)
assert network_util.wait_net_service("127.0.0.1", port)
def test_serve_daemon(self):
"""Test serving a galaxy tool via a daemon Galaxy process."""
port = network_util.get_free_port()
cat_path = os.path.join(TEST_REPOS_DIR, "single_tool", "cat.xml")
config = galaxy_serve(
self.test_context,
[for_path(cat_path)],
install_galaxy=True,
port=port,
daemon=True,
no_dependency_resolution=True,
)
assert network_util.wait_net_service(
"localhost",
config.port,
timeout=.1,
)
config_dict = config.gi.config.get_config()
def _serve(ctx, runnables, **kwds):
engine = kwds.get("engine", "galaxy")
if engine == "docker_galaxy":
kwds["dockerize"] = True
daemon = kwds.get("daemon", False)
if daemon:
kwds["no_cleanup"] = True
port = kwds.get("port", None)
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")