From d12097c35c59df5c6d6edb4bc14ff4b26dcfcc46 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Thu, 15 Oct 2026 02:07:42 +0200 Subject: [PATCH] Give each embedded Pulsar MQ test instance its own AMQP queues The module-level tool test instance in test_pulsar_embedded_mq.py stays up until pytest leaves the module, so it is still consuming from the broker while the purge test classes in the same file run. All of them used the default pulsar_ queue names, and the older instance's embedded Pulsar could take the new instance's setup message and stage the job in its own directory; the job then failed with tool_script.sh missing. --- test/integration/test_pulsar_embedded_mq.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/test/integration/test_pulsar_embedded_mq.py b/test/integration/test_pulsar_embedded_mq.py index b9376a10a16..c7a674d4192 100644 --- a/test/integration/test_pulsar_embedded_mq.py +++ b/test/integration/test_pulsar_embedded_mq.py @@ -3,6 +3,7 @@ import os import string import tempfile +import uuid import pytest @@ -27,8 +28,10 @@ runners: conda_auto_init: false conda_auto_install: false message_queue_url: ${amqp_url} + amqp_key_prefix: ${amqp_key_prefix} staging_directory: ${jobs_directory} amqp_url: ${amqp_url} + amqp_key_prefix: ${amqp_key_prefix} execution: default: pulsar_mq_environment environments: @@ -57,7 +60,12 @@ def _handle_galaxy_config_kwds(cls, config): 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) + amqp_url=AMQP_URL, + # The module's tool test instance outlives the classes that run after it; queues of its + # own keep its embedded Pulsar from taking their jobs. + amqp_key_prefix=f"pulsar_{uuid.uuid4().hex[:12]}_", + 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)