mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Drop redundant completion hook test, improve remaining test
This commit is contained in:
@@ -80,69 +80,6 @@ class TestWorkflowCompletionExportHook(PosixFileSourceSetup, IntegrationTestCase
|
||||
# Verify the zip contains expected rocrate metadata
|
||||
with zipfile.ZipFile(export_path, "r") as zf:
|
||||
names = zf.namelist()
|
||||
datasets = [n for n in names if n.startswith("datasets/") and not n.endswith("/")]
|
||||
assert len(datasets) == 2, f"Expected 2 datasets, got {len(datasets)}: {datasets}"
|
||||
assert "ro-crate-metadata.json" in names, f"Missing ro-crate-metadata.json in {names}"
|
||||
|
||||
def test_export_with_multiple_outputs(self):
|
||||
"""Test export works correctly with multiple workflow outputs."""
|
||||
workflow = """
|
||||
class: GalaxyWorkflow
|
||||
name: Multi-Output Completion Export Test
|
||||
inputs:
|
||||
input1: data
|
||||
steps:
|
||||
cat1:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input1
|
||||
cat2:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input1
|
||||
outputs:
|
||||
output1:
|
||||
outputSource: cat1/out_file1
|
||||
output2:
|
||||
outputSource: cat2/out_file1
|
||||
"""
|
||||
export_filename = "multi_output_export.rocrate.zip"
|
||||
target_uri = f"gxfiles://posix_test/{export_filename}"
|
||||
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self.workflow_populator.run_workflow(
|
||||
workflow,
|
||||
test_data={"input1": "test data content"},
|
||||
history_id=history_id,
|
||||
wait=True,
|
||||
assert_ok=True,
|
||||
extra_invocation_kwds={
|
||||
"on_complete": [
|
||||
{
|
||||
"export_to_file_source": {
|
||||
"target_uri": target_uri,
|
||||
"format": "rocrate.zip",
|
||||
"include_files": True,
|
||||
}
|
||||
}
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
# Wait for completion
|
||||
self.workflow_populator.wait_for_invocation_and_completion(summary.invocation_id, timeout=60)
|
||||
|
||||
# Wait for export
|
||||
export_path = os.path.join(self.root_dir, export_filename)
|
||||
|
||||
def check_export():
|
||||
return export_path if os.path.exists(export_path) else None
|
||||
|
||||
wait_on(check_export, "export file", timeout=60)
|
||||
|
||||
# Verify export contains dataset files
|
||||
with zipfile.ZipFile(export_path, "r") as zf:
|
||||
names = zf.namelist()
|
||||
dataset_files = [n for n in names if n.startswith("datasets/")]
|
||||
# Should have input + 2 outputs = 3 datasets
|
||||
assert (
|
||||
len(dataset_files) >= 3
|
||||
), f"Expected at least 3 dataset files, found {len(dataset_files)}: {dataset_files}"
|
||||
|
||||
Reference in New Issue
Block a user