mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-21 13:50:20 +08:00
John Chilton
f6f9f9a891
Also make sure output collections in non-mapping scenarios are also properly renamed as well.
2839 lines
113 KiB
Python
2839 lines
113 KiB
Python
from __future__ import print_function
|
|
|
|
import json
|
|
import time
|
|
from collections import namedtuple
|
|
from json import dumps
|
|
from uuid import uuid4
|
|
|
|
import yaml
|
|
from requests import delete, put
|
|
|
|
from base import api # noqa: I100,I202
|
|
from base.populators import ( # noqa: I100
|
|
DatasetCollectionPopulator,
|
|
DatasetPopulator,
|
|
flakey,
|
|
skip_without_tool,
|
|
wait_on,
|
|
WorkflowPopulator
|
|
)
|
|
from galaxy.exceptions import error_codes # noqa: I201
|
|
from galaxy.tools.verify.test_data import TestDataResolver
|
|
|
|
|
|
SIMPLE_NESTED_WORKFLOW_YAML = """
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: outer_input
|
|
outputs:
|
|
- id: outer_output
|
|
source: second_cat#out_file1
|
|
steps:
|
|
- tool_id: cat1
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: outer_input
|
|
- run:
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: inner_input
|
|
outputs:
|
|
- id: workflow_output
|
|
source: random_lines#out_file1
|
|
steps:
|
|
- tool_id: random_lines1
|
|
label: random_lines
|
|
state:
|
|
num_lines: 1
|
|
input:
|
|
$link: inner_input
|
|
seed_source:
|
|
seed_source_selector: set_seed
|
|
seed: asdf
|
|
label: nested_workflow
|
|
connect:
|
|
inner_input: first_cat#out_file1
|
|
- tool_id: cat1
|
|
label: second_cat
|
|
state:
|
|
input1:
|
|
$link: nested_workflow#workflow_output
|
|
queries:
|
|
- input2:
|
|
$link: nested_workflow#workflow_output
|
|
"""
|
|
|
|
|
|
class BaseWorkflowsApiTestCase(api.ApiTestCase):
|
|
# TODO: Find a new file for this class.
|
|
|
|
def setUp(self):
|
|
super(BaseWorkflowsApiTestCase, self).setUp()
|
|
self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
|
|
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
|
|
self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor)
|
|
|
|
def _assert_user_has_workflow_with_name(self, name):
|
|
names = self._workflow_names()
|
|
assert name in names, "No workflows with name %s in users workflows <%s>" % (name, names)
|
|
|
|
def _workflow_names(self):
|
|
index_response = self._get("workflows")
|
|
self._assert_status_code_is(index_response, 200)
|
|
names = [w["name"] for w in index_response.json()]
|
|
return names
|
|
|
|
def import_workflow(self, workflow, **kwds):
|
|
upload_response = self.workflow_populator.import_workflow(workflow, **kwds)
|
|
return upload_response
|
|
|
|
def _upload_yaml_workflow(self, has_yaml, **kwds):
|
|
return self.workflow_populator.upload_yaml_workflow(has_yaml, **kwds)
|
|
|
|
def _setup_workflow_run(self, workflow=None, inputs_by='step_id', history_id=None, workflow_id=None):
|
|
if not workflow_id:
|
|
workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
if not history_id:
|
|
history_id = self.dataset_populator.new_history()
|
|
hda1 = self.dataset_populator.new_dataset(history_id, content="1 2 3")
|
|
hda2 = self.dataset_populator.new_dataset(history_id, content="4 5 6")
|
|
workflow_request = dict(
|
|
history="hist_id=%s" % history_id,
|
|
workflow_id=workflow_id,
|
|
)
|
|
label_map = {
|
|
'WorkflowInput1': self._ds_entry(hda1),
|
|
'WorkflowInput2': self._ds_entry(hda2)
|
|
}
|
|
if inputs_by == 'step_id':
|
|
ds_map = self._build_ds_map(workflow_id, label_map)
|
|
workflow_request["ds_map"] = ds_map
|
|
elif inputs_by == "step_index":
|
|
index_map = {
|
|
'0': self._ds_entry(hda1),
|
|
'1': self._ds_entry(hda2)
|
|
}
|
|
workflow_request["inputs"] = dumps(index_map)
|
|
workflow_request["inputs_by"] = 'step_index'
|
|
elif inputs_by == "name":
|
|
workflow_request["inputs"] = dumps(label_map)
|
|
workflow_request["inputs_by"] = 'name'
|
|
elif inputs_by in ["step_uuid", "uuid_implicitly"]:
|
|
uuid_map = {
|
|
workflow["steps"]["0"]["uuid"]: self._ds_entry(hda1),
|
|
workflow["steps"]["1"]["uuid"]: self._ds_entry(hda2),
|
|
}
|
|
workflow_request["inputs"] = dumps(uuid_map)
|
|
if inputs_by == "step_uuid":
|
|
workflow_request["inputs_by"] = "step_uuid"
|
|
|
|
return workflow_request, history_id
|
|
|
|
def _build_ds_map(self, workflow_id, label_map):
|
|
workflow_inputs = self._workflow_inputs(workflow_id)
|
|
ds_map = {}
|
|
for key, value in workflow_inputs.items():
|
|
label = value["label"]
|
|
if label in label_map:
|
|
ds_map[key] = label_map[label]
|
|
return dumps(ds_map)
|
|
|
|
def _ds_entry(self, hda):
|
|
src = 'hda'
|
|
if 'history_content_type' in hda and hda['history_content_type'] == "dataset_collection":
|
|
src = 'hdca'
|
|
return dict(src=src, id=hda["id"])
|
|
|
|
def _workflow_inputs(self, uploaded_workflow_id):
|
|
workflow_show_resposne = self._get("workflows/%s" % uploaded_workflow_id)
|
|
self._assert_status_code_is(workflow_show_resposne, 200)
|
|
workflow_inputs = workflow_show_resposne.json()["inputs"]
|
|
return workflow_inputs
|
|
|
|
def _invocation_details(self, workflow_id, invocation_id):
|
|
invocation_details_response = self._get("workflows/%s/usage/%s" % (workflow_id, invocation_id))
|
|
self._assert_status_code_is(invocation_details_response, 200)
|
|
invocation_details = invocation_details_response.json()
|
|
return invocation_details
|
|
|
|
def _run_jobs(self, has_workflow, history_id=None, wait=True, source_type=None, jobs_descriptions=None, expected_response=200, assert_ok=True):
|
|
def read_test_data(test_dict):
|
|
test_data_resolver = TestDataResolver()
|
|
filename = test_data_resolver.get_filename(test_dict["value"])
|
|
content = open(filename, "r").read()
|
|
return content
|
|
|
|
if history_id is None:
|
|
history_id = self.history_id
|
|
workflow_id = self._upload_yaml_workflow(
|
|
has_workflow, source_type=source_type
|
|
)
|
|
if jobs_descriptions is None:
|
|
assert source_type != "path"
|
|
jobs_descriptions = yaml.safe_load(has_workflow)
|
|
|
|
test_data = jobs_descriptions.get("test_data", {})
|
|
|
|
label_map = {}
|
|
inputs = {}
|
|
has_uploads = False
|
|
|
|
for key, value in test_data.items():
|
|
is_dict = isinstance(value, dict)
|
|
if is_dict and ("elements" in value or value.get("type", None) in ["list:paired", "list", "paired"]):
|
|
elements_data = value.get("elements", [])
|
|
elements = []
|
|
for element_data in elements_data:
|
|
identifier = element_data["identifier"]
|
|
input_type = element_data.get("type", "raw")
|
|
if input_type == "File":
|
|
content = read_test_data(element_data)
|
|
else:
|
|
content = element_data["content"]
|
|
elements.append((identifier, content))
|
|
# TODO: make this collection_type
|
|
collection_type = value["type"]
|
|
new_collection_kwds = {}
|
|
if "name" in value:
|
|
new_collection_kwds["name"] = value["name"]
|
|
if collection_type == "list:paired":
|
|
hdca = self.dataset_collection_populator.create_list_of_pairs_in_history(history_id, **new_collection_kwds).json()
|
|
elif collection_type == "list":
|
|
hdca = self.dataset_collection_populator.create_list_in_history(history_id, contents=elements, **new_collection_kwds).json()
|
|
else:
|
|
hdca = self.dataset_collection_populator.create_pair_in_history(history_id, contents=elements, **new_collection_kwds).json()
|
|
label_map[key] = self._ds_entry(hdca)
|
|
inputs[key] = hdca
|
|
has_uploads = True
|
|
elif is_dict and "type" in value:
|
|
input_type = value["type"]
|
|
if input_type == "File":
|
|
content = read_test_data(value)
|
|
new_dataset_kwds = {
|
|
"content": content
|
|
}
|
|
if "name" in value:
|
|
new_dataset_kwds["name"] = value["name"]
|
|
if "file_type" in value:
|
|
new_dataset_kwds["file_type"] = value["file_type"]
|
|
hda = self.dataset_populator.new_dataset(history_id, **new_dataset_kwds)
|
|
label_map[key] = self._ds_entry(hda)
|
|
has_uploads = True
|
|
elif input_type == "raw":
|
|
label_map[key] = value["value"]
|
|
inputs[key] = value["value"]
|
|
elif not is_dict:
|
|
has_uploads = True
|
|
hda = self.dataset_populator.new_dataset(history_id, content=value)
|
|
label_map[key] = self._ds_entry(hda)
|
|
inputs[key] = hda
|
|
else:
|
|
raise ValueError("Invalid test_data def %" % test_data)
|
|
workflow_request = dict(
|
|
history="hist_id=%s" % history_id,
|
|
workflow_id=workflow_id,
|
|
)
|
|
workflow_request["inputs"] = dumps(label_map)
|
|
workflow_request["inputs_by"] = 'name'
|
|
if has_uploads:
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
url = "workflows/%s/usage" % (workflow_id)
|
|
invocation_response = self._post(url, data=workflow_request)
|
|
self._assert_status_code_is(invocation_response, expected_response)
|
|
invocation = invocation_response.json()
|
|
invocation_id = invocation.get('id')
|
|
if invocation_id:
|
|
# Wait for workflow to become fully scheduled and then for all jobs
|
|
# complete.
|
|
if wait:
|
|
self.workflow_populator.wait_for_workflow(workflow_id, invocation_id, history_id, assert_ok=assert_ok)
|
|
jobs = self._history_jobs(history_id)
|
|
return RunJobsSummary(
|
|
history_id=history_id,
|
|
workflow_id=workflow_id,
|
|
invocation_id=invocation_id,
|
|
inputs=inputs,
|
|
jobs=jobs,
|
|
invocation=invocation,
|
|
workflow_request=workflow_request
|
|
)
|
|
|
|
def _history_jobs(self, history_id):
|
|
return self._get("jobs", {"history_id": history_id, "order_by": "create_time"}).json()
|
|
|
|
def _assert_history_job_count(self, history_id, n):
|
|
jobs = self._history_jobs(history_id)
|
|
self.assertEqual(len(jobs), n)
|
|
|
|
|
|
# Workflow API TODO:
|
|
# - Allow history_id as param to workflow run action. (hist_id)
|
|
# - Allow post to workflows/<workflow_id>/run in addition to posting to
|
|
# /workflows with id in payload.
|
|
# - Much more testing obviously, always more testing.
|
|
class WorkflowsApiTestCase(BaseWorkflowsApiTestCase):
|
|
|
|
def setUp(self):
|
|
super(WorkflowsApiTestCase, self).setUp()
|
|
|
|
def test_show_valid(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("dummy")
|
|
workflow_id = self.workflow_populator.simple_workflow("test_regular")
|
|
show_response = self._get("workflows/%s" % workflow_id, {"style": "instance"})
|
|
workflow = show_response.json()
|
|
self._assert_looks_like_instance_workflow_representation(workflow)
|
|
assert len(workflow["steps"]) == 3
|
|
self.assertEqual(sorted(step["id"] for step in workflow["steps"].values()), [0, 1, 2])
|
|
|
|
show_response = self._get("workflows/%s" % workflow_id, {"legacy": True})
|
|
workflow = show_response.json()
|
|
self._assert_looks_like_instance_workflow_representation(workflow)
|
|
assert len(workflow["steps"]) == 3
|
|
# Can't reay say what the legacy IDs are but must be greater than 3 because dummy
|
|
# workflow was created first in this instance.
|
|
self.assertNotEqual(sorted(step["id"] for step in workflow["steps"].values()), [0, 1, 2])
|
|
|
|
def test_show_invalid_key_is_400(self):
|
|
show_response = self._get("workflows/%s" % self._random_key())
|
|
self._assert_status_code_is(show_response, 400)
|
|
|
|
def test_cannot_show_private_workflow(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("test_not_importportable")
|
|
with self._different_user():
|
|
show_response = self._get("workflows/%s" % workflow_id)
|
|
self._assert_status_code_is(show_response, 403)
|
|
|
|
def test_delete(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("test_delete")
|
|
workflow_name = "test_delete"
|
|
self._assert_user_has_workflow_with_name(workflow_name)
|
|
workflow_url = self._api_url("workflows/%s" % workflow_id, use_key=True)
|
|
delete_response = delete(workflow_url)
|
|
self._assert_status_code_is(delete_response, 200)
|
|
# Make sure workflow is no longer in index by default.
|
|
assert workflow_name not in self._workflow_names()
|
|
|
|
def test_other_cannot_delete(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("test_other_delete")
|
|
with self._different_user():
|
|
workflow_url = self._api_url("workflows/%s" % workflow_id, use_key=True)
|
|
delete_response = delete(workflow_url)
|
|
self._assert_status_code_is(delete_response, 403)
|
|
|
|
def test_index(self):
|
|
index_response = self._get("workflows")
|
|
self._assert_status_code_is(index_response, 200)
|
|
assert isinstance(index_response.json(), list)
|
|
|
|
def test_upload(self):
|
|
self.__test_upload(use_deprecated_route=False)
|
|
|
|
def test_upload_deprecated(self):
|
|
self.__test_upload(use_deprecated_route=True)
|
|
|
|
def test_import_tools_requires_admin(self):
|
|
response = self.__test_upload(import_tools=True, assert_ok=False)
|
|
assert response.status_code == 403
|
|
|
|
def __test_upload(self, use_deprecated_route=False, name="test_import", workflow=None, assert_ok=True, import_tools=False):
|
|
if workflow is None:
|
|
workflow = self.workflow_populator.load_workflow(name=name)
|
|
data = dict(
|
|
workflow=dumps(workflow),
|
|
)
|
|
if import_tools:
|
|
data["import_tools"] = import_tools
|
|
if use_deprecated_route:
|
|
route = "workflows/upload"
|
|
else:
|
|
route = "workflows"
|
|
upload_response = self._post(route, data=data)
|
|
if assert_ok:
|
|
self._assert_status_code_is(upload_response, 200)
|
|
self._assert_user_has_workflow_with_name(name)
|
|
return upload_response
|
|
|
|
def test_update(self):
|
|
original_workflow = self.workflow_populator.load_workflow(name="test_import")
|
|
uuids = {}
|
|
labels = {}
|
|
|
|
for order_index, step_dict in original_workflow["steps"].items():
|
|
uuid = str(uuid4())
|
|
step_dict["uuid"] = uuid
|
|
uuids[order_index] = uuid
|
|
label = "label_%s" % order_index
|
|
step_dict["label"] = label
|
|
labels[order_index] = label
|
|
|
|
def check_label_and_uuid(order_index, step_dict):
|
|
assert order_index in uuids
|
|
assert order_index in labels
|
|
|
|
self.assertEqual(uuids[order_index], step_dict["uuid"])
|
|
self.assertEqual(labels[order_index], step_dict["label"])
|
|
|
|
upload_response = self.__test_upload(workflow=original_workflow)
|
|
workflow_id = upload_response.json()["id"]
|
|
|
|
def update(workflow_object):
|
|
put_response = self._update_workflow(workflow_id, workflow_object)
|
|
self._assert_status_code_is(put_response, 200)
|
|
return put_response
|
|
|
|
workflow_content = self._download_workflow(workflow_id)
|
|
steps = workflow_content["steps"]
|
|
|
|
def tweak_step(step):
|
|
order_index, step_dict = step
|
|
check_label_and_uuid(order_index, step_dict)
|
|
assert step_dict['position']['top'] != 1
|
|
assert step_dict['position']['left'] != 1
|
|
step_dict['position'] = {'top': 1, 'left': 1}
|
|
|
|
map(tweak_step, steps.items())
|
|
|
|
update(workflow_content)
|
|
|
|
def check_step(step):
|
|
order_index, step_dict = step
|
|
check_label_and_uuid(order_index, step_dict)
|
|
assert step_dict['position']['top'] == 1
|
|
assert step_dict['position']['left'] == 1
|
|
|
|
updated_workflow_content = self._download_workflow(workflow_id)
|
|
map(check_step, updated_workflow_content['steps'].items())
|
|
|
|
# Re-update against original worklfow...
|
|
update(original_workflow)
|
|
|
|
updated_workflow_content = self._download_workflow(workflow_id)
|
|
|
|
# Make sure the positions have been updated.
|
|
map(tweak_step, updated_workflow_content['steps'].items())
|
|
|
|
def test_update_no_tool_id(self):
|
|
workflow_object = self.workflow_populator.load_workflow(name="test_import")
|
|
upload_response = self.__test_upload(workflow=workflow_object)
|
|
workflow_id = upload_response.json()["id"]
|
|
del workflow_object["steps"]["2"]["tool_id"]
|
|
put_response = self._update_workflow(workflow_id, workflow_object)
|
|
self._assert_status_code_is(put_response, 400)
|
|
|
|
def test_update_missing_tool(self):
|
|
# Create allows missing tools, update doesn't currently...
|
|
workflow_object = self.workflow_populator.load_workflow(name="test_import")
|
|
upload_response = self.__test_upload(workflow=workflow_object)
|
|
workflow_id = upload_response.json()["id"]
|
|
workflow_object["steps"]["2"]["tool_id"] = "cat-not-found"
|
|
put_response = self._update_workflow(workflow_id, workflow_object)
|
|
self._assert_status_code_is(put_response, 400)
|
|
|
|
def test_require_unique_step_uuids(self):
|
|
workflow_dup_uuids = self.workflow_populator.load_workflow(name="test_import")
|
|
uuid0 = str(uuid4())
|
|
for step_dict in workflow_dup_uuids["steps"].values():
|
|
step_dict["uuid"] = uuid0
|
|
response = self.workflow_populator.create_workflow_response(workflow_dup_uuids)
|
|
self._assert_status_code_is(response, 400)
|
|
|
|
def test_require_unique_step_labels(self):
|
|
workflow_dup_label = self.workflow_populator.load_workflow(name="test_import")
|
|
for step_dict in workflow_dup_label["steps"].values():
|
|
step_dict["label"] = "my duplicated label"
|
|
response = self.workflow_populator.create_workflow_response(workflow_dup_label)
|
|
self._assert_status_code_is(response, 400)
|
|
|
|
def test_import_deprecated(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("test_import_published_deprecated", publish=True)
|
|
with self._different_user():
|
|
other_import_response = self.__import_workflow(workflow_id)
|
|
self._assert_status_code_is(other_import_response, 200)
|
|
self._assert_user_has_workflow_with_name("imported: test_import_published_deprecated")
|
|
|
|
def test_import_annotations(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("test_import_annotations", publish=True)
|
|
with self._different_user():
|
|
other_import_response = self.__import_workflow(workflow_id)
|
|
self._assert_status_code_is(other_import_response, 200)
|
|
|
|
# Test annotations preserved during upload and copied over during
|
|
# import.
|
|
other_id = other_import_response.json()["id"]
|
|
imported_workflow = self._show_workflow(other_id)
|
|
assert imported_workflow["annotation"] == "simple workflow"
|
|
step_annotations = set(step["annotation"] for step in imported_workflow["steps"].values())
|
|
assert "input1 description" in step_annotations
|
|
|
|
def test_import_subworkflows(self):
|
|
def get_subworkflow_content_id(workflow_id):
|
|
workflow_contents = self._download_workflow(workflow_id, style="editor")
|
|
steps = workflow_contents['steps']
|
|
subworkflow_step = next(s for s in steps.values() if s["type"] == "subworkflow")
|
|
return subworkflow_step['content_id']
|
|
|
|
workflow_id = self._upload_yaml_workflow(SIMPLE_NESTED_WORKFLOW_YAML, publish=True)
|
|
subworkflow_content_id = get_subworkflow_content_id(workflow_id)
|
|
with self._different_user():
|
|
other_import_response = self.__import_workflow(workflow_id)
|
|
self._assert_status_code_is(other_import_response, 200)
|
|
imported_workflow_id = other_import_response.json()["id"]
|
|
imported_subworkflow_content_id = get_subworkflow_content_id(imported_workflow_id)
|
|
assert subworkflow_content_id != imported_subworkflow_content_id
|
|
|
|
def test_not_importable_prevents_import(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("test_not_importportable")
|
|
with self._different_user():
|
|
other_import_response = self.__import_workflow(workflow_id)
|
|
self._assert_status_code_is(other_import_response, 403)
|
|
|
|
def test_import_published(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("test_import_published", publish=True)
|
|
with self._different_user():
|
|
other_import_response = self.__import_workflow(workflow_id, deprecated_route=True)
|
|
self._assert_status_code_is(other_import_response, 200)
|
|
self._assert_user_has_workflow_with_name("imported: test_import_published")
|
|
|
|
def test_export(self):
|
|
uploaded_workflow_id = self.workflow_populator.simple_workflow("test_for_export")
|
|
downloaded_workflow = self._download_workflow(uploaded_workflow_id)
|
|
assert downloaded_workflow["name"] == "test_for_export"
|
|
assert len(downloaded_workflow["steps"]) == 3
|
|
first_input = downloaded_workflow["steps"]["0"]["inputs"][0]
|
|
assert first_input["name"] == "WorkflowInput1"
|
|
assert first_input["description"] == "input1 description"
|
|
self._assert_has_keys(downloaded_workflow, "a_galaxy_workflow", "format-version", "annotation", "uuid", "steps")
|
|
for step in downloaded_workflow["steps"].values():
|
|
self._assert_has_keys(
|
|
step,
|
|
'id',
|
|
'type',
|
|
'tool_id',
|
|
'tool_version',
|
|
'name',
|
|
'tool_state',
|
|
'annotation',
|
|
'inputs',
|
|
'workflow_outputs',
|
|
'outputs'
|
|
)
|
|
if step['type'] == "tool":
|
|
self._assert_has_keys(step, "post_job_actions")
|
|
|
|
def test_export_editor(self):
|
|
uploaded_workflow_id = self.workflow_populator.simple_workflow("test_for_export")
|
|
downloaded_workflow = self._download_workflow(uploaded_workflow_id, style="editor")
|
|
self._assert_has_keys(downloaded_workflow, "name", "steps", "upgrade_messages")
|
|
for step in downloaded_workflow["steps"].values():
|
|
self._assert_has_keys(
|
|
step,
|
|
'id',
|
|
'type',
|
|
'content_id',
|
|
'name',
|
|
'tool_state',
|
|
'tooltip',
|
|
'data_inputs',
|
|
'data_outputs',
|
|
'config_form',
|
|
'annotation',
|
|
'post_job_actions',
|
|
'workflow_outputs',
|
|
'uuid',
|
|
'label',
|
|
)
|
|
|
|
def test_import_missing_tool(self):
|
|
workflow = self.workflow_populator.load_workflow_from_resource(name="test_workflow_missing_tool")
|
|
workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
workflow_description = self._show_workflow(workflow_id)
|
|
steps = workflow_description["steps"]
|
|
missing_tool_steps = [v for v in steps.values() if v['tool_id'] == 'cat_missing_tool']
|
|
assert len(missing_tool_steps) == 1
|
|
|
|
def test_import_no_tool_id(self):
|
|
# Import works with missing tools, but not with absent content/tool id.
|
|
workflow = self.workflow_populator.load_workflow_from_resource(name="test_workflow_missing_tool")
|
|
del workflow["steps"]["2"]["tool_id"]
|
|
create_response = self.__test_upload(workflow=workflow, assert_ok=False)
|
|
self._assert_status_code_is(create_response, 400)
|
|
|
|
def test_import_export_with_runtime_inputs(self):
|
|
workflow = self.workflow_populator.load_workflow_from_resource(name="test_workflow_with_runtime_input")
|
|
workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
downloaded_workflow = self._download_workflow(workflow_id)
|
|
assert len(downloaded_workflow["steps"]) == 2
|
|
runtime_input = downloaded_workflow["steps"]["1"]["inputs"][0]
|
|
assert runtime_input["description"].startswith("runtime parameter for tool")
|
|
assert runtime_input["name"] == "num_lines"
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_run_workflow_by_index(self):
|
|
self.__run_cat_workflow(inputs_by='step_index')
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_run_workflow_by_uuid(self):
|
|
self.__run_cat_workflow(inputs_by='step_uuid')
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_run_workflow_by_uuid_implicitly(self):
|
|
self.__run_cat_workflow(inputs_by='uuid_implicitly')
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_run_workflow_by_name(self):
|
|
self.__run_cat_workflow(inputs_by='name')
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_run_workflow(self):
|
|
self.__run_cat_workflow(inputs_by='step_id')
|
|
|
|
@skip_without_tool("multiple_versions")
|
|
def test_run_versioned_tools(self):
|
|
history_01_id = self.dataset_populator.new_history()
|
|
workflow_version_01 = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- tool_id: multiple_versions
|
|
tool_version: "0.1"
|
|
state:
|
|
inttest: 0
|
|
""")
|
|
self.__invoke_workflow(history_01_id, workflow_version_01)
|
|
self.dataset_populator.wait_for_history(history_01_id, assert_ok=True)
|
|
|
|
history_02_id = self.dataset_populator.new_history()
|
|
workflow_version_02 = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- tool_id: multiple_versions
|
|
tool_version: "0.2"
|
|
state:
|
|
inttest: 1
|
|
""")
|
|
self.__invoke_workflow(history_02_id, workflow_version_02)
|
|
self.dataset_populator.wait_for_history(history_02_id, assert_ok=True)
|
|
|
|
def __run_cat_workflow(self, inputs_by):
|
|
workflow = self.workflow_populator.load_workflow(name="test_for_run")
|
|
workflow["steps"]["0"]["uuid"] = str(uuid4())
|
|
workflow["steps"]["1"]["uuid"] = str(uuid4())
|
|
workflow_request, history_id = self._setup_workflow_run(workflow, inputs_by=inputs_by)
|
|
# TODO: This should really be a post to workflows/<workflow_id>/run or
|
|
# something like that.
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
invocation_id = run_workflow_response.json()["id"]
|
|
invocation = self._invocation_details(workflow_request["workflow_id"], invocation_id)
|
|
assert invocation["state"] == "scheduled", invocation
|
|
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
|
|
@skip_without_tool("collection_creates_pair")
|
|
def test_workflow_run_output_collections(self):
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- label: text_input
|
|
type: input
|
|
- label: split_up
|
|
tool_id: collection_creates_pair
|
|
state:
|
|
input1:
|
|
$link: text_input
|
|
- tool_id: collection_paired_test
|
|
state:
|
|
f1:
|
|
$link: split_up#paired_output
|
|
""")
|
|
history_id = self.dataset_populator.new_history()
|
|
hda1 = self.dataset_populator.new_dataset(history_id, content="a\nb\nc\nd\n")
|
|
inputs = {
|
|
'0': self._ds_entry(hda1),
|
|
}
|
|
invocation_id = self.__invoke_workflow(history_id, workflow_id, inputs)
|
|
self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
self.assertEqual("a\nc\nb\nd\n", self.dataset_populator.get_history_dataset_content(history_id, hid=0))
|
|
|
|
@skip_without_tool("job_properties")
|
|
@skip_without_tool("identifier_multiple_in_conditional")
|
|
def test_workflow_resume_from_failed_step(self):
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- tool_id: job_properties
|
|
state:
|
|
thebool: true
|
|
failbool: true
|
|
- tool_id: identifier_multiple_in_conditional
|
|
state:
|
|
outer_cond:
|
|
cond_param_outer: true
|
|
inner_cond:
|
|
cond_param_inner: true
|
|
input1:
|
|
$link: 0#out_file1
|
|
""")
|
|
history_id = self.dataset_populator.new_history()
|
|
invocation_id = self.__invoke_workflow(history_id, workflow_id)
|
|
self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id, assert_ok=False)
|
|
failed_dataset_one = self.dataset_populator.get_history_dataset_details(history_id, hid=1, wait=True, assert_ok=False)
|
|
assert failed_dataset_one['state'] == 'error', failed_dataset_one
|
|
paused_dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=5, wait=True, assert_ok=False)
|
|
assert paused_dataset['state'] == 'paused', paused_dataset
|
|
inputs = {"thebool": "false",
|
|
"failbool": "false",
|
|
"rerun_remap_job_id": failed_dataset_one['creating_job']}
|
|
self.dataset_populator.run_tool(tool_id='job_properties',
|
|
inputs=inputs,
|
|
history_id=history_id,
|
|
assert_ok=True)
|
|
unpaused_dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=5, wait=True, assert_ok=False)
|
|
assert unpaused_dataset['state'] == 'ok'
|
|
|
|
@skip_without_tool("job_properties")
|
|
@skip_without_tool("identifier_multiple_in_conditional")
|
|
def test_workflow_resume_from_failed_step_with_hdca_input(self):
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- tool_id: job_properties
|
|
state:
|
|
thebool: true
|
|
failbool: true
|
|
- tool_id: identifier_collection
|
|
state:
|
|
input1:
|
|
$link: 0#list_output
|
|
""")
|
|
with self.dataset_populator.test_history() as history_id:
|
|
invocation_id = self.__invoke_workflow(history_id, workflow_id)
|
|
self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id, assert_ok=False)
|
|
failed_dataset_one = self.dataset_populator.get_history_dataset_details(history_id, hid=1, wait=True, assert_ok=False)
|
|
assert failed_dataset_one['state'] == 'error', failed_dataset_one
|
|
paused_dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=5, wait=True, assert_ok=False)
|
|
assert paused_dataset['state'] == 'paused', paused_dataset
|
|
inputs = {"thebool": "false",
|
|
"failbool": "false",
|
|
"rerun_remap_job_id": failed_dataset_one['creating_job']}
|
|
self.dataset_populator.run_tool(tool_id='job_properties',
|
|
inputs=inputs,
|
|
history_id=history_id,
|
|
assert_ok=True)
|
|
unpaused_dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=5, wait=True,
|
|
assert_ok=False)
|
|
assert unpaused_dataset['state'] == 'ok'
|
|
|
|
@skip_without_tool("fail_identifier")
|
|
@skip_without_tool("identifier_multiple_in_conditional")
|
|
def test_workflow_resume_with_mapped_over_input(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
job_summary = self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- label: input_datasets
|
|
type: input_collection
|
|
- label: fail_identifier_1
|
|
tool_id: fail_identifier
|
|
state:
|
|
input1:
|
|
$link: input_datasets
|
|
failbool: true
|
|
- tool_id: identifier_collection
|
|
state:
|
|
input1:
|
|
$link: fail_identifier_1#out_file1
|
|
test_data:
|
|
input_datasets:
|
|
type: list
|
|
elements:
|
|
- identifier: fail
|
|
value: 1.fastq
|
|
type: File
|
|
- identifier: success
|
|
value: 1.fastq
|
|
type: File
|
|
""", history_id=history_id, assert_ok=False, wait=False)
|
|
self.wait_for_invocation_and_jobs(history_id, job_summary.workflow_id, job_summary.invocation_id, assert_ok=False)
|
|
history_contents = self.dataset_populator._get_contents_request(history_id=history_id).json()
|
|
paused_dataset = history_contents[-1]
|
|
failed_dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=5, assert_ok=False)
|
|
assert paused_dataset['state'] == 'paused', paused_dataset
|
|
assert failed_dataset['state'] == 'error', failed_dataset
|
|
inputs = {"input1": {'values': [{'src': 'hda',
|
|
'id': history_contents[0]['id']}]
|
|
},
|
|
"failbool": "false",
|
|
"rerun_remap_job_id": failed_dataset['creating_job']}
|
|
self.dataset_populator.run_tool(tool_id='fail_identifier',
|
|
inputs=inputs,
|
|
history_id=history_id,
|
|
assert_ok=True)
|
|
unpaused_dataset = self.dataset_populator.get_history_dataset_details(history_id, wait=True, assert_ok=False)
|
|
assert unpaused_dataset['state'] == 'ok'
|
|
contents = self.dataset_populator.get_history_dataset_content(history_id, hid=7, assert_ok=False)
|
|
assert contents == 'fail\nsuccess\n', contents
|
|
|
|
@skip_without_tool("collection_creates_pair")
|
|
def test_workflow_run_output_collection_mapping(self):
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- type: input_collection
|
|
- tool_id: collection_creates_pair
|
|
state:
|
|
input1:
|
|
$link: 0
|
|
- tool_id: collection_paired_test
|
|
state:
|
|
f1:
|
|
$link: 1#paired_output
|
|
- tool_id: cat_list
|
|
state:
|
|
input1:
|
|
$link: 2#out1
|
|
""")
|
|
history_id = self.dataset_populator.new_history()
|
|
hdca1 = self.dataset_collection_populator.create_list_in_history(history_id, contents=["a\nb\nc\nd\n", "e\nf\ng\nh\n"]).json()
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
inputs = {
|
|
'0': self._ds_entry(hdca1),
|
|
}
|
|
invocation_id = self.__invoke_workflow(history_id, workflow_id, inputs)
|
|
self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
self.assertEqual("a\nc\nb\nd\ne\ng\nf\nh\n", self.dataset_populator.get_history_dataset_content(history_id, hid=0))
|
|
|
|
@skip_without_tool("collection_split_on_column")
|
|
def test_workflow_run_dynamic_output_collections(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- label: text_input1
|
|
type: input
|
|
- label: text_input2
|
|
type: input
|
|
- label: cat_inputs
|
|
tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: text_input1
|
|
queries:
|
|
- input2:
|
|
$link: text_input2
|
|
- label: split_up
|
|
tool_id: collection_split_on_column
|
|
state:
|
|
input1:
|
|
$link: cat_inputs#out_file1
|
|
- tool_id: cat_list
|
|
state:
|
|
input1:
|
|
$link: split_up#split_output
|
|
""")
|
|
hda1 = self.dataset_populator.new_dataset(history_id, content="samp1\t10.0\nsamp2\t20.0\n")
|
|
hda2 = self.dataset_populator.new_dataset(history_id, content="samp1\t30.0\nsamp2\t40.0\n")
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
inputs = {
|
|
'0': self._ds_entry(hda1),
|
|
'1': self._ds_entry(hda2),
|
|
}
|
|
invocation_id = self.__invoke_workflow(history_id, workflow_id, inputs)
|
|
self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
|
|
details = self.dataset_populator.get_history_dataset_details(history_id, hid=0)
|
|
last_item_hid = details["hid"]
|
|
assert last_item_hid == 7, "Expected 7 history items, got %s" % last_item_hid
|
|
content = self.dataset_populator.get_history_dataset_content(history_id, hid=0)
|
|
self.assertEqual("10.0\n30.0\n20.0\n40.0\n", content)
|
|
|
|
@skip_without_tool("collection_split_on_column")
|
|
@skip_without_tool("min_repeat")
|
|
def test_workflow_run_dynamic_output_collections_2(self):
|
|
# A more advanced output collection workflow, testing regression of
|
|
# https://github.com/galaxyproject/galaxy/issues/776
|
|
with self.dataset_populator.test_history() as history_id:
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- label: test_input_1
|
|
type: input
|
|
- label: test_input_2
|
|
type: input
|
|
- label: test_input_3
|
|
type: input
|
|
- label: split_up
|
|
tool_id: collection_split_on_column
|
|
state:
|
|
input1:
|
|
$link: test_input_2
|
|
- label: min_repeat
|
|
tool_id: min_repeat
|
|
state:
|
|
queries:
|
|
- input:
|
|
$link: test_input_1
|
|
queries2:
|
|
- input2:
|
|
$link: split_up#split_output
|
|
""")
|
|
hda1 = self.dataset_populator.new_dataset(history_id, content="samp1\t10.0\nsamp2\t20.0\n")
|
|
hda2 = self.dataset_populator.new_dataset(history_id, content="samp1\t20.0\nsamp2\t40.0\n")
|
|
hda3 = self.dataset_populator.new_dataset(history_id, content="samp1\t30.0\nsamp2\t60.0\n")
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
inputs = {
|
|
'0': self._ds_entry(hda1),
|
|
'1': self._ds_entry(hda2),
|
|
'2': self._ds_entry(hda3),
|
|
}
|
|
invocation_id = self.__invoke_workflow(history_id, workflow_id, inputs)
|
|
self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
|
|
collection_details = self.dataset_populator.get_history_collection_details(history_id, hid=7)
|
|
assert collection_details["populated_state"] == "ok"
|
|
content = self.dataset_populator.get_history_dataset_content(history_id, hid=11)
|
|
self.assertEqual(content.strip(), "samp1\t10.0\nsamp2\t20.0")
|
|
|
|
@skip_without_tool("collection_split_on_column")
|
|
def test_workflow_run_dynamic_output_collections_3(self):
|
|
# Test a workflow that create a list:list:list followed by a mapping step.
|
|
history_id = self.dataset_populator.new_history()
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- label: text_input1
|
|
type: input
|
|
- label: text_input2
|
|
type: input
|
|
- label: cat_inputs
|
|
tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: text_input1
|
|
queries:
|
|
- input2:
|
|
$link: text_input2
|
|
- label: split_up_1
|
|
tool_id: collection_split_on_column
|
|
state:
|
|
input1:
|
|
$link: cat_inputs#out_file1
|
|
- label: split_up_2
|
|
tool_id: collection_split_on_column
|
|
state:
|
|
input1:
|
|
$link: split_up_1#split_output
|
|
- tool_id: cat
|
|
state:
|
|
input1:
|
|
$link: split_up_2#split_output
|
|
""")
|
|
hda1 = self.dataset_populator.new_dataset(history_id, content="samp1\t10.0\nsamp2\t20.0\n")
|
|
hda2 = self.dataset_populator.new_dataset(history_id, content="samp1\t30.0\nsamp2\t40.0\n")
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
inputs = {
|
|
'0': self._ds_entry(hda1),
|
|
'1': self._ds_entry(hda2),
|
|
}
|
|
invocation_id = self.__invoke_workflow(history_id, workflow_id, inputs)
|
|
self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
|
|
|
|
@skip_without_tool("mapper")
|
|
@skip_without_tool("pileup")
|
|
def test_workflow_metadata_validation_0(self):
|
|
# Testing regression of
|
|
# https://github.com/galaxyproject/galaxy/issues/1514
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- label: input_fastqs
|
|
type: input_collection
|
|
- label: reference
|
|
type: input
|
|
- label: map_over_mapper
|
|
tool_id: mapper
|
|
state:
|
|
input1:
|
|
$link: input_fastqs
|
|
reference:
|
|
$link: reference
|
|
- label: pileup
|
|
tool_id: pileup
|
|
state:
|
|
input1:
|
|
$link: map_over_mapper#out_file1
|
|
reference:
|
|
$link: reference
|
|
test_data:
|
|
input_fastqs:
|
|
type: list
|
|
elements:
|
|
- identifier: samp1
|
|
value: 1.fastq
|
|
type: File
|
|
- identifier: samp2
|
|
value: 1.fastq
|
|
type: File
|
|
reference:
|
|
value: 1.fasta
|
|
type: File
|
|
""", history_id=history_id)
|
|
|
|
def test_run_subworkflow_simple(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
workflow_run_description = """%s
|
|
|
|
test_data:
|
|
outer_input:
|
|
value: 1.bed
|
|
type: File
|
|
""" % SIMPLE_NESTED_WORKFLOW_YAML
|
|
self._run_jobs(workflow_run_description, history_id=history_id)
|
|
|
|
content = self.dataset_populator.get_history_dataset_content(history_id)
|
|
self.assertEqual("chr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\nchr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\n", content)
|
|
|
|
@skip_without_tool("cat1")
|
|
@skip_without_tool("collection_paired_test")
|
|
def test_workflow_run_zip_collections(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- label: test_input_1
|
|
type: input
|
|
- label: test_input_2
|
|
type: input
|
|
- label: first_cat
|
|
tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: test_input_1
|
|
- label: zip_it
|
|
tool_id: "__ZIP_COLLECTION__"
|
|
state:
|
|
input_forward:
|
|
$link: first_cat#out_file1
|
|
input_reverse:
|
|
$link: test_input_2
|
|
- label: concat_pair
|
|
tool_id: collection_paired_test
|
|
state:
|
|
f1:
|
|
$link: zip_it#output
|
|
""")
|
|
hda1 = self.dataset_populator.new_dataset(history_id, content="samp1\t10.0\nsamp2\t20.0\n")
|
|
hda2 = self.dataset_populator.new_dataset(history_id, content="samp1\t20.0\nsamp2\t40.0\n")
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
inputs = {
|
|
'0': self._ds_entry(hda1),
|
|
'1': self._ds_entry(hda2),
|
|
}
|
|
invocation_id = self.__invoke_workflow(history_id, workflow_id, inputs)
|
|
self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
|
|
content = self.dataset_populator.get_history_dataset_content(history_id)
|
|
self.assertEqual(content.strip(), "samp1\t10.0\nsamp2\t20.0\nsamp1\t20.0\nsamp2\t40.0")
|
|
|
|
def test_filter_failed_mapping(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
summary = self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- type: collection
|
|
label: input_c
|
|
steps:
|
|
- label: mixed_collection
|
|
tool_id: exit_code_from_file
|
|
state:
|
|
input:
|
|
$link: input_c
|
|
|
|
- label: filtered_collection
|
|
tool_id: "__FILTER_FAILED_DATASETS__"
|
|
state:
|
|
input:
|
|
$link: mixed_collection#out_file1
|
|
|
|
- tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: filtered_collection
|
|
test_data:
|
|
input_c:
|
|
type: list
|
|
elements:
|
|
- identifier: i1
|
|
content: "0"
|
|
- identifier: i2
|
|
content: "1"
|
|
""", history_id=history_id, wait=True, assert_ok=False)
|
|
jobs = summary.jobs
|
|
|
|
def filter_jobs_by_tool(tool_id):
|
|
return [j for j in summary.jobs if j["tool_id"] == tool_id]
|
|
|
|
assert len(filter_jobs_by_tool("upload1")) == 2, jobs
|
|
assert len(filter_jobs_by_tool("exit_code_from_file")) == 2, jobs
|
|
assert len(filter_jobs_by_tool("__FILTER_FAILED_DATASETS__")) == 1, jobs
|
|
# Follow proves one job was filtered out of the result of cat1
|
|
assert len(filter_jobs_by_tool("cat1")) == 1, jobs
|
|
|
|
def test_workflow_request(self):
|
|
workflow = self.workflow_populator.load_workflow(name="test_for_queue")
|
|
workflow_request, history_id = self._setup_workflow_run(workflow)
|
|
url = "workflows/%s/usage" % (workflow_request["workflow_id"])
|
|
del workflow_request["workflow_id"]
|
|
run_workflow_response = self._post(url, data=workflow_request)
|
|
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
# Give some time for workflow to get scheduled before scanning the history.
|
|
time.sleep(5)
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
|
|
def test_workflow_output_dataset(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
summary = self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
outputs:
|
|
- id: wf_output_1
|
|
source: first_cat#out_file1
|
|
steps:
|
|
- tool_id: cat1
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
|
|
test_data:
|
|
input1: "hello world"
|
|
""", history_id=history_id)
|
|
workflow_id = summary.workflow_id
|
|
invocation_id = summary.invocation_id
|
|
invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id))
|
|
self._assert_status_code_is(invocation_response, 200)
|
|
invocation = invocation_response.json()
|
|
self._assert_has_keys(invocation , "id", "outputs", "output_collections")
|
|
assert len(invocation["output_collections"]) == 0
|
|
assert len(invocation["outputs"]) == 1
|
|
output_content = self.dataset_populator.get_history_dataset_content(history_id, dataset_id=invocation["outputs"]["wf_output_1"]["id"])
|
|
assert "hello world" == output_content.strip()
|
|
|
|
def test_workflow_output_dataset_collection(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
summary = self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
type: data_collection_input
|
|
collection_type: list
|
|
outputs:
|
|
- id: wf_output_1
|
|
source: first_cat#out_file1
|
|
steps:
|
|
- tool_id: cat
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
test_data:
|
|
input1:
|
|
type: list
|
|
name: the_dataset_list
|
|
elements:
|
|
- identifier: el1
|
|
value: 1.fastq
|
|
type: File
|
|
""", history_id=history_id)
|
|
workflow_id = summary.workflow_id
|
|
invocation_id = summary.invocation_id
|
|
invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id))
|
|
self._assert_status_code_is(invocation_response, 200)
|
|
invocation = invocation_response.json()
|
|
self._assert_has_keys(invocation , "id", "outputs", "output_collections")
|
|
assert len(invocation["output_collections"]) == 1
|
|
assert len(invocation["outputs"]) == 0
|
|
output_content = self.dataset_populator.get_history_collection_details(history_id, content_id=invocation["output_collections"]["wf_output_1"]["id"])
|
|
self._assert_has_keys(output_content , "id", "elements")
|
|
assert output_content["collection_type"] == "list"
|
|
elements = output_content["elements"]
|
|
assert len(elements) == 1
|
|
elements0 = elements[0]
|
|
assert elements0["element_identifier"] == "el1"
|
|
|
|
def test_worklfow_input_mapping(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
summary = self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
outputs:
|
|
- id: wf_output_1
|
|
source: first_cat#out_file1
|
|
steps:
|
|
- tool_id: cat
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
test_data:
|
|
input1:
|
|
type: list
|
|
name: the_dataset_list
|
|
elements:
|
|
- identifier: el1
|
|
value: 1.fastq
|
|
type: File
|
|
- identifier: el2
|
|
value: 1.fastq
|
|
type: File
|
|
""", history_id=history_id)
|
|
workflow_id = summary.workflow_id
|
|
invocation_id = summary.invocation_id
|
|
invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id))
|
|
self._assert_status_code_is(invocation_response, 200)
|
|
invocation = invocation_response.json()
|
|
self._assert_has_keys(invocation , "id", "outputs", "output_collections")
|
|
assert len(invocation["output_collections"]) == 1
|
|
assert len(invocation["outputs"]) == 0
|
|
output_content = self.dataset_populator.get_history_collection_details(history_id, content_id=invocation["output_collections"]["wf_output_1"]["id"])
|
|
self._assert_has_keys(output_content , "id", "elements")
|
|
elements = output_content["elements"]
|
|
assert len(elements) == 2
|
|
elements0 = elements[0]
|
|
assert elements0["element_identifier"] == "el1"
|
|
|
|
@skip_without_tool("collection_creates_pair")
|
|
def test_workflow_run_input_mapping_with_output_collections(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
summary = self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
outputs:
|
|
- id: wf_output_1
|
|
source: split_up#paired_output
|
|
steps:
|
|
- label: text_input
|
|
type: input
|
|
- label: split_up
|
|
tool_id: collection_creates_pair
|
|
state:
|
|
input1:
|
|
$link: text_input
|
|
test_data:
|
|
text_input:
|
|
type: list
|
|
name: the_dataset_list
|
|
elements:
|
|
- identifier: el1
|
|
value: 1.fastq
|
|
type: File
|
|
- identifier: el2
|
|
value: 1.fastq
|
|
type: File
|
|
""", history_id=history_id)
|
|
workflow_id = summary.workflow_id
|
|
invocation_id = summary.invocation_id
|
|
invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id))
|
|
self._assert_status_code_is(invocation_response, 200)
|
|
invocation = invocation_response.json()
|
|
self._assert_has_keys(invocation , "id", "outputs", "output_collections")
|
|
assert len(invocation["output_collections"]) == 1
|
|
assert len(invocation["outputs"]) == 0
|
|
output_content = self.dataset_populator.get_history_collection_details(history_id, content_id=invocation["output_collections"]["wf_output_1"]["id"])
|
|
self._assert_has_keys(output_content , "id", "elements")
|
|
assert output_content["collection_type"] == "list:paired", output_content
|
|
elements = output_content["elements"]
|
|
assert len(elements) == 2
|
|
elements0 = elements[0]
|
|
assert elements0["element_identifier"] == "el1"
|
|
|
|
def test_workflow_run_input_mapping_with_subworkflows(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
summary = self._run_jobs("""%s
|
|
|
|
test_data:
|
|
outer_input:
|
|
type: list
|
|
name: the_dataset_list
|
|
elements:
|
|
- identifier: el1
|
|
value: 1.fastq
|
|
type: File
|
|
- identifier: el2
|
|
value: 1.fastq
|
|
type: File
|
|
""" % SIMPLE_NESTED_WORKFLOW_YAML, history_id=history_id)
|
|
workflow_id = summary.workflow_id
|
|
invocation_id = summary.invocation_id
|
|
invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id))
|
|
self._assert_status_code_is(invocation_response, 200)
|
|
invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id))
|
|
self._assert_status_code_is(invocation_response, 200)
|
|
invocation = invocation_response.json()
|
|
self._assert_has_keys(invocation , "id", "outputs", "output_collections")
|
|
assert len(invocation["output_collections"]) == 1, invocation
|
|
assert len(invocation["outputs"]) == 0
|
|
output_content = self.dataset_populator.get_history_collection_details(history_id, content_id=invocation["output_collections"]["outer_output"]["id"])
|
|
self._assert_has_keys(output_content , "id", "elements")
|
|
assert output_content["collection_type"] == "list", output_content
|
|
elements = output_content["elements"]
|
|
assert len(elements) == 2
|
|
elements0 = elements[0]
|
|
assert elements0["element_identifier"] == "el1"
|
|
|
|
@skip_without_tool("cat_list")
|
|
@skip_without_tool("random_lines1")
|
|
@skip_without_tool("split")
|
|
def test_subworkflow_recover_mapping_1(self):
|
|
# This test case tests an outer workflow continues to scheduling and handle
|
|
# collection mapping properly after the last step of a subworkflow requires delayed
|
|
# evaluation. Testing rescheduling and propagating connections within a subworkflow
|
|
# is handled by the next test case.
|
|
with self.dataset_populator.test_history() as history_id:
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: outer_input
|
|
outputs:
|
|
- id: outer_output
|
|
source: second_cat#out_file1
|
|
steps:
|
|
- tool_id: cat1
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: outer_input
|
|
- run:
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: inner_input
|
|
outputs:
|
|
- id: workflow_output
|
|
source: random_lines#out_file1
|
|
steps:
|
|
- tool_id: random_lines1
|
|
label: random_lines
|
|
state:
|
|
num_lines: 2
|
|
input:
|
|
$link: inner_input
|
|
seed_source:
|
|
seed_source_selector: set_seed
|
|
seed: asdf
|
|
label: nested_workflow
|
|
connect:
|
|
inner_input: first_cat#out_file1
|
|
- tool_id: split
|
|
label: split
|
|
state:
|
|
input1:
|
|
$link: nested_workflow#workflow_output
|
|
- tool_id: cat_list
|
|
label: second_cat
|
|
state:
|
|
input1:
|
|
$link: split#output
|
|
|
|
test_data:
|
|
outer_input:
|
|
value: 1.bed
|
|
type: File
|
|
""", history_id=history_id, wait=True)
|
|
self.assertEqual("chr16\t142908\t143003\tCCDS10397.1_cds_0_0_chr16_142909_f\t0\t+\nchr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\n", self.dataset_populator.get_history_dataset_content(history_id))
|
|
|
|
@skip_without_tool("cat_list")
|
|
@skip_without_tool("random_lines1")
|
|
@skip_without_tool("split")
|
|
def test_subworkflow_recover_mapping_2(self):
|
|
# Like the above test case, this test case tests an outer workflow continues to
|
|
# schedule and handle collection mapping properly after a subworkflow needs to be
|
|
# delayed, but this also tests recovering and handling scheduling within the subworkflow
|
|
# since the delayed step (split) isn't the last step of the subworkflow.
|
|
with self.dataset_populator.test_history() as history_id:
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: outer_input
|
|
outputs:
|
|
- id: outer_output
|
|
source: second_cat#out_file1
|
|
steps:
|
|
- tool_id: cat1
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: outer_input
|
|
- run:
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: inner_input
|
|
outputs:
|
|
- id: workflow_output
|
|
source: inner_cat#out_file1
|
|
steps:
|
|
- tool_id: random_lines1
|
|
label: random_lines
|
|
state:
|
|
num_lines: 2
|
|
input:
|
|
$link: inner_input
|
|
seed_source:
|
|
seed_source_selector: set_seed
|
|
seed: asdf
|
|
- tool_id: split
|
|
label: split
|
|
state:
|
|
input1:
|
|
$link: random_lines#out_file1
|
|
- tool_id: cat1
|
|
label: inner_cat
|
|
state:
|
|
input1:
|
|
$link: split#output
|
|
|
|
label: nested_workflow
|
|
connect:
|
|
inner_input: first_cat#out_file1
|
|
- tool_id: cat_list
|
|
label: second_cat
|
|
state:
|
|
input1:
|
|
$link: nested_workflow#workflow_output
|
|
|
|
test_data:
|
|
outer_input:
|
|
value: 1.bed
|
|
type: File
|
|
""", history_id=history_id, wait=True)
|
|
self.assertEqual("chr16\t142908\t143003\tCCDS10397.1_cds_0_0_chr16_142909_f\t0\t+\nchr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\n", self.dataset_populator.get_history_dataset_content(history_id))
|
|
|
|
@skip_without_tool("cat_list")
|
|
@skip_without_tool("random_lines1")
|
|
@skip_without_tool("split")
|
|
def test_recover_mapping_in_subworkflow(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: outer_input
|
|
outputs:
|
|
- id: outer_output
|
|
source: second_cat#out_file1
|
|
steps:
|
|
- tool_id: cat1
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: outer_input
|
|
- run:
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: inner_input
|
|
outputs:
|
|
- id: workflow_output
|
|
source: split#output
|
|
steps:
|
|
- tool_id: random_lines1
|
|
label: random_lines
|
|
state:
|
|
num_lines: 2
|
|
input:
|
|
$link: inner_input
|
|
seed_source:
|
|
seed_source_selector: set_seed
|
|
seed: asdf
|
|
- tool_id: split
|
|
label: split
|
|
state:
|
|
input1:
|
|
$link: random_lines#out_file1
|
|
label: nested_workflow
|
|
connect:
|
|
inner_input: first_cat#out_file1
|
|
- tool_id: cat_list
|
|
label: second_cat
|
|
state:
|
|
input1:
|
|
$link: nested_workflow#workflow_output
|
|
|
|
test_data:
|
|
outer_input:
|
|
value: 1.bed
|
|
type: File
|
|
""", history_id=history_id, wait=True)
|
|
self.assertEqual("chr16\t142908\t143003\tCCDS10397.1_cds_0_0_chr16_142909_f\t0\t+\nchr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\n", self.dataset_populator.get_history_dataset_content(history_id))
|
|
|
|
@skip_without_tool("empty_list")
|
|
@skip_without_tool("count_list")
|
|
@skip_without_tool("random_lines1")
|
|
def test_empty_list_mapping(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
outputs:
|
|
- id: count_list
|
|
source: count_list#out_file1
|
|
steps:
|
|
- tool_id: empty_list
|
|
label: empty_list
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
- tool_id: random_lines1
|
|
label: random_lines
|
|
state:
|
|
num_lines: 2
|
|
input:
|
|
$link: empty_list#output
|
|
seed_source:
|
|
seed_source_selector: set_seed
|
|
seed: asdf
|
|
- tool_id: count_list
|
|
label: count_list
|
|
state:
|
|
input1:
|
|
$link: random_lines#out_file1
|
|
|
|
test_data:
|
|
input1:
|
|
value: 1.bed
|
|
type: File
|
|
""", history_id=history_id, wait=True)
|
|
self.assertEqual("0\n", self.dataset_populator.get_history_dataset_content(history_id))
|
|
|
|
@skip_without_tool("empty_list")
|
|
@skip_without_tool("count_multi_file")
|
|
@skip_without_tool("random_lines1")
|
|
def test_empty_list_reduction(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
outputs:
|
|
- id: count_multi_file
|
|
source: count_multi_file#out_file1
|
|
steps:
|
|
- tool_id: empty_list
|
|
label: empty_list
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
- tool_id: random_lines1
|
|
label: random_lines
|
|
state:
|
|
num_lines: 2
|
|
input:
|
|
$link: empty_list#output
|
|
seed_source:
|
|
seed_source_selector: set_seed
|
|
seed: asdf
|
|
- tool_id: count_multi_file
|
|
label: count_multi_file
|
|
state:
|
|
input1:
|
|
$link: random_lines#out_file1
|
|
|
|
test_data:
|
|
input1:
|
|
value: 1.bed
|
|
type: File
|
|
""", history_id=history_id, wait=True)
|
|
self.assertEqual("0\n", self.dataset_populator.get_history_dataset_content(history_id))
|
|
|
|
@skip_without_tool("cat")
|
|
def test_cancel_new_workflow_when_history_deleted(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
# Invoke a workflow with a pause step.
|
|
uploaded_workflow_id, invocation_id = self._invoke_paused_workflow(history_id)
|
|
|
|
# There is no pause of anything in here, so likely the invocation is
|
|
# is still in a new state. If it isn't that is fine, continue with the
|
|
# test it will just happen to test the same thing as below.
|
|
|
|
# Wait for all the datasets to complete, make sure the workflow invocation
|
|
# is not complete.
|
|
self._assert_invocation_non_terminal(uploaded_workflow_id, invocation_id)
|
|
|
|
self._delete("histories/%s" % history_id)
|
|
|
|
invocation_cancelled = self._wait_for_invocation_state(uploaded_workflow_id, invocation_id, 'cancelled')
|
|
assert invocation_cancelled, "Workflow state is not cancelled..."
|
|
|
|
@skip_without_tool("cat")
|
|
def test_cancel_ready_workflow_when_history_deleted(self):
|
|
# Same as previous test but make sure invocation isn't a new state before
|
|
# cancelling.
|
|
with self.dataset_populator.test_history() as history_id:
|
|
# Invoke a workflow with a pause step.
|
|
uploaded_workflow_id, invocation_id = self._invoke_paused_workflow(history_id)
|
|
|
|
# Wait for at least one scheduling step.
|
|
self._wait_for_invocation_non_new(uploaded_workflow_id, invocation_id)
|
|
|
|
# Wait for all the datasets to complete, make sure the workflow invocation
|
|
# is not complete.
|
|
self._assert_invocation_non_terminal(uploaded_workflow_id, invocation_id)
|
|
|
|
self._delete("histories/%s" % history_id)
|
|
|
|
invocation_cancelled = self._wait_for_invocation_state(uploaded_workflow_id, invocation_id, 'cancelled')
|
|
assert invocation_cancelled, "Workflow state is not cancelled..."
|
|
|
|
@skip_without_tool("cat")
|
|
def test_workflow_pause(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
# Invoke a workflow with a pause step.
|
|
uploaded_workflow_id, invocation_id = self._invoke_paused_workflow(history_id)
|
|
|
|
# Wait for at least one scheduling step.
|
|
self._wait_for_invocation_non_new(uploaded_workflow_id, invocation_id)
|
|
|
|
# Make sure the history didn't enter a failed state in there.
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
|
|
# Assert the workflow hasn't finished scheduling, we can be pretty sure we
|
|
# are at the pause step in this case then.
|
|
self._assert_invocation_non_terminal(uploaded_workflow_id, invocation_id)
|
|
|
|
# Review the paused steps to allow the workflow to continue.
|
|
self.__review_paused_steps(uploaded_workflow_id, invocation_id, order_index=2, action=True)
|
|
|
|
# Wait for the workflow to finish scheduling and ensure both the invocation
|
|
# and the history are in valid states.
|
|
invocation_scheduled = self._wait_for_invocation_state(uploaded_workflow_id, invocation_id, 'scheduled')
|
|
assert invocation_scheduled, "Workflow state is not scheduled..."
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
|
|
@skip_without_tool("cat")
|
|
def test_workflow_pause_cancel(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
# Invoke a workflow with a pause step.
|
|
uploaded_workflow_id, invocation_id = self._invoke_paused_workflow(history_id)
|
|
|
|
# Wait for at least one scheduling step.
|
|
self._wait_for_invocation_non_new(uploaded_workflow_id, invocation_id)
|
|
|
|
# Make sure the history didn't enter a failed state in there.
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
|
|
# Assert the workflow hasn't finished scheduling, we can be pretty sure we
|
|
# are at the pause step in this case then.
|
|
self._assert_invocation_non_terminal(uploaded_workflow_id, invocation_id)
|
|
|
|
# Review the paused workflow and cancel it at the paused step.
|
|
self.__review_paused_steps(uploaded_workflow_id, invocation_id, order_index=2, action=False)
|
|
|
|
# Ensure the workflow eventually becomes cancelled.
|
|
invocation_cancelled = self._wait_for_invocation_state(uploaded_workflow_id, invocation_id, 'cancelled')
|
|
assert invocation_cancelled, "Workflow state is not cancelled..."
|
|
|
|
@skip_without_tool("head")
|
|
def test_workflow_map_reduce_pause(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_map_reduce_pause")
|
|
uploaded_workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
hda1 = self.dataset_populator.new_dataset(history_id, content="reviewed\nunreviewed")
|
|
hdca1 = self.dataset_collection_populator.create_list_in_history(history_id, contents=["1\n2\n3", "4\n5\n6"]).json()
|
|
index_map = {
|
|
'0': self._ds_entry(hda1),
|
|
'1': self._ds_entry(hdca1),
|
|
}
|
|
invocation_id = self.__invoke_workflow(history_id, uploaded_workflow_id, index_map)
|
|
|
|
# Wait for at least one scheduling step.
|
|
self._wait_for_invocation_non_new(uploaded_workflow_id, invocation_id)
|
|
|
|
# Make sure the history didn't enter a failed state in there.
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
|
|
# Assert the workflow hasn't finished scheduling, we can be pretty sure we
|
|
# are at the pause step in this case then.
|
|
self._assert_invocation_non_terminal(uploaded_workflow_id, invocation_id)
|
|
|
|
self.__review_paused_steps(uploaded_workflow_id, invocation_id, order_index=4, action=True)
|
|
self.wait_for_invocation_and_jobs(history_id, uploaded_workflow_id, invocation_id)
|
|
invocation = self._invocation_details(uploaded_workflow_id, invocation_id)
|
|
assert invocation['state'] == 'scheduled'
|
|
self.assertEqual("reviewed\n1\nreviewed\n4\n", self.dataset_populator.get_history_dataset_content(history_id))
|
|
|
|
@skip_without_tool("cat")
|
|
def test_cancel_workflow_invocation(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
# Invoke a workflow with a pause step.
|
|
uploaded_workflow_id, invocation_id = self._invoke_paused_workflow(history_id)
|
|
|
|
# Wait for at least one scheduling step.
|
|
self._wait_for_invocation_non_new(uploaded_workflow_id, invocation_id)
|
|
|
|
# Make sure the history didn't enter a failed state in there.
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
|
|
# Assert the workflow hasn't finished scheduling, we can be pretty sure we
|
|
# are at the pause step in this case then.
|
|
self._assert_invocation_non_terminal(uploaded_workflow_id, invocation_id)
|
|
|
|
invocation_url = self._api_url("workflows/%s/usage/%s" % (uploaded_workflow_id, invocation_id), use_key=True)
|
|
delete_response = delete(invocation_url)
|
|
self._assert_status_code_is(delete_response, 200)
|
|
|
|
invocation = self._invocation_details(uploaded_workflow_id, invocation_id)
|
|
assert invocation['state'] == 'cancelled'
|
|
|
|
def test_run_with_implicit_connection(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
run_summary = self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- label: test_input
|
|
type: input
|
|
- label: first_cat
|
|
tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: test_input
|
|
- label: the_pause
|
|
type: pause
|
|
connect:
|
|
input:
|
|
- first_cat#out_file1
|
|
- label: second_cat
|
|
tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: the_pause
|
|
- label: third_cat
|
|
tool_id: random_lines1
|
|
connect:
|
|
$step: second_cat
|
|
state:
|
|
num_lines: 1
|
|
input:
|
|
$link: test_input
|
|
seed_source:
|
|
seed_source_selector: set_seed
|
|
seed: asdf
|
|
test_data:
|
|
test_input: "hello world"
|
|
""", history_id=history_id, wait=False)
|
|
history_id = run_summary.history_id
|
|
workflow_id = run_summary.workflow_id
|
|
invocation_id = run_summary.invocation_id
|
|
# Wait for first two jobs to be scheduled - upload and first cat.
|
|
wait_on(lambda: len(self._history_jobs(history_id)) >= 2 or None, "history jobs")
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
invocation = self._invocation_details(workflow_id, invocation_id)
|
|
assert invocation['state'] != 'scheduled', invocation
|
|
# Expect two jobs - the upload and first cat. randomlines shouldn't run
|
|
# it is implicitly dependent on second cat.
|
|
self._assert_history_job_count(history_id, 2)
|
|
|
|
self.__review_paused_steps(workflow_id, invocation_id, order_index=2, action=True)
|
|
self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
|
|
self._assert_history_job_count(history_id, 4)
|
|
|
|
def test_run_with_validated_parameter_connection_valid(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
run_summary = self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- label: text_input
|
|
type: text
|
|
steps:
|
|
- tool_id: validation_repeat
|
|
state:
|
|
r2:
|
|
- text:
|
|
$link: text_input
|
|
test_data:
|
|
text_input:
|
|
value: "abd"
|
|
type: raw
|
|
""", history_id=history_id, wait=True)
|
|
time.sleep(10)
|
|
self.workflow_populator.wait_for_invocation(run_summary.workflow_id, run_summary.invocation_id)
|
|
jobs = self._history_jobs(history_id)
|
|
assert len(jobs) == 1
|
|
|
|
def test_run_with_validated_parameter_connection_invalid(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- label: text_input
|
|
type: text
|
|
steps:
|
|
- tool_id: validation_repeat
|
|
state:
|
|
r2:
|
|
- text:
|
|
$link: text_input
|
|
test_data:
|
|
text_input:
|
|
value: ""
|
|
type: raw
|
|
""", history_id=history_id, wait=True, assert_ok=False)
|
|
|
|
def test_run_with_text_connection(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- label: data_input
|
|
type: data
|
|
- label: text_input
|
|
type: text
|
|
steps:
|
|
- label: randomlines
|
|
tool_id: random_lines1
|
|
state:
|
|
num_lines: 1
|
|
input:
|
|
$link: data_input
|
|
seed_source:
|
|
seed_source_selector: set_seed
|
|
seed:
|
|
$link: text_input
|
|
test_data:
|
|
data_input:
|
|
value: 1.bed
|
|
type: File
|
|
text_input:
|
|
value: asdf
|
|
type: raw
|
|
""", history_id=history_id)
|
|
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
content = self.dataset_populator.get_history_dataset_content(history_id)
|
|
self.assertEqual("chr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\n", content)
|
|
|
|
def wait_for_invocation_and_jobs(self, history_id, workflow_id, invocation_id, assert_ok=True):
|
|
state = self.workflow_populator.wait_for_invocation(workflow_id, invocation_id)
|
|
if assert_ok:
|
|
assert state == "scheduled", state
|
|
time.sleep(.5)
|
|
self.dataset_populator.wait_for_history_jobs(history_id, assert_ok=assert_ok)
|
|
time.sleep(.5)
|
|
|
|
@flakey
|
|
@skip_without_tool('cat1')
|
|
def test_workflow_rerun_with_use_cached_job(self):
|
|
workflow = self.workflow_populator.load_workflow(name="test_for_run")
|
|
# We launch a workflow
|
|
with self.dataset_populator.test_history() as history_id:
|
|
workflow_request, _ = self._setup_workflow_run(workflow, history_id=history_id)
|
|
run_workflow_response = self._post("workflows", data=workflow_request).json()
|
|
# We copy the workflow inputs to a new history
|
|
new_workflow_request = workflow_request.copy()
|
|
new_ds_map = json.loads(new_workflow_request['ds_map'])
|
|
with self.dataset_populator.test_history() as new_history_id:
|
|
for key, input_values in run_workflow_response['inputs'].items():
|
|
copy_payload = {"content": input_values['id'], "source": "hda", "type": "dataset"}
|
|
copy_response = self._post("histories/%s/contents" % new_history_id, data=copy_payload).json()
|
|
new_ds_map[key]['id'] = copy_response['id']
|
|
new_workflow_request['ds_map'] = json.dumps(new_ds_map)
|
|
new_workflow_request['history'] = "hist_id=%s" % new_history_id
|
|
new_workflow_request['use_cached_job'] = True
|
|
# We run the workflow again, it should not produce any new outputs
|
|
new_workflow_response = self._post("workflows", data=new_workflow_request).json()
|
|
first_wf_output = self._get("datasets/%s" % run_workflow_response['outputs'][0]).json()
|
|
second_wf_output = self._get("datasets/%s" % new_workflow_response['outputs'][0]).json()
|
|
assert first_wf_output['file_name'] == second_wf_output['file_name'], \
|
|
"first output :\n%s\nsecond output: %s" % (first_wf_output, second_wf_output)
|
|
|
|
@skip_without_tool('cat1')
|
|
def test_nested_workflow_rerun_with_use_cached_job(self):
|
|
with self.dataset_populator.test_history() as history_id_one, self.dataset_populator.test_history() as history_id_two:
|
|
workflow_run_description = """%s
|
|
|
|
test_data:
|
|
outer_input:
|
|
value: 1.bed
|
|
type: File
|
|
""" % SIMPLE_NESTED_WORKFLOW_YAML
|
|
run_jobs_summary = self._run_jobs(workflow_run_description, history_id=history_id_one)
|
|
self.dataset_populator.wait_for_history(history_id_one, assert_ok=True)
|
|
workflow_request = run_jobs_summary.workflow_request
|
|
# We copy the inputs to a new history and re-reun the workflow
|
|
inputs = json.loads(workflow_request['inputs'])
|
|
dataset_type = inputs['outer_input']['src']
|
|
dataset_id = inputs['outer_input']['id']
|
|
copy_payload = {"content": dataset_id, "source": dataset_type, "type": "dataset"}
|
|
copy_response = self._post("histories/%s/contents" % history_id_two, data=copy_payload)
|
|
self._assert_status_code_is(copy_response, 200)
|
|
new_dataset_id = copy_response.json()['id']
|
|
inputs['outer_input']['id'] = new_dataset_id
|
|
workflow_request['use_cached_job'] = True
|
|
workflow_request['history'] = "hist_id={history_id_two}".format(history_id_two=history_id_two)
|
|
workflow_request['inputs'] = json.dumps(inputs)
|
|
run_workflow_response = self._post("workflows", data=run_jobs_summary.workflow_request).json()
|
|
self.workflow_populator.wait_for_workflow(workflow_request['workflow_id'],
|
|
run_workflow_response['id'],
|
|
history_id_two,
|
|
assert_ok=True)
|
|
# Now make sure that the HDAs in each history point to the same dataset instances
|
|
history_one_contents = self.__history_contents(history_id_one)
|
|
history_two_contents = self.__history_contents(history_id_two)
|
|
assert len(history_one_contents) == len(history_two_contents)
|
|
for i, (item_one, item_two) in enumerate(zip(history_one_contents, history_two_contents)):
|
|
assert item_one['dataset_id'] == item_two['dataset_id'], \
|
|
'Dataset ids should match, but "%s" and "%s" are not the same for History item %i.' % (item_one['dataset_id'],
|
|
item_two['dataset_id'],
|
|
i + 1)
|
|
|
|
def test_cannot_run_inaccessible_workflow(self):
|
|
workflow = self.workflow_populator.load_workflow(name="test_for_run_cannot_access")
|
|
workflow_request, history_id = self._setup_workflow_run(workflow)
|
|
with self._different_user():
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 403)
|
|
|
|
def test_400_on_invalid_workflow_id(self):
|
|
workflow = self.workflow_populator.load_workflow(name="test_for_run_does_not_exist")
|
|
workflow_request, history_id = self._setup_workflow_run(workflow)
|
|
workflow_request["workflow_id"] = self._random_key()
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 400)
|
|
|
|
def test_cannot_run_against_other_users_history(self):
|
|
workflow = self.workflow_populator.load_workflow(name="test_for_run_does_not_exist")
|
|
workflow_request, history_id = self._setup_workflow_run(workflow)
|
|
with self._different_user():
|
|
other_history_id = self.dataset_populator.new_history()
|
|
workflow_request["history"] = "hist_id=%s" % other_history_id
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 403)
|
|
|
|
@skip_without_tool("cat")
|
|
@skip_without_tool("cat_list")
|
|
def test_workflow_run_with_matching_lists(self):
|
|
workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_matching_lists")
|
|
workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
history_id = self.dataset_populator.new_history()
|
|
hdca1 = self.dataset_collection_populator.create_list_in_history(history_id, contents=[("sample1-1", "1 2 3"), ("sample2-1", "7 8 9")]).json()
|
|
hdca2 = self.dataset_collection_populator.create_list_in_history(history_id, contents=[("sample1-2", "4 5 6"), ("sample2-2", "0 a b")]).json()
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
label_map = {"list1": self._ds_entry(hdca1), "list2": self._ds_entry(hdca2)}
|
|
workflow_request = dict(
|
|
history="hist_id=%s" % history_id,
|
|
workflow_id=workflow_id,
|
|
ds_map=self._build_ds_map(workflow_id, label_map),
|
|
)
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
self.assertEqual("1 2 3\n4 5 6\n7 8 9\n0 a b\n", self.dataset_populator.get_history_dataset_content(history_id))
|
|
|
|
def test_workflow_stability(self):
|
|
# Run this index stability test with following command:
|
|
# ./run_tests.sh test/api/test_workflows.py:WorkflowsApiTestCase.test_workflow_stability
|
|
num_tests = 1
|
|
for workflow_file in ["test_workflow_topoambigouity", "test_workflow_topoambigouity_auto_laidout"]:
|
|
workflow = self.workflow_populator.load_workflow_from_resource(workflow_file)
|
|
last_step_map = self._step_map(workflow)
|
|
for i in range(num_tests):
|
|
uploaded_workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
downloaded_workflow = self._download_workflow(uploaded_workflow_id)
|
|
step_map = self._step_map(downloaded_workflow)
|
|
assert step_map == last_step_map
|
|
last_step_map = step_map
|
|
|
|
def _step_map(self, workflow):
|
|
# Build dict mapping 'tep index to input name.
|
|
step_map = {}
|
|
for step_index, step in workflow["steps"].items():
|
|
if step["type"] == "data_input":
|
|
step_map[step_index] = step["inputs"][0]["name"]
|
|
return step_map
|
|
|
|
def test_empty_create(self):
|
|
response = self._post("workflows")
|
|
self._assert_status_code_is(response, 400)
|
|
self._assert_error_code_is(response, error_codes.USER_REQUEST_MISSING_PARAMETER)
|
|
|
|
def test_invalid_create_multiple_types(self):
|
|
data = {
|
|
'shared_workflow_id': '1234567890abcdef',
|
|
'from_history_id': '1234567890abcdef'
|
|
}
|
|
response = self._post("workflows", data)
|
|
self._assert_status_code_is(response, 400)
|
|
self._assert_error_code_is(response, error_codes.USER_REQUEST_INVALID_PARAMETER)
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_run_with_pja(self):
|
|
workflow = self.workflow_populator.load_workflow(name="test_for_pja_run", add_pja=True)
|
|
workflow_request, history_id = self._setup_workflow_run(workflow, inputs_by='step_index')
|
|
workflow_request["replacement_params"] = dumps(dict(replaceme="was replaced"))
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
content = self.dataset_populator.get_history_dataset_details(history_id, wait=True, assert_ok=True)
|
|
assert content["name"] == "foo was replaced"
|
|
|
|
@skip_without_tool("output_filter")
|
|
def test_optional_workflow_output(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
run_object = self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs: []
|
|
outputs:
|
|
- id: wf_output_1
|
|
source: output_filter#out_1
|
|
steps:
|
|
- tool_id: output_filter
|
|
label: output_filter
|
|
state:
|
|
produce_out_1: False
|
|
filter_text_1: '1'
|
|
test_data: {}
|
|
""", history_id=history_id, wait=False)
|
|
self.wait_for_invocation_and_jobs(history_id, run_object.workflow_id, run_object.invocation_id)
|
|
contents = self.__history_contents(history_id)
|
|
assert len(contents) == 1
|
|
okay_dataset = contents[0]
|
|
assert okay_dataset["state"] == "ok"
|
|
|
|
@skip_without_tool("cat")
|
|
def test_run_rename_on_mapped_over_dataset(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
type: data_collection_input
|
|
collection_type: list
|
|
steps:
|
|
- tool_id: cat
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
outputs:
|
|
out_file1:
|
|
rename: "my new name"
|
|
test_data:
|
|
input1:
|
|
type: list
|
|
name: the_dataset_list
|
|
elements:
|
|
- identifier: el1
|
|
value: 1.fastq
|
|
type: File
|
|
""", history_id=history_id)
|
|
content = self.dataset_populator.get_history_dataset_details(history_id, hid=4, wait=True, assert_ok=True)
|
|
name = content["name"]
|
|
assert name == "my new name", name
|
|
assert content["history_content_type"] == "dataset"
|
|
content = self.dataset_populator.get_history_collection_details(history_id, hid=3, wait=True, assert_ok=True)
|
|
name = content["name"]
|
|
assert content["history_content_type"] == "dataset_collection", content
|
|
assert name == "my new name", name
|
|
|
|
@skip_without_tool("cat")
|
|
def test_run_rename_based_on_inputs_on_mapped_over_dataset(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
type: data_collection_input
|
|
collection_type: list
|
|
steps:
|
|
- tool_id: cat
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
outputs:
|
|
out_file1:
|
|
rename: "#{input1} suffix"
|
|
test_data:
|
|
input1:
|
|
type: list
|
|
name: the_dataset_list
|
|
elements:
|
|
- identifier: el1
|
|
value: 1.fastq
|
|
type: File
|
|
""", history_id=history_id)
|
|
content = self.dataset_populator.get_history_collection_details(history_id, hid=3, wait=True, assert_ok=True)
|
|
name = content["name"]
|
|
assert content["history_content_type"] == "dataset_collection", content
|
|
assert name == "the_dataset_list suffix", name
|
|
|
|
@skip_without_tool("collection_creates_pair")
|
|
def test_run_rename_collection_output(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
steps:
|
|
- tool_id: collection_creates_pair
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
outputs:
|
|
paired_output:
|
|
rename: "my new name"
|
|
test_data:
|
|
input1:
|
|
value: 1.fasta
|
|
type: File
|
|
name: fasta1
|
|
""", history_id=history_id)
|
|
details1 = self.dataset_populator.get_history_collection_details(history_id, hid=4, wait=True, assert_ok=True)
|
|
|
|
assert details1["name"] == "my new name", details1
|
|
assert details1["history_content_type"] == "dataset_collection"
|
|
|
|
@skip_without_tool("create_2")
|
|
def test_run_rename_multiple_outputs(self):
|
|
with self.dataset_populator.test_history() as history_id:
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs: []
|
|
steps:
|
|
- tool_id: create_2
|
|
state:
|
|
sleep_time: 0
|
|
outputs:
|
|
out_file1:
|
|
rename: "my new name"
|
|
out_file2:
|
|
rename: "my other new name"
|
|
test_data: {}
|
|
""", history_id=history_id)
|
|
details1 = self.dataset_populator.get_history_dataset_details(history_id, hid=1, wait=True, assert_ok=True)
|
|
details2 = self.dataset_populator.get_history_dataset_details(history_id, hid=2)
|
|
|
|
assert details1["name"] == "my new name"
|
|
assert details2["name"] == "my other new name"
|
|
|
|
@skip_without_tool("cat")
|
|
def test_run_rename_based_on_input(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
steps:
|
|
- tool_id: cat
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
outputs:
|
|
out_file1:
|
|
rename: "#{input1 | basename} suffix"
|
|
test_data:
|
|
input1:
|
|
value: 1.fasta
|
|
type: File
|
|
name: fasta1
|
|
""", history_id=history_id)
|
|
content = self.dataset_populator.get_history_dataset_details(history_id, wait=True, assert_ok=True)
|
|
name = content["name"]
|
|
assert name == "fasta1 suffix", name
|
|
|
|
@skip_without_tool("cat")
|
|
def test_run_rename_based_on_input_recursive(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
steps:
|
|
- tool_id: cat
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
outputs:
|
|
out_file1:
|
|
rename: "#{input1} #{input1 | upper} suffix"
|
|
test_data:
|
|
input1:
|
|
value: 1.fasta
|
|
type: File
|
|
name: '#{input1}'
|
|
""", history_id=history_id)
|
|
content = self.dataset_populator.get_history_dataset_details(history_id, wait=True, assert_ok=True)
|
|
name = content["name"]
|
|
assert name == "#{input1} #{INPUT1} suffix", name
|
|
|
|
@skip_without_tool("cat")
|
|
def test_run_rename_based_on_input_repeat(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
- id: input2
|
|
steps:
|
|
- tool_id: cat
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
queries:
|
|
- input2:
|
|
$link: input2
|
|
outputs:
|
|
out_file1:
|
|
rename: "#{queries_0.input2| basename} suffix"
|
|
test_data:
|
|
input1:
|
|
value: 1.fasta
|
|
type: File
|
|
name: fasta1
|
|
input2:
|
|
value: 1.fasta
|
|
type: File
|
|
name: fasta2
|
|
""", history_id=history_id)
|
|
content = self.dataset_populator.get_history_dataset_details(history_id, wait=True, assert_ok=True)
|
|
name = content["name"]
|
|
assert name == "fasta2 suffix", name
|
|
|
|
@skip_without_tool("mapper2")
|
|
def test_run_rename_based_on_input_conditional(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: fasta_input
|
|
- id: fastq_input
|
|
steps:
|
|
- tool_id: mapper2
|
|
state:
|
|
fastq_input:
|
|
fastq_input_selector: single
|
|
fastq_input1:
|
|
$link: fastq_input
|
|
reference:
|
|
$link: fasta_input
|
|
outputs:
|
|
out_file1:
|
|
# Wish it was qualified for conditionals but it doesn't seem to be. -John
|
|
# rename: "#{fastq_input.fastq_input1 | basename} suffix"
|
|
rename: "#{fastq_input1 | basename} suffix"
|
|
test_data:
|
|
fasta_input:
|
|
value: 1.fasta
|
|
type: File
|
|
name: fasta1
|
|
file_type: fasta
|
|
fastq_input:
|
|
value: 1.fastqsanger
|
|
type: File
|
|
name: fastq1
|
|
file_type: fastqsanger
|
|
""", history_id=history_id)
|
|
content = self.dataset_populator.get_history_dataset_details(history_id, wait=True, assert_ok=True)
|
|
name = content["name"]
|
|
assert name == "fastq1 suffix", name
|
|
|
|
@skip_without_tool("mapper2")
|
|
def test_run_rename_based_on_input_collection(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: fasta_input
|
|
- id: fastq_inputs
|
|
steps:
|
|
- tool_id: mapper2
|
|
state:
|
|
fastq_input:
|
|
fastq_input_selector: paired_collection
|
|
fastq_input1:
|
|
$link: fastq_inputs
|
|
reference:
|
|
$link: fasta_input
|
|
outputs:
|
|
out_file1:
|
|
# Wish it was qualified for conditionals but it doesn't seem to be. -John
|
|
# rename: "#{fastq_input.fastq_input1 | basename} suffix"
|
|
rename: "#{fastq_input1} suffix"
|
|
test_data:
|
|
fasta_input:
|
|
value: 1.fasta
|
|
type: File
|
|
name: fasta1
|
|
file_type: fasta
|
|
fastq_inputs:
|
|
type: list
|
|
name: the_dataset_pair
|
|
elements:
|
|
- identifier: forward
|
|
value: 1.fastq
|
|
type: File
|
|
- identifier: reverse
|
|
value: 1.fastq
|
|
type: File
|
|
""", history_id=history_id)
|
|
content = self.dataset_populator.get_history_dataset_details(history_id, wait=True, assert_ok=True)
|
|
name = content["name"]
|
|
assert name == "the_dataset_pair suffix", name
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_run_with_runtime_pja(self):
|
|
workflow = self.workflow_populator.load_workflow(name="test_for_pja_runtime")
|
|
uuid0, uuid1, uuid2 = str(uuid4()), str(uuid4()), str(uuid4())
|
|
workflow["steps"]["0"]["uuid"] = uuid0
|
|
workflow["steps"]["1"]["uuid"] = uuid1
|
|
workflow["steps"]["2"]["uuid"] = uuid2
|
|
workflow_request, history_id = self._setup_workflow_run(workflow, inputs_by='step_index')
|
|
workflow_request["replacement_params"] = dumps(dict(replaceme="was replaced"))
|
|
|
|
pja_map = {
|
|
"RenameDatasetActionout_file1": dict(
|
|
action_type="RenameDatasetAction",
|
|
output_name="out_file1",
|
|
action_arguments=dict(newname="foo ${replaceme}"),
|
|
)
|
|
}
|
|
workflow_request["parameters"] = dumps({
|
|
uuid2: {"__POST_JOB_ACTIONS__": pja_map}
|
|
})
|
|
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
content = self.dataset_populator.get_history_dataset_details(history_id, wait=True, assert_ok=True)
|
|
assert content["name"] == "foo was replaced", content["name"]
|
|
|
|
# Test for regression of previous behavior where runtime post job actions
|
|
# would be added to the original workflow post job actions.
|
|
workflow_id = workflow_request["workflow_id"]
|
|
downloaded_workflow = self._download_workflow(workflow_id)
|
|
pjas = list(downloaded_workflow["steps"]["2"]["post_job_actions"].values())
|
|
assert len(pjas) == 0, len(pjas)
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_run_with_delayed_runtime_pja(self):
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- label: test_input
|
|
type: input
|
|
- label: first_cat
|
|
tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: test_input
|
|
- label: the_pause
|
|
type: pause
|
|
connect:
|
|
input:
|
|
- first_cat#out_file1
|
|
- label: second_cat
|
|
tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: the_pause
|
|
""")
|
|
downloaded_workflow = self._download_workflow(workflow_id)
|
|
print(downloaded_workflow)
|
|
uuid_dict = dict((int(index), step["uuid"]) for index, step in downloaded_workflow["steps"].items())
|
|
history_id = self.dataset_populator.new_history()
|
|
hda = self.dataset_populator.new_dataset(history_id, content="1 2 3")
|
|
self.dataset_populator.wait_for_history(history_id)
|
|
inputs = {
|
|
'0': self._ds_entry(hda),
|
|
}
|
|
print(inputs)
|
|
uuid2 = uuid_dict[3]
|
|
workflow_request = {}
|
|
workflow_request["replacement_params"] = dumps(dict(replaceme="was replaced"))
|
|
pja_map = {
|
|
"RenameDatasetActionout_file1": dict(
|
|
action_type="RenameDatasetAction",
|
|
output_name="out_file1",
|
|
action_arguments=dict(newname="foo ${replaceme}"),
|
|
)
|
|
}
|
|
workflow_request["parameters"] = dumps({
|
|
uuid2: {"__POST_JOB_ACTIONS__": pja_map}
|
|
})
|
|
invocation_id = self.__invoke_workflow(history_id, workflow_id, inputs=inputs, request=workflow_request)
|
|
|
|
time.sleep(2)
|
|
self.dataset_populator.wait_for_history(history_id)
|
|
self.__review_paused_steps(workflow_id, invocation_id, order_index=2, action=True)
|
|
|
|
self.workflow_populator.wait_for_workflow(workflow_id, invocation_id, history_id)
|
|
time.sleep(1)
|
|
content = self.dataset_populator.get_history_dataset_details(history_id)
|
|
assert content["name"] == "foo was replaced", content["name"]
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_delete_intermediate_datasets_pja_1(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
inputs:
|
|
- id: input1
|
|
outputs:
|
|
- id: wf_output_1
|
|
source: third_cat#out_file1
|
|
steps:
|
|
- tool_id: cat1
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: input1
|
|
- tool_id: cat1
|
|
label: second_cat
|
|
state:
|
|
input1:
|
|
$link: first_cat#out_file1
|
|
- tool_id: cat1
|
|
label: third_cat
|
|
state:
|
|
input1:
|
|
$link: second_cat#out_file1
|
|
outputs:
|
|
out_file1:
|
|
delete_intermediate_datasets: true
|
|
test_data:
|
|
input1: "hello world"
|
|
""", history_id=history_id)
|
|
hda1 = self.dataset_populator.get_history_dataset_details(history_id, hid=1)
|
|
hda2 = self.dataset_populator.get_history_dataset_details(history_id, hid=2)
|
|
hda3 = self.dataset_populator.get_history_dataset_details(history_id, hid=3)
|
|
hda4 = self.dataset_populator.get_history_dataset_details(history_id, hid=4)
|
|
assert not hda1["deleted"]
|
|
assert hda2["deleted"]
|
|
# I think hda3 should be deleted, but the inputs to
|
|
# steps with workflow outputs are not deleted.
|
|
# assert hda3["deleted"]
|
|
print(hda3["deleted"])
|
|
assert not hda4["deleted"]
|
|
|
|
@skip_without_tool("random_lines1")
|
|
def test_run_replace_params_by_tool(self):
|
|
workflow_request, history_id = self._setup_random_x2_workflow("test_for_replace_tool_params")
|
|
workflow_request["parameters"] = dumps(dict(random_lines1=dict(num_lines=5)))
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
# Would be 8 and 6 without modification
|
|
self.__assert_lines_hid_line_count_is(history_id, 2, 5)
|
|
self.__assert_lines_hid_line_count_is(history_id, 3, 5)
|
|
|
|
@skip_without_tool("random_lines1")
|
|
def test_run_replace_params_by_uuid(self):
|
|
workflow_request, history_id = self._setup_random_x2_workflow("test_for_replace_tool_params")
|
|
workflow_request["parameters"] = dumps({
|
|
"58dffcc9-bcb7-4117-a0e1-61513524b3b1": dict(num_lines=4),
|
|
"58dffcc9-bcb7-4117-a0e1-61513524b3b2": dict(num_lines=3),
|
|
})
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
# Would be 8 and 6 without modification
|
|
self.__assert_lines_hid_line_count_is(history_id, 2, 4)
|
|
self.__assert_lines_hid_line_count_is(history_id, 3, 3)
|
|
|
|
@skip_without_tool("cat1")
|
|
@skip_without_tool("addValue")
|
|
def test_run_batch(self):
|
|
workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_batch")
|
|
workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
history_id = self.dataset_populator.new_history()
|
|
hda1 = self.dataset_populator.new_dataset(history_id, content="1 2 3")
|
|
hda2 = self.dataset_populator.new_dataset(history_id, content="4 5 6")
|
|
hda3 = self.dataset_populator.new_dataset(history_id, content="7 8 9")
|
|
hda4 = self.dataset_populator.new_dataset(history_id, content="10 11 12")
|
|
parameters = {
|
|
"0": {"input": {"batch": True, "values": [{"id" : hda1.get("id"), "hid": hda1.get("hid"), "src": "hda"},
|
|
{"id" : hda2.get("id"), "hid": hda2.get("hid"), "src": "hda"},
|
|
{"id" : hda3.get("id"), "hid": hda2.get("hid"), "src": "hda"},
|
|
{"id" : hda4.get("id"), "hid": hda2.get("hid"), "src": "hda"}]}},
|
|
"1": {"input": {"batch": False, "values": [{"id" : hda1.get("id"), "hid": hda1.get("hid"), "src": "hda"}]}, "exp": "2"}}
|
|
workflow_request = {
|
|
"history_id" : history_id,
|
|
"batch" : True,
|
|
"parameters_normalized": True,
|
|
"parameters" : dumps(parameters),
|
|
}
|
|
invocation_response = self._post("workflows/%s/usage" % workflow_id, data=workflow_request)
|
|
self._assert_status_code_is(invocation_response, 200)
|
|
time.sleep(5)
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
r1 = "1 2 3\t1\n1 2 3\t2\n"
|
|
r2 = "4 5 6\t1\n1 2 3\t2\n"
|
|
r3 = "7 8 9\t1\n1 2 3\t2\n"
|
|
r4 = "10 11 12\t1\n1 2 3\t2\n"
|
|
t1 = self.dataset_populator.get_history_dataset_content(history_id, hid=7)
|
|
t2 = self.dataset_populator.get_history_dataset_content(history_id, hid=10)
|
|
t3 = self.dataset_populator.get_history_dataset_content(history_id, hid=13)
|
|
t4 = self.dataset_populator.get_history_dataset_content(history_id, hid=16)
|
|
self.assertEqual(r1, t1)
|
|
self.assertEqual(r2, t2)
|
|
self.assertEqual(r3, t3)
|
|
self.assertEqual(r4, t4)
|
|
|
|
@skip_without_tool("validation_default")
|
|
def test_parameter_substitution_sanitization(self):
|
|
substitions = dict(input1="\" ; echo \"moo")
|
|
run_workflow_response, history_id = self._run_validation_workflow_with_substitions(substitions)
|
|
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
self.assertEqual("__dq__ X echo __dq__moo\n", self.dataset_populator.get_history_dataset_content(history_id, hid=1))
|
|
|
|
@skip_without_tool("validation_repeat")
|
|
def test_parameter_substitution_validation_value_errors_0(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- tool_id: validation_repeat
|
|
state:
|
|
r2:
|
|
- text: "abd"
|
|
""")
|
|
workflow_request = dict(
|
|
history="hist_id=%s" % history_id,
|
|
parameters=dumps(dict(validation_repeat={"r2_0|text": ""}))
|
|
)
|
|
url = "workflows/%s/invocations" % workflow_id
|
|
invocation_response = self._post(url, data=workflow_request)
|
|
# Take a valid stat and make it invalid, assert workflow won't run.
|
|
self._assert_status_code_is(invocation_response, 400)
|
|
|
|
@skip_without_tool("validation_default")
|
|
def test_parameter_substitution_validation_value_errors_1(self):
|
|
substitions = dict(select_param="\" ; echo \"moo")
|
|
run_workflow_response, history_id = self._run_validation_workflow_with_substitions(substitions)
|
|
|
|
self._assert_status_code_is(run_workflow_response, 400)
|
|
|
|
@skip_without_tool("validation_repeat")
|
|
def test_workflow_import_state_validation_1(self):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- tool_id: validation_repeat
|
|
state:
|
|
r2:
|
|
- text: ""
|
|
""", history_id=history_id, wait=False, expected_response=400)
|
|
|
|
def _run_validation_workflow_with_substitions(self, substitions):
|
|
workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_validation_1")
|
|
uploaded_workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
history_id = self.dataset_populator.new_history()
|
|
workflow_request = dict(
|
|
history="hist_id=%s" % history_id,
|
|
workflow_id=uploaded_workflow_id,
|
|
parameters=dumps(dict(validation_default=substitions))
|
|
)
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
return run_workflow_response, history_id
|
|
|
|
@skip_without_tool("random_lines1")
|
|
def test_run_replace_params_by_steps(self):
|
|
workflow_request, history_id, steps = self._setup_random_x2_workflow_steps("test_for_replace_step_params")
|
|
params = dumps({str(steps[1]["id"]): dict(num_lines=5)})
|
|
workflow_request["parameters"] = params
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
# Would be 8 and 6 without modification
|
|
self.__assert_lines_hid_line_count_is(history_id, 2, 8)
|
|
self.__assert_lines_hid_line_count_is(history_id, 3, 5)
|
|
|
|
@skip_without_tool("random_lines1")
|
|
def test_run_replace_params_nested(self):
|
|
workflow_request, history_id, steps = self._setup_random_x2_workflow_steps("test_for_replace_step_params_nested")
|
|
seed_source = dict(
|
|
seed_source_selector="set_seed",
|
|
seed="moo",
|
|
)
|
|
params = dumps({str(steps[0]["id"]): dict(num_lines=1, seed_source=seed_source),
|
|
str(steps[1]["id"]): dict(num_lines=1, seed_source=seed_source)})
|
|
workflow_request["parameters"] = params
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
|
|
self.assertEqual("3\n", self.dataset_populator.get_history_dataset_content(history_id))
|
|
|
|
def test_pja_import_export(self):
|
|
workflow = self.workflow_populator.load_workflow(name="test_for_pja_import", add_pja=True)
|
|
uploaded_workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
downloaded_workflow = self._download_workflow(uploaded_workflow_id)
|
|
self._assert_has_keys(downloaded_workflow["steps"], "0", "1", "2")
|
|
pjas = list(downloaded_workflow["steps"]["2"]["post_job_actions"].values())
|
|
assert len(pjas) == 1, len(pjas)
|
|
pja = pjas[0]
|
|
self._assert_has_keys(pja, "action_type", "output_name", "action_arguments")
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_only_own_invocations_accessible(self):
|
|
workflow_id, usage = self._run_workflow_once_get_invocation("test_usage")
|
|
with self._different_user():
|
|
usage_details_response = self._get("workflows/%s/usage/%s" % (workflow_id, usage["id"]))
|
|
self._assert_status_code_is(usage_details_response, 403)
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_invocation_usage(self):
|
|
workflow_id, usage = self._run_workflow_once_get_invocation("test_usage")
|
|
invocation_id = usage["id"]
|
|
usage_details = self._invocation_details(workflow_id, invocation_id)
|
|
# Assert some high-level things about the structure of data returned.
|
|
self._assert_has_keys(usage_details, "inputs", "steps")
|
|
invocation_steps = usage_details["steps"]
|
|
for step in invocation_steps:
|
|
self._assert_has_keys(step, "workflow_step_id", "order_index", "id")
|
|
an_invocation_step = invocation_steps[0]
|
|
step_id = an_invocation_step["id"]
|
|
step_response = self._get("workflows/%s/usage/%s/steps/%s" % (workflow_id, invocation_id, step_id))
|
|
self._assert_status_code_is(step_response, 200)
|
|
self._assert_has_keys(step_response.json(), "id", "order_index")
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_invocations_accessible_imported_workflow(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("test_usage", publish=True)
|
|
with self._different_user():
|
|
other_import_response = self.__import_workflow(workflow_id)
|
|
self._assert_status_code_is(other_import_response, 200)
|
|
other_id = other_import_response.json()["id"]
|
|
workflow_request, history_id = self._setup_workflow_run(workflow_id=other_id)
|
|
response = self._get("workflows/%s/usage" % other_id)
|
|
self._assert_status_code_is(response, 200)
|
|
assert len(response.json()) == 0
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
run_workflow_response = run_workflow_response.json()
|
|
invocation_id = run_workflow_response['id']
|
|
usage_details_response = self._get("workflows/%s/usage/%s" % (other_id, invocation_id))
|
|
self._assert_status_code_is(usage_details_response, 200)
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_invocations_accessible_published_workflow(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("test_usage", publish=True)
|
|
with self._different_user():
|
|
workflow_request, history_id = self._setup_workflow_run(workflow_id=workflow_id)
|
|
workflow_request['workflow_id'] = workflow_request.pop('workflow_id')
|
|
response = self._get("workflows/%s/usage" % workflow_id)
|
|
self._assert_status_code_is(response, 200)
|
|
assert len(response.json()) == 0
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
run_workflow_response = run_workflow_response.json()
|
|
invocation_id = run_workflow_response['id']
|
|
usage_details_response = self._get("workflows/%s/usage/%s" % (workflow_id, invocation_id))
|
|
self._assert_status_code_is(usage_details_response, 200)
|
|
|
|
@skip_without_tool("cat1")
|
|
def test_invocations_not_accessible_by_different_user_for_published_workflow(self):
|
|
workflow_id = self.workflow_populator.simple_workflow("test_usage", publish=True)
|
|
workflow_request, history_id = self._setup_workflow_run(workflow_id=workflow_id)
|
|
workflow_request['workflow_id'] = workflow_request.pop('workflow_id')
|
|
response = self._get("workflows/%s/usage" % workflow_id)
|
|
self._assert_status_code_is(response, 200)
|
|
assert len(response.json()) == 0
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
run_workflow_response = run_workflow_response.json()
|
|
invocation_id = run_workflow_response['id']
|
|
with self._different_user():
|
|
usage_details_response = self._get("workflows/%s/usage/%s" % (workflow_id, invocation_id))
|
|
self._assert_status_code_is(usage_details_response, 403)
|
|
|
|
def _invoke_paused_workflow(self, history_id):
|
|
workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_pause")
|
|
workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
hda1 = self.dataset_populator.new_dataset(history_id, content="1 2 3")
|
|
index_map = {
|
|
'0': self._ds_entry(hda1),
|
|
}
|
|
invocation_id = self.__invoke_workflow(
|
|
history_id,
|
|
workflow_id,
|
|
index_map,
|
|
)
|
|
return workflow_id, invocation_id
|
|
|
|
def _wait_for_invocation_non_new(self, workflow_id, invocation_id):
|
|
target_state_reached = False
|
|
for i in range(50):
|
|
invocation = self._invocation_details(workflow_id, invocation_id)
|
|
if invocation['state'] != 'new':
|
|
target_state_reached = True
|
|
break
|
|
|
|
time.sleep(.25)
|
|
|
|
return target_state_reached
|
|
|
|
def _assert_invocation_non_terminal(self, workflow_id, invocation_id):
|
|
invocation = self._invocation_details(workflow_id, invocation_id)
|
|
assert invocation['state'] in ['ready', 'new'], invocation
|
|
|
|
def _wait_for_invocation_state(self, workflow_id, invocation_id, target_state):
|
|
target_state_reached = False
|
|
for i in range(25):
|
|
invocation = self._invocation_details(workflow_id, invocation_id)
|
|
if invocation['state'] == target_state:
|
|
target_state_reached = True
|
|
break
|
|
|
|
time.sleep(.5)
|
|
|
|
return target_state_reached
|
|
|
|
def _update_workflow(self, workflow_id, workflow_object):
|
|
data = dict(
|
|
workflow=workflow_object
|
|
)
|
|
raw_url = 'workflows/%s' % workflow_id
|
|
url = self._api_url(raw_url, use_key=True)
|
|
put_response = put(url, data=dumps(data))
|
|
return put_response
|
|
|
|
def _invocation_step_details(self, workflow_id, invocation_id, step_id):
|
|
invocation_step_response = self._get("workflows/%s/usage/%s/steps/%s" % (workflow_id, invocation_id, step_id))
|
|
self._assert_status_code_is(invocation_step_response, 200)
|
|
invocation_step_details = invocation_step_response.json()
|
|
return invocation_step_details
|
|
|
|
def _execute_invocation_step_action(self, workflow_id, invocation_id, step_id, action):
|
|
raw_url = "workflows/%s/usage/%s/steps/%s" % (workflow_id, invocation_id, step_id)
|
|
url = self._api_url(raw_url, use_key=True)
|
|
payload = dumps(dict(action=action))
|
|
action_response = put(url, data=payload)
|
|
self._assert_status_code_is(action_response, 200)
|
|
invocation_step_details = action_response.json()
|
|
return invocation_step_details
|
|
|
|
def _run_workflow_once_get_invocation(self, name):
|
|
workflow = self.workflow_populator.load_workflow(name=name)
|
|
workflow_request, history_id = self._setup_workflow_run(workflow)
|
|
workflow_id = workflow_request["workflow_id"]
|
|
response = self._get("workflows/%s/usage" % workflow_id)
|
|
self._assert_status_code_is(response, 200)
|
|
assert len(response.json()) == 0
|
|
run_workflow_response = self._post("workflows", data=workflow_request)
|
|
self._assert_status_code_is(run_workflow_response, 200)
|
|
|
|
response = self._get("workflows/%s/usage" % workflow_id)
|
|
self._assert_status_code_is(response, 200)
|
|
usages = response.json()
|
|
assert len(usages) == 1
|
|
return workflow_id, usages[0]
|
|
|
|
def _setup_random_x2_workflow_steps(self, name):
|
|
workflow_request, history_id = self._setup_random_x2_workflow("test_for_replace_step_params")
|
|
random_line_steps = self._random_lines_steps(workflow_request)
|
|
return workflow_request, history_id, random_line_steps
|
|
|
|
def _random_lines_steps(self, workflow_request):
|
|
workflow_summary_response = self._get("workflows/%s" % workflow_request["workflow_id"])
|
|
self._assert_status_code_is(workflow_summary_response, 200)
|
|
steps = workflow_summary_response.json()["steps"]
|
|
return sorted((step for step in steps.values() if step["tool_id"] == "random_lines1"), key=lambda step: step["id"])
|
|
|
|
def _setup_random_x2_workflow(self, name):
|
|
workflow = self.workflow_populator.load_random_x2_workflow(name)
|
|
uploaded_workflow_id = self.workflow_populator.create_workflow(workflow)
|
|
workflow_inputs = self._workflow_inputs(uploaded_workflow_id)
|
|
key = next(iter(workflow_inputs.keys()))
|
|
history_id = self.dataset_populator.new_history()
|
|
ten_lines = "\n".join(str(_) for _ in range(10))
|
|
hda1 = self.dataset_populator.new_dataset(history_id, content=ten_lines)
|
|
workflow_request = dict(
|
|
history="hist_id=%s" % history_id,
|
|
workflow_id=uploaded_workflow_id,
|
|
ds_map=dumps({
|
|
key: self._ds_entry(hda1),
|
|
}),
|
|
)
|
|
return workflow_request, history_id
|
|
|
|
def __review_paused_steps(self, uploaded_workflow_id, invocation_id, order_index, action=True):
|
|
invocation = self._invocation_details(uploaded_workflow_id, invocation_id)
|
|
invocation_steps = invocation["steps"]
|
|
pause_steps = [s for s in invocation_steps if s['order_index'] == order_index]
|
|
for pause_step in pause_steps:
|
|
pause_step_id = pause_step['id']
|
|
|
|
self._execute_invocation_step_action(uploaded_workflow_id, invocation_id, pause_step_id, action=action)
|
|
|
|
def __assert_lines_hid_line_count_is(self, history, hid, lines):
|
|
contents_url = "histories/%s/contents" % history
|
|
history_contents = self.__history_contents(history)
|
|
hda_summary = next(hc for hc in history_contents if hc["hid"] == hid)
|
|
hda_info_response = self._get("%s/%s" % (contents_url, hda_summary["id"]))
|
|
self._assert_status_code_is(hda_info_response, 200)
|
|
self.assertEqual(hda_info_response.json()["metadata_data_lines"], lines)
|
|
|
|
def __history_contents(self, history_id):
|
|
contents_url = "histories/%s/contents" % history_id
|
|
history_contents_response = self._get(contents_url)
|
|
self._assert_status_code_is(history_contents_response, 200)
|
|
return history_contents_response.json()
|
|
|
|
def __invoke_workflow(self, *args, **kwds):
|
|
return self.workflow_populator.invoke_workflow(*args, **kwds)
|
|
|
|
def __import_workflow(self, workflow_id, deprecated_route=False):
|
|
if deprecated_route:
|
|
route = "workflows/import"
|
|
import_data = dict(
|
|
workflow_id=workflow_id,
|
|
)
|
|
else:
|
|
route = "workflows"
|
|
import_data = dict(
|
|
shared_workflow_id=workflow_id,
|
|
)
|
|
return self._post(route, import_data)
|
|
|
|
def _download_workflow(self, workflow_id, style=None):
|
|
params = {}
|
|
if style:
|
|
params = {"style": style}
|
|
download_response = self._get("workflows/%s/download" % workflow_id, params)
|
|
self._assert_status_code_is(download_response, 200)
|
|
downloaded_workflow = download_response.json()
|
|
return downloaded_workflow
|
|
|
|
def _show_workflow(self, workflow_id):
|
|
show_response = self._get("workflows/%s" % workflow_id)
|
|
self._assert_status_code_is(show_response, 200)
|
|
return show_response.json()
|
|
|
|
def _assert_looks_like_instance_workflow_representation(self, workflow):
|
|
self._assert_has_keys(
|
|
workflow,
|
|
'url',
|
|
'owner',
|
|
'inputs',
|
|
'annotation',
|
|
'steps'
|
|
)
|
|
for step in workflow["steps"].values():
|
|
self._assert_has_keys(
|
|
step,
|
|
'id',
|
|
'type',
|
|
'tool_id',
|
|
'tool_version',
|
|
'annotation',
|
|
'tool_inputs',
|
|
'input_steps',
|
|
)
|
|
|
|
|
|
RunJobsSummary = namedtuple('RunJobsSummary', ['history_id', 'workflow_id', 'invocation_id', 'inputs', 'jobs', 'invocation', 'workflow_request'])
|