feat: fail fast at preflight when job_definition and job_role are combined

A pre-existing job definition (--aws-batch-job-definition) bakes in its own job
role, so combining it with --aws-batch-job-role can never take effect. Reject
the combination at the top of _preflight_validate — before _validate_job_role
would waste an iam:GetRole call on a role that will be ignored.

The per-job check in BatchJobBuilder is retained: a per-rule
aws_batch_job_definition resource is not visible at preflight time and can only
be caught at submission time.

Author: nh13

snakemake_executor_plugin_aws_batch/__init__.py

@@ -349,10 +349,26 @@ def _preflight_validate(self) -> None:
 
         Best-effort about *uncertainty*: a transient API error or a missing
         describe/iam permission degrades to a warning and the workflow proceeds.
-        Only a confirmed-bad configuration (disabled/invalid queue or compute
-        environment, ``maxvCpus=0``, or a non-existent job role) raises, before
-        any job is submitted.
+        Only a confirmed-bad configuration raises, before any job is submitted:
+        combining ``--aws-batch-job-role`` with ``--aws-batch-job-definition``
+        (checked first), a disabled/invalid queue or compute environment,
+        ``maxvCpus=0``, or a non-existent job role.
         """
+        # Reject --aws-batch-job-role + --aws-batch-job-definition early, before
+        # _validate_job_role wastes an iam:GetRole call on a role that can never
+        # be used: a pre-existing job definition bakes in its own role. The
+        # per-job check in BatchJobBuilder must remain — a per-rule
+        # aws_batch_job_definition resource is not visible here at preflight time.
+        if getattr(self.settings, "job_definition", None) and getattr(
+            self.settings, "job_role", None
+        ):
+            raise WorkflowError(
+                "Cannot combine --aws-batch-job-role with "
+                "--aws-batch-job-definition: the job role is baked into a "
+                "pre-existing job definition and cannot be overridden per job. "
+                "Remove --aws-batch-job-role or omit --aws-batch-job-definition."
+            )
+
         problems = self._queue_problems()
         if problems:
             raise WorkflowError(

tests/test_preflight.py

@@ -239,6 +239,49 @@ def test_no_queue_configured_does_not_raise(self):
         ex._validate_job_role.assert_called_once()
 
 
+class TestPreflightJobDefinitionJobRoleCombo:
+    """`job_definition` + `job_role` must fail fast, before any queue/role call.
+
+    A pre-existing job definition bakes in its own role, so combining it with
+    `--aws-batch-job-role` can never work. The per-rule resource is caught later
+    in BatchJobBuilder; this covers the setting-level combo at preflight.
+    """
+
+    def test_both_set_raises_before_queue_or_role_checks(self):
+        ex = _executor(
+            job_definition="my-job-def",
+            job_role="arn:aws:iam::1:role/my-role",
+        )
+        ex._queue_problems = MagicMock()
+        ex._validate_job_role = MagicMock()
+        with pytest.raises(WorkflowError) as excinfo:
+            ex._preflight_validate()
+        msg = str(excinfo.value)
+        assert "--aws-batch-job-role" in msg
+        assert "--aws-batch-job-definition" in msg
+        # Fail fast: neither the queue describe nor the iam:GetRole check runs.
+        ex._queue_problems.assert_not_called()
+        ex._validate_job_role.assert_not_called()
+
+    def test_job_definition_only_does_not_raise(self):
+        ex = _with_queue(
+            _executor(job_definition="my-job-def"),
+            {"state": "ENABLED", "status": "VALID", "computeEnvironmentOrder": []},
+        )
+        ex._validate_job_role = MagicMock()
+        ex._preflight_validate()  # must not raise
+        ex._validate_job_role.assert_called_once()
+
+    def test_job_role_only_preserves_existing_behaviour(self):
+        ex = _with_queue(
+            _executor(job_role="arn:aws:iam::1:role/my-role"),
+            {"state": "ENABLED", "status": "VALID", "computeEnvironmentOrder": []},
+        )
+        ex._validate_job_role = MagicMock()
+        ex._preflight_validate()  # must not raise
+        ex._validate_job_role.assert_called_once()
+
+
 class TestValidateJobRole:
     def _iam(self, **get_role_kwargs) -> MagicMock:
         return MagicMock(get_role=MagicMock(**get_role_kwargs))