Add unit tests with a regression check

2026-09-24 16:30:27 +08:00 · 2026-06-11 11:55:11 -04:00
parent ad8d849718
commit cacaebb283
2 changed files with 122 additions and 49 deletions
@@ -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
@@ -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"