Runtime value checking for LaunchSettings and BatchSettings

smartsim/settings/arguments/batch/lsf.py

@@ -57,8 +57,11 @@ def set_walltime(self, walltime: str) -> None:
         :param walltime: Time in hh:mm format, e.g. "10:00" for 10 hours,
                          if time is supplied in hh:mm:ss format, seconds
                          will be ignored and walltime will be set as ``hh:mm``
+        :raises TypeError: if not type str
         """
         # For compatibility with other launchers, as explained in docstring
+        if not isinstance(walltime, str):
+            raise TypeError("walltime argument was not of type str")
         if walltime:
             if len(walltime.split(":")) > 2:
                 walltime = ":".join(walltime.split(":")[:2])
@@ -72,7 +75,13 @@ def set_smts(self, smts: int) -> None:
         takes precedence.
 
         :param smts: SMT (e.g on Summit: 1, 2, or 4)
+        :raises TypeError: if not an int
+        :raises ValueError: if not positive int
         """
+        if not isinstance(smts, int):
+            raise TypeError("smts argument was not of type int")
+        if smts <= 0:
+            raise ValueError("smts must be a positive value")
         self.set("alloc_flags", str(smts))
 
     def set_project(self, project: str) -> None:
@@ -81,7 +90,10 @@ def set_project(self, project: str) -> None:
         This sets ``-P``.
 
         :param time: project name
+        :raises TypeError: if not a str
         """
+        if not isinstance(project, str):
+            raise TypeError("project argument was not of type str")
         self.set("P", project)
 
     def set_account(self, account: str) -> None:
@@ -90,7 +102,10 @@ def set_account(self, account: str) -> None:
         this function is an alias for `set_project`.
 
         :param account: project name
+        :raises TypeError: if not a str
         """
+        if not isinstance(account, str):
+            raise TypeError("account argument was not of type str")
         return self.set_project(account)
 
     def set_nodes(self, num_nodes: int) -> None:
@@ -99,7 +114,13 @@ def set_nodes(self, num_nodes: int) -> None:
         This sets ``-nnodes``.
 
         :param nodes: number of nodes
+        :raises TypeError: if not an int
+        :raises ValueError: if not positive int
         """
+        if not isinstance(num_nodes, int):
+            raise TypeError("num_nodes argument was not of type int")
+        if num_nodes <= 0:
+            raise ValueError("Number of nodes must be a positive value")
         self.set("nnodes", str(num_nodes))
 
     def set_hostlist(self, host_list: t.Union[str, t.List[str]]) -> None:
@@ -110,10 +131,11 @@ def set_hostlist(self, host_list: t.Union[str, t.List[str]]) -> None:
         """
         if isinstance(host_list, str):
             host_list = [host_list.strip()]
-        if not isinstance(host_list, list):
+        if not (
+            isinstance(host_list, list)
+            and all(isinstance(item, str) for item in host_list)
+        ):
             raise TypeError("host_list argument must be a list of strings")
-        if not all(isinstance(host, str) for host in host_list):
-            raise TypeError("host_list argument must be list of strings")
         self.set("m", '"' + " ".join(host_list) + '"')
 
     def set_tasks(self, tasks: int) -> None:
@@ -122,7 +144,13 @@ def set_tasks(self, tasks: int) -> None:
         This sets ``-n``
 
         :param tasks: number of tasks
+        :raises TypeError: if not an int
+        :raises ValueError: if not positive int
         """
+        if not isinstance(tasks, int):
+            raise TypeError("tasks argument was not of type int")
+        if tasks <= 0:
+            raise ValueError("Number of tasks must be a positive value")
         self.set("n", str(tasks))
 
     def set_queue(self, queue: str) -> None:
@@ -131,7 +159,10 @@ def set_queue(self, queue: str) -> None:
         This sets ``-q``
 
         :param queue: The queue to submit the job on
+        :raises TypeError: if not a str
         """
+        if not isinstance(queue, str):
+            raise TypeError("queue argument was not of type str")
         self.set("q", queue)
 
     def format_batch_args(self) -> t.List[str]:

smartsim/settings/arguments/batch/pbs.py

@@ -26,6 +26,7 @@
 
 from __future__ import annotations
 
+import re
 import typing as t
 from copy import deepcopy
 
@@ -61,7 +62,13 @@ def set_nodes(self, num_nodes: int) -> None:
         nodes here is sets the 'nodes' resource.
 
         :param num_nodes: number of nodes
+        :raises TypeError: if not an int
+        :raises ValueError: if not positive int
         """
+        if not isinstance(num_nodes, int):
+            raise TypeError("num_nodes argument was not of type int")
+        if num_nodes <= 0:
+            raise ValueError("Number of nodes must be a positive value")
 
         self.set("nodes", str(num_nodes))
 
@@ -73,9 +80,10 @@ def set_hostlist(self, host_list: t.Union[str, t.List[str]]) -> None:
         """
         if isinstance(host_list, str):
             host_list = [host_list.strip()]
-        if not isinstance(host_list, list):
-            raise TypeError("host_list argument must be a list of strings")
-        if not all(isinstance(host, str) for host in host_list):
+        if not (
+            isinstance(host_list, list)
+            and all(isinstance(item, str) for item in host_list)
+        ):
             raise TypeError("host_list argument must be a list of strings")
         self.set("hostname", ",".join(host_list))
 
@@ -89,15 +97,26 @@ def set_walltime(self, walltime: str) -> None:
         this value will be overridden
 
         :param walltime: wall time
+        :raises ValueError: if walltime format is invalid
+        :raises TypeError: if not type str
         """
-        self.set("walltime", walltime)
+        if not isinstance(walltime, str):
+            raise TypeError("walltime argument was not of type str")
+        pattern = r"^\d{2}:\d{2}:\d{2}$"
+        if walltime and re.match(pattern, walltime):
+            self.set("walltime", walltime)
+        else:
+            raise ValueError("Invalid walltime format. Please use 'HH:MM:SS' format.")
 
     def set_queue(self, queue: str) -> None:
         """Set the queue for the batch job
 
         :param queue: queue name
+        :raises TypeError: if not a str
         """
-        self.set("q", str(queue))
+        if not isinstance(queue, str):
+            raise TypeError("queue argument was not of type str")
+        self.set("q", queue)
 
     def set_ncpus(self, num_cpus: int) -> None:
         """Set the number of cpus obtained in each node.
@@ -107,15 +126,24 @@ def set_ncpus(self, num_cpus: int) -> None:
         this value will be overridden
 
         :param num_cpus: number of cpus per node in select
+        :raises TypeError: if not an int
+        :raises ValueError: if not positive int
         """
+        if not isinstance(num_cpus, int):
+            raise TypeError("num_cpus argument was not of type int")
+        if num_cpus <= 0:
+            raise ValueError("Number of CPUs must be a positive value")
         self.set("ppn", str(num_cpus))
 
     def set_account(self, account: str) -> None:
         """Set the account for this batch job
 
         :param acct: account id
+        :raises TypeError: if not a str
         """
-        self.set("A", str(account))
+        if not isinstance(account, str):
+            raise TypeError("account argument was not of type str")
+        self.set("A", account)
 
     def format_batch_args(self) -> t.List[str]:
         """Get the formatted batch arguments for a preview

smartsim/settings/arguments/batch/slurm.py

@@ -56,7 +56,11 @@ def set_walltime(self, walltime: str) -> None:
         format = "HH:MM:SS"
 
         :param walltime: wall time
+        :raises ValueError: if walltime format is invalid
+        :raises TypeError: if not str
         """
+        if not isinstance(walltime, str):
+            raise TypeError("walltime argument was not of type str")
         pattern = r"^\d{2}:\d{2}:\d{2}$"
         if walltime and re.match(pattern, walltime):
             self.set("time", str(walltime))
@@ -69,7 +73,13 @@ def set_nodes(self, num_nodes: int) -> None:
         This sets ``--nodes``.
 
         :param num_nodes: number of nodes
+        :raises TypeError: if not an int
+        :raises ValueError: if not positive int
         """
+        if not isinstance(num_nodes, int):
+            raise TypeError("num_nodes argument was not of type int")
+        if num_nodes <= 0:
+            raise ValueError("Number of nodes must be a positive value")
         self.set("nodes", str(num_nodes))
 
     def set_account(self, account: str) -> None:
@@ -78,7 +88,10 @@ def set_account(self, account: str) -> None:
         This sets ``--account``.
 
         :param account: account id
+        :raises TypeError: if not a str
         """
+        if not isinstance(account, str):
+            raise TypeError("account argument was not of type str")
         self.set("account", account)
 
     def set_partition(self, partition: str) -> None:
@@ -87,16 +100,22 @@ def set_partition(self, partition: str) -> None:
         This sets ``--partition``.
 
         :param partition: partition name
+        :raises TypeError: if not a str
         """
-        self.set("partition", str(partition))
+        if not isinstance(partition, str):
+            raise TypeError("partition argument was not of type str")
+        self.set("partition", partition)
 
     def set_queue(self, queue: str) -> None:
         """alias for set_partition
 
         Sets the partition for the slurm batch job
 
         :param queue: the partition to run the batch job on
+        :raises TypeError: if not a str
         """
+        if not isinstance(queue, str):
+            raise TypeError("queue argument was not of type str")
         return self.set_partition(queue)
 
     def set_cpus_per_task(self, cpus_per_task: int) -> None:
@@ -105,7 +124,13 @@ def set_cpus_per_task(self, cpus_per_task: int) -> None:
         This sets ``--cpus-per-task``
 
         :param num_cpus: number of cpus to use per task
+        :raises TypeError: if not int
+        :raises ValueError: if not positive int
         """
+        if not isinstance(cpus_per_task, int):
+            raise TypeError("cpus_per_task argument was not of type int")
+        if cpus_per_task <= 0:
+            raise ValueError("CPUs per task must be a positive value")
         self.set("cpus-per-task", str(cpus_per_task))
 
     def set_hostlist(self, host_list: t.Union[str, t.List[str]]) -> None:
@@ -118,10 +143,13 @@ def set_hostlist(self, host_list: t.Union[str, t.List[str]]) -> None:
         """
         if isinstance(host_list, str):
             host_list = [host_list.strip()]
-        if not isinstance(host_list, list):
+
+        if not (
+            isinstance(host_list, list)
+            and all(isinstance(item, str) for item in host_list)
+        ):
             raise TypeError("host_list argument must be a list of strings")
-        if not all(isinstance(host, str) for host in host_list):
-            raise TypeError("host_list argument must be list of strings")
+
         self.set("nodelist", ",".join(host_list))
 
     def format_batch_args(self) -> t.List[str]:

smartsim/settings/batch_settings.py

@@ -117,6 +117,18 @@
             """The scheduler type"""
         except ValueError:
             raise ValueError(f"Invalid scheduler type: {batch_scheduler}") from None
+
+        if batch_args is not None:
+            if not (
+                isinstance(batch_args, dict)
+                and all(
+                    isinstance(key, str) and isinstance(val, (str, type(None)))
+                    for key, val in batch_args.items()
+                )
+            ):
+                raise TypeError(
+                    "batch_args argument was not of type dict of str and str or None"
+                )
         self._arguments = self._get_arguments(batch_args)
         """The BatchSettings child class based on scheduler type"""
         self.env_vars = env_vars or {}
@@ -140,6 +152,16 @@
     @env_vars.setter
     def env_vars(self, value: t.Dict[str, str | None]) -> None:
         """Set the environment variables."""
+
+        if not (
+            isinstance(value, t.Dict)
+            and all(
+                isinstance(key, str) and isinstance(val, (str, type(None)))
+                for key, val in value.items()
+            )
+        ):
+            raise TypeError("env_vars argument was not of type dict of str and str")
+
         self._env_vars = copy.deepcopy(value)
 
     def _get_arguments(self, batch_args: StringArgument | None) -> BatchArguments:

smartsim/settings/launch_settings.py

@@ -126,6 +126,18 @@
             """The launcher type"""
         except ValueError:
             raise ValueError(f"Invalid launcher type: {launcher}")
+
+        if launch_args is not None:
+            if not (
+                isinstance(launch_args, t.Dict)
+                and all(
+                    isinstance(key, str) and isinstance(val, (str, type(None)))
+                    for key, val in launch_args.items()
+                )
+            ):
+                raise TypeError(
+                    "launch_args argument was not of type dict of str and str or None"
+                )
         self._arguments = self._get_arguments(launch_args)
         """The LaunchSettings child class based on launcher type"""
         self.env_vars = env_vars or {}
@@ -165,6 +177,15 @@
 
         :param value: The new environment mapping
         """
+        if not (
+            isinstance(value, dict)
+            and all(
+                isinstance(key, str) and isinstance(val, (str, type(None)))
+                for key, val in value.items()
+            )
+        ):
+            raise TypeError("env_vars argument was not of type dict of str and str")
+
         self._env_vars = copy.deepcopy(value)
 
     def _get_arguments(self, launch_args: StringArgument | None) -> LaunchArguments:
@@ -209,6 +230,7 @@
         :param env_vars: environment variables to update or add
         :raises TypeError: if env_vars values cannot be coerced to strings
         """
+
         # Coerce env_vars values to str as a convenience to user
         for env, val in env_vars.items():
             if not isinstance(env, str):