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

Drop useless and/or redundant completion tests

Browse Source
This commit is contained in:
mvdbeek
2026-01-21 14:44:54 +01:00
parent 513476dc11
commit c4a444ea05
1 changed files with 8 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy_test/api/test_workflow_completion.py
+8 -48
View File
@@ -11,7 +11,10 @@ from galaxy_test.base.populators import (
DatasetPopulator,
WorkflowPopulator,
)
from galaxy_test.base.workflow_fixtures import WORKFLOW_SIMPLE_CAT_TWICE
from galaxy_test.base.workflow_fixtures import (
WORKFLOW_RUNTIME_PARAMETER_AFTER_PAUSE,
WORKFLOW_SIMPLE_CAT_TWICE,
)
from ._framework import ApiTestCase
@@ -26,40 +29,18 @@ class TestWorkflowCompletionEndpoint(ApiTestCase, UsesCeleryTasks):
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
def test_completion_endpoint_exists(self):
"""Test that the completion endpoint exists and returns proper response."""
with self.dataset_populator.test_history() as history_id:
summary = self.workflow_populator.run_workflow(
WORKFLOW_SIMPLE_CAT_TWICE,
test_data={"input1": "hello world"},
history_id=history_id,
wait=True,
assert_ok=True,
)
# The endpoint should exist and return 200
completion = self.workflow_populator.get_invocation_completion(summary.invocation_id)
# Completion may or may not exist yet depending on timing
if completion is not None:
assert "completion_time" in completion
assert "all_jobs_ok" in completion
def test_completion_endpoint_returns_none_before_completion(self):
"""Test that completion endpoint returns None for incomplete invocation."""
with self.dataset_populator.test_history() as history_id:
summary = self.workflow_populator.run_workflow(
WORKFLOW_SIMPLE_CAT_TWICE,
WORKFLOW_RUNTIME_PARAMETER_AFTER_PAUSE,
test_data={"input1": "hello world"},
history_id=history_id,
wait=False, # Don't wait for completion
)
# Check immediately - might be None or already complete depending on timing
completion = self.workflow_populator.get_invocation_completion(summary.invocation_id)
# Either None (not complete) or has completion data
if completion is not None:
assert "completion_time" in completion
assert "all_jobs_ok" in completion
assert completion is None
def test_monitor_transitions_to_completed(self):
"""Test that background monitor transitions invocation to COMPLETED state."""
@@ -77,7 +58,6 @@ class TestWorkflowCompletionEndpoint(ApiTestCase, UsesCeleryTasks):
# Verify completion record exists
completion = self.workflow_populator.get_invocation_completion(summary.invocation_id)
assert completion is not None, "Expected completion record to exist"
assert completion["all_jobs_ok"] is True
def test_completion_response_structure(self):
"""Test that completion response has correct structure."""
@@ -98,29 +78,9 @@ class TestWorkflowCompletionEndpoint(ApiTestCase, UsesCeleryTasks):
assert completion is not None, "Expected completion record to exist after waiting"
assert "completion_time" in completion
assert "job_state_summary" in completion
assert "all_jobs_ok" in completion
assert "hooks_executed" in completion
assert isinstance(completion["job_state_summary"], dict)
assert isinstance(completion["hooks_executed"], list)
def test_on_complete_actions_stored(self):
"""Test that on_complete actions are stored when invoking workflow."""
with self.dataset_populator.test_history() as history_id:
# Run workflow with on_complete actions (new format with action objects)
summary = self.workflow_populator.run_workflow(
WORKFLOW_SIMPLE_CAT_TWICE,
test_data={"input1": "hello world"},
history_id=history_id,
wait=True,
assert_ok=True,
extra_invocation_kwds={"on_complete": [{"send_notification": {}}]},
)
# Wait for monitor to process (completion transition)
invocation = self.workflow_populator.wait_for_invocation_and_completion(summary.invocation_id, timeout=30)
# Verify state transitioned to completed
assert invocation["state"] == "completed"
assert completion["job_state_summary"] == {"ok": 1}
assert completion["hooks_executed"] == []
def test_completion_export_config_accepted(self):
"""Test that export_to_file_source configuration is accepted when invoking workflow."""