feat: new improved cli validation

Author: cmeesters

snakemake_executor_plugin_slurm/__init__.py

@@ -46,7 +46,11 @@
 from .efficiency_report import create_efficiency_report
 from .submit_string import get_submit_command
 from .partitions import read_partition_file, get_best_partition
-from .validation import validate_slurm_extra, validate_executor_settings
+from .validation import (
+    validate_slurm_extra,
+    validate_executor_settings,
+    validate_status_command_settings,
+)
 
 
 def _get_status_command_default():
@@ -266,8 +270,7 @@ class ExecutorSettings(ExecutorSettingsBase):
 
     def __post_init__(self):
         """Validate settings after initialization."""
-        # Run all validation checks
-        validate_executor_settins(self)
+        validate_executor_settings(self)
 
 
 # Required:
@@ -329,61 +332,10 @@ def __post_init__(self, test_mode: bool = False):
                 else None
             )
         atexit.register(self.clean_old_logs)
-
-        # Validate status_command configuration if the field exists
-        self._validate_status_command_settings()
-
-    def _validate_status_command_settings(self):
-        """Validate and provide feedback about status_command configuration."""
-        if hasattr(self.workflow.executor_settings, "status_command"):
-            status_command = self.workflow.executor_settings.status_command
-            if status_command:
-                min_job_age = get_min_job_age()
-                sacct_available = is_query_tool_available("sacct")
-
-                # Threshold: 3x initial status check interval (default 40s)
-                initial_interval = getattr(
-                    self.workflow.executor_settings,
-                    "init_seconds_before_status_checks",
-                    40,
-                )
-                dynamic_check_threshold = 3 * initial_interval
-
-                if not sacct_available and status_command == "sacct":
-                    self.logger.warning(
-                        "The 'sacct' command is not available on this system. "
-                        "Using 'squeue' instead for job status queries."
-                    )
-                elif sacct_available and min_job_age is not None:
-                    if (
-                        min_job_age < dynamic_check_threshold
-                        and status_command == "squeue"
-                    ):
-                        self.logger.warning(
-                            f"MinJobAge is {min_job_age} seconds "
-                            f"(< {dynamic_check_threshold}s). "
-                            f"This may cause 'squeue' to miss recently finished jobs "
-                            "that have been purged from slurmctld, leading to job "
-                            "status queries being impossible with 'squeue'. "
-                            "Consider using 'sacct' instead or let your admini- "
-                            "strator increase MinJobAge. "
-                            "(Threshold is 3x status check interval: 3 x "
-                            f"{initial_interval}s = "
-                            f"{dynamic_check_threshold}s)"
-                        )
-                    elif (
-                        min_job_age >= dynamic_check_threshold
-                        and status_command == "sacct"
-                    ):
-                        self.logger.warning(
-                            f"MinJobAge is {min_job_age} seconds (>= "
-                            f"{dynamic_check_threshold}s). "
-                            f"The 'squeue' command should work reliably for "
-                            "status queries. "
-                            "(Threshold is 3x status check interval: 3 x "
-                            f"{initial_interval}s = "
-                            f"{dynamic_check_threshold}s)"
-                        )
+        # moved validation to validation.py
+        validate_status_command_settings(
+            self.workflow.executor_settings, self.logger
+        )
 
     def get_status_command(self):
         """Get the status command to use, with fallback logic."""
@@ -654,13 +606,32 @@ async def check_active_jobs(
         missing_sacct_status = set()
 
         # decide which status command to use
-        status_command = self.get_status_command()
-        # Getting the actual command with parameters.
-        # Here, the command will be a list generated with
-        # shlex.split().
-        if status_command == "sacct":
+        status_command_name = self.get_status_command()
+        min_job_age = get_min_job_age()
+        initial_interval = getattr(
+            self.workflow.executor_settings,
+            "init_seconds_before_status_checks",
+            40,
+        )
+        dynamic_check_threshold = 3 * initial_interval
+        if status_command_name == "squeue":
+            if (
+                min_job_age is None
+                or min_job_age < dynamic_check_threshold
+            ) and is_query_tool_available("sacct"):
+                self.logger.info(
+                    "Falling back to 'sacct' for status queries "
+                    f"(MinJobAge={min_job_age}; threshold={dynamic_check_threshold}s)."
+                )
+                status_command_name = "sacct"
+        if status_command_name == "sacct" and not is_query_tool_available("sacct"):
+            self.logger.info(
+                "'sacct' unavailable, using 'squeue' for status queries."
+            )
+            status_command_name = "squeue"
+        if status_command_name == "sacct":
             status_command = query_job_status_sacct(self.run_uuid)
-        elif status_command == "squeue":
+        else:
             status_command = query_job_status_squeue(self.run_uuid)
 
         # this code is inspired by the snakemake profile:

snakemake_executor_plugin_slurm/validation.py

@@ -5,6 +5,7 @@
 import re
 from pathlib import Path
 from snakemake_interface_common.exceptions import WorkflowError
+from .job_status_query import get_min_job_age, is_query_tool_available
 
 
 def get_forbidden_slurm_options():
@@ -76,33 +77,58 @@ def validate_slurm_extra(job):
             )
 
 
-def validate_executor_settings(settings):
+def validate_status_command_settings(settings, logger):
+    """Emit warnings about status_command sensibility."""
+    if not hasattr(settings, "status_command"):
+        return
+    status_command = settings.status_command
+    if not status_command:
+        return
+    min_job_age = get_min_job_age()
+    sacct_available = is_query_tool_available("sacct")
+    initial_interval = getattr(
+        settings,
+        "init_seconds_before_status_checks",
+        40,
+    )
+    dynamic_check_threshold = 3 * initial_interval
+    if not sacct_available and status_command == "sacct":
+        logger.warning(
+            "The 'sacct' command is not available. Falling back to 'squeue'."
+        )
+    elif sacct_available and min_job_age is not None:
+        if min_job_age < dynamic_check_threshold and status_command == "squeue":
+            logger.warning(
+                f"MinJobAge {min_job_age}s (< {dynamic_check_threshold}s). "
+                "This may cause 'squeue' to miss recently finished jobs. "
+                "Consider using 'sacct' or increasing MinJobAge."
+            )
+        elif min_job_age >= dynamic_check_threshold and status_command == "sacct":
+            logger.warning(
+                f"MinJobAge {min_job_age}s (>= {dynamic_check_threshold}s). "
+                "'squeue' should work reliably for status queries."
+            )
+
+
+def validate_executor_settings(settings, logger=None):
     """
     Validate ExecutorSettings fields for correctness
     (better user feedback in case of wrong inputs)
-
-    Args:
-        settings: ExecutorSettins instance to validate
-
-    Raises:
-        WorkflowError - if any validation check fails
     """
     # status_command: only allow known values
     if settings.status_command is not None:
         if settings.status_command not in {"sacct", "squeue"}:
             raise WorkflowError(
                 "Invalid status command. Allowed values are 'sacct' or 'squeue'."
             )
-
-    # status_attempts: require positive integer
+    # status_attempts
     if settings.status_attempts is not None:
         if (
             not isinstance(settings.status_attempts, int)
             or settings.status_attempts < 1
         ):
             raise WorkflowError("status_attempts must be a positive integer")
-
-    # init_settings_befor_status_checks: require non-negative integer
+    # init_seconds_before_status_checks
     if settings.init_seconds_before_status_checks is not None:
         if (
             not isinstance(settings.init_seconds_before_status_checks, int)
@@ -111,8 +137,7 @@ def validate_executor_settings(settings):
             raise WorkflowError(
                 "init-seconds-before-status-checks must be a positive integer."
             )
-
-    # efficiency_threshold, if set must be within (0, 1]
+    # efficiency_threshold
     if settings.efficiency_threshold is not None:
         try:
             thr = float(settings.efficiency_threshold)
@@ -124,18 +149,19 @@ def validate_executor_settings(settings):
             raise WorkflowError(
                 "efficiency-threshold must be a number in range (0, 1]."
             )
-
-    # partition_config: if provided, it must exist
+    # partition_config
     if settings.partition_config is not None:
         p = Path(settings.partition_config)
         if not p.exists():
             raise WorkflowError(
                 f"Partition configuration file not found, given was {p}."
             )
-
-    # delete_logfiles_older_than: if provided, should be reasonable
+    # delete_logfiles_older_than
     if settings.delete_logfiles_older_than is not None:
         if not isinstance(settings.delete_logfiles_older_than, int):
             raise WorkflowError(
                 "delete-logfiles-older-than must be an integer (days)."
             )
+    # status_command warnings (optional logger)
+    if logger:
+        validate_status_command_settings(settings, logger)