opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python3: test/api/

Browse Source
This commit is contained in:
Junzhou Wang
2016-06-25 17:31:20 -04:00
parent ac07c62381
commit 7d7396f18f
6 changed files with 64 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/api/helpers.py
+5 -5
View File
@@ -3,7 +3,7 @@ from operator import itemgetter
import time
import json
import StringIO
from six import StringIO
from pkg_resources import resource_string
# Simple workflow that takes an input and call cat wrapper on it.
@@ -27,7 +27,7 @@ def skip_without_tool( tool_id ):
index = api_test_case.galaxy_interactor.get( "tools", data=dict(in_panel=False) )
tools = index.json()
# In panels by default, so flatten out sections...
tool_ids = map( itemgetter( "id" ), tools )
tool_ids = [itemgetter( "id" )(_) for _ in tools]
return tool_ids
def wrapped_method( api_test_case, *args, **kwargs ):
@@ -110,7 +110,7 @@ class BaseDatasetPopulator( object ):
'dbkey': dbkey,
'file_type': file_type,
}
if isinstance( content, file ):
if hasattr(content, 'read'):
upload_params[ "files_0|file_data"] = content
else:
upload_params[ 'files_0|url_paste' ] = content
@@ -329,7 +329,7 @@ class LibraryPopulator( object ):
"db_key": kwds.get( "db_key", "?" ),
}
files = {
"files_0|file_data": kwds.get( "file", StringIO.StringIO( kwds.get( "contents", "TestData" ) ) ),
"files_0|file_data": kwds.get( "file", StringIO( kwds.get( "contents", "TestData" ) ) ),
}
return create_data, files
@@ -438,7 +438,7 @@ class BaseDatasetCollectionPopulator( object ):
def __datasets( self, history_id, count, contents=None ):
datasets = []
for i in xrange( count ):
for i in range( count ):
new_kwds = {}
if contents:
new_kwds[ "content" ] = contents[ i ]
test/api/test_datasets.py
+3 -2
View File
@@ -1,3 +1,4 @@
from __future__ import print_function
import textwrap
from base import api
@@ -12,8 +13,8 @@ class DatasetsApiTestCase( api.ApiTestCase, TestsDatasets ):
def test_index( self ):
index_response = self._get( "datasets" )
print index_response
print dir( index_response )
print(index_response)
print(dir( index_response ))
self._assert_status_code_is( index_response, 501 )
def test_show( self ):
test/api/test_tool_data.py
+4 -4
View File
@@ -1,7 +1,7 @@
""" Tests for the tool data API.
"""
from __future__ import print_function
from base import api
import operator
@@ -14,14 +14,14 @@ class ToolDataApiTestCase( api.ApiTestCase ):
def test_list(self):
index_response = self._get( "tool_data", admin=True )
self._assert_status_code_is( index_response, 200 )
print index_response.content
print(index_response.content)
index = index_response.json()
assert "testalpha" in map(operator.itemgetter("name"), index)
assert "testalpha" in [operator.itemgetter("name")(_) for _ in index]
def test_show(self):
show_response = self._get( "tool_data/testalpha", admin=True )
self._assert_status_code_is( show_response, 200 )
print show_response.content
print(show_response.content)
data_table = show_response.json()
assert data_table["columns"] == ["value", "name", "path"]
first_entry = data_table["fields"][0]
test/api/test_workflow_extraction.py
+13 -12
View File
@@ -1,3 +1,4 @@
from __future__ import print_function
from collections import namedtuple
import functools
from json import dumps, loads
@@ -18,12 +19,12 @@ class WorkflowExtractionApiTestCase( BaseWorkflowsApiTestCase ):
# Run the simple test workflow and extract it back out from history
cat1_job_id = self.__setup_and_run_cat1_workflow( history_id=self.history_id )
contents = self._history_contents()
input_hids = map( lambda c: c[ "hid" ], contents[ 0:2 ] )
input_hids = [c[ "hid" ] for c in contents[ 0:2 ]]
downloaded_workflow = self._extract_and_download_workflow(
dataset_ids=input_hids,
job_ids=[ cat1_job_id ],
)
self.assertEquals( downloaded_workflow[ "name" ], "test import from history" )
self.assertEqual( downloaded_workflow[ "name" ], "test import from history" )
self.__assert_looks_like_cat1_example_workflow( downloaded_workflow )
def test_extract_with_copied_inputs( self ):
@@ -42,7 +43,7 @@ class WorkflowExtractionApiTestCase( BaseWorkflowsApiTestCase ):
for old_dataset in old_contents:
self.__copy_content_to_history( self.history_id, old_dataset )
new_contents = self._history_contents()
input_hids = map( lambda c: c[ "hid" ], new_contents[ (offset + 0):(offset + 2) ] )
input_hids = [c[ "hid" ] for c in new_contents[ (offset + 0):(offset + 2) ]]
cat1_job_id = self.__job_id( self.history_id, new_contents[ (offset + 2) ][ "id" ] )
downloaded_workflow = self._extract_and_download_workflow(
dataset_ids=input_hids,
@@ -141,7 +142,7 @@ test_data:
collection_step = self._get_steps_of_type( downloaded_workflow, "data_collection_input", expected_len=1 )[ 0 ]
collection_step_state = loads( collection_step[ "tool_state" ] )
self.assertEquals( collection_step_state[ "collection_type" ], u"paired" )
self.assertEqual( collection_step_state[ "collection_type" ], "paired" )
@skip_without_tool( "cat_collection" )
def test_subcollection_mapping( self ):
@@ -169,7 +170,7 @@ test_data:
dataset_collection_ids=[ jobs_summary.inputs["text_input1"]["hid"] ],
job_ids=[ job1_id, job2_id ],
)
print jobs_summary.inputs["text_input1"]
print(jobs_summary.inputs["text_input1"])
self.__check_workflow(
downloaded_workflow,
step_count=3,
@@ -181,7 +182,7 @@ test_data:
collection_step = self._get_steps_of_type( downloaded_workflow, "data_collection_input", expected_len=1 )[ 0 ]
collection_step_state = loads( collection_step[ "tool_state" ] )
self.assertEquals( collection_step_state[ "collection_type" ], u"list:paired" )
self.assertEqual( collection_step_state[ "collection_type" ], "list:paired" )
@skip_without_tool( "collection_split_on_column" )
def test_extract_workflow_with_output_collections( self ):
@@ -214,7 +215,7 @@ test_data:
text_input2: "samp1\t30.0\nsamp2\t40.0\n"
""")
tool_ids = [ "cat1", "collection_split_on_column", "cat_list" ]
job_ids = map( functools.partial(self._job_id_for_tool, jobs_summary.jobs ), tool_ids )
job_ids = list(map( functools.partial(self._job_id_for_tool, jobs_summary.jobs ), tool_ids ))
downloaded_workflow = self._extract_and_download_workflow(
dataset_ids=[ "1", "2" ],
job_ids=job_ids,
@@ -264,7 +265,7 @@ test_data:
content: "samp1\t30.0\nsamp2\t40.0\n"
""")
tool_ids = [ "cat1", "collection_creates_pair", "cat_collection", "cat_list" ]
job_ids = map( functools.partial(self._job_id_for_tool, jobs_summary.jobs ), tool_ids )
job_ids = list(map( functools.partial(self._job_id_for_tool, jobs_summary.jobs ), tool_ids ))
downloaded_workflow = self._extract_and_download_workflow(
dataset_collection_ids=[ "3" ],
job_ids=job_ids,
@@ -282,7 +283,7 @@ test_data:
return self._job_for_tool( jobs, tool_id )[ "id" ]
def _job_for_tool( self, jobs, tool_id ):
tool_jobs = filter( lambda j: j["tool_id"] == tool_id, jobs )
tool_jobs = [j for j in jobs if j["tool_id"] == tool_id]
if not tool_jobs:
assert False, "Failed to find job for tool %s" % tool_id
# if len( tool_jobs ) > 1:
@@ -330,8 +331,8 @@ test_data:
input1 = tool_step[ "input_connections" ][ "input1" ]
input2 = tool_step[ "input_connections" ][ "queries_0|input2" ]
self.assertEquals( input_steps[ 0 ][ "id" ], input1[ "id" ] )
self.assertEquals( input_steps[ 1 ][ "id" ], input2[ "id" ] )
self.assertEqual( input_steps[ 0 ][ "id" ], input1[ "id" ] )
self.assertEqual( input_steps[ 1 ][ "id" ], input2[ "id" ] )
def _history_contents( self, history_id=None ):
if history_id is None:
@@ -368,7 +369,7 @@ test_data:
collection_steps = self._get_steps_of_type( downloaded_workflow, "data_collection_input", expected_len=1 )
collection_step = collection_steps[ 0 ]
collection_step_state = loads( collection_step[ "tool_state" ] )
self.assertEquals( collection_step_state[ "collection_type" ], u"paired" )
self.assertEqual( collection_step_state[ "collection_type" ], "paired" )
collect_step_idx = collection_step[ "id" ]
return collect_step_idx
test/api/test_workflows.py
+34 -34
View File
@@ -81,7 +81,7 @@ class BaseWorkflowsApiTestCase( api.ApiTestCase, ImporterGalaxyInterface ):
def _workflow_names( self ):
index_response = self._get( "workflows" )
self._assert_status_code_is( index_response, 200 )
names = map( lambda w: w[ "name" ], index_response.json() )
names = [w[ "name" ] for w in index_response.json()]
return names
# Import importer interface...
@@ -140,7 +140,7 @@ class BaseWorkflowsApiTestCase( api.ApiTestCase, ImporterGalaxyInterface ):
def _build_ds_map( self, workflow_id, label_map ):
workflow_inputs = self._workflow_inputs( workflow_id )
ds_map = {}
for key, value in workflow_inputs.iteritems():
for key, value in workflow_inputs.items():
label = value[ "label" ]
if label in label_map:
ds_map[ key ] = label_map[ label ]
@@ -275,7 +275,7 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ):
workflow = show_response.json()
self._assert_looks_like_instance_workflow_representation( workflow )
assert len(workflow["steps"]) == 3
self.assertEquals(sorted([step["id"] for step in workflow["steps"].values()]), [0, 1, 2])
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()
@@ -283,7 +283,7 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ):
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.assertNotEquals(sorted([step["id"] for step in workflow["steps"].values()]), [0, 1, 2])
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() )
@@ -344,7 +344,7 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ):
uuids = {}
labels = {}
for order_index, step_dict in original_workflow["steps"].iteritems():
for order_index, step_dict in original_workflow["steps"].items():
uuid = str(uuid4())
step_dict["uuid"] = uuid
uuids[order_index] = uuid
@@ -356,8 +356,8 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ):
assert order_index in uuids
assert order_index in labels
self.assertEquals(uuids[order_index], step_dict["uuid"])
self.assertEquals(labels[order_index], step_dict["label"])
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"]
@@ -377,7 +377,7 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ):
assert step_dict['position']['left'] != 1
step_dict['position'] = {'top': 1, 'left': 1}
map(tweak_step, steps.iteritems())
list(map(tweak_step, iter(steps.items())))
update(workflow_content)
@@ -388,7 +388,7 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ):
assert step_dict['position']['left'] == 1
updated_workflow_content = self._download_workflow(workflow_id)
map(check_step, updated_workflow_content['steps'].iteritems())
list(map(check_step, iter(updated_workflow_content['steps'].items())))
# Re-update against original worklfow...
update(original_workflow)
@@ -396,7 +396,7 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ):
updated_workflow_content = self._download_workflow(workflow_id)
# Make sure the positions have been updated.
map(tweak_step, updated_workflow_content['steps'].iteritems())
list(map(tweak_step, iter(updated_workflow_content['steps'].items())))
def test_update_no_tool_id( self ):
workflow_object = self.workflow_populator.load_workflow( name="test_import" )
@@ -448,7 +448,7 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ):
other_id = other_import_response.json()["id"]
imported_workflow = self._show_workflow( other_id )
assert imported_workflow["annotation"] == "simple workflow"
step_annotations = set(map(lambda step: step["annotation"], imported_workflow["steps"].values()))
step_annotations = set([step["annotation"] for step in imported_workflow["steps"].values()])
assert "input1 description" in step_annotations
def test_import_subworkflows( self ):
@@ -536,7 +536,7 @@ class WorkflowsApiTestCase( BaseWorkflowsApiTestCase ):
workflow_id = self.workflow_populator.create_workflow( workflow )
workflow_description = self._show_workflow( workflow_id )
steps = workflow_description["steps"]
missing_tool_steps = filter(lambda v: v['tool_id'] == 'cat_missing_tool', steps.values())
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 ):
@@ -642,7 +642,7 @@ steps:
self.__invoke_workflow( history_id, workflow_id, inputs )
time.sleep( 5 )
self.dataset_populator.wait_for_history( history_id, assert_ok=True )
self.assertEquals("a\nc\nb\nd\n", self.dataset_populator.get_history_dataset_content( history_id, hid=0 ) )
self.assertEqual("a\nc\nb\nd\n", self.dataset_populator.get_history_dataset_content( history_id, hid=0 ) )
@skip_without_tool( "collection_creates_pair" )
def test_workflow_run_output_collection_mapping(self):
@@ -672,7 +672,7 @@ steps:
self.__invoke_workflow( history_id, workflow_id, inputs )
time.sleep( 5 )
self.dataset_populator.wait_for_history( history_id, assert_ok=True )
self.assertEquals("a\nc\nb\nd\ne\ng\nf\nh\n", self.dataset_populator.get_history_dataset_content( history_id, hid=0 ) )
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):
@@ -715,7 +715,7 @@ steps:
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.assertEquals("10.0\n30.0\n20.0\n40.0\n", content )
self.assertEqual("10.0\n30.0\n20.0\n40.0\n", content )
@skip_without_tool( "collection_split_on_column" )
@skip_without_tool( "min_repeat" )
@@ -759,7 +759,7 @@ steps:
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, hid=7 )
self.assertEquals(content.strip(), "samp1\t10.0\nsamp2\t20.0")
self.assertEqual(content.strip(), "samp1\t10.0\nsamp2\t20.0")
@skip_without_tool( "mapper" )
@skip_without_tool( "pileup" )
@@ -808,7 +808,7 @@ test_data:
self._run_jobs(SIMPLE_NESTED_WORKFLOW_YAML, history_id=history_id)
content = self.dataset_populator.get_history_dataset_content( history_id )
self.assertEquals("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)
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" )
@@ -851,7 +851,7 @@ steps:
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.assertEquals(content.strip(), "samp1\t10.0\nsamp2\t20.0\nsamp1\t20.0\nsamp2\t40.0")
self.assertEqual(content.strip(), "samp1\t10.0\nsamp2\t20.0\nsamp1\t20.0\nsamp2\t40.0")
def test_workflow_request( self ):
workflow = self.workflow_populator.load_workflow( name="test_for_queue" )
@@ -943,7 +943,7 @@ steps:
self.dataset_populator.wait_for_history( history_id, assert_ok=True )
invocation = self._invocation_details( uploaded_workflow_id, invocation_id )
assert invocation[ 'state' ] == 'scheduled'
self.assertEquals("reviewed\n1\nreviewed\n4\n", self.dataset_populator.get_history_dataset_content( history_id ) )
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 ):
@@ -1097,7 +1097,7 @@ test_data:
self.dataset_populator.wait_for_history( history_id, assert_ok=True )
content = self.dataset_populator.get_history_dataset_content( history_id )
self.assertEquals("chr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\n", content)
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 ):
self.workflow_populator.wait_for_invocation( workflow_id, invocation_id )
@@ -1146,7 +1146,7 @@ test_data:
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.assertEquals("1 2 3\n4 5 6\n7 8 9\n0 a b\n", self.dataset_populator.get_history_dataset_content( history_id ) )
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:
@@ -1167,7 +1167,7 @@ test_data:
def _step_map(self, workflow):
# Build dict mapping 'tep index to input name.
step_map = {}
for step_index, step in workflow["steps"].iteritems():
for step_index, step in workflow["steps"].items():
if step[ "type" ] == "data_input":
step_map[step_index] = step["inputs"][0]["name"]
return step_map
@@ -1226,7 +1226,7 @@ test_data:
# would be added to the original workflow post job actions.
workflow_id = workflow_request["workflow_id"]
downloaded_workflow = self._download_workflow( workflow_id )
pjas = downloaded_workflow[ "steps" ][ "2" ][ "post_job_actions" ].values()
pjas = list(downloaded_workflow[ "steps" ][ "2" ][ "post_job_actions" ].values())
assert len( pjas ) == 0, len( pjas )
@skip_without_tool( "cat1" )
@@ -1253,15 +1253,15 @@ steps:
$link: the_pause
""")
downloaded_workflow = self._download_workflow( workflow_id )
print downloaded_workflow
uuid_dict = dict( map( lambda (index, step): ( int( index ), step["uuid"] ), downloaded_workflow["steps"].iteritems() ) )
print(downloaded_workflow)
uuid_dict = dict( [( int( index_step[0] ), index_step[1]["uuid"] ) for index_step in iter(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
print(inputs)
uuid2 = uuid_dict[ 3 ]
workflow_request = {}
workflow_request[ "replacement_params" ] = dumps( dict( replaceme="was replaced" ) )
@@ -1327,7 +1327,7 @@ test_data:
# I think hda3 should be deleted, but the inputs to
# steps with workflow outputs are not deleted.
# assert hda3["deleted"]
print hda3["deleted"]
print(hda3["deleted"])
assert not hda4["deleted"]
@skip_without_tool( "random_lines1" )
@@ -1361,7 +1361,7 @@ test_data:
run_workflow_response, history_id = self._run_validation_workflow_with_substitions( substitions )
self.dataset_populator.wait_for_history( history_id, assert_ok=True )
self.assertEquals("__dq__ X echo __dq__moo\n", self.dataset_populator.get_history_dataset_content( history_id, hid=1 ) )
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 ):
@@ -1439,14 +1439,14 @@ steps:
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.assertEquals("3\n", self.dataset_populator.get_history_dataset_content( history_id ) )
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 = downloaded_workflow[ "steps" ][ "2" ][ "post_job_actions" ].values()
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" )
@@ -1523,15 +1523,15 @@ steps:
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( filter(lambda step: step["tool_id"] == "random_lines1", steps.values()), key=lambda step: step["id"] )
return sorted( [step for step in list(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 = workflow_inputs.keys()[ 0 ]
key = list(workflow_inputs.keys())[ 0 ]
history_id = self.dataset_populator.new_history()
ten_lines = "\n".join( map( str, range( 10 ) ) )
ten_lines = "\n".join( map( str, list(range( 10)) ) )
hda1 = self.dataset_populator.new_dataset( history_id, content=ten_lines )
workflow_request = dict(
history="hist_id=%s" % history_id,
@@ -1558,7 +1558,7 @@ steps:
hda_summary = filter( lambda hc: hc[ "hid" ] == hid, history_contents_response.json() )[ 0 ]
hda_info_response = self._get( "%s/%s" % ( contents_url, hda_summary[ "id" ] ) )
self._assert_status_code_is( hda_info_response, 200 )
self.assertEquals( hda_info_response.json()[ "metadata_data_lines" ], lines )
self.assertEqual( hda_info_response.json()[ "metadata_data_lines" ], lines )
def __invoke_workflow( self, history_id, workflow_id, inputs={}, request={}, assert_ok=True ):
request["history"] = "hist_id=%s" % history_id,