Fix up test cases

Add a sleep so we can delay output purging until command line is
templated. If we don't sleep we generate a command line where the
output path is just `''`. That needs another fix!

test/functional/tools/all_output_types.xml

@@ -1,7 +1,9 @@
 <tool id="all_output_types" name="all_output_types" version="1.0.0" profile="24.0">
     <command><![CDATA[
+        sleep $sleep_param &&
         echo hi > output.txt &&
-        cp output.txt '$static_output' &&
+        echo hi > '$static_output' &&
+        echo hi > '$static_output_2' &&
         cp '$c1' galaxy.json
         ]]>
     </command>
@@ -12,9 +14,13 @@
     "info": "my dynamic info"
   }}
         </configfile>
-      </configfiles>
+    </configfiles>
+    <inputs>
+        <param name="sleep_param" type="integer" value="0" />
+    </inputs>
     <outputs>
         <data name="static_output" format="txt" />
+        <data name="static_output_2" format="txt" />
         <data name="output_workdir" from_work_dir="output.txt" format="txt" />
         <data name="output_tool_supplied_metadata" from_work_dir="output.txt" format="auto" />
         <data format="txt" name="discovered_output">

test/integration/objectstore/_purged_handling.py

@@ -0,0 +1,45 @@
+import time
+
+from galaxy_test.base.populators import DatasetPopulator
+
+
+def purge_while_job_running(dataset_populator: DatasetPopulator, sleep_before_purge=4):
+    with dataset_populator.test_history() as history_id:
+        response = dataset_populator.run_tool(
+            "all_output_types",
+            inputs={
+                "sleep_param": 5,
+            },
+            history_id=history_id,
+        )
+        job = dataset_populator.get_job_details(response["jobs"][0]["id"], full=True).json()
+        # Sleep a second to make sure we template command before purging output
+        time.sleep(sleep_before_purge)
+        hda_ids = []
+        for output_name, output in job["outputs"].items():
+            if output_name == "static_output_2":
+                # We need to keep one output so the job doesn't get cancelled
+                continue
+            dataset_populator.delete_dataset(
+                history_id=history_id, content_id=output["id"], purge=True, wait_for_purge=True
+            )
+            hda_ids.append(output["id"])
+        for output_collection in job["output_collections"].values():
+            hdca = dataset_populator.get_history_collection_details(
+                history_id=history_id, content_id=output_collection["id"]
+            )
+            for element in hdca["elements"]:
+                hda_id = element["object"]["id"]
+                dataset_populator.delete_dataset(
+                    history_id=history_id, content_id=hda_id, purge=True, wait_for_purge=True
+                )
+                hda_ids.append(hda_id)
+        dataset_populator.wait_for_job(job["id"], assert_ok=True)
+        # Now make sure we can't find the datasets
+        for hda_id in hda_ids:
+            exception = None
+            try:
+                dataset_populator.get_history_dataset_content(history_id=history_id, dataset_id=hda_id)
+            except AssertionError as e:
+                exception = e
+            assert exception and "The dataset you are attempting to view has been purged" in str(exception)

test/integration/test_extended_metadata.py

@@ -5,6 +5,7 @@
     LibraryPopulator,
 )
 from galaxy_test.driver import integration_util
+from .objectstore.test_purged_handling import purge_while_job_running
 
 TEST_TOOL_IDS = [
     "from_work_dir_glob",
@@ -47,6 +48,7 @@
 
 class TestExtendedMetadataIntegration(integration_util.IntegrationTestCase):
     dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
 
     def setUp(self):
         super().setUp()
@@ -95,6 +97,9 @@ def test_fetch_data_library(self):
         assert dataset["file_ext"] == "bed", dataset
         assert dataset["created_from_basename"] == "4.bed"
 
+    def test_purge_while_job_running(self):
+        purge_while_job_running(self.dataset_populator)
+
 
 class TestExtendedMetadataDeferredIntegration(integration_util.IntegrationTestCase):
     dataset_populator: DatasetPopulator

test/integration/test_pulsar_embedded_mq.py

@@ -7,7 +7,9 @@
 import pytest
 
 from galaxy.util import safe_makedirs
+from galaxy_test.base.populators import DatasetPopulator
 from galaxy_test.driver import integration_util
+from .objectstore.test_purged_handling import purge_while_job_running
 
 SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
 EMBEDDED_PULSAR_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "embedded_pulsar_mq_job_conf.yml")
@@ -46,34 +48,60 @@
 """
 
 
+def _handle_galaxy_config_kwds(cls, config):
+    amqp_url = os.environ.get("GALAXY_TEST_AMQP_URL", None)
+    if amqp_url is None:
+        pytest.skip("External AMQP URL not configured for test")
+
+    jobs_directory = os.path.join(cls._test_driver.mkdtemp(), "pulsar_staging")
+    safe_makedirs(jobs_directory)
+    job_conf_template = string.Template(JOB_CONF_TEMPLATE)
+    job_conf_str = job_conf_template.substitute(
+        amqp_url=AMQP_URL, jobs_directory=jobs_directory, galaxy_home=os.path.join(SCRIPT_DIRECTORY, os.pardir)
+    )
+    with tempfile.NamedTemporaryFile(suffix="_mq_job_conf.yml", mode="w", delete=False) as job_conf:
+        job_conf.write(job_conf_str)
+    config["job_config_file"] = job_conf.name
+    infrastructure_url = "http://localhost:$GALAXY_WEB_PORT"
+    config["galaxy_infrastructure_url"] = infrastructure_url
+
+
+class TestEmbeddedMessageQueuePulsarPurge(integration_util.IntegrationTestCase):
+    dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        _handle_galaxy_config_kwds(cls, config)
+
+    def test_purge_while_job_running(self):
+        purge_while_job_running(self.dataset_populator)
+
+
 class EmbeddedMessageQueuePulsarIntegrationInstance(integration_util.IntegrationInstance):
     """Describe a Galaxy test instance with embedded pulsar configured.
 
     $ Setup RabbitMQ (e.g. https://www.rabbitmq.com/install-homebrew.html)
     $ GALAXY_TEST_AMQP_URL='amqp://guest:guest@localhost:5672//' pytest -s test/integration/test_pulsar_embedded_mq.py
     """
 
+    dataset_populator: DatasetPopulator
     framework_tool_and_types = True
 
     @classmethod
     def handle_galaxy_config_kwds(cls, config):
-        amqp_url = os.environ.get("GALAXY_TEST_AMQP_URL", None)
-        if amqp_url is None:
-            pytest.skip("External AMQP URL not configured for test")
-
-        jobs_directory = os.path.join(cls._test_driver.mkdtemp(), "pulsar_staging")
-        safe_makedirs(jobs_directory)
-        job_conf_template = string.Template(JOB_CONF_TEMPLATE)
-        job_conf_str = job_conf_template.substitute(
-            amqp_url=AMQP_URL, jobs_directory=jobs_directory, galaxy_home=os.path.join(SCRIPT_DIRECTORY, os.pardir)
-        )
-        with tempfile.NamedTemporaryFile(suffix="_mq_job_conf.yml", mode="w", delete=False) as job_conf:
-            job_conf.write(job_conf_str)
-        config["job_config_file"] = job_conf.name
-        infrastructure_url = "http://localhost:$GALAXY_WEB_PORT"
-        config["galaxy_infrastructure_url"] = infrastructure_url
+        _handle_galaxy_config_kwds(cls, config)
+
+    def test_purge_while_job_running(self):
+        purge_while_job_running(self.dataset_populator)
 
 
 instance = integration_util.integration_module_instance(EmbeddedMessageQueuePulsarIntegrationInstance)
 
-test_tools = integration_util.integration_tool_runner(["simple_constructs", "composite_output_tests"])
+test_tools = integration_util.integration_tool_runner(
+    ["simple_constructs", "composite_output_tests", "all_output_types"]
+)