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

Replace `self.assert* method calls with plain asserts in lib/galaxy_test/`

Browse Source 
using https://github.com/pytest-dev/unittest2pytest with the following command:

```
unittest2pytest -f self_assert -w -n lib/galaxy_test/
```
This commit is contained in:
Nicola Soranzo
2022-09-21 23:50:04 +01:00
parent 6dce7be430
commit a69103dd2c
24 changed files with 302 additions and 317 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy_test/api/test_dataset_collections.py
+4 -4
View File
@@ -88,8 +88,8 @@ class DatasetCollectionApiTestCase(ApiTestCase):
assert pair_1_element["element_index"] == 0, pair_1_element
pair_1_object = pair_1_element["object"]
self._assert_has_keys(pair_1_object, "collection_type", "elements", "element_count")
self.assertEqual(pair_1_object["collection_type"], "paired")
self.assertEqual(pair_1_object["populated"], True)
assert pair_1_object["collection_type"] == "paired"
assert pair_1_object["populated"] is True
pair_elements = pair_1_object["elements"]
assert len(pair_elements) == 2
pair_1_element_1 = pair_elements[0]
@@ -233,7 +233,7 @@ class DatasetCollectionApiTestCase(ApiTestCase):
}
self.dataset_populator.fetch(payload)
hdca = self._assert_one_collection_created_in_history()
self.assertEqual(hdca["name"], "Test upload")
assert hdca["name"] == "Test upload"
hdca_tags = hdca["tags"]
assert len(hdca_tags) == 1
assert "name:collection1" in hdca_tags
@@ -262,7 +262,7 @@ class DatasetCollectionApiTestCase(ApiTestCase):
}
self.dataset_populator.fetch(payload)
hdca = self._assert_one_collection_created_in_history()
self.assertEqual(hdca["name"], "Test upload")
assert hdca["name"] == "Test upload"
assert len(hdca["elements"]) == 1, hdca
element0 = hdca["elements"][0]
assert element0["element_identifier"] == "samp1"
lib/galaxy_test/api/test_histories.py
+3 -3
View File
@@ -32,7 +32,7 @@ class BaseHistories:
post_data = dict(name=name)
create_response = self._post("histories", data=post_data).json()
self._assert_has_keys(create_response, "name", "id")
self.assertEqual(create_response["name"], name)
assert create_response["name"] == name
return create_response
def _assert_history_length(self, history_id, n):
@@ -56,14 +56,14 @@ class HistoriesApiTestCase(ApiTestCase, BaseHistories):
# Make sure new history appears in index of user's histories.
index_response = self._get("histories").json()
indexed_history = [h for h in index_response if h["id"] == created_id][0]
self.assertEqual(indexed_history["name"], "TestHistory1")
assert indexed_history["name"] == "TestHistory1"
def test_create_history_json(self):
name = "TestHistoryJson"
post_data = dict(name=name)
create_response = self._post("histories", data=post_data, json=True).json()
self._assert_has_keys(create_response, "name", "id")
self.assertEqual(create_response["name"], name)
assert create_response["name"] == name
return create_response
def test_show_history(self):
lib/galaxy_test/api/test_history_contents.py
+1 -1
View File
@@ -897,7 +897,7 @@ class HistoryContentsApiTestCase(ApiTestCase):
)
self._assert_status_code_is(contents_response, 200)
collection = contents_response.json()[0]
self.assertCountEqual(collection["elements_datatypes"], expected_datatypes)
assert sorted(collection["elements_datatypes"]) == sorted(expected_datatypes)
class HistoryContentsApiNearTestCase(ApiTestCase):
lib/galaxy_test/api/test_jobs.py
+1 -1
View File
@@ -943,7 +943,7 @@ steps:
search_payload = self._search_payload(history_id=history_id, tool_id=tool_id, inputs=inputs)
empty_search_response = self._post("jobs/search", data=search_payload)
self._assert_status_code_is(empty_search_response, 200)
self.assertEqual(len(empty_search_response.json()), 0)
assert len(empty_search_response.json()) == 0
tool_response = self._post("tools", data=search_payload)
self.dataset_populator.wait_for_tool_run(history_id, run_response=tool_response)
self._search(search_payload, expected_search_count=1)
lib/galaxy_test/api/test_pages.py
+7 -7
View File
@@ -98,9 +98,9 @@ steps:
self._assert_status_code_is(show_response, 200)
show_json = show_response.json()
self._assert_has_keys(show_json, "slug", "title", "id")
self.assertEqual(show_json["slug"], "invocation-report")
self.assertEqual(show_json["title"], "Invocation Report")
self.assertEqual(show_json["content_format"], "markdown")
assert show_json["slug"] == "invocation-report"
assert show_json["title"] == "Invocation Report"
assert show_json["content_format"] == "markdown"
markdown_content = show_json["content"]
assert "## Workflow Outputs" in markdown_content
assert "## Workflow Inputs" in markdown_content
@@ -307,10 +307,10 @@ steps:
self._assert_status_code_is(show_response, 200)
show_json = show_response.json()
self._assert_has_keys(show_json, "slug", "title", "id")
self.assertEqual(show_json["slug"], "pagetoshow")
self.assertEqual(show_json["title"], "MY PAGE")
self.assertEqual(show_json["content"], "<p>Page!</p>")
self.assertEqual(show_json["content_format"], "html")
assert show_json["slug"] == "pagetoshow"
assert show_json["title"] == "MY PAGE"
assert show_json["content"] == "<p>Page!</p>"
assert show_json["content_format"] == "html"
def test_403_on_unowner_show(self):
response_json = self._create_valid_page_as("others_page_show@bx.psu.edu", "otherspageshow")
lib/galaxy_test/api/test_tools.py
+160 -162
View File
@@ -347,8 +347,8 @@ class ToolsTestCase(ApiTestCase, TestsTools):
put_response.raise_for_status()
response = get(user_info_url).json()
self.assertEqual(len(response["addresses"]), 1)
self.assertEqual(response["addresses"][0]["desc"], cool_name_with_quote)
assert len(response["addresses"]) == 1
assert response["addresses"][0]["desc"] == cool_name_with_quote
hda1 = self.dataset_populator.new_dataset(history_id, content="1\t2\t3", name=cool_name_with_quote)
assert hda1["name"] == cool_name_with_quote
@@ -549,7 +549,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
response = self._run("__UNZIP_COLLECTION__", history_id, inputs, assert_ok=True)
outputs = response["outputs"]
self.assertEqual(len(outputs), 2)
assert len(outputs) == 2
output_forward = outputs[0]
output_reverse = outputs[1]
output_forward_content = self.dataset_populator.get_history_dataset_content(
@@ -605,7 +605,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
}
response = self._run("__UNZIP_COLLECTION__", history_id, inputs, assert_ok=True)
implicit_collections = response["implicit_collections"]
self.assertEqual(len(implicit_collections), 2)
assert len(implicit_collections) == 2
unzipped_hdca = self.dataset_populator.get_history_collection_details(
history_id, hid=implicit_collections[0]["hid"]
)
@@ -622,7 +622,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
response = self._run("__ZIP_COLLECTION__", history_id, inputs, assert_ok=True)
output_collections = response["output_collections"]
self.assertEqual(len(output_collections), 1)
assert len(output_collections) == 1
self.dataset_populator.wait_for_job(response["jobs"][0]["id"], assert_ok=True)
zipped_hdca = self.dataset_populator.get_history_collection_details(
history_id, hid=output_collections[0]["hid"]
@@ -678,7 +678,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
response = self._run("__ZIP_COLLECTION__", history_id, inputs, assert_ok=True)
implicit_collections = response["implicit_collections"]
self.assertEqual(len(implicit_collections), 1)
assert len(implicit_collections) == 1
self.dataset_populator.wait_for_job(response["jobs"][0]["id"], assert_ok=True)
zipped_hdca = self.dataset_populator.get_history_collection_details(
history_id, hid=implicit_collections[0]["hid"]
@@ -694,7 +694,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
response = self.dataset_populator.run_exit_code_from_file(history_id, ok_hdca_id)
mixed_implicit_collections = response["implicit_collections"]
self.assertEqual(len(mixed_implicit_collections), 1)
assert len(mixed_implicit_collections) == 1
mixed_hdca_hid = mixed_implicit_collections[0]["hid"]
mixed_hdca = self.dataset_populator.get_history_collection_details(
history_id, hid=mixed_hdca_hid, wait=False
@@ -725,7 +725,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
response = self.dataset_populator.run_exit_code_from_file(history_id, ok_hdca_id)
mixed_implicit_collections = response["implicit_collections"]
self.assertEqual(len(mixed_implicit_collections), 1)
assert len(mixed_implicit_collections) == 1
mixed_hdca_hid = mixed_implicit_collections[0]["hid"]
mixed_hdca = self.dataset_populator.get_history_collection_details(
history_id, hid=mixed_hdca_hid, wait=False
@@ -774,7 +774,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
filter_output_collections = response["output_collections"]
if batch:
return response["implicit_collections"][0]
self.assertEqual(len(filter_output_collections), 1)
assert len(filter_output_collections) == 1
filtered_hid = filter_output_collections[0]["hid"]
filtered_hdca = self.dataset_populator.get_history_collection_details(history_id, hid=filtered_hid, wait=False)
return filtered_hdca
@@ -787,7 +787,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self.dataset_populator.wait_for_history(history_id)
response = self._run("__APPLY_RULES__", history_id, inputs, assert_ok=True)
output_collections = response["output_collections"]
self.assertEqual(len(output_collections), 1)
assert len(output_collections) == 1
output_hid = output_collections[0]["hid"]
output_hdca = self.dataset_populator.get_history_collection_details(history_id, hid=output_hid, wait=False)
example["check"](output_hdca, self.dataset_populator)
@@ -874,10 +874,10 @@ class ToolsTestCase(ApiTestCase, TestsTools):
input1=dataset_to_param(new_dataset),
)
outputs = self._cat1_outputs(history_id, inputs=inputs)
self.assertEqual(len(outputs), 1)
assert len(outputs) == 1
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "Cat1Test")
assert output1_content.strip() == "Cat1Test"
@skip_without_tool("cat1")
@uses_test_history(require_new=True)
@@ -909,10 +909,10 @@ class ToolsTestCase(ApiTestCase, TestsTools):
input1=[dataset_to_param(new_dataset)],
)
outputs = self._cat1_outputs(history_id, inputs=inputs)
self.assertEqual(len(outputs), 1)
assert len(outputs) == 1
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "Cat1Testlistified")
assert output1_content.strip() == "Cat1Testlistified"
@skip_without_tool("multiple_versions")
@uses_test_history(require_new=False)
@@ -922,10 +922,10 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = self._run_and_get_outputs(
tool_id="multiple_versions", history_id=history_id, tool_version=version
)
self.assertEqual(len(outputs), 1)
assert len(outputs) == 1
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), f"Version {version}")
assert output1_content.strip() == f"Version {version}"
@skip_without_tool("multiple_versions")
@uses_test_history(require_new=False)
@@ -958,10 +958,10 @@ class ToolsTestCase(ApiTestCase, TestsTools):
input1={"batch": False, "values": [dataset_to_param(new_dataset)]},
)
outputs = self._cat1_outputs(history_id, inputs=inputs)
self.assertEqual(len(outputs), 1)
assert len(outputs) == 1
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "123")
assert output1_content.strip() == "123"
@skip_without_tool("cat1")
@uses_test_history(require_new=False)
@@ -986,7 +986,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
assert_inputs(inputs, can_be_used=False)
outputs = self._cat1_outputs(history_id, inputs=inputs)
self.assertEqual(len(outputs), 1)
assert len(outputs) == 1
output1 = outputs[0]
inputs_2 = dict(
@@ -1175,7 +1175,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self._assert_has_keys(output_collection, "id", "name", "elements", "populated")
assert not output_collection["populated"]
assert len(output_collection["elements"]) == 0
self.assertEqual(output_collection["name"], "Table split on first column")
assert output_collection["name"] == "Table split on first column"
self.dataset_populator.wait_for_job(create["jobs"][0]["id"], assert_ok=True)
get_collection_response = self._get(
@@ -1186,7 +1186,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
output_collection = get_collection_response.json()
self._assert_has_keys(output_collection, "id", "name", "elements", "populated")
assert output_collection["populated"]
self.assertEqual(output_collection["name"], "Table split on first column")
assert output_collection["name"] == "Table split on first column"
assert len(output_collection["elements"]) == 2
output_element_0 = output_collection["elements"][0]
@@ -1228,7 +1228,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
output_content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(output_content, "Hello World\n")
assert output_content == "Hello World\n"
def test_dynamic_tool_from_path(self):
# Create tool.
@@ -1244,7 +1244,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
output_content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(output_content, "Hello World 2\n")
assert output_content == "Hello World 2\n"
def test_dynamic_tool_no_id(self):
# Create tool.
@@ -1257,7 +1257,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
output_content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(output_content, "Hello World 2\n")
assert output_content == "Hello World 2\n"
def test_show_dynamic_tools(self):
# Create tool.
@@ -1316,10 +1316,10 @@ class ToolsTestCase(ApiTestCase, TestsTools):
new_dataset2 = self.dataset_populator.new_dataset(history_id, content="Cat2Test")
inputs = {"input1": dataset_to_param(new_dataset1), "queries_0|input2": dataset_to_param(new_dataset2)}
outputs = self._cat1_outputs(history_id, inputs=inputs)
self.assertEqual(len(outputs), 1)
assert len(outputs) == 1
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "Cat1Test\nCat2Test")
assert output1_content.strip() == "Cat1Test\nCat2Test"
@skip_without_tool("mapper_two")
@uses_test_history(require_new=False)
@@ -1375,13 +1375,13 @@ class ToolsTestCase(ApiTestCase, TestsTools):
def _check_cat1_multirun(self, history_id, inputs):
outputs = self._cat1_outputs(history_id, inputs=inputs)
self.assertEqual(len(outputs), 2)
assert len(outputs) == 2
output1 = outputs[0]
output2 = outputs[1]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
output2_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output2)
self.assertEqual(output1_content.strip(), "123")
self.assertEqual(output2_content.strip(), "456")
assert output1_content.strip() == "123"
assert output2_content.strip() == "456"
@skip_without_tool("random_lines1")
@uses_test_history(require_new=False)
@@ -1422,7 +1422,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
"queries_0|input2": {"batch": True, "values": second_two},
}
outputs = self._cat1_outputs(history_id, inputs=inputs)
self.assertEqual(len(outputs), 2)
assert len(outputs) == 2
outputs_contents = [
self.dataset_populator.get_history_dataset_content(history_id, dataset=o).strip() for o in outputs
]
@@ -1440,7 +1440,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs_contents = [
self.dataset_populator.get_history_dataset_content(history_id, dataset=o).strip() for o in outputs
]
self.assertEqual(len(outputs), 4)
assert len(outputs) == 4
assert "123\n789" in outputs_contents
assert "456\n0ab" in outputs_contents
assert "123\n0ab" in outputs_contents
@@ -1533,19 +1533,19 @@ class ToolsTestCase(ApiTestCase, TestsTools):
implicit_collections = create["implicit_collections"]
collections = create["output_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(implicit_collections), 0)
self.assertEqual(len(collections), 1)
assert len(jobs) == 1
assert len(implicit_collections) == 0
assert len(collections) == 1
output_collection = collections[0]
return output_collection
def _assert_elements_are(self, collection, *args):
elements = collection["elements"]
self.assertEqual(len(elements), len(args))
assert len(elements) == len(args)
for index, element in enumerate(elements):
arg = args[index]
self.assertEqual(arg, element["element_identifier"])
assert arg == element["element_identifier"]
return elements
def _verify_element(self, history_id, element, **props):
@@ -1555,14 +1555,14 @@ class ToolsTestCase(ApiTestCase, TestsTools):
expected_contents = props["contents"]
contents = self.dataset_populator.get_history_dataset_content(history_id, dataset_id=object_id)
self.assertEqual(contents, expected_contents)
assert contents == expected_contents
del props["contents"]
if props:
details = self.dataset_populator.get_history_dataset_details(history_id, dataset_id=object_id)
for key, value in props.items():
self.assertEqual(details[key], value)
assert details[key] == value
def _setup_repeat_multirun(self):
history_id = self.dataset_populator.new_history()
@@ -1577,13 +1577,13 @@ class ToolsTestCase(ApiTestCase, TestsTools):
def _check_repeat_multirun(self, history_id, inputs):
outputs = self._cat1_outputs(history_id, inputs=inputs)
self.assertEqual(len(outputs), 2)
assert len(outputs) == 2
output1 = outputs[0]
output2 = outputs[1]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
output2_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output2)
self.assertEqual(output1_content.strip(), "Common\n123")
self.assertEqual(output2_content.strip(), "Common\n456")
assert output1_content.strip() == "Common\n123"
assert output2_content.strip() == "Common\n456"
def _setup_two_multiruns(self):
history_id = self.dataset_populator.new_history()
@@ -1619,9 +1619,9 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 0)
self.assertEqual(len(outputs), 0)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 0
assert len(outputs) == 0
assert len(implicit_collections) == 1
empty_output = implicit_collections[0]
assert empty_output["name"] == "Concatenate datasets on collection 1", empty_output
@@ -1639,9 +1639,9 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 2)
self.assertEqual(len(outputs), 2)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 2
assert len(outputs) == 2
assert len(implicit_collections) == 1
output1 = outputs[0]
output2 = outputs[1]
output1_details = self.dataset_populator.get_history_dataset_details(history_id, dataset=output1)
@@ -1658,9 +1658,9 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 2)
self.assertEqual(len(outputs), 2)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 2
assert len(outputs) == 2
assert len(implicit_collections) == 1
for output in outputs:
assert output["file_ext"] == "txt", output
@@ -1678,8 +1678,8 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("output_filter_with_input", history_id, inputs).json()
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 3)
self.assertEqual(len(implicit_collections), 3)
assert len(jobs) == 3
assert len(implicit_collections) == 3
self._check_implicit_collection_populated(create)
@skip_without_tool("output_filter_with_input_optional")
@@ -1694,9 +1694,9 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("output_filter_with_input_optional", history_id, inputs).json()
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
assert len(jobs) == 1
self.dataset_populator.wait_for_job(jobs[0]["id"], assert_ok=True)
self.assertEqual(len(implicit_collections), 1)
assert len(implicit_collections) == 1
self._check_implicit_collection_populated(create)
@skip_without_tool("output_filter_with_input")
@@ -1713,8 +1713,8 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("output_filter_with_input", history_id, inputs).json()
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 3)
self.assertEqual(len(implicit_collections), 2)
assert len(jobs) == 3
assert len(implicit_collections) == 2
self._check_implicit_collection_populated(create)
@skip_without_tool("Cut1")
@@ -1730,9 +1730,9 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 2)
self.assertEqual(len(outputs), 2)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 2
assert len(outputs) == 2
assert len(implicit_collections) == 1
output1 = outputs[0]
output2 = outputs[1]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
@@ -1749,14 +1749,12 @@ class ToolsTestCase(ApiTestCase, TestsTools):
inputs = {"input": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]}}
create = self._run("collection_creates_dynamic_list_of_pairs", history_id, inputs).json()
implicit_collections = create["implicit_collections"]
self.assertEqual(len(implicit_collections), 1)
self.assertEqual(implicit_collections[0]["collection_type"], "list:list:paired")
self.assertEqual(implicit_collections[0]["elements"][0]["object"]["element_count"], None)
assert len(implicit_collections) == 1
assert implicit_collections[0]["collection_type"] == "list:list:paired"
assert implicit_collections[0]["elements"][0]["object"]["element_count"] is None
self.dataset_populator.wait_for_job(create["jobs"][0]["id"], assert_ok=True)
hdca = self._get(f"histories/{history_id}/contents/dataset_collections/{implicit_collections[0]['id']}").json()
self.assertEqual(
hdca["elements"][0]["object"]["elements"][0]["object"]["elements"][0]["element_identifier"], "forward"
)
assert hdca["elements"][0]["object"]["elements"][0]["object"]["elements"][0]["element_identifier"] == "forward"
def _bed_list(self, history_id):
bed1_contents = open(self.get_filename("1.bed")).read()
@@ -1778,15 +1776,15 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 2)
self.assertEqual(len(outputs), 2)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 2
assert len(outputs) == 2
assert len(implicit_collections) == 1
output1 = outputs[0]
output2 = outputs[1]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
output2_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output2)
self.assertEqual(output1_content.strip(), "forward")
self.assertEqual(output2_content.strip(), "reverse")
assert output1_content.strip() == "forward"
assert output2_content.strip() == "reverse"
@skip_without_tool("identifier_single")
@uses_test_history(require_new=False)
@@ -1801,12 +1799,12 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
self.assertEqual(len(implicit_collections), 0)
assert len(jobs) == 1
assert len(outputs) == 1
assert len(implicit_collections) == 0
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "Plain HDA")
assert output1_content.strip() == "Plain HDA"
@skip_without_tool("identifier_multiple")
@uses_test_history(require_new=False)
@@ -1828,12 +1826,12 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
self.assertEqual(len(implicit_collections), 0)
assert len(jobs) == 1
assert len(outputs) == 1
assert len(implicit_collections) == 0
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "forward\nreverse")
assert output1_content.strip() == "forward\nreverse"
@skip_without_tool("identifier_in_conditional")
@uses_test_history(require_new=False)
@@ -1848,12 +1846,12 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
self.assertEqual(len(implicit_collections), 0)
assert len(jobs) == 1
assert len(outputs) == 1
assert len(implicit_collections) == 0
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "forward\nreverse")
assert output1_content.strip() == "forward\nreverse"
@skip_without_tool("identifier_in_conditional")
@uses_test_history(require_new=False)
@@ -1871,12 +1869,12 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
self.assertEqual(len(implicit_collections), 0)
assert len(jobs) == 1
assert len(outputs) == 1
assert len(implicit_collections) == 0
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "forward\nreverse")
assert output1_content.strip() == "forward\nreverse"
@skip_without_tool("identifier_multiple_in_repeat")
@uses_test_history(require_new=False)
@@ -1891,12 +1889,12 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
self.assertEqual(len(implicit_collections), 0)
assert len(jobs) == 1
assert len(outputs) == 1
assert len(implicit_collections) == 0
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "forward\nreverse")
assert output1_content.strip() == "forward\nreverse"
@skip_without_tool("identifier_in_conditional")
@uses_test_history(require_new=False)
@@ -1913,15 +1911,15 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 2)
self.assertEqual(len(outputs), 2)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 2
assert len(outputs) == 2
assert len(implicit_collections) == 1
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "forward")
assert output1_content.strip() == "forward"
output2 = outputs[1]
output2_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output2)
self.assertEqual(output2_content.strip(), "reverse")
assert output2_content.strip() == "reverse"
@skip_without_tool("identifier_multiple_in_conditional")
@uses_test_history(require_new=False)
@@ -1936,12 +1934,12 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
self.assertEqual(len(implicit_collections), 0)
assert len(jobs) == 1
assert len(outputs) == 1
assert len(implicit_collections) == 0
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "forward\nreverse")
assert output1_content.strip() == "forward\nreverse"
@skip_without_tool("identifier_multiple_in_repeat")
@uses_test_history(require_new=False)
@@ -1956,12 +1954,12 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
self.assertEqual(len(implicit_collections), 0)
assert len(jobs) == 1
assert len(outputs) == 1
assert len(implicit_collections) == 0
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "forward\nreverse")
assert output1_content.strip() == "forward\nreverse"
@skip_without_tool("identifier_single_in_repeat")
@uses_test_history(require_new=False)
@@ -1973,8 +1971,8 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = create_response.json()
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 2)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 2
assert len(implicit_collections) == 1
output_collection = implicit_collections[0]
elements = output_collection["elements"]
assert len(elements) == 2
@@ -1995,12 +1993,12 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
self.assertEqual(len(implicit_collections), 0)
assert len(jobs) == 1
assert len(outputs) == 1
assert len(implicit_collections) == 0
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "Normal HDA1")
assert output1_content.strip() == "Normal HDA1"
@skip_without_tool("identifier_multiple")
@uses_test_history(require_new=False)
@@ -2014,12 +2012,12 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
self.assertEqual(len(implicit_collections), 0)
assert len(jobs) == 1
assert len(outputs) == 1
assert len(implicit_collections) == 0
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "Normal HDA1\nNormal HDA2")
assert output1_content.strip() == "Normal HDA1\nNormal HDA2"
@skip_without_tool("identifier_collection")
@uses_test_history(require_new=False)
@@ -2046,11 +2044,11 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = create_response.json()
outputs = create["outputs"]
jobs = create["jobs"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
assert len(jobs) == 1
assert len(outputs) == 1
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
self.assertEqual(output1_content.strip(), "\n".join(d["name"] for d in element_identifiers))
assert output1_content.strip() == "\n".join(d["name"] for d in element_identifiers)
@skip_without_tool("identifier_in_actions")
@uses_test_history(require_new=False)
@@ -2101,8 +2099,8 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("collection_paired_structured_like", history_id, inputs, assert_ok=True)
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 2)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 2
assert len(implicit_collections) == 1
implicit_collection = implicit_collections[0]
assert implicit_collection["collection_type"] == "list:paired", implicit_collection["collection_type"]
outer_elements = implicit_collection["elements"]
@@ -2120,8 +2118,8 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("collection_paired_conditional_structured_like", history_id, inputs, assert_ok=True)
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 2)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 2
assert len(implicit_collections) == 1
implicit_collection = implicit_collections[0]
assert implicit_collection["collection_type"] == "list:paired", implicit_collection["collection_type"]
outer_elements = implicit_collection["elements"]
@@ -2132,9 +2130,9 @@ class ToolsTestCase(ApiTestCase, TestsTools):
outputs = create["outputs"]
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 4)
self.assertEqual(len(outputs), 4)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 4
assert len(outputs) == 4
assert len(implicit_collections) == 1
implicit_collection = implicit_collections[0]
self._assert_has_keys(implicit_collection, "collection_type", "elements")
assert implicit_collection["collection_type"] == "list:paired"
@@ -2147,7 +2145,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
assert first_object["collection_type"] == "paired"
assert len(first_object["elements"]) == 2
first_object_forward_element = first_object["elements"][0]
self.assertEqual(outputs[0]["id"], first_object_forward_element["object"]["id"])
assert outputs[0]["id"] == first_object_forward_element["object"]["id"]
@skip_without_tool("cat1")
@uses_test_history(require_new=False)
@@ -2165,17 +2163,17 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self._assert_status_code_is(response, 200)
response_object = response.json()
outputs = response_object["outputs"]
self.assertEqual(len(outputs), 2)
assert len(outputs) == 2
output1 = outputs[0]
output2 = outputs[1]
self.dataset_populator.wait_for_history(history_id)
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
output2_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output2)
self.assertEqual(output1_content.strip(), "123\n789")
self.assertEqual(output2_content.strip(), "456\n0ab")
assert output1_content.strip() == "123\n789"
assert output2_content.strip() == "456\n0ab"
self.assertEqual(len(response_object["jobs"]), 2)
self.assertEqual(len(response_object["implicit_collections"]), 1)
assert len(response_object["jobs"]) == 2
assert len(response_object["implicit_collections"]) == 1
@skip_without_tool("cat1")
@uses_test_history(require_new=False)
@@ -2190,13 +2188,13 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self._assert_status_code_is(response, 200)
response_object = response.json()
outputs = response_object["outputs"]
self.assertEqual(len(outputs), 4)
assert len(outputs) == 4
self.assertEqual(len(response_object["jobs"]), 4)
assert len(response_object["jobs"]) == 4
implicit_collections = response_object["implicit_collections"]
self.assertEqual(len(implicit_collections), 1)
assert len(implicit_collections) == 1
implicit_collection = implicit_collections[0]
self.assertEqual(implicit_collection["collection_type"], "paired:paired")
assert implicit_collection["collection_type"] == "paired:paired"
outer_elements = implicit_collection["elements"]
assert len(outer_elements) == 2
@@ -2227,7 +2225,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
for (element, expected_contents) in expected_contents_list:
dataset_id = element["object"]["id"]
contents = self.dataset_populator.get_history_dataset_content(history_id, dataset_id=dataset_id)
self.assertEqual(expected_contents, contents)
assert expected_contents == contents
@skip_without_tool("cat1")
@uses_test_history(require_new=False)
@@ -2247,10 +2245,10 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self._assert_status_code_is(response, 200)
response_object = response.json()
outputs = response_object["outputs"]
self.assertEqual(len(outputs), 2)
assert len(outputs) == 2
self.assertEqual(len(response_object["jobs"]), 2)
self.assertEqual(len(response_object["implicit_collections"]), 1)
assert len(response_object["jobs"]) == 2
assert len(response_object["implicit_collections"]) == 1
@skip_without_tool("identifier_source")
def test_default_identifier_source_map_over(self):
@@ -2287,8 +2285,8 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("collection_creates_pair", history_id, inputs, assert_ok=True)
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 2)
self.assertEqual(len(implicit_collections), 1)
assert len(jobs) == 2
assert len(implicit_collections) == 1
implicit_collection = implicit_collections[0]
assert implicit_collection["collection_type"] == "list:paired", implicit_collection
outer_elements = implicit_collection["elements"]
@@ -2319,7 +2317,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
]
for i in range(4):
contents = self.dataset_populator.get_history_dataset_content(history_id, dataset_id=pair_ids[i])
self.assertEqual(expected_contents[i], contents)
assert expected_contents[i] == contents
@skip_without_tool("cat1")
def test_cannot_map_over_incompatible_collections(self):
@@ -2436,8 +2434,8 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("multi_data_param", history_id, inputs, assert_ok=True)
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(implicit_collections), 2)
assert len(jobs) == 1
assert len(implicit_collections) == 2
output_hdca = self.dataset_populator.get_history_collection_details(
history_id, hid=implicit_collections[0]["hid"]
)
@@ -2482,7 +2480,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
}
create = self._run("column_multi_param", history_id, inputs, assert_ok=True)
jobs = create["jobs"]
self.assertEqual(len(jobs), 1)
assert len(jobs) == 1
content = self.dataset_populator.get_history_dataset_content(history_id, hid=3)
assert content.strip() == "hg17", content
@@ -2496,8 +2494,8 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("multi_data_repeat", history_id, inputs, assert_ok=True)
outputs = create["outputs"]
jobs = create["jobs"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
assert len(jobs) == 1
assert len(outputs) == 1
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
assert output1_content.strip() == "123\n456", output1_content
@@ -2512,8 +2510,8 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("multi_data_repeat", history_id, inputs, assert_ok=True)
outputs = create["outputs"]
jobs = create["jobs"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 1)
assert len(jobs) == 1
assert len(outputs) == 1
output1 = outputs[0]
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
assert output1_content.strip() == "123\n456", output1_content
@@ -2532,20 +2530,20 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("multi_data_param", history_id, inputs, assert_ok=True)
outputs = create["outputs"]
jobs = create["jobs"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 2)
assert len(jobs) == 1
assert len(outputs) == 2
output1, output2 = outputs
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
output2_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output2)
self.assertEqual(output1_content.strip(), "123\n456\nTestData123\nTestData123\nTestData123")
self.assertEqual(output2_content.strip(), "123\n456")
assert output1_content.strip() == "123\n456\nTestData123\nTestData123\nTestData123"
assert output2_content.strip() == "123\n456"
def _check_simple_reduce_job(self, history_id, inputs):
create = self._run("multi_data_param", history_id, inputs, assert_ok=True)
outputs = create["outputs"]
jobs = create["jobs"]
self.assertEqual(len(jobs), 1)
self.assertEqual(len(outputs), 2)
assert len(jobs) == 1
assert len(outputs) == 2
output1, output2 = outputs
output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
output2_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output2)
@@ -2654,10 +2652,10 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
response = self._run("collection_cat_group_tag", history_id, inputs, assert_ok=True)
outputs = response["outputs"]
self.assertEqual(len(outputs), 1)
assert len(outputs) == 1
output = outputs[0]
output_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output)
self.assertEqual(output_content.strip(), "123\n456")
assert output_content.strip() == "123\n456"
@skip_without_tool("collection_cat_group_tag_multiple")
@uses_test_history(require_new=False)
@@ -2670,10 +2668,10 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
response = self._run("collection_cat_group_tag_multiple", history_id, inputs, assert_ok=True)
outputs = response["outputs"]
self.assertEqual(len(outputs), 1)
assert len(outputs) == 1
output = outputs[0]
output_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output)
self.assertEqual(output_content.strip(), "123\n456\n456\n0ab")
assert output_content.strip() == "123\n456\n456\n0ab"
@skip_without_tool("expression_forty_two")
def test_galaxy_expression_tool_simplest(self):
@@ -2682,7 +2680,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self._assert_status_code_is(run_response, 200)
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
output_content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(output_content, "42")
assert output_content == "42"
@skip_without_tool("expression_parse_int")
def test_galaxy_expression_tool_simple(self):
@@ -2694,7 +2692,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self._assert_status_code_is(run_response, 200)
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
output_content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(output_content, "7")
assert output_content == "7"
@skip_without_tool("expression_log_line_count")
def test_galaxy_expression_metadata(self):
@@ -2709,7 +2707,7 @@ class ToolsTestCase(ApiTestCase, TestsTools):
self._assert_status_code_is(run_response, 200)
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
output_content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(output_content, "3")
assert output_content == "3"
@skip_without_tool("cat1")
@uses_test_history(require_new=False)
@@ -2981,9 +2979,9 @@ class ToolsTestCase(ApiTestCase, TestsTools):
create = self._run("collection_paired_structured_like", history_id, inputs, assert_ok=True)
jobs = create["jobs"]
implicit_collections = create["implicit_collections"]
self.assertEqual(len(jobs), 2)
assert len(jobs) == 2
self.dataset_populator.wait_for_jobs(jobs, assert_ok=True)
self.assertEqual(len(implicit_collections), 1)
assert len(implicit_collections) == 1
implicit_collection = implicit_collections[0]
assert implicit_collection["collection_type"] == "list:paired", implicit_collection["collection_type"]
outer_elements = implicit_collection["elements"]
lib/galaxy_test/api/test_tools_upload.py
+23 -23
View File
@@ -51,22 +51,22 @@ class ToolsUploadTestCase(ApiTestCase):
def test_upload_posix_newline_fixes_by_default(self):
windows_content = ONE_TO_SIX_ON_WINDOWS
result_content = self._upload_and_get_content(windows_content)
self.assertEqual(result_content, ONE_TO_SIX_WITH_TABS)
assert result_content == ONE_TO_SIX_WITH_TABS
def test_fetch_posix_unaltered(self):
windows_content = ONE_TO_SIX_ON_WINDOWS
result_content = self._upload_and_get_content(windows_content, api="fetch")
self.assertEqual(result_content, ONE_TO_SIX_ON_WINDOWS)
assert result_content == ONE_TO_SIX_ON_WINDOWS
def test_upload_disable_posix_fix(self):
windows_content = ONE_TO_SIX_ON_WINDOWS
result_content = self._upload_and_get_content(windows_content, to_posix_lines=None)
self.assertEqual(result_content, windows_content)
assert result_content == windows_content
def test_fetch_post_lines_option(self):
windows_content = ONE_TO_SIX_ON_WINDOWS
result_content = self._upload_and_get_content(windows_content, api="fetch", to_posix_lines=True)
self.assertEqual(result_content, ONE_TO_SIX_WITH_TABS)
assert result_content == ONE_TO_SIX_WITH_TABS
# Test how trailing new lines are added
# - upload1 adds by default because to_posix_lines is on by default
@@ -75,47 +75,47 @@ class ToolsUploadTestCase(ApiTestCase):
def test_post_lines_trailing(self):
input_content = ONE_TO_SIX_WITH_TABS_NO_TRAILING_NEWLINE
result_content = self._upload_and_get_content(input_content)
self.assertEqual(result_content, ONE_TO_SIX_WITH_TABS)
assert result_content == ONE_TO_SIX_WITH_TABS
def test_post_lines_trailing_off(self):
input_content = ONE_TO_SIX_WITH_TABS_NO_TRAILING_NEWLINE
result_content = self._upload_and_get_content(input_content, to_posix_lines=False)
self.assertEqual(result_content, ONE_TO_SIX_WITH_TABS_NO_TRAILING_NEWLINE)
assert result_content == ONE_TO_SIX_WITH_TABS_NO_TRAILING_NEWLINE
def test_fetch_post_lines_trailing_off_by_default(self):
input_content = ONE_TO_SIX_WITH_TABS_NO_TRAILING_NEWLINE
result_content = self._upload_and_get_content(input_content, api="fetch")
self.assertEqual(result_content, ONE_TO_SIX_WITH_TABS_NO_TRAILING_NEWLINE)
assert result_content == ONE_TO_SIX_WITH_TABS_NO_TRAILING_NEWLINE
def test_fetch_post_lines_trailing_if_to_posix(self):
input_content = ONE_TO_SIX_WITH_TABS_NO_TRAILING_NEWLINE
result_content = self._upload_and_get_content(input_content, api="fetch", to_posix_lines=True)
self.assertEqual(result_content, ONE_TO_SIX_WITH_TABS)
assert result_content == ONE_TO_SIX_WITH_TABS
def test_upload_tab_to_space_off_by_default(self):
table = ONE_TO_SIX_WITH_SPACES
result_content = self._upload_and_get_content(table)
self.assertEqual(result_content, table)
assert result_content == table
def test_fetch_tab_to_space_off_by_default(self):
table = ONE_TO_SIX_WITH_SPACES
result_content = self._upload_and_get_content(table, api="fetch")
self.assertEqual(result_content, table)
assert result_content == table
def test_upload_tab_to_space(self):
table = ONE_TO_SIX_WITH_SPACES
result_content = self._upload_and_get_content(table, space_to_tab="Yes")
self.assertEqual(result_content, ONE_TO_SIX_WITH_TABS)
assert result_content == ONE_TO_SIX_WITH_TABS
def test_fetch_tab_to_space(self):
table = ONE_TO_SIX_WITH_SPACES
result_content = self._upload_and_get_content(table, api="fetch", space_to_tab=True)
self.assertEqual(result_content, ONE_TO_SIX_WITH_TABS)
assert result_content == ONE_TO_SIX_WITH_TABS
def test_fetch_tab_to_space_doesnt_swap_newlines(self):
table = ONE_TO_SIX_WITH_SPACES_ON_WINDOWS
result_content = self._upload_and_get_content(table, api="fetch", space_to_tab=True)
self.assertEqual(result_content, ONE_TO_SIX_ON_WINDOWS)
assert result_content == ONE_TO_SIX_ON_WINDOWS
def test_fetch_compressed_with_explicit_type(self):
fastqgz_path = TestDataResolver().get_filename("1.fastqsanger.gz")
@@ -191,42 +191,42 @@ class ToolsUploadTestCase(ApiTestCase):
rdata_path = TestDataResolver().get_filename("1.RData")
with open(rdata_path, "rb") as fh:
rdata_metadata = self._upload_and_get_details(fh, file_type="auto")
self.assertEqual(rdata_metadata["file_ext"], "rdata")
assert rdata_metadata["file_ext"] == "rdata"
@skip_without_datatype("csv")
def test_csv_upload(self):
csv_path = TestDataResolver().get_filename("1.csv")
with open(csv_path, "rb") as fh:
csv_metadata = self._upload_and_get_details(fh, file_type="csv")
self.assertEqual(csv_metadata["file_ext"], "csv")
assert csv_metadata["file_ext"] == "csv"
@skip_without_datatype("csv")
def test_csv_upload_auto(self):
csv_path = TestDataResolver().get_filename("1.csv")
with open(csv_path, "rb") as fh:
csv_metadata = self._upload_and_get_details(fh, file_type="auto")
self.assertEqual(csv_metadata["file_ext"], "csv")
assert csv_metadata["file_ext"] == "csv"
@skip_without_datatype("csv")
def test_csv_fetch(self):
csv_path = TestDataResolver().get_filename("1.csv")
with open(csv_path, "rb") as fh:
csv_metadata = self._upload_and_get_details(fh, api="fetch", ext="csv", to_posix_lines=True)
self.assertEqual(csv_metadata["file_ext"], "csv")
assert csv_metadata["file_ext"] == "csv"
@skip_without_datatype("csv")
def test_csv_sniff_fetch(self):
csv_path = TestDataResolver().get_filename("1.csv")
with open(csv_path, "rb") as fh:
csv_metadata = self._upload_and_get_details(fh, api="fetch", ext="auto", to_posix_lines=True)
self.assertEqual(csv_metadata["file_ext"], "csv")
assert csv_metadata["file_ext"] == "csv"
@skip_without_datatype("tiff")
def test_image_upload_auto(self):
tiff_path = TestDataResolver().get_filename("1.tiff")
with open(tiff_path, "rb") as fh:
tiff_metadata = self._upload_and_get_details(fh, file_type="auto")
self.assertEqual(tiff_metadata["file_ext"], "tiff")
assert tiff_metadata["file_ext"] == "tiff"
@uses_test_history(require_new=False)
def test_newlines_stage_fetch(self, history_id):
@@ -241,7 +241,7 @@ class ToolsUploadTestCase(ApiTestCase):
dataset = datasets[0][0]
content = self.dataset_populator.get_history_dataset_content(history_id=history_id, dataset=dataset)
# By default this appends the newline.
self.assertEqual(content, "This is a line of text.\n")
assert content == "This is a line of text.\n"
@uses_test_history(require_new=False)
def test_stage_object(self, history_id):
@@ -251,7 +251,7 @@ class ToolsUploadTestCase(ApiTestCase):
)
dataset = datasets[0][0]
content = self.dataset_populator.get_history_dataset_content(history_id=history_id, dataset=dataset)
self.assertEqual(content.strip(), '"randomstr"')
assert content.strip() == '"randomstr"'
@uses_test_history(require_new=False)
def test_stage_object_fetch(self, history_id):
@@ -259,7 +259,7 @@ class ToolsUploadTestCase(ApiTestCase):
inputs, datasets = stage_inputs(self.galaxy_interactor, history_id, job, use_path_paste=False)
dataset = datasets[0][0]
content = self.dataset_populator.get_history_dataset_content(history_id=history_id, dataset=dataset)
self.assertEqual(content, '"randomstr"')
assert content == '"randomstr"'
@uses_test_history(require_new=False)
def test_newlines_stage_fetch_configured(self, history_id):
@@ -277,7 +277,7 @@ class ToolsUploadTestCase(ApiTestCase):
dataset = datasets[0][0]
content = self.dataset_populator.get_history_dataset_content(history_id=history_id, dataset=dataset)
# By default this appends the newline, but we disabled with 'to_posix_lines=False' above.
self.assertEqual(content, "This is a line of text.")
assert content == "This is a line of text."
details = self.dataset_populator.get_history_dataset_details(history_id=history_id, dataset=dataset)
assert details["genome_build"] == "hg19"
lib/galaxy_test/api/test_users.py
+13 -13
View File
@@ -52,7 +52,7 @@ class UsersApiTestCase(ApiTestCase):
update_response = self.__update(user, username=new_name)
self._assert_status_code_is(update_response, 200)
update_json = update_response.json()
self.assertEqual(update_json["username"], new_name)
assert update_json["username"] == new_name
# too short
update_response = self.__update(user, username="")
@@ -75,7 +75,7 @@ class UsersApiTestCase(ApiTestCase):
update_response = put(update_url, data=json.dumps(dict(username=new_name)))
self._assert_status_code_is(update_response, 200)
update_json = update_response.json()
self.assertEqual(update_json["username"], new_name)
assert update_json["username"] == new_name
def test_delete_user(self):
user = self._setup_user(TEST_USER_EMAIL_DELETE)
@@ -111,27 +111,27 @@ class UsersApiTestCase(ApiTestCase):
user = self._setup_user(TEST_USER_EMAIL)
url = self.__url("information/inputs", user)
response = get(url).json()
self.assertEqual(response["username"], user["username"])
self.assertEqual(response["email"], TEST_USER_EMAIL)
assert response["username"] == user["username"]
assert response["email"] == TEST_USER_EMAIL
put(url, data=json.dumps(dict(username="newname", email="new@email.email")))
response = get(url).json()
self.assertEqual(response["username"], "newname")
self.assertEqual(response["email"], "new@email.email")
assert response["username"] == "newname"
assert response["email"] == "new@email.email"
put(url, data=json.dumps(dict(username=user["username"], email=TEST_USER_EMAIL)))
response = get(url).json()
self.assertEqual(response["username"], user["username"])
self.assertEqual(response["email"], TEST_USER_EMAIL)
assert response["username"] == user["username"]
assert response["email"] == TEST_USER_EMAIL
put(url, data=json.dumps({"address_0|desc": "_desc"}))
response = get(url).json()
self.assertEqual(len(response["addresses"]), 1)
self.assertEqual(response["addresses"][0]["desc"], "_desc")
assert len(response["addresses"]) == 1
assert response["addresses"][0]["desc"] == "_desc"
def test_create_api_key(self):
user = self._setup_user(TEST_USER_EMAIL)
user_id = user["id"]
response = self._put(f"users/{user_id}/api_key/inputs", admin=True)
self._assert_status_code_is_ok(response)
self.assertEqual(response.json()["inputs"][0]["name"], "api-key")
assert response.json()["inputs"][0]["name"] == "api-key"
@skip_without_tool("cat1")
def test_favorites(self):
@@ -140,7 +140,7 @@ class UsersApiTestCase(ApiTestCase):
url = self._api_url(f"users/{user['id']}/favorites/tools", params=dict(key=self.master_api_key))
put_response = put(url, data=json.dumps({"object_id": "cat1"}))
self._assert_status_code_is_ok(put_response)
self.assertEqual(put_response.json()["tools"][0], "cat1")
assert put_response.json()["tools"][0] == "cat1"
# not implemented for workflows yet
url = self._api_url(f"users/{user['id']}/favorites/workflows", params=dict(key=self.master_api_key))
put_response = put(url, data=json.dumps({"object_id": "14ds68f4sda68gf46dsag4"}))
@@ -149,7 +149,7 @@ class UsersApiTestCase(ApiTestCase):
url = self._api_url(f"users/{user['id']}/favorites/tools/cat1", params=dict(key=self.master_api_key))
delete_response = delete(url)
self._assert_status_code_is_ok(delete_response)
self.assertEqual(delete_response.json()["tools"], [])
assert delete_response.json()["tools"] == []
# delete non-existing tool favorite
url = self._api_url(
f"users/{user['id']}/favorites/tools/madeuptoolthatdoes/not/exist/in/favs",
lib/galaxy_test/api/test_workflow_extraction.py
+6 -6
View File
@@ -32,7 +32,7 @@ class WorkflowExtractionApiTestCase(BaseWorkflowsApiTestCase):
dataset_ids=input_hids,
job_ids=[cat1_job_id],
)
self.assertEqual(downloaded_workflow["name"], "test import from history")
assert downloaded_workflow["name"] == "test import from history"
self.__assert_looks_like_cat1_example_workflow(downloaded_workflow)
@summarize_instance_history_on_error
@@ -222,7 +222,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.assertEqual(collection_step_state["collection_type"], "paired")
assert collection_step_state["collection_type"] == "paired"
@skip_without_tool("cat_collection")
def test_subcollection_mapping(self):
@@ -264,7 +264,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.assertEqual(collection_step_state["collection_type"], "list:paired")
assert collection_step_state["collection_type"] == "list:paired"
@skip_without_tool("cat_list")
@skip_without_tool("collection_creates_dynamic_nested")
@@ -450,8 +450,8 @@ test_data:
input1 = tool_step["input_connections"]["input1"]
input2 = tool_step["input_connections"]["queries_0|input2"]
self.assertEqual(input_steps[0]["id"], input1["id"])
self.assertEqual(input_steps[1]["id"], input2["id"])
assert input_steps[0]["id"] == input1["id"]
assert input_steps[1]["id"] == input2["id"]
def _history_contents(self, history_id=None):
if history_id is None:
@@ -484,7 +484,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.assertEqual(collection_step_state["collection_type"], "paired")
assert collection_step_state["collection_type"] == "paired"
collect_step_idx = collection_step["id"]
return collect_step_idx
lib/galaxy_test/api/test_workflows.py
+48 -61
View File
@@ -223,7 +223,7 @@ class BaseWorkflowsApiTestCase(ApiTestCase, RunsWorkflowFixtures):
def _assert_history_job_count(self, history_id, n):
jobs = self._history_jobs(history_id)
self.assertEqual(len(jobs), n)
assert len(jobs) == n
def _download_workflow(self, workflow_id, style=None, history_id=None):
return self.workflow_populator.download_workflow(workflow_id, style=style, history_id=history_id)
@@ -314,7 +314,7 @@ class WorkflowsApiTestCase(BaseWorkflowsApiTestCase, ChangeDatatypeTestCase):
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])
assert sorted(step["id"] for step in workflow["steps"].values()) == [0, 1, 2]
show_response = self._get(f"workflows/{workflow_id}", {"legacy": True})
workflow = show_response.json()
@@ -322,7 +322,7 @@ class WorkflowsApiTestCase(BaseWorkflowsApiTestCase, ChangeDatatypeTestCase):
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])
assert sorted(step["id"] for step in workflow["steps"].values()) != [0, 1, 2]
def test_show_invalid_key_is_400(self):
show_response = self._get(f"workflows/{self._random_key()}")
@@ -768,8 +768,8 @@ class WorkflowsApiTestCase(BaseWorkflowsApiTestCase, ChangeDatatypeTestCase):
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"])
assert uuids[order_index] == step_dict["uuid"]
assert labels[order_index] == step_dict["label"]
upload_response = self.__test_upload(workflow=original_workflow)
workflow_id = upload_response.json()["id"]
@@ -1347,7 +1347,7 @@ steps:
def test_workflow_run_output_collections(self) -> None:
with self.dataset_populator.test_history() as history_id:
self._run_workflow(WORKFLOW_WITH_OUTPUT_COLLECTION, history_id=history_id)
self.assertEqual("a\nc\nb\nd\n", self.dataset_populator.get_history_dataset_content(history_id, hid=0))
assert "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")
@@ -1642,9 +1642,7 @@ steps:
}
invocation_id = self.__invoke_workflow(workflow_id, inputs=inputs, history_id=history_id)
self.workflow_populator.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
self.assertEqual(
"a\nc\nb\nd\ne\ng\nf\nh\n", self.dataset_populator.get_history_dataset_content(history_id, hid=0)
)
assert "a\nc\nb\nd\ne\ng\nf\nh\n" == self.dataset_populator.get_history_dataset_content(history_id, hid=0)
@skip_without_tool("cat_list")
@skip_without_tool("collection_split_on_column")
@@ -1655,7 +1653,7 @@ steps:
last_item_hid = details["hid"]
assert last_item_hid == 7, f"Expected 7 history items, got {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)
assert "10.0\n30.0\n20.0\n40.0\n" == content
@skip_without_tool("collection_split_on_column")
@skip_without_tool("min_repeat")
@@ -1696,7 +1694,7 @@ steps:
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")
assert content.strip() == "samp1\t10.0\nsamp2\t20.0"
@skip_without_tool("cat")
@skip_without_tool("collection_split_on_column")
@@ -1885,9 +1883,9 @@ outer_input:
invocation_id = summary.invocation_id
content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(
"chrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n",
content,
assert (
content
== "chrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n"
)
steps = self.workflow_populator.get_invocation(invocation_id)["steps"]
assert sum(1 for step in steps if step["subworkflow_invocation_id"] is None) == 3
@@ -2034,12 +2032,12 @@ test_data:
"""
summary = self._run_workflow(workflow_text, test_data=test_data, history_id=history_id)
jobs = summary.jobs
assert len(jobs) == 4, "4 jobs expected, got %d jobs" % len(jobs)
assert len(jobs) == 4, f"4 jobs expected, got {len(jobs)} jobs"
content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(
"chrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n",
content,
assert (
content
== "chrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n"
)
run_test(NESTED_WORKFLOW_AUTO_LABELS_MODERN_SYNTAX)
@@ -2080,7 +2078,7 @@ steps:
invocation_id = self.__invoke_workflow(workflow_id, inputs=inputs, history_id=history_id)
self.workflow_populator.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")
assert content.strip() == "samp1\t10.0\nsamp2\t20.0\nsamp1\t20.0\nsamp2\t40.0"
@skip_without_tool("collection_paired_test")
def test_workflow_flatten(self):
@@ -2214,7 +2212,7 @@ steps:
invocation_id = summary.invocation_id
bco = self.workflow_populator.get_biocompute_object(invocation_id)
self.workflow_populator.validate_biocompute_object(bco)
self.assertEqual(bco["provenance_domain"]["name"], "Simple Workflow")
assert bco["provenance_domain"]["name"] == "Simple Workflow"
@skip_without_tool("__APPLY_RULES__")
def test_workflow_run_apply_rules(self):
@@ -2615,11 +2613,11 @@ test_data:
wait=True,
round_trip_format_conversion=True,
)
self.assertEqual(
"chr6\t108722976\t108723115\tCCDS5067.1_cds_0_0_chr6_108722977_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n",
self.dataset_populator.get_history_dataset_content(history_id),
assert (
self.dataset_populator.get_history_dataset_content(history_id)
== "chr6\t108722976\t108723115\tCCDS5067.1_cds_0_0_chr6_108722977_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n"
)
# self.assertEqual("chr16\t142908\t143003\tCCDS10397.1_cds_0_0_chr16_142909_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n", self.dataset_populator.get_history_dataset_content(history_id))
# assert self.dataset_populator.get_history_dataset_content(history_id) == "chr16\t142908\t143003\tCCDS10397.1_cds_0_0_chr16_142909_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n"
@skip_without_tool("cat_list")
@skip_without_tool("random_lines1")
@@ -2686,9 +2684,9 @@ outer_input:
wait=True,
round_trip_format_conversion=True,
)
self.assertEqual(
"chr6\t108722976\t108723115\tCCDS5067.1_cds_0_0_chr6_108722977_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n",
self.dataset_populator.get_history_dataset_content(history_id),
assert (
self.dataset_populator.get_history_dataset_content(history_id)
== "chr6\t108722976\t108723115\tCCDS5067.1_cds_0_0_chr6_108722977_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n"
)
@skip_without_tool("cat_list")
@@ -2747,9 +2745,9 @@ outer_input:
wait=True,
round_trip_format_conversion=True,
)
self.assertEqual(
"chr6\t108722976\t108723115\tCCDS5067.1_cds_0_0_chr6_108722977_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n",
self.dataset_populator.get_history_dataset_content(history_id),
assert (
self.dataset_populator.get_history_dataset_content(history_id)
== "chr6\t108722976\t108723115\tCCDS5067.1_cds_0_0_chr6_108722977_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n"
)
@skip_without_tool("empty_list")
@@ -2792,7 +2790,7 @@ input1:
history_id=history_id,
wait=True,
)
self.assertEqual("0\n", self.dataset_populator.get_history_dataset_content(history_id))
assert "0\n" == self.dataset_populator.get_history_dataset_content(history_id)
@skip_without_tool("random_lines1")
def test_change_datatype_collection_map_over(self):
@@ -2889,7 +2887,7 @@ input1:
wait=True,
round_trip_format_conversion=True,
)
self.assertEqual("0\n", self.dataset_populator.get_history_dataset_content(history_id))
assert "0\n" == self.dataset_populator.get_history_dataset_content(history_id)
@skip_without_tool("cat")
def test_cancel_new_workflow_when_history_deleted(self):
@@ -3008,9 +3006,7 @@ input1:
self.workflow_populator.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)
)
assert "reviewed\n1\nreviewed\n4\n" == self.dataset_populator.get_history_dataset_content(history_id)
@skip_without_tool("cat")
def test_cancel_workflow_invocation(self):
@@ -3406,7 +3402,7 @@ text_input:
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual("chrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n", content)
assert "chrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n" == content
def test_run_with_numeric_input_connection(self):
history_id = self.dataset_populator.new_history()
@@ -3612,10 +3608,8 @@ outer_input:
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,
), 'Dataset ids should match, but "{}" and "{}" are not the same for History item {}.'.format(
item_one["dataset_id"], item_two["dataset_id"], i + 1
)
def test_cannot_run_inaccessible_workflow(self):
@@ -3668,9 +3662,7 @@ outer_input:
self.workflow_populator.invoke_workflow_and_wait(
workflow_id, history_id=history_id, request=workflow_request
)
self.assertEqual(
"1 2 3\n4 5 6\n7 8 9\n0 a b\n", self.dataset_populator.get_history_dataset_content(history_id)
)
assert "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:
@@ -4858,10 +4850,10 @@ steps:
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)
assert r1 == t1
assert r2 == t2
assert r3 == t3
assert r4 == t4
@skip_without_tool("cat1")
@skip_without_tool("addValue")
@@ -4910,10 +4902,10 @@ steps:
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)
assert r1 == t1
assert r2 == t2
assert r3 == t3
assert r4 == t4
@skip_without_tool("validation_default")
def test_parameter_substitution_sanitization(self):
@@ -4921,9 +4913,7 @@ steps:
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)
)
assert "__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):
@@ -5014,7 +5004,7 @@ steps:
)
workflow_request["parameters"] = params
self.workflow_populator.invoke_workflow_and_wait(workflow_id, request=workflow_request)
self.assertEqual("2\n", self.dataset_populator.get_history_dataset_content(history_id))
assert "2\n" == self.dataset_populator.get_history_dataset_content(history_id)
@skip_without_tool("random_lines1")
def test_run_replace_params_nested_normalized(self):
@@ -5030,7 +5020,7 @@ steps:
workflow_request["parameters"] = params
workflow_request["parameters_normalized"] = False
self.workflow_populator.invoke_workflow_and_wait(workflow_id, request=workflow_request)
self.assertEqual("2\n", self.dataset_populator.get_history_dataset_content(history_id))
assert "2\n" == self.dataset_populator.get_history_dataset_content(history_id)
@skip_without_tool("random_lines1")
def test_run_replace_params_over_default(self):
@@ -5512,7 +5502,7 @@ input_c:
hda_summary = next(hc for hc in history_contents if hc["hid"] == hid)
hda_info_response = self._get(f"{contents_url}/{hda_summary['id']}")
self._assert_status_code_is(hda_info_response, 200)
self.assertEqual(hda_info_response.json()["metadata_data_lines"], lines)
assert hda_info_response.json()["metadata_data_lines"] == lines
def __history_contents(self, history_id):
contents_url = f"histories/{history_id}/contents"
@@ -5606,7 +5596,4 @@ test_data:
inputs=json.dumps({"input1": self._ds_entry(hda1)}),
)
self.workflow_populator.invoke_workflow_and_wait(workflow_id, history_id=history_id, request=workflow_request)
self.assertEqual(
"Hello World Second!\nhello world 2\n",
self.dataset_populator.get_history_dataset_content(history_id),
)
assert self.dataset_populator.get_history_dataset_content(history_id) == "Hello World Second!\nhello world 2\n"
lib/galaxy_test/api/test_workflows_from_yaml.py
+6 -6
View File
@@ -78,13 +78,13 @@ input1: "hello world"
round_trip_format_conversion=True,
)
contents1 = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(contents1.strip(), "hello world\nhello world")
assert contents1.strip() == "hello world\nhello world"
def test_outputs(self):
workflow_id = self._upload_yaml_workflow(WORKFLOW_WITH_OUTPUTS, round_trip_format_conversion=True)
workflow = self._get(f"workflows/{workflow_id}/download").json()
self.assertEqual(workflow["steps"]["1"]["workflow_outputs"][0]["output_name"], "out_file1")
self.assertEqual(workflow["steps"]["1"]["workflow_outputs"][0]["label"], "wf_output_1")
assert workflow["steps"]["1"]["workflow_outputs"][0]["output_name"] == "out_file1"
assert workflow["steps"]["1"]["workflow_outputs"][0]["label"] == "wf_output_1"
workflow = self.workflow_populator.download_workflow(workflow_id, style="format2")
def test_runtime_inputs(self):
@@ -160,7 +160,7 @@ steps:
assert subworkflow_connection["input_subworkflow_step_id"] == 0
# 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)
# assert content == "chr5\t131424298\t131424460\tCCDS4149.1_cds_0_0_chr5_131424299_f\t0\t+\n"
def test_subworkflow_duplicate(self):
duplicate_subworkflow_invocate_wf = """
@@ -339,7 +339,7 @@ test_data:
)
content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(content, "hello world\nhello world 2\n")
assert content == "hello world\nhello world 2\n"
def test_workflow_import_tool(self):
history_id = self.dataset_populator.new_history()
@@ -347,7 +347,7 @@ test_data:
jobs_descriptions = {"test_data": {"input1": "hello world"}}
self._run_jobs(workflow_path, source_type="path", jobs_descriptions=jobs_descriptions, history_id=history_id)
content = self.dataset_populator.get_history_dataset_content(history_id)
self.assertEqual(content, "hello world\nhello world 2\n")
assert content == "hello world\nhello world 2\n"
def test_parameter_default_rep(self):
workflow = self._upload_and_download(WORKFLOW_PARAMETER_INPUT_INTEGER_DEFAULT)
lib/galaxy_test/selenium/test_admin_app.py
+2 -2
View File
@@ -138,13 +138,13 @@ class AdminAppTestCase(SeleniumTestCase):
self.sleep_for(self.wait_types.UX_TRANSITION)
# Make sure the job lock has been toggled.
new_label = lock_label.wait_for_text()
self.assertNotEqual(new_label, original_label)
assert new_label != original_label
self.screenshot("admin_jobs_locked")
lock_label.wait_for_and_click()
self.sleep_for(self.wait_types.UX_TRANSITION)
self.screenshot("admin_jobs_unlocked")
# And confirm that it has toggled back to what it was.
self.assertEqual(lock_label.wait_for_text(), original_label)
assert lock_label.wait_for_text() == original_label
@selenium_test
def test_admin_server_display(self):
lib/galaxy_test/selenium/test_custom_builds.py
+2 -2
View File
@@ -27,9 +27,9 @@ class CustomBuildsTestcase(SharedStateSeleniumTestCase):
actual_builds = self.get_custom_builds()
intersection = set(actual_builds).intersection(expected_builds)
if present:
self.assertEqual(intersection, set(expected_builds))
assert intersection == set(expected_builds)
else:
self.assertEqual(intersection, set())
assert intersection == set()
def _login(self):
self.home() # ensure Galaxy is loaded
lib/galaxy_test/selenium/test_histories_list.py
+6 -6
View File
@@ -22,7 +22,7 @@ class SavedHistoriesTestCase(SharedStateSeleniumTestCase):
@retry_assertion_during_transitions
def assert_history_name_switched():
self.assertEqual(self.history_panel_name(), self.history2_name)
assert self.history_panel_name() == self.history2_name
assert_history_name_switched()
@@ -32,7 +32,7 @@ class SavedHistoriesTestCase(SharedStateSeleniumTestCase):
self.navigate_to_histories_page()
self.click_grid_popup_option(self.history2_name, "View")
history_name = self.wait_for_selector("[data-description='name display']")
self.assertEqual(history_name.text, self.history2_name)
assert history_name.text == self.history2_name
@selenium_test
def test_history_publish(self):
@@ -155,7 +155,7 @@ class SavedHistoriesTestCase(SharedStateSeleniumTestCase):
# Filter out histories created by other tests
actual_histories = [x for x in actual_histories if x in self.all_histories]
self.assertEqual(actual_histories, expected_histories)
assert actual_histories == expected_histories
@selenium_test
def test_standard_search(self):
@@ -238,16 +238,16 @@ class SavedHistoriesTestCase(SharedStateSeleniumTestCase):
if not sort_matters:
actual_histories = set(actual_histories)
expected_histories = set(expected_histories)
self.assertEqual(actual_histories, expected_histories)
assert actual_histories == expected_histories
@retry_assertion_during_transitions
def assert_histories_in_grid(self, expected_histories, present=True):
actual_histories = self.get_histories()
intersection = set(actual_histories).intersection(expected_histories)
if present:
self.assertEqual(intersection, set(expected_histories))
assert intersection == set(expected_histories)
else:
self.assertEqual(intersection, set())
assert intersection == set()
def get_histories(self):
return self.histories_get_history_names()
lib/galaxy_test/selenium/test_history_panel.py
+1 -1
View File
@@ -244,7 +244,7 @@ class HistoryPanelTestCase(SeleniumTestCase):
@retry_assertion_during_transitions
def assert_name_changed(self):
name = self.history_panel_name()
self.assertEqual(name, NEW_HISTORY_NAME)
assert name == NEW_HISTORY_NAME
def _refresh(self):
if self.is_beta_history():
lib/galaxy_test/selenium/test_invocation_grid.py
+1 -1
View File
@@ -47,4 +47,4 @@ class InvocationGridSeleniumTestCase(SeleniumTestCase, TestsGalaxyPagers):
@retry_assertion_during_transitions
def _assert_showing_n_invocations(self, n):
self.assertEqual(len(self.invocation_index_table_elements()), n)
assert len(self.invocation_index_table_elements()) == n
lib/galaxy_test/selenium/test_library_landing.py
+2 -2
View File
@@ -87,7 +87,7 @@ class LibraryLandingTestCase(SeleniumTestCase):
@retry_assertion_during_transitions
def _assert_names_are(self, expected_names):
names = [e.find_element(self.by.CSS_SELECTOR, "td a").text for e in self.libraries_index_table_elements()]
self.assertEqual(names, expected_names)
assert names == expected_names
@retry_assertion_during_transitions
def _assert_at_least_one_library_displayed(self):
@@ -95,7 +95,7 @@ class LibraryLandingTestCase(SeleniumTestCase):
@retry_assertion_during_transitions
def _assert_num_displayed_libraries_is(self, n):
self.assertEqual(n, self._num_displayed_libraries())
assert n == self._num_displayed_libraries()
def _num_displayed_libraries(self):
return len(self.libraries_index_table_elements())
lib/galaxy_test/selenium/test_personal_information.py
+7 -7
View File
@@ -13,25 +13,25 @@ class ManageInformationTestCase(SeleniumTestCase):
'Not available.'
"""
self.login()
self.assertEqual(self.get_api_key(), "Not available.")
assert self.get_api_key() == "Not available."
api_key = self.get_api_key()
self.navigate_to_user_preferences()
self.components.preferences.manage_api_key.wait_for_and_click()
self.sleep_for(self.wait_types.UX_TRANSITION)
api_key_input = self.components.preferences.api_key_input.wait_for_visible()
# Assert that what's rendered on screen is what the API is returning
self.assertEqual(api_key_input.get_property("value"), api_key)
assert api_key_input.get_property("value") == api_key
self.components.preferences.get_new_key.wait_for_and_click()
self.sleep_for(self.wait_types.UX_TRANSITION)
new_api_key = self.get_api_key()
api_key_input = self.components.preferences.api_key_input.wait_for_visible()
# And assert that this has now changed, and still renders correctly
self.assertEqual(new_api_key, api_key_input.get_property("value"))
assert new_api_key == api_key_input.get_property("value")
@selenium_test
def test_change_email(self):
def assert_email(email_to_check):
self.assertTrue(email_to_check == self.components.preferences.current_email.wait_for_text())
assert email_to_check == self.components.preferences.current_email.wait_for_text()
email = self._get_random_email()
self.register(email)
@@ -46,7 +46,7 @@ class ManageInformationTestCase(SeleniumTestCase):
new_email = self._get_random_email()
# new email should be different from initially registered
self.assertTrue(email != new_email)
assert email != new_email
email_input_field = self.components.preferences.email_input.wait_for_visible()
self.clear_input_field_and_write(email_input_field, new_email)
@@ -64,7 +64,7 @@ class ManageInformationTestCase(SeleniumTestCase):
return self.components.preferences.username_input.wait_for_visible()
def assert_public_name(expected_name):
self.assertTrue(expected_name == get_name_input_field().get_attribute("value"))
assert expected_name == get_name_input_field().get_attribute("value")
public_name = "user-public-name"
self.register(username=public_name)
@@ -123,7 +123,7 @@ class ManageInformationTestCase(SeleniumTestCase):
# check if address was saved correctly
for input_field_label in address_fields.keys():
input_field = self.get_address_input_field(get_address_form(), input_field_label)
self.assertTrue(input_field.get_attribute("value") == address_fields[input_field_label])
assert input_field.get_attribute("value") == address_fields[input_field_label]
def navigate_to_manage_information(self):
self.navigate_to_user_preferences()
lib/galaxy_test/selenium/test_published_histories_grid.py
+4 -4
View File
@@ -105,7 +105,7 @@ class HistoryGridTestCase(SharedStateSeleniumTestCase):
if tag_button_text == target_tag_button_text:
break
self.assertEqual(tag_button_text, target_tag_button_text)
assert tag_button_text == target_tag_button_text
tag_button.click()
self.assert_grid_histories_are([self.history1_name, self.history3_name], False)
@@ -139,16 +139,16 @@ class HistoryGridTestCase(SharedStateSeleniumTestCase):
if not sort_matters:
actual_histories = set(actual_histories)
expected_histories = set(expected_histories)
self.assertEqual(actual_histories, expected_histories)
assert actual_histories == expected_histories
@retry_assertion_during_transitions
def assert_histories_in_grid(self, expected_histories, present=True):
actual_histories = self.get_histories()
intersection = set(actual_histories).intersection(expected_histories)
if present:
self.assertEqual(intersection, set(expected_histories))
assert intersection == set(expected_histories)
else:
self.assertEqual(intersection, set())
assert intersection == set()
def set_filter(self, selector, value):
filter_input = self.wait_for_selector_clickable(selector)
lib/galaxy_test/selenium/test_sign_out.py
+1 -1
View File
@@ -17,7 +17,7 @@ class SignOutTestCase(SeleniumTestCase):
self.components.sign_out.cancel_button.wait_for_and_click()
assert self.is_logged_in()
new_email = self.driver.find_element(By.ID, "user-preferences-current-email").text
self.assertTrue(email == new_email)
assert email == new_email
self.components.preferences.sign_out.wait_for_and_click()
self.components.sign_out.sign_out_button.wait_for_and_click()
self.sleep_for(self.wait_types.UX_TRANSITION)
lib/galaxy_test/selenium/test_tool_describing_tours.py
+1 -1
View File
@@ -70,7 +70,7 @@ class ToolDescribingToursTestCase(SeleniumTestCase):
def _ensure_tdt_available(self):
"""Skip a test if the webhook TDT doesn't appear."""
response = self.api_get("webhooks", raw=True)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
data = response.json()
webhooks = [x["id"] for x in data]
if "tour_generator" not in webhooks:
lib/galaxy_test/selenium/test_workflow_editor.py
+1 -1
View File
@@ -724,7 +724,7 @@ steps:
def assert_workflow_bookmarked_status(target_status):
name_matches = [c.text == new_workflow_name for c in self.components.tool_panel.workflow_names.all()]
status = any(name_matches)
self.assertTrue(status == target_status)
assert status == target_status
new_workflow_name = self.workflow_create_new(clear_placeholder=True)
lib/galaxy_test/selenium/test_workflow_management.py
+1 -1
View File
@@ -75,7 +75,7 @@ class WorkflowManagementTestCase(SeleniumTestCase, TestsGalaxyPagers, UsesWorkfl
@retry_assertion_during_transitions
def check_tags():
self.assertEqual(self.workflow_index_tags(), ["cooltag"])
assert self.workflow_index_tags() == ["cooltag"]
check_tags()
self.screenshot("workflow_manage_tags")
lib/galaxy_test/selenium/test_workflow_run.py
+1 -1
View File
@@ -189,7 +189,7 @@ steps:
)
self.workflow_run_wait_for_ok(hid=7)
content = self.dataset_populator.get_history_dataset_content(history_id, hid=7)
self.assertEqual("10.0\n30.0\n20.0\n40.0\n", content)
assert "10.0\n30.0\n20.0\n40.0\n" == content
@selenium_test
@managed_history