diff --git a/lib/galaxy/jobs/runners/pulsar.py b/lib/galaxy/jobs/runners/pulsar.py index 7a2af874882..ea9819bc2b5 100644 --- a/lib/galaxy/jobs/runners/pulsar.py +++ b/lib/galaxy/jobs/runners/pulsar.py @@ -197,6 +197,18 @@ PULSAR_PARAM_SPECS = dict( map=int, default=None, ), + relay_url=dict( + map=specs.to_str_or_none, + default=None, + ), + relay_username=dict( + map=specs.to_str_or_none, + default=None, + ), + relay_password=dict( + map=specs.to_str_or_none, + default=None, + ), ) @@ -260,7 +272,7 @@ class PulsarJobRunner(AsynchronousJobRunner[AsynchronousJobState]): client_manager_kwargs[kwd] = self.runner_params[kwd] for kwd in self.runner_params.keys(): - if kwd.startswith("amqp_") or kwd.startswith("transport_"): + if kwd.startswith("amqp_") or kwd.startswith("transport_") or kwd.startswith("relay_"): client_manager_kwargs[kwd] = self.runner_params[kwd] return client_manager_kwargs diff --git a/lib/galaxy_test/driver/integration_util.py b/lib/galaxy_test/driver/integration_util.py index 34b0f25b333..4ab96b0a618 100644 --- a/lib/galaxy_test/driver/integration_util.py +++ b/lib/galaxy_test/driver/integration_util.py @@ -8,6 +8,7 @@ testing configuration. import os import re import string +import subprocess import sys from collections.abc import Iterator from typing import ( @@ -49,6 +50,56 @@ SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__)) VAULT_CONF = os.path.join(SCRIPT_DIRECTORY, "vault_conf.yml") +def docker_run(image, name, *args, detach=True, remove=True, ports=None, env_vars: Optional[dict[str, str]] = None): + cmd = ["docker", "run"] + + if ports: + for host_port, container_port in ports: + cmd.extend(["-p", f"{host_port}:{container_port}"]) + + if detach: + cmd.append("-d") + + cmd.extend(["--name", name]) + + if remove: + cmd.append("--rm") + if env_vars: + for key, value in env_vars.items(): + cmd.extend(["-e", f"{key}={value}"]) + + cmd.append(image) + cmd.extend(args) + print("Running docker command:", " ".join(cmd)) + + subprocess.check_call(cmd) + + +def docker_exec(container_name, *args, output=True): + cmd = ["docker", "exec", container_name] + cmd.extend(args) + + if output: + return subprocess.check_output(cmd) + else: + subprocess.check_call(cmd) + + +def docker_ip_address(container_name): + cmd = [ + "docker", + "inspect", + "-f", + "{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}", + container_name, + ] + return subprocess.check_output(cmd).decode("utf-8").strip() + + +def docker_rm(container_name): + subprocess.check_call(["docker", "rm", "-f", container_name]) + + def skip_if_jenkins(cls): if os.environ.get("BUILD_NUMBER", ""): return skip diff --git a/test/integration/objectstore/_base.py b/test/integration/objectstore/_base.py index d8cefdf1f52..df94dbb00a5 100644 --- a/test/integration/objectstore/_base.py +++ b/test/integration/objectstore/_base.py @@ -6,6 +6,12 @@ import time from galaxy_test.base.populators import DatasetPopulator from galaxy_test.driver import integration_util +from galaxy_test.driver.integration_util import ( + docker_exec, + docker_ip_address, + docker_rm, + docker_run, +) OBJECT_STORE_HOST = os.environ.get("GALAXY_INTEGRATION_OBJECT_STORE_HOST", "127.0.0.1") OBJECT_STORE_PORT = int(os.environ.get("GALAXY_INTEGRATION_OBJECT_STORE_PORT", 9000)) @@ -377,49 +383,3 @@ def await_oneprovider_demo_readiness(op_container_name): def get_onedata_access_token(oz_container_name): return docker_exec(oz_container_name, "demo-access-token").decode("utf-8").strip() - - -def docker_run(image, name, *args, detach=True, remove=True, ports=None): - cmd = ["docker", "run"] - - if ports: - for container_port, host_port in ports: - cmd.extend(["-p", f"{container_port}:{host_port}"]) - - if detach: - cmd.append("-d") - - cmd.extend(["--name", name]) - - if remove: - cmd.append("--rm") - - cmd.append(image) - cmd.extend(args) - - subprocess.check_call(cmd) - - -def docker_exec(container_name, *args, output=True): - cmd = ["docker", "exec", container_name] - cmd.extend(args) - - if output: - return subprocess.check_output(cmd) - else: - subprocess.check_call(cmd) - - -def docker_ip_address(container_name): - cmd = [ - "docker", - "inspect", - "-f", - "{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}", - container_name, - ] - return subprocess.check_output(cmd).decode("utf-8").strip() - - -def docker_rm(container_name): - subprocess.check_call(["docker", "rm", "-f", container_name]) diff --git a/test/integration/test_pulsar_embedded_relay.py b/test/integration/test_pulsar_embedded_relay.py new file mode 100644 index 00000000000..50601bf1128 --- /dev/null +++ b/test/integration/test_pulsar_embedded_relay.py @@ -0,0 +1,109 @@ +"""Integration tests for the Pulsar embedded runner with outputs to working directory.""" + +import os +import string +import tempfile +import uuid + +from galaxy.util import safe_makedirs +from galaxy_test.base.populators import DatasetPopulator +from galaxy_test.driver import integration_util +from galaxy_test.driver.integration_util import ( + docker_rm, + docker_run, +) + +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") +AMQP_URL = os.environ.get("GALAXY_TEST_PULSAR_RELAY") + +JOB_CONF_TEMPLATE = """ +runners: + local: + load: galaxy.jobs.runners.local:LocalJobRunner + workers: 1 + pulsar: + load: galaxy.jobs.runners.pulsar:PulsarEmbeddedMQJobRunner + pulsar_app_config: + tool_dependency_dir: none + conda_auto_init: false + conda_auto_install: false + message_queue_url: ${amqp_url} + message_queue_username: admin + message_queue_password: ${relay_password} + staging_directory: ${jobs_directory} + relay_url: ${amqp_url} + relay_username: admin + relay_password: ${relay_password} +execution: + default: pulsar_mq_environment + environments: + pulsar_mq_environment: + runner: pulsar + rewrite_parameters: true + dependency_resolution: none + default_file_action: remote_transfer + jobs_directory: ${jobs_directory} + remote_metadata: true + remote_property_galaxy_home: ${galaxy_home} + local_environment: + runner: local +tools: + - id: __DATA_FETCH__ + environment: local_environment +""" + + +def _handle_galaxy_config_kwds(cls, config): + random_password = uuid.uuid4().hex + docker_run( + "mvdbeek/pulsar-relay", + cls.container_name, + env_vars={"PULSAR_BOOTSTRAP_ADMIN_USERNAME": "admin", "PULSAR_BOOTSTRAP_ADMIN_PASSWORD": random_password}, + remove=False, + ports=[(9001, 8080)], + ) + 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="http://localhost:9001", + jobs_directory=jobs_directory, + galaxy_home=os.path.join(SCRIPT_DIRECTORY, os.pardir), + relay_password=random_password, + ) + 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 EmbeddedPulsarRelayIntegrationInstance(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 + """ + + container_name = "pulsar-relay" + dataset_populator: DatasetPopulator + framework_tool_and_types = True + + @classmethod + def handle_galaxy_config_kwds(cls, config): + _handle_galaxy_config_kwds(cls, config) + + @classmethod + def tearDownClass(cls): + try: + super().tearDownClass() + finally: + docker_rm(cls.container_name) + + +instance = integration_util.integration_module_instance(EmbeddedPulsarRelayIntegrationInstance) + +test_tools = integration_util.integration_tool_runner( + ["simple_constructs", "composite_output_tests", "all_output_types"] +)