opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-01 15:37:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

use history_id fixture instead of instance variable with the same name

Browse Source
This commit is contained in:
Michael R. Crusoe
2022-11-20 15:15:51 +01:00
committed by Nicola Soranzo
parent dd656aa3b3
commit 95ea3a2d95
9 changed files with 165 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/integration/objectstore/test_objectstore_datatype_upload.py
+12 -4
View File
@@ -196,23 +196,31 @@ idle_connection_irods_instance = integration_util.integration_module_instance(Ir
@pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys()))
def test_upload_datatype_dos_disk_and_disk(distributed_instance, test_data, temp_file):
upload_datatype_helper(distributed_instance, test_data, temp_file)
assert distributed_instance.dataset_populator
with distributed_instance.dataset_populator.test_history() as history_id:
upload_datatype_helper(distributed_instance, test_data, temp_file, history_id)
@pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys()))
def test_upload_datatype_irods(irods_instance, test_data, temp_file):
upload_datatype_helper(irods_instance, test_data, temp_file, True)
assert irods_instance.dataset_populator
with irods_instance.dataset_populator.test_history() as history_id:
upload_datatype_helper(irods_instance, test_data, temp_file, True, history_id)
@pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys()))
def test_upload_datatype_dos_irods_and_disk(distributed_and_irods_instance, test_data, temp_file):
upload_datatype_helper(distributed_and_irods_instance, test_data, temp_file)
assert distributed_and_irods_instance.dataset_populator
with distributed_and_irods_instance.dataset_populator.test_history() as history_id:
upload_datatype_helper(distributed_and_irods_instance, test_data, temp_file, history_id)
@pytest.mark.parametrize("test_data", SINGLE_TEST_CASE.values(), ids=list(SINGLE_TEST_CASE.keys()))
def test_upload_datatype_irods_idle_connections(idle_connection_irods_instance, test_data, temp_file):
# Upload a file to iRods
upload_datatype_helper(idle_connection_irods_instance, test_data, temp_file, True)
assert idle_connection_irods_instance.dataset_populator
with idle_connection_irods_instance.dataset_populator.test_history() as history_id:
upload_datatype_helper(idle_connection_irods_instance, test_data, temp_file, True, history_id)
# Get Irods object store's connection pool
connection_pool = idle_connection_irods_instance._test_driver.app.object_store.session.pool
test/integration/test_datatype_upload.py
+7 -7
View File
@@ -58,10 +58,12 @@ TEST_CASES = collect_test_data(registry)
@pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys()))
def test_upload_datatype_auto(instance, test_data, temp_file, celery_session_worker, celery_session_app):
upload_datatype_helper(instance, test_data, temp_file)
assert instance.dataset_populator
with instance.dataset_populator.test_history() as history_id:
upload_datatype_helper(instance, test_data, temp_file, history_id)
def upload_datatype_helper(instance, test_data, temp_file, delete_cache_dir=False):
def upload_datatype_helper(instance, test_data, temp_file, history_id: str, delete_cache_dir=False):
is_compressed = False
for is_method in (is_bz2, is_gzip, is_zip):
is_compressed = is_method(test_data.path)
@@ -73,15 +75,13 @@ def upload_datatype_helper(instance, test_data, temp_file, delete_cache_dir=Fals
else:
file_type = test_data.datatype.file_ext
dataset = instance.dataset_populator.new_dataset(
instance.history_id,
history_id,
content=content,
wait=False,
file_type=file_type,
auto_decompress=True,
)
dataset = instance.dataset_populator.get_history_dataset_details(
instance.history_id, dataset=dataset, assert_ok=False
)
dataset = instance.dataset_populator.get_history_dataset_details(history_id, dataset=dataset, assert_ok=False)
expected_file_ext = test_data.datatype.file_ext
# State might be error if the datatype can't be uploaded
if dataset["state"] == "error" and not test_data.uploadable:
@@ -114,7 +114,7 @@ def upload_datatype_helper(instance, test_data, temp_file, delete_cache_dir=Fals
# download file and verify it hasn't been manipulated
temp_file.write(
instance.dataset_populator.get_history_dataset_content(
history_id=instance.history_id, dataset=dataset, type="bytes", assert_ok=False, raw=True
history_id=history_id, dataset=dataset, type="bytes", assert_ok=False, raw=True
)
)
temp_file.flush()
test/integration/test_fail_job_tool_unavailable.py
+5 -6
View File
@@ -15,7 +15,6 @@ class TestFailJobWhenToolUnavailable(integration_util.IntegrationTestCase):
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
self.history_id = self.dataset_populator.new_history()
@classmethod
def handle_galaxy_config_kwds(
@@ -27,7 +26,7 @@ class TestFailJobWhenToolUnavailable(integration_util.IntegrationTestCase):
# Disable tool dependency resolution.
config["tool_dependency_dir"] = "none"
def test_fail_job_when_tool_unavailable(self):
def test_fail_job_when_tool_unavailable(self, history_id: str):
self.workflow_populator.run_workflow(
"""
class: GalaxyWorkflow
@@ -48,20 +47,20 @@ steps:
input2:
$link: sleep/output1
""",
history_id=self.history_id,
history_id=history_id,
assert_ok=False,
wait=False,
)
# Wait until workflow is fully scheduled, otherwise can't test effect of removing tool from queued job
time.sleep(5)
self._app.toolbox.remove_tool_by_id("cat1")
self.dataset_populator.wait_for_history(self.history_id, assert_ok=False)
state_details = self.galaxy_interactor.get(f"histories/{self.history_id}").json()["state_details"]
self.dataset_populator.wait_for_history(history_id, assert_ok=False)
state_details = self.galaxy_interactor.get(f"histories/{history_id}").json()["state_details"]
assert state_details["running"] == 0
assert state_details["ok"] == 1
assert state_details["error"] == 1
failed_hda = self.dataset_populator.get_history_dataset_details(
history_id=self.history_id, assert_ok=False, details=True
history_id=history_id, assert_ok=False, details=True
)
assert failed_hda["state"] == "error"
job = self.galaxy_interactor.get("jobs/{}".format(failed_hda["creating_job"])).json()
test/integration/test_history_import_export.py
+17 -18
View File
@@ -102,14 +102,13 @@ class TestImportExportHistoryContentsViaTasksIntegration(IntegrationTestCase, Us
def setUp(self):
super().setUp()
self._set_up_populators()
self.history_id = self.dataset_populator.new_history()
def _set_up_populators(self):
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor)
def test_export_and_imported_discarded(self):
hda1 = self.dataset_populator.new_dataset(self.history_id, wait=True)
def test_export_and_imported_discarded(self, history_id: str):
hda1 = self.dataset_populator.new_dataset(history_id, wait=True)
second_history_id, as_list = self.dataset_populator.reupload_contents(hda1)
@@ -119,9 +118,9 @@ class TestImportExportHistoryContentsViaTasksIntegration(IntegrationTestCase, Us
assert new_hda["state"] == "discarded"
assert not new_hda["deleted"]
def test_export_and_imported_discarded_bam(self):
def test_export_and_imported_discarded_bam(self, history_id: str):
contents = self.dataset_populator.new_dataset(
self.history_id,
history_id,
content=open(self.test_data_resolver.get_filename("1.bam"), "rb"),
file_type="bam",
wait=True,
@@ -133,9 +132,9 @@ class TestImportExportHistoryContentsViaTasksIntegration(IntegrationTestCase, Us
assert new_hda["state"] == "discarded"
assert not new_hda["deleted"]
def test_import_as_discarded_from_dict(self):
def test_import_as_discarded_from_dict(self, history_id: str):
as_list = self.dataset_populator.create_contents_from_store(
self.history_id,
history_id,
store_dict=one_hda_model_store_dict(
include_source=False,
),
@@ -146,12 +145,12 @@ class TestImportExportHistoryContentsViaTasksIntegration(IntegrationTestCase, Us
assert new_hda["state"] == "discarded"
assert not new_hda["deleted"]
contents_response = self._get(f"histories/{self.history_id}/contents?v=dev")
contents_response = self._get(f"histories/{history_id}/contents?v=dev")
contents_response.raise_for_status()
def test_import_as_deferred_from_discarded_with_source_dict(self):
def test_import_as_deferred_from_discarded_with_source_dict(self, history_id: str):
as_list = self.dataset_populator.create_contents_from_store(
self.history_id,
history_id,
store_dict=one_hda_model_store_dict(
include_source=True,
),
@@ -162,18 +161,18 @@ class TestImportExportHistoryContentsViaTasksIntegration(IntegrationTestCase, Us
assert new_hda["state"] == "deferred"
assert not new_hda["deleted"]
contents_response = self._get(f"histories/{self.history_id}/contents?v=dev")
contents_response = self._get(f"histories/{history_id}/contents?v=dev")
contents_response.raise_for_status()
def test_export_and_imported_discarded_collection(self):
def test_export_and_imported_discarded_collection(self, history_id: str):
create_response = self.dataset_collection_populator.create_list_in_history(
history_id=self.history_id,
history_id=history_id,
direct_upload=True,
wait=True,
).json()
self.dataset_populator.wait_for_history(self.history_id)
self.dataset_populator.wait_for_history(history_id)
contents = create_response["outputs"][0]
temp_tar = self.dataset_populator.download_contents_to_store(self.history_id, contents, "tgz")
temp_tar = self.dataset_populator.download_contents_to_store(history_id, contents, "tgz")
with tarfile.open(name=temp_tar) as tf:
assert "datasets_attrs.txt" in tf.getnames()
assert "collections_attrs.txt" in tf.getnames()
@@ -188,9 +187,9 @@ class TestImportExportHistoryContentsViaTasksIntegration(IntegrationTestCase, Us
hdcas = [e for e in as_list if e["history_content_type"] == "dataset_collection"]
assert len(hdcas) == 1
def test_import_as_deferred_from_dict(self):
def test_import_as_deferred_from_dict(self, history_id: str):
as_list = self.dataset_populator.create_contents_from_store(
self.history_id,
history_id,
store_dict=deferred_hda_model_store_dict(),
)
assert len(as_list) == 1
@@ -199,5 +198,5 @@ class TestImportExportHistoryContentsViaTasksIntegration(IntegrationTestCase, Us
assert new_hda["state"] == "deferred"
assert not new_hda["deleted"]
contents_response = self._get(f"histories/{self.history_id}/contents?v=dev")
contents_response = self._get(f"histories/{history_id}/contents?v=dev")
contents_response.raise_for_status()
test/integration/test_job_recovery.py
-1
View File
@@ -17,7 +17,6 @@ class TestJobRecoveryBeforeHandledIntegeration(integration_util.IntegrationTestC
def setUp(self):
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.history_id = self.dataset_populator.new_history()
@classmethod
def handle_galaxy_config_kwds(cls, config):
test/integration/test_page_revision_json_encoding.py
+6 -7
View File
@@ -17,29 +17,28 @@ class TestPageJsonEncodingIntegration(integration_util.IntegrationTestCase):
def setUp(self):
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.history_id = self.dataset_populator.new_history()
def test_page_encoding(self):
def test_page_encoding(self, history_id: str):
request = dict(
slug="mypage",
title="MY PAGE",
content=f"""<p>Page!<div class="embedded-item" id="History-{self.history_id}"></div></p>""",
content=f"""<p>Page!<div class="embedded-item" id="History-{history_id}"></div></p>""",
)
page_response = self._post("pages", request, json=True)
api_asserts.assert_status_code_is_ok(page_response)
sa_session = self._app.model.context
page_revision = sa_session.query(model.PageRevision).filter_by(content_format="html").all()[0]
assert '''id="History-1"''' in page_revision.content, page_revision.content
assert f'''id="History-{self.history_id}"''' not in page_revision.content, page_revision.content
assert f'''id="History-{history_id}"''' not in page_revision.content, page_revision.content
show_page_response = self._get("pages/{}".format(page_response.json()["id"]))
api_asserts.assert_status_code_is_ok(show_page_response)
content = show_page_response.json()["content"]
assert '''id="History-1"''' not in content, content
assert f'''id="History-{self.history_id}"''' in content, content
assert f'''id="History-{history_id}"''' in content, content
def test_page_encoding_markdown(self):
dataset = self.dataset_populator.new_dataset(self.history_id)
def test_page_encoding_markdown(self, history_id: str):
dataset = self.dataset_populator.new_dataset(history_id)
dataset_id = dataset["id"]
request = dict(
slug="mypage-markdown",
test/integration/test_upload_configuration_options.py
+96 -97
View File
@@ -54,11 +54,10 @@ class BaseUploadContentConfigurationInstance(integration_util.IntegrationInstanc
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.library_populator = LibraryPopulator(self.galaxy_interactor)
self.history_id = self.dataset_populator.new_history()
def fetch_target(self, target, assert_ok=False, attach_test_file=False, wait=False):
def fetch_target(self, target, history_id: str, assert_ok=False, attach_test_file=False, wait=False):
payload: Dict[str, Any] = {
"history_id": self.history_id,
"history_id": history_id,
"targets": [target],
}
if attach_test_file:
@@ -85,25 +84,25 @@ class BaseUploadContentConfigurationTestCase(BaseUploadContentConfigurationInsta
class TestInvalidFetchRequests(BaseUploadContentConfigurationTestCase):
def test_in_place_not_allowed(self):
def test_in_place_not_allowed(self, history_id):
elements = [{"src": "files", "in_place": False}]
target = {
"destination": {"type": "hdca"},
"elements": elements,
"collection_type": "list",
}
response = self.fetch_target(target, attach_test_file=True)
response = self.fetch_target(target, history_id=history_id, attach_test_file=True)
self._assert_status_code_is(response, 400)
assert "in_place" in response.json()["err_msg"]
def test_files_not_attached(self):
def test_files_not_attached(self, history_id: str):
elements = [{"src": "files"}]
target = {
"destination": {"type": "hdca"},
"elements": elements,
"collection_type": "list",
}
response = self.fetch_target(target)
response = self.fetch_target(target, history_id=history_id)
self._assert_status_code_is(response, 400)
assert "Failed to find uploaded file matching target" in response.json()["err_msg"]
@@ -114,9 +113,9 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase):
super().handle_galaxy_config_kwds(config)
config["allow_path_paste"] = True
def test_disallowed_for_primary_file(self):
def test_disallowed_for_primary_file(self, history_id):
payload = self.dataset_populator.upload_payload(
self.history_id, f"file://{TEST_DATA_DIRECTORY}/1.RData", file_type="binary"
history_id, f"file://{TEST_DATA_DIRECTORY}/1.RData", file_type="binary"
)
create_response = self.dataset_populator.tools_post(payload)
# Ideally this would be 403 but the tool API endpoint isn't using
@@ -124,11 +123,11 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase):
assert create_response.status_code >= 400
@skip_without_datatype("velvet")
def test_disallowed_for_composite_file(self):
def test_disallowed_for_composite_file(self, history_id):
path = os.path.join(TEST_DATA_DIRECTORY, "1.txt")
assert os.path.exists(path)
payload = self.dataset_populator.upload_payload(
self.history_id,
history_id,
"sequences content",
file_type="velvet",
extra_inputs={
@@ -155,7 +154,7 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase):
assert response.status_code == 403, response.json()
assert os.path.exists(path)
def test_disallowed_for_fetch(self):
def test_disallowed_for_fetch(self, history_id: str):
path = os.path.join(TEST_DATA_DIRECTORY, "1.txt")
assert os.path.exists(path)
elements = [{"src": "path", "path": path}]
@@ -164,11 +163,11 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase):
"elements": elements,
"collection_type": "list",
}
response = self.fetch_target(target)
response = self.fetch_target(target, history_id=history_id)
self._assert_status_code_is(response, 403)
assert os.path.exists(path)
def test_disallowed_for_fetch_urls(self):
def test_disallowed_for_fetch_urls(self, history_id: str):
path = os.path.join(TEST_DATA_DIRECTORY, "1.txt")
assert os.path.exists(path)
elements = [{"src": "url", "url": f"file://{path}"}]
@@ -177,7 +176,7 @@ class TestNonAdminsCannotPasteFilePath(BaseUploadContentConfigurationTestCase):
"elements": elements,
"collection_type": "list",
}
response = self.fetch_target(target)
response = self.fetch_target(target, history_id=history_id)
self._assert_status_code_is(response, 403)
assert os.path.exists(path)
@@ -191,9 +190,9 @@ class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase):
super().handle_galaxy_config_kwds(config)
config["allow_path_paste"] = True
def test_admin_path_paste(self):
def test_admin_path_paste(self, history_id):
payload = self.dataset_populator.upload_payload(
self.history_id,
history_id,
f"file://{TEST_DATA_DIRECTORY}/random-file",
)
create_response = self.dataset_populator.tools_post(payload)
@@ -227,7 +226,7 @@ class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase):
# We should probably verify the linking, but this was enough for now to exhibit
# https://github.com/galaxyproject/galaxy/issues/8756
def test_admin_fetch(self):
def test_admin_fetch(self, history_id: str):
path = os.path.join(TEST_DATA_DIRECTORY, "1.txt")
elements = [{"src": "path", "path": path}]
target = {
@@ -235,11 +234,11 @@ class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase):
"elements": elements,
"collection_type": "list",
}
response = self.fetch_target(target)
response = self.fetch_target(target, history_id=history_id)
self._assert_status_code_is(response, 200)
assert os.path.exists(path)
def test_admin_fetch_file_url(self):
def test_admin_fetch_file_url(self, history_id: str):
path = os.path.join(TEST_DATA_DIRECTORY, "1.txt")
elements = [{"src": "url", "url": f"file://{path}"}]
target = {
@@ -247,7 +246,7 @@ class TestAdminsCanPasteFilePaths(BaseUploadContentConfigurationTestCase):
"elements": elements,
"collection_type": "list",
}
response = self.fetch_target(target)
response = self.fetch_target(target, history_id=history_id)
self._assert_status_code_is(response, 200)
assert os.path.exists(path)
@@ -261,18 +260,18 @@ class TestDefaultBinaryContentFilters(BaseUploadContentConfigurationTestCase):
super().handle_galaxy_config_kwds(config)
config["allow_path_paste"] = True
def test_random_binary_allowed(self):
def test_random_binary_allowed(self, history_id: str):
dataset = self.dataset_populator.new_dataset(
self.history_id, f"file://{TEST_DATA_DIRECTORY}/random-file", file_type="auto", wait=True
history_id, f"file://{TEST_DATA_DIRECTORY}/random-file", file_type="auto", wait=True
)
dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=dataset)
dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=dataset)
assert dataset["file_ext"] == "binary", dataset
def test_gzipped_html_content_blocked_by_default(self):
def test_gzipped_html_content_blocked_by_default(self, history_id: str):
dataset = self.dataset_populator.new_dataset(
self.history_id, f"file://{TEST_DATA_DIRECTORY}/bad.html.gz", file_type="auto", wait=True, assert_ok=False
history_id, f"file://{TEST_DATA_DIRECTORY}/bad.html.gz", file_type="auto", wait=True, assert_ok=False
)
dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=dataset, assert_ok=False)
dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=dataset, assert_ok=False)
assert dataset["file_size"] == 0
@@ -286,11 +285,11 @@ class TestDisableContentChecking(BaseUploadContentConfigurationTestCase):
config["allow_path_paste"] = True
config["check_upload_content"] = False
def test_gzipped_html_content_now_allowed(self):
def test_gzipped_html_content_now_allowed(self, history_id: str):
dataset = self.dataset_populator.new_dataset(
self.history_id, f"file://{TEST_DATA_DIRECTORY}/bad.html.gz", file_type="auto", wait=True
history_id, f"file://{TEST_DATA_DIRECTORY}/bad.html.gz", file_type="auto", wait=True
)
dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=dataset)
dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=dataset)
# Same file was empty above!
assert dataset["file_size"] != 0
@@ -304,37 +303,37 @@ class TestAutoDecompress(BaseUploadContentConfigurationTestCase):
super().handle_galaxy_config_kwds(config)
config["allow_path_paste"] = True
def test_auto_decompress_off(self):
def test_auto_decompress_off(self, history_id: str):
dataset = self.dataset_populator.new_dataset(
self.history_id,
history_id,
f"file://{TEST_DATA_DIRECTORY}/1.sam.gz",
file_type="auto",
auto_decompress=False,
wait=True,
)
dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=dataset)
dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=dataset)
assert dataset["file_ext"] == "binary", dataset
def test_auto_decompress_on(self):
def test_auto_decompress_on(self, history_id: str):
dataset = self.dataset_populator.new_dataset(
self.history_id, f"file://{TEST_DATA_DIRECTORY}/1.sam.gz", file_type="auto", wait=True
history_id, f"file://{TEST_DATA_DIRECTORY}/1.sam.gz", file_type="auto", wait=True
)
dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=dataset)
dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=dataset)
assert dataset["file_ext"] == "sam", dataset
class TestLocalAddressWhitelisting(BaseUploadContentConfigurationTestCase):
def test_blocked_url_for_primary_file(self):
payload = self.dataset_populator.upload_payload(self.history_id, "http://localhost/", file_type="txt")
def test_blocked_url_for_primary_file(self, history_id: str):
payload = self.dataset_populator.upload_payload(history_id, "http://localhost/", file_type="txt")
create_response = self.dataset_populator.tools_post(payload)
# Ideally this would be 403 but the tool API endpoint isn't using
# the newer API decorator that handles those details.
assert create_response.status_code >= 400
@skip_without_datatype("velvet")
def test_blocked_url_for_composite_file(self):
def test_blocked_url_for_composite_file(self, history_id: str):
payload = self.dataset_populator.upload_payload(
self.history_id,
history_id,
"sequences content",
file_type="velvet",
extra_inputs={
@@ -349,14 +348,14 @@ class TestLocalAddressWhitelisting(BaseUploadContentConfigurationTestCase):
# the newer API decorator that handles those details.
assert create_response.status_code >= 400
def test_blocked_url_for_fetch(self):
def test_blocked_url_for_fetch(self, history_id: str):
elements = [{"src": "url", "url": "http://localhost"}]
target = {
"destination": {"type": "hdca"},
"elements": elements,
"collection_type": "list",
}
response = self.fetch_target(target)
response = self.fetch_target(target, history_id=history_id)
self._assert_status_code_is(response, 403)
@@ -377,10 +376,10 @@ class BaseFtpUploadConfigurationTestCase(BaseUploadContentConfigurationTestCase)
def ftp_dir(cls):
return cls.temp_config_dir("ftp")
def _check_content(self, dataset, content, ext="txt"):
dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=dataset)
def _check_content(self, dataset, content, history_id: str, ext="txt"):
dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=dataset)
assert dataset["file_ext"] == ext, dataset
content = self.dataset_populator.get_history_dataset_content(self.history_id, dataset=dataset)
content = self.dataset_populator.get_history_dataset_content(history_id, dataset=dataset)
assert content == content, content
def _get_user_ftp_path(self):
@@ -398,7 +397,7 @@ class BaseFtpUploadConfigurationTestCase(BaseUploadContentConfigurationTestCase)
if not os.path.exists(path):
os.makedirs(path)
def _run_purgable_upload(self):
def _run_purgable_upload(self, history_id: str):
# Purge setting is actually used with a fairly specific set of parameters - see:
# https://github.com/galaxyproject/galaxy/issues/5361
content = "hello world\n"
@@ -409,14 +408,14 @@ class BaseFtpUploadConfigurationTestCase(BaseUploadContentConfigurationTestCase)
assert os.path.exists(ftp_path)
# gotta set to_posix_lines to None currently to force purging of non-binary data.
dataset = self.dataset_populator.new_dataset(
self.history_id, ftp_files="test", file_type="txt", to_posix_lines=None, wait=True
history_id, ftp_files="test", file_type="txt", to_posix_lines=None, wait=True
)
self._check_content(dataset, content)
self._check_content(dataset, content, history_id=history_id)
return ftp_path
class TestSimpleFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase):
def test_ftp_upload(self):
def test_ftp_upload(self, history_id: str):
content = "hello world\n"
ftp_path = self._write_ftp_file(content)
ftp_files = self.dataset_populator.get_remote_files()
@@ -426,11 +425,11 @@ class TestSimpleFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase):
# set to_posix_lines to False to exercise purging - by default this file type wouldn't
# be purged.
dataset = self.dataset_populator.new_dataset(
self.history_id, ftp_files="test", file_type="txt", to_posix_lines=False, wait=True
history_id, ftp_files="test", file_type="txt", to_posix_lines=False, wait=True
)
self._check_content(dataset, content)
self._check_content(dataset, content, history_id=history_id)
def test_ftp_fetch(self):
def test_ftp_fetch(self, history_id: str):
content = "hello world\n"
ftp_path = self._write_ftp_file(content)
ftp_files = self.dataset_populator.get_remote_files()
@@ -444,13 +443,13 @@ class TestSimpleFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase):
"collection_type": "list",
"name": "cool collection",
}
response = self.fetch_target(target, assert_ok=True, wait=True)
response = self.fetch_target(target, history_id=history_id, assert_ok=True, wait=True)
response_object = response.json()
assert "output_collections" in response_object
output_collections = response_object["output_collections"]
assert len(output_collections) == 1, response_object
dataset = self.dataset_populator.get_history_dataset_details(self.history_id, hid=2)
self._check_content(dataset, content)
dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=2)
self._check_content(dataset, content, history_id=history_id)
class TestExplicitEmailAsIdentifierFtpUploadConfiguration(TestSimpleFtpUploadConfiguration):
@@ -483,8 +482,8 @@ class TestDisableFtpPurgeUploadConfiguration(BaseFtpUploadConfigurationTestCase)
def handle_extra_ftp_config(cls, config):
config["ftp_upload_purge"] = "False"
def test_ftp_uploads_not_purged(self):
ftp_path = self._run_purgable_upload()
def test_ftp_uploads_not_purged(self, history_id: str):
ftp_path = self._run_purgable_upload(history_id=history_id)
# Purge is disabled, this better still be here.
assert os.path.exists(ftp_path)
@@ -494,13 +493,13 @@ class TestEnableFtpPurgeUploadConfiguration(BaseFtpUploadConfigurationTestCase):
def handle_extra_ftp_config(cls, config):
config["ftp_upload_purge"] = "True"
def test_ftp_uploads_not_purged(self):
ftp_path = self._run_purgable_upload()
def test_ftp_uploads_not_purged(self, history_id: str):
ftp_path = self._run_purgable_upload(history_id=history_id)
assert not os.path.exists(ftp_path)
class TestAdvancedFtpUploadFetch(BaseFtpUploadConfigurationTestCase):
def test_fetch_ftp_directory(self):
def test_fetch_ftp_directory(self, history_id: str):
dir_path = self._get_user_ftp_path()
self._write_file(os.path.join(dir_path, "subdir"), "content 1", filename="1")
self._write_file(os.path.join(dir_path, "subdir"), "content 22", filename="2")
@@ -512,14 +511,14 @@ class TestAdvancedFtpUploadFetch(BaseFtpUploadConfigurationTestCase):
"ftp_path": "subdir",
"collection_type": "list",
}
self.fetch_target(target, assert_ok=True, wait=True)
hdca = self.dataset_populator.get_history_collection_details(self.history_id, hid=1)
self.fetch_target(target, history_id=history_id, assert_ok=True, wait=True)
hdca = self.dataset_populator.get_history_collection_details(history_id, hid=1)
assert len(hdca["elements"]) == 3, hdca
element0 = hdca["elements"][0]
assert element0["element_identifier"] == "1", hdca
assert element0["object"]["file_size"] == 9, element0
def test_fetch_nested_elements_from(self):
def test_fetch_nested_elements_from(self, history_id: str):
dir_path = self._get_user_ftp_path()
self._write_file(os.path.join(dir_path, "subdir1"), "content 1", filename="1")
self._write_file(os.path.join(dir_path, "subdir1"), "content 22", filename="2")
@@ -545,9 +544,9 @@ class TestAdvancedFtpUploadFetch(BaseFtpUploadConfigurationTestCase):
"elements": elements,
"collection_type": "list:list",
}
self.fetch_target(target, assert_ok=True, wait=True)
self.fetch_target(target, history_id=history_id, assert_ok=True, wait=True)
hdca = self.dataset_populator.get_history_collection_details(
self.history_id, history_content_type="dataset_collection"
history_id, history_content_type="dataset_collection"
)
assert len(hdca["elements"]) == 2, hdca
element0 = hdca["elements"][0]
@@ -555,13 +554,13 @@ class TestAdvancedFtpUploadFetch(BaseFtpUploadConfigurationTestCase):
class TestUploadOptionsFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase):
def test_upload_api_option_space_to_tab(self):
def test_upload_api_option_space_to_tab(self, history_id: str):
self._write_user_ftp_file("0.txt", ONE_TO_SIX_WITH_SPACES)
self._write_user_ftp_file("1.txt", ONE_TO_SIX_WITH_SPACES)
self._write_user_ftp_file("2.txt", ONE_TO_SIX_WITH_SPACES)
payload = self.dataset_populator.upload_payload(
self.history_id,
history_id,
ftp_files="0.txt",
file_type="tabular",
dbkey="hg19",
@@ -579,26 +578,26 @@ class TestUploadOptionsFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase
},
)
run_response = self.dataset_populator.tools_post(payload)
self.dataset_populator.wait_for_tool_run(self.history_id, run_response)
self.dataset_populator.wait_for_tool_run(history_id, run_response)
datasets = run_response.json()["outputs"]
assert len(datasets) == 3, datasets
content = self.dataset_populator.get_history_dataset_content(self.history_id, dataset=datasets[0])
content = self.dataset_populator.get_history_dataset_content(history_id, dataset=datasets[0])
assert content == ONE_TO_SIX_WITH_TABS
content = self.dataset_populator.get_history_dataset_content(self.history_id, dataset=datasets[1])
content = self.dataset_populator.get_history_dataset_content(history_id, dataset=datasets[1])
assert content == ONE_TO_SIX_WITH_SPACES
content = self.dataset_populator.get_history_dataset_content(self.history_id, dataset=datasets[2])
content = self.dataset_populator.get_history_dataset_content(history_id, dataset=datasets[2])
assert content == ONE_TO_SIX_WITH_TABS
def test_upload_api_option_to_posix_lines(self):
def test_upload_api_option_to_posix_lines(self, history_id: str):
self._write_user_ftp_file("0.txt", ONE_TO_SIX_ON_WINDOWS)
self._write_user_ftp_file("1.txt", ONE_TO_SIX_ON_WINDOWS)
self._write_user_ftp_file("2.txt", ONE_TO_SIX_ON_WINDOWS)
payload = self.dataset_populator.upload_payload(
self.history_id,
history_id,
ftp_files="0.txt",
file_type="tabular",
dbkey="hg19",
@@ -616,44 +615,44 @@ class TestUploadOptionsFtpUploadConfiguration(BaseFtpUploadConfigurationTestCase
},
)
run_response = self.dataset_populator.tools_post(payload)
self.dataset_populator.wait_for_tool_run(self.history_id, run_response)
self.dataset_populator.wait_for_tool_run(history_id, run_response)
datasets = run_response.json()["outputs"]
assert len(datasets) == 3, datasets
content = self.dataset_populator.get_history_dataset_content(self.history_id, dataset=datasets[0])
content = self.dataset_populator.get_history_dataset_content(history_id, dataset=datasets[0])
assert content == ONE_TO_SIX_WITH_TABS
content = self.dataset_populator.get_history_dataset_content(self.history_id, dataset=datasets[1])
content = self.dataset_populator.get_history_dataset_content(history_id, dataset=datasets[1])
assert content == ONE_TO_SIX_ON_WINDOWS
content = self.dataset_populator.get_history_dataset_content(self.history_id, dataset=datasets[2])
content = self.dataset_populator.get_history_dataset_content(history_id, dataset=datasets[2])
assert content == ONE_TO_SIX_WITH_TABS
def test_upload_option_auto_decompress_default(self):
def test_upload_option_auto_decompress_default(self, history_id: str):
self._copy_to_user_ftp_file("1.sam.gz")
payload = self.dataset_populator.upload_payload(
self.history_id,
history_id,
ftp_files="1.sam.gz",
file_type="auto",
)
run_response = self.dataset_populator.tools_post(payload)
self.dataset_populator.wait_for_tool_run(self.history_id, run_response)
self.dataset_populator.wait_for_tool_run(history_id, run_response)
datasets = run_response.json()["outputs"]
dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=datasets[0])
dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=datasets[0])
assert dataset["file_ext"] == "sam", dataset
def test_upload_option_auto_decompress_off(self):
def test_upload_option_auto_decompress_off(self, history_id: str):
self._copy_to_user_ftp_file("1.sam.gz")
payload = self.dataset_populator.upload_payload(
self.history_id,
history_id,
ftp_files="1.sam.gz",
file_type="auto",
auto_decompress=False,
)
run_response = self.dataset_populator.tools_post(payload)
self.dataset_populator.wait_for_tool_run(self.history_id, run_response)
self.dataset_populator.wait_for_tool_run(history_id, run_response)
datasets = run_response.json()["outputs"]
dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=datasets[0])
dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=datasets[0])
assert dataset["file_ext"] != "sam", dataset
def _copy_to_user_ftp_file(self, test_data_path):
@@ -851,7 +850,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase):
dataset = self.library_populator.get_library_contents_with_path(library["id"], "/dir1/file3")
assert dataset["file_size"] == 11, dataset
def test_fetch_history_compressed_type(self):
def test_fetch_history_compressed_type(self, history_id: str):
destination = {"type": "hdas"}
archive = self.test_data_resolver.get_filename("1.fastqsanger.gz")
targets = [
@@ -861,7 +860,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase):
}
]
payload = {
"history_id": self.history_id, # TODO: Shouldn't be needed :(
"history_id": history_id, # TODO: Shouldn't be needed :(
"targets": targets,
}
fetch_response = self.dataset_populator.fetch(payload)
@@ -870,7 +869,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase):
assert len(outputs) == 1
output = outputs[0]
assert output["name"] == "1.fastqsanger.gz"
contents_response = self.dataset_populator._get_contents_request(self.history_id)
contents_response = self.dataset_populator._get_contents_request(history_id)
assert contents_response.status_code == 200
contents = contents_response.json()
assert len(contents) == 1, contents
@@ -878,7 +877,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase):
assert contents[0]["name"] == "1.fastqsanger.gz", contents[0]
assert contents[0]["hid"] == 1, contents[0]
def test_fetch_recursive_archive_history(self):
def test_fetch_recursive_archive_history(self, history_id: str):
destination = {"type": "hdas"}
archive = self.test_data_resolver.get_filename("testdir1.zip")
targets = [
@@ -890,16 +889,16 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase):
}
]
payload = {
"history_id": self.history_id, # TODO: Shouldn't be needed :(
"history_id": history_id, # TODO: Shouldn't be needed :(
"targets": targets,
}
self.dataset_populator.fetch(payload)
contents_response = self.dataset_populator._get_contents_request(self.history_id)
contents_response = self.dataset_populator._get_contents_request(history_id)
assert contents_response.status_code == 200
contents = contents_response.json()
assert len(contents) == 3
def test_fetch_recursive_archive_to_library(self):
def test_fetch_recursive_archive_to_library(self, history_id: str):
bed_test_data_path = self.test_data_resolver.get_filename("testdir1.zip")
targets = [
{
@@ -910,7 +909,7 @@ class TestFetchByPath(BaseUploadContentConfigurationTestCase):
}
]
payload = {
"history_id": self.history_id, # TODO: Shouldn't be needed :(
"history_id": history_id, # TODO: Shouldn't be needed :(
"targets": targets,
}
self.dataset_populator.fetch(payload)
@@ -931,19 +930,19 @@ class TestDirectoryAndCompressedTypes(BaseUploadContentConfigurationTestCase):
super().handle_galaxy_config_kwds(config)
config["allow_path_paste"] = True
def test_tar_to_directory(self):
def test_tar_to_directory(self, history_id: str):
dataset = self.dataset_populator.new_dataset(
self.history_id,
history_id,
f"file://{TEST_DATA_DIRECTORY}/testdir.tar",
file_type="tar",
auto_decompress=False,
wait=True,
)
dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=dataset)
dataset = self.dataset_populator.get_history_dataset_details(history_id, dataset=dataset)
assert dataset["file_ext"] == "tar", dataset
response = self.dataset_populator.run_tool(
tool_id="CONVERTER_tar_to_directory",
inputs={"input1": {"src": "hda", "id": dataset["id"]}},
history_id=self.history_id,
history_id=history_id,
)
self.dataset_populator.wait_for_job(response["jobs"][0]["id"])
test/integration/test_work_queue_put_failure.py
+4 -5
View File
@@ -22,7 +22,6 @@ class TestWorkQueuePutFailure(integration_util.IntegrationTestCase):
def setUp(self):
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.history_id = self.dataset_populator.new_history()
@classmethod
def handle_galaxy_config_kwds(
@@ -37,10 +36,10 @@ class TestWorkQueuePutFailure(integration_util.IntegrationTestCase):
# Disable tool dependency resolution.
config["tool_dependency_dir"] = "none"
def test_job_fails(self):
def test_job_fails(self, history_id: str):
# Set fetch_data to false so we don't bypass the job queue
self.dataset_populator.new_dataset(self.history_id, fetch_data=False, content="1 2 3")
self.dataset_populator.wait_for_history(self.history_id, assert_ok=False)
state_details = self.galaxy_interactor.get(f"histories/{self.history_id}").json()["state_details"]
self.dataset_populator.new_dataset(history_id, fetch_data=False, content="1 2 3")
self.dataset_populator.wait_for_history(history_id, assert_ok=False)
state_details = self.galaxy_interactor.get(f"histories/{history_id}").json()["state_details"]
assert state_details["running"] == 0
assert state_details["error"] == 1
test/integration/test_workflow_handler_configuration.py
+18 -20
View File
@@ -105,15 +105,13 @@ class BaseWorkflowHandlerConfigurationTestCase(integration_util.IntegrationTestC
super().setUp()
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
self.history_id = self.dataset_populator.new_history()
@classmethod
def handle_galaxy_config_kwds(cls, config):
config["job_config_file"] = config_file(WORKFLOW_HANDLER_JOB_CONFIG_TEMPLATE, assign_with=cls.assign_with)
def _invoke_n_workflows(self, n):
def _invoke_n_workflows(self, n, history_id: str):
workflow_id = self.workflow_populator.upload_yaml_workflow(PAUSE_WORKFLOW)
history_id = self.history_id
hda1 = self.dataset_populator.new_dataset(history_id, content="1 2 3")
index_map = {"0": dict(src="hda", id=hda1["id"])}
request = {}
@@ -124,11 +122,11 @@ class BaseWorkflowHandlerConfigurationTestCase(integration_util.IntegrationTestC
for _ in range(n):
self._post(url, data=request)
def _get_workflow_invocations(self):
def _get_workflow_invocations(self, history_id: str):
# Consider exposing handler via the API to reduce breaking
# into Galaxy's internal state.
app = self._app
history_id = app.security.decode_id(self.history_id)
history_id = app.security.decode_id(history_id)
sa_session = app.model.context.current
history = sa_session.query(app.model.History).get(history_id)
workflow_invocations = history.workflow_invocations
@@ -144,9 +142,9 @@ class TestHistoryRestrictionConfiguration(BaseWorkflowHandlerConfigurationTestCa
# Assign with db-preassign. Would also work with grabbing assignment, but we don't start grabber.
assign_with = "db-preassign"
def test_history_to_handler_restriction(self):
self._invoke_n_workflows(10)
workflow_invocations = self._get_workflow_invocations()
def test_history_to_handler_restriction(self, history_id: str):
self._invoke_n_workflows(10, history_id)
workflow_invocations = self._get_workflow_invocations(history_id)
assert len(workflow_invocations) == 10
# Verify all 10 assigned to same handler - there would be a
# 1 in 10^10 chance for this to occur randomly.
@@ -166,9 +164,9 @@ class TestHistoryParallelConfiguration(BaseWorkflowHandlerConfigurationTestCase)
super().handle_galaxy_config_kwds(config)
config["parallelize_workflow_scheduling_within_histories"] = True
def test_workflows_spread_across_multiple_handlers(self):
self._invoke_n_workflows(20)
workflow_invocations = self._get_workflow_invocations()
def test_workflows_spread_across_multiple_handlers(self, history_id: str):
self._invoke_n_workflows(20, history_id)
workflow_invocations = self._get_workflow_invocations(history_id)
assert len(workflow_invocations) == 20
handlers = set()
for workflow_invocation in workflow_invocations:
@@ -192,9 +190,9 @@ class TestWorkflowSchedulerHandlerAssignment(BaseWorkflowHandlerConfigurationTes
WORKFLOW_SCHEDULERS_CONFIG_TEMPLATE, assign_with=cls.assign_with
)
def test_handler_assignment(self):
self._invoke_n_workflows(1)
workflow_invocations = self._get_workflow_invocations()
def test_handler_assignment(self, history_id: str):
self._invoke_n_workflows(1, history_id)
workflow_invocations = self._get_workflow_invocations(history_id)
assert WORKFLOW_SCHEDULER_HANDLER_PATTERN.match(workflow_invocations[0].handler)
@@ -236,10 +234,10 @@ class TestJobHandlerAsWorkflowHandlerWithDbSkipLocked(BaseWorkflowHandlerConfigu
super().handle_galaxy_config_kwds(config)
config["server_name"] = "handler0"
def test_handler_assignment(self):
self._invoke_n_workflows(1)
def test_handler_assignment(self, history_id: str):
self._invoke_n_workflows(1, history_id)
time.sleep(2)
workflow_invocations = self._get_workflow_invocations()
workflow_invocations = self._get_workflow_invocations(history_id)
assert JOB_HANDLER_PATTERN.match(workflow_invocations[0].handler)
def test_default_job_handler_is_workflow_handler(self):
@@ -285,10 +283,10 @@ class TestWorkflowSchedulerHandlerAssignmentDbSkipLocked(TestExplicitWorkflowHan
assign_with = "db-skip-locked"
def test_handler_assignment(self):
self._invoke_n_workflows(1)
def test_handler_assignment(self, history_id: str):
self._invoke_n_workflows(1, history_id)
time.sleep(2)
workflow_invocations = self._get_workflow_invocations()
workflow_invocations = self._get_workflow_invocations(history_id)
assert WORKFLOW_SCHEDULER_HANDLER_PATTERN.match(workflow_invocations[0].handler)