mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
John Chilton
b4c77b7617
This is a slight tightening up of the ad-hoc, experimental workflow YAML definition used by the test framework (and by Kyle, but lets just admit Kyle is part of Galaxy's test framework). This format is still defined entirely client side by transcoding the YAML or python object description into real (or format 1) Galaxy workflows - so these cannot realisticaly be declared part of Galaxy's public interface and can remain experimental and subject to change. In addition to refactoring the code implementing these workflows for upstream modifications to explore new features, the format itself has been made slightly more stringent in two ways: - 'steps' must now be explicit (previously converted auto-convert list to dict). - Ensure the workflow declared a 'class' is defined and the class is 'GalaxyWorkflow'. This will help with discovery of workflow objects with downstream tooling (planemo for workflows?) and brings the Galaxy definition slightly more inline with the CWL definition for workflows (very slightly).
145 lines
3.5 KiB
Python
145 lines
3.5 KiB
Python
|
|
from .test_workflows import BaseWorkflowsApiTestCase
|
|
|
|
|
|
class WorkflowsFromYamlApiTestCase( BaseWorkflowsApiTestCase ):
|
|
|
|
def setUp( self ):
|
|
super( WorkflowsFromYamlApiTestCase, self ).setUp()
|
|
|
|
def test_simple_upload(self):
|
|
workflow_id = self._upload_yaml_workflow("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- type: input
|
|
- tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: 0
|
|
- tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: 1#out_file1
|
|
- tool_id: random_lines1
|
|
state:
|
|
num_lines: 10
|
|
input:
|
|
$link: 2#out_file1
|
|
seed_source:
|
|
seed_source_selector: set_seed
|
|
seed: asdf
|
|
__current_case__: 1
|
|
""")
|
|
self._get("workflows/%s/download" % workflow_id).content
|
|
|
|
# FIXME: This test fails on some machines due to (we're guessing) yaml loading
|
|
# order being not guaranteed and inconsistent across platforms. The workflow
|
|
# yaml loader probably needs to enforce order using something like the
|
|
# approach described here:
|
|
# https://stackoverflow.com/questions/13297744/pyyaml-control-ordering-of-items-called-by-yaml-load
|
|
# def test_multiple_input( self ):
|
|
# history_id = self.dataset_populator.new_history()
|
|
# self._run_jobs("""
|
|
# steps:
|
|
# - type: input
|
|
# label: input1
|
|
# - type: input
|
|
# label: input2
|
|
# - tool_id: cat_list
|
|
# state:
|
|
# input1:
|
|
# - $link: input1
|
|
# - $link: input2
|
|
# test_data:
|
|
# input1: "hello world"
|
|
# input2: "123"
|
|
# """, history_id=history_id)
|
|
# contents1 = self.dataset_populator.get_history_dataset_content(history_id)
|
|
# assert contents1 == "hello world\n123\n"
|
|
|
|
def test_simple_output_actions( self ):
|
|
history_id = self.dataset_populator.new_history()
|
|
self._run_jobs("""
|
|
class: GalaxyWorkflow
|
|
steps:
|
|
- type: input
|
|
label: input1
|
|
- tool_id: cat1
|
|
label: first_cat
|
|
state:
|
|
input1:
|
|
$link: 0
|
|
outputs:
|
|
out_file1:
|
|
hide: true
|
|
rename: "the new value"
|
|
- tool_id: cat1
|
|
state:
|
|
input1:
|
|
$link: first_cat#out_file1
|
|
test_data:
|
|
input1: "hello world"
|
|
""", history_id=history_id)
|
|
|
|
details1 = self.dataset_populator.get_history_dataset_details(history_id, hid=2)
|
|
assert not details1["visible"]
|
|
assert details1["name"] == "the new value", details1
|
|
details2 = self.dataset_populator.get_history_dataset_details(history_id, hid=3)
|
|
assert details2["visible"]
|
|
|
|
def test_pause( 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
|
|
""")
|
|
print self._get("workflows/%s/download" % workflow_id).json()
|
|
|
|
def test_implicit_connections( 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
|
|
- label: third_cat
|
|
tool_id: cat1
|
|
connect:
|
|
$step: second_cat
|
|
state:
|
|
input1:
|
|
$link: test_input
|
|
""")
|
|
workflow = self._get("workflows/%s/download" % workflow_id).json()
|
|
print workflow
|