diff --git a/lib/galaxy/jobs/runners/gcp_batch.py b/lib/galaxy/jobs/runners/gcp_batch.py index 438c6572aaa..694f0d3fbd4 100644 --- a/lib/galaxy/jobs/runners/gcp_batch.py +++ b/lib/galaxy/jobs/runners/gcp_batch.py @@ -43,6 +43,56 @@ log = logging.getLogger(__name__) __all__ = ("GoogleCloudBatchJobRunner",) +# Runner parameter specifications. The defaults defined here are served lazily by +# the ParamsWithSpecs defaultdict via __missing__, which is only triggered by +# subscript access (runner_params[key]) -- not by .get(). See _get_job_params. +RUNNER_PARAM_SPECS: dict[str, dict[str, Any]] = { + "project_id": dict(map=str, default=None), + "region": dict(map=str, default="us-central1"), + "zone": dict(map=str, default=None), + "service_account_file": dict(map=str, default=None), + "service_account_email": dict(map=str, default=None), + "machine_type": dict(map=str, default="n2-standard-4"), + "boot_disk_size_gb": dict(map=int, default=100), + "boot_disk_type": dict(map=str, default="pd-standard"), + "max_retry_count": dict(map=int, default=3), + "max_run_duration": dict(map=str, default=DEFAULT_MAX_RUN_DURATION), + "polling_interval": dict(map=int, default=30), + # Volume configuration (generic format: "server:/remote_path:/mount_path[:ro],...") + "gcp_batch_volumes": dict(map=str, default=None), + # Extra docker volume mounts (format: "/host/path:/container/path[:ro],...") + "docker_extra_volumes": dict(map=str, default=None), + # Network configuration for NFS access + "network": dict(map=str, default="default"), + "subnet": dict(map=str, default="default"), + # Compute resource configuration (defaults - will be overridden by job requirements) + "vcpu": dict(map=float, default=1.0), + "memory_mib": dict(map=int, default=DEFAULT_MEMORY_MIB), + # Job-specific resource requests (same as Kubernetes runner) + "requests_cpu": dict(map=str, default=None), + "requests_memory": dict(map=str, default=None), + "limits_cpu": dict(map=str, default=None), + "limits_memory": dict(map=str, default=None), + # Container execution settings + "use_container": dict(map=bool, default=True), + "galaxy_user_id": dict( + map=str, valid=lambda s: s == "$uid" or isinstance(s, int) or not s or str(s).isdigit(), default=None + ), + "galaxy_group_id": dict( + map=str, valid=lambda s: s == "$gid" or isinstance(s, int) or not s or str(s).isdigit(), default=None + ), + # Custom VM image (optional) + "custom_vm_image": dict(map=str, default=None), + # Job cleanup: if true, delete GCP Batch jobs after Galaxy marks them complete + "delete_completed_jobs": dict(map=bool, default=True), + # Prefix for GCP Batch job IDs (helps identify which Galaxy server submitted a job) + "job_id_prefix": dict(map=str, default="galaxy-job"), + # Object store fallback (for future use) + "use_object_store": dict(map=bool, default=False), + "object_store_path": dict(map=str, default=None), +} + + class GoogleCloudBatchJobRunner(AsynchronousJobRunner): """ Job runner that submits jobs to Google Cloud Batch. @@ -54,54 +104,7 @@ class GoogleCloudBatchJobRunner(AsynchronousJobRunner): """Initialize the Google Cloud Batch job runner.""" log.debug("Starting GoogleCloudBatchJobRunner.__init__") - # Define runner parameter specifications - runner_param_specs = { - "project_id": dict(map=str, default=None), - "region": dict(map=str, default="us-central1"), - "zone": dict(map=str, default=None), - "service_account_file": dict(map=str, default=None), - "service_account_email": dict(map=str, default=None), - "machine_type": dict(map=str, default="n2-standard-4"), - "boot_disk_size_gb": dict(map=int, default=100), - "boot_disk_type": dict(map=str, default="pd-standard"), - "max_retry_count": dict(map=int, default=3), - "max_run_duration": dict(map=str, default=DEFAULT_MAX_RUN_DURATION), - "polling_interval": dict(map=int, default=30), - # Volume configuration (generic format: "server:/remote_path:/mount_path[:ro],...") - "gcp_batch_volumes": dict(map=str, default=None), - # Extra docker volume mounts (format: "/host/path:/container/path[:ro],...") - "docker_extra_volumes": dict(map=str, default=None), - # Network configuration for NFS access - "network": dict(map=str, default="default"), - "subnet": dict(map=str, default="default"), - # Compute resource configuration (defaults - will be overridden by job requirements) - "vcpu": dict(map=float, default=1.0), - "memory_mib": dict(map=int, default=DEFAULT_MEMORY_MIB), - # Job-specific resource requests (same as Kubernetes runner) - "requests_cpu": dict(map=str, default=None), - "requests_memory": dict(map=str, default=None), - "limits_cpu": dict(map=str, default=None), - "limits_memory": dict(map=str, default=None), - # Container execution settings - "use_container": dict(map=bool, default=True), - "galaxy_user_id": dict( - map=str, valid=lambda s: s == "$uid" or isinstance(s, int) or not s or str(s).isdigit(), default=None - ), - "galaxy_group_id": dict( - map=str, valid=lambda s: s == "$gid" or isinstance(s, int) or not s or str(s).isdigit(), default=None - ), - # Custom VM image (optional) - "custom_vm_image": dict(map=str, default=None), - # Job cleanup: if true, delete GCP Batch jobs after Galaxy marks them complete - "delete_completed_jobs": dict(map=bool, default=True), - # Prefix for GCP Batch job IDs (helps identify which Galaxy server submitted a job) - "job_id_prefix": dict(map=str, default="galaxy-job"), - # Object store fallback (for future use) - "use_object_store": dict(map=bool, default=False), - "object_store_path": dict(map=str, default=None), - } - - kwargs.update({"runner_param_specs": runner_param_specs}) + kwargs.update({"runner_param_specs": RUNNER_PARAM_SPECS}) super().__init__(app, nworkers, **kwargs) # Initialize Google Cloud Batch client diff --git a/test/unit/app/jobs/test_gcp_batch_runner.py b/test/unit/app/jobs/test_gcp_batch_runner.py index 6d9b019ac3d..b75a4a85465 100644 --- a/test/unit/app/jobs/test_gcp_batch_runner.py +++ b/test/unit/app/jobs/test_gcp_batch_runner.py @@ -9,7 +9,10 @@ from typing import ( import pytest from galaxy.jobs.runners import RunnerParams -from galaxy.jobs.runners.gcp_batch import GoogleCloudBatchJobRunner +from galaxy.jobs.runners.gcp_batch import ( + GoogleCloudBatchJobRunner, + RUNNER_PARAM_SPECS, +) from galaxy.jobs.runners.util.gcp_batch import ( convert_cpu_to_milli, convert_duration_to_seconds, @@ -357,3 +360,70 @@ class TestMonitorSleepTime: # ParamsWithSpecs.__missing__; .get() would yield None and raise in max(). runner = _sleep_runner(RunnerParams(specs={"polling_interval": dict(map=int, default=30)}, params={})) assert runner.monitor_sleep_time == 30 + + +def _make_runner(runner_params=None): + """Build a GoogleCloudBatchJobRunner without running __init__ (no GCP client). + + _get_job_params only depends on self.runner_params, so we set that directly. + """ + runner = object.__new__(GoogleCloudBatchJobRunner) + runner.runner_params = RunnerParams(specs=RUNNER_PARAM_SPECS, params=runner_params or {}) + return runner + + +# Keys that _get_job_params copies from the destination / runner config. Derived +# dynamically (rather than hard-coded) so the parametrized tests below automatically +# cover any parameter added to _get_job_params in the future. +JOB_PARAM_KEYS = sorted(_make_runner()._get_job_params(SimpleNamespace(params={})).keys()) + + +class TestGetJobParams: + """Tests for GoogleCloudBatchJobRunner._get_job_params default resolution.""" + + @pytest.mark.parametrize("key", JOB_PARAM_KEYS) + def test_unset_param_falls_back_to_spec_default(self, key): + """Every copied parameter resolves to its spec default when nothing overrides it. + + This is the regression guard: .get() on the RunnerParams defaultdict would + bypass __missing__ and yield None instead of the configured default. + """ + runner = _make_runner() + destination = SimpleNamespace(params={}) + + params = runner._get_job_params(destination) + + assert params[key] == RUNNER_PARAM_SPECS[key]["default"] + + @pytest.mark.parametrize("key", JOB_PARAM_KEYS) + def test_destination_overrides_every_param(self, key): + """A value on the job destination takes precedence over the spec default for every param. + + Destination params are not passed through the RunnerParams spec mapping, so a + plain string sentinel is a valid override for every key. + """ + runner = _make_runner() + sentinel = "destination-sentinel-value" + destination = SimpleNamespace(params={key: sentinel}) + + params = runner._get_job_params(destination) + + assert params[key] == sentinel + + def test_runner_config_overrides_default(self): + """A value set in the runner (plugin) config is used when the destination is silent.""" + runner = _make_runner({"job_id_prefix": "from-config"}) + destination = SimpleNamespace(params={}) + + params = runner._get_job_params(destination) + + assert params["job_id_prefix"] == "from-config" + + def test_destination_overrides_runner_config(self): + """Destination params win over runner config, which wins over the spec default.""" + runner = _make_runner({"job_id_prefix": "from-config"}) + destination = SimpleNamespace(params={"job_id_prefix": "from-destination"}) + + params = runner._get_job_params(destination) + + assert params["job_id_prefix"] == "from-destination"