Merge pull request #336 from pyiron/validate

Move validation routines to separate submodule

Author: jan-janssen

pysqa/utils/basic.py

@@ -4,7 +4,6 @@
 import getpass
 import importlib
 import os
-import re
 from typing import List, Optional, Tuple, Union
 
 import pandas
@@ -13,6 +12,7 @@
 
 from pysqa.utils.execute import execute_command
 from pysqa.utils.queues import Queues
+from pysqa.utils.validate import value_error_if_none, value_in_range
 
 
 class BasisQueueAdapter(object):
@@ -372,15 +372,15 @@ def check_queue_parameters(
         """
         if active_queue is None:
             active_queue = self._config["queues"][queue]
-        cores = self._value_in_range(
+        cores = value_in_range(
             value=cores,
             value_min=active_queue["cores_min"],
             value_max=active_queue["cores_max"],
         )
-        run_time_max = self._value_in_range(
+        run_time_max = value_in_range(
             value=run_time_max, value_max=active_queue["run_time_max"]
         )
-        memory_max = self._value_in_range(
+        memory_max = value_in_range(
             value=memory_max, value_max=active_queue["memory_max"]
         )
         return cores, run_time_max, memory_max
@@ -465,7 +465,7 @@ def _job_submission_template(
         """
         if queue is None:
             queue = self._config["queue_primary"]
-        self._value_error_if_none(value=command)
+        value_error_if_none(value=command)
         if queue not in self.queue_list:
             raise ValueError(
                 "The queue "
@@ -567,113 +567,3 @@ def _load_templates(queue_lst_dict: dict, directory: str = ".") -> None:
                             + error.message,
                             lineno=error.lineno,
                         )
-
-    @staticmethod
-    def _value_error_if_none(value: str) -> None:
-        """
-        Raise a ValueError if the value is None or not a string.
-
-        Args:
-            value (str/None): The value to check.
-
-        Raises:
-            ValueError: If the value is None.
-            TypeError: If the value is not a string.
-        """
-        if value is None:
-            raise ValueError("Value cannot be None.")
-        if not isinstance(value, str):
-            raise TypeError()
-
-    @classmethod
-    def _value_in_range(
-        cls,
-        value: Union[int, float, None],
-        value_min: Union[int, float, None] = None,
-        value_max: Union[int, float, None] = None,
-    ) -> Union[int, float, None]:
-        """
-        Check if a value is within a specified range.
-
-        Args:
-            value (int/float/None): The value to check.
-            value_min (int/float/None): The minimum value. Defaults to None.
-            value_max (int/float/None): The maximum value. Defaults to None.
-
-        Returns:
-            int/float/None: The value if it is within the range, otherwise the minimum or maximum value.
-        """
-
-        if value is not None:
-            value_, value_min_, value_max_ = [
-                (
-                    cls._memory_spec_string_to_value(v)
-                    if v is not None and isinstance(v, str)
-                    else v
-                )
-                for v in (value, value_min, value_max)
-            ]
-            # ATTENTION: '60000' is interpreted as '60000M' since default magnitude is 'M'
-            # ATTENTION: int('60000') is interpreted as '60000B' since _memory_spec_string_to_value return the size in
-            # ATTENTION: bytes, as target_magnitude = 'b'
-            # We want to compare the the actual (k,m,g)byte value if there is any
-            if value_min_ is not None and value_ < value_min_:
-                return value_min
-            if value_max_ is not None and value_ > value_max_:
-                return value_max
-            return value
-        else:
-            if value_min is not None:
-                return value_min
-            if value_max is not None:
-                return value_max
-            return value
-
-    @staticmethod
-    def _is_memory_string(value: str) -> bool:
-        """
-        Check if a string specifies a certain amount of memory.
-
-        Args:
-            value (str): The string to check.
-
-        Returns:
-            bool: True if the string matches a memory specification, False otherwise.
-        """
-        memory_spec_pattern = r"[0-9]+[bBkKmMgGtT]?"
-        return re.findall(memory_spec_pattern, value)[0] == value
-
-    @classmethod
-    def _memory_spec_string_to_value(
-        cls, value: str, default_magnitude: str = "m", target_magnitude: str = "b"
-    ) -> Union[int, float]:
-        """
-        Converts a valid memory string (tested by _is_memory_string) into an integer/float value of desired
-        magnitude `default_magnitude`. If it is a plain integer string (e.g.: '50000') it will be interpreted with
-        the magnitude passed in by the `default_magnitude`. The output will rescaled to `target_magnitude`
-
-        Args:
-            value (str): The string to convert.
-            default_magnitude (str): The magnitude for interpreting plain integer strings [b, B, k, K, m, M, g, G, t, T]. Defaults to "m".
-            target_magnitude (str): The magnitude to which the output value should be converted [b, B, k, K, m, M, g, G, t, T]. Defaults to "b".
-
-        Returns:
-            Union[int, float]: The value of the string in `target_magnitude` units.
-        """
-        magnitude_mapping = {"b": 0, "k": 1, "m": 2, "g": 3, "t": 4}
-        if cls._is_memory_string(value):
-            integer_pattern = r"[0-9]+"
-            magnitude_pattern = r"[bBkKmMgGtT]+"
-            integer_value = int(re.findall(integer_pattern, value)[0])
-
-            magnitude = re.findall(magnitude_pattern, value)
-            if len(magnitude) > 0:
-                magnitude = magnitude[0].lower()
-            else:
-                magnitude = default_magnitude.lower()
-            # Convert it to default magnitude = megabytes
-            return (integer_value * 1024 ** magnitude_mapping[magnitude]) / (
-                1024 ** magnitude_mapping[target_magnitude]
-            )
-        else:
-            return value

pysqa/utils/validate.py

@@ -0,0 +1,111 @@
+import re
+from typing import Union
+
+
+def value_error_if_none(value: str) -> None:
+    """
+    Raise a ValueError if the value is None or not a string.
+
+    Args:
+        value (str/None): The value to check.
+
+    Raises:
+        ValueError: If the value is None.
+        TypeError: If the value is not a string.
+    """
+    if value is None:
+        raise ValueError("Value cannot be None.")
+    if not isinstance(value, str):
+        raise TypeError()
+
+
+def value_in_range(
+    value: Union[int, float, None],
+    value_min: Union[int, float, None] = None,
+    value_max: Union[int, float, None] = None,
+) -> Union[int, float, None]:
+    """
+    Check if a value is within a specified range.
+
+    Args:
+        value (int/float/None): The value to check.
+        value_min (int/float/None): The minimum value. Defaults to None.
+        value_max (int/float/None): The maximum value. Defaults to None.
+
+    Returns:
+        int/float/None: The value if it is within the range, otherwise the minimum or maximum value.
+    """
+
+    if value is not None:
+        value_, value_min_, value_max_ = [
+            (
+                _memory_spec_string_to_value(v)
+                if v is not None and isinstance(v, str)
+                else v
+            )
+            for v in (value, value_min, value_max)
+        ]
+        # ATTENTION: '60000' is interpreted as '60000M' since default magnitude is 'M'
+        # ATTENTION: int('60000') is interpreted as '60000B' since _memory_spec_string_to_value return the size in
+        # ATTENTION: bytes, as target_magnitude = 'b'
+        # We want to compare the the actual (k,m,g)byte value if there is any
+        if value_min_ is not None and value_ < value_min_:
+            return value_min
+        if value_max_ is not None and value_ > value_max_:
+            return value_max
+        return value
+    else:
+        if value_min is not None:
+            return value_min
+        if value_max is not None:
+            return value_max
+        return value
+
+
+def _is_memory_string(value: str) -> bool:
+    """
+    Check if a string specifies a certain amount of memory.
+
+    Args:
+        value (str): The string to check.
+
+    Returns:
+        bool: True if the string matches a memory specification, False otherwise.
+    """
+    memory_spec_pattern = r"[0-9]+[bBkKmMgGtT]?"
+    return re.findall(memory_spec_pattern, value)[0] == value
+
+
+def _memory_spec_string_to_value(
+    value: str, default_magnitude: str = "m", target_magnitude: str = "b"
+) -> Union[int, float]:
+    """
+    Converts a valid memory string (tested by _is_memory_string) into an integer/float value of desired
+    magnitude `default_magnitude`. If it is a plain integer string (e.g.: '50000') it will be interpreted with
+    the magnitude passed in by the `default_magnitude`. The output will rescaled to `target_magnitude`
+
+    Args:
+        value (str): The string to convert.
+        default_magnitude (str): The magnitude for interpreting plain integer strings [b, B, k, K, m, M, g, G, t, T]. Defaults to "m".
+        target_magnitude (str): The magnitude to which the output value should be converted [b, B, k, K, m, M, g, G, t, T]. Defaults to "b".
+
+    Returns:
+        Union[int, float]: The value of the string in `target_magnitude` units.
+    """
+    magnitude_mapping = {"b": 0, "k": 1, "m": 2, "g": 3, "t": 4}
+    if _is_memory_string(value):
+        integer_pattern = r"[0-9]+"
+        magnitude_pattern = r"[bBkKmMgGtT]+"
+        integer_value = int(re.findall(integer_pattern, value)[0])
+
+        magnitude = re.findall(magnitude_pattern, value)
+        if len(magnitude) > 0:
+            magnitude = magnitude[0].lower()
+        else:
+            magnitude = default_magnitude.lower()
+        # Convert it to default magnitude = megabytes
+        return (integer_value * 1024 ** magnitude_mapping[magnitude]) / (
+            1024 ** magnitude_mapping[target_magnitude]
+        )
+    else:
+        return value

tests/test_basic.py

@@ -6,6 +6,7 @@
 from jinja2.exceptions import TemplateSyntaxError
 from pysqa import QueueAdapter
 from pysqa.utils.basic import BasisQueueAdapter
+from pysqa.utils.validate import value_in_range
 
 __author__ = "Jan Janssen"
 __copyright__ = "Copyright 2019, Jan Janssen"
@@ -36,45 +37,35 @@ def test_bad_queue_type(self):
             BasisQueueAdapter(config={"queue_type": "error", "queues": {}})
 
     def test_memory_string_comparison(self):
-        self.assertEqual(BasisQueueAdapter._value_in_range(1023, value_min="1K"), "1K")
-        self.assertEqual(BasisQueueAdapter._value_in_range(1035, value_min="1K"), 1035)
-        self.assertEqual(BasisQueueAdapter._value_in_range(1035, value_max="1K"), "1K")
+        self.assertEqual(value_in_range(1023, value_min="1K"), "1K")
+        self.assertEqual(value_in_range(1035, value_min="1K"), 1035)
+        self.assertEqual(value_in_range(1035, value_max="1K"), "1K")
+        self.assertEqual(value_in_range("1035", value_min="1K"), "1035")
         self.assertEqual(
-            BasisQueueAdapter._value_in_range("1035", value_min="1K"), "1035"
-        )
-        self.assertEqual(
-            BasisQueueAdapter._value_in_range(
-                "60000M", value_min="1K", value_max="50G"
-            ),
+            value_in_range("60000M", value_min="1K", value_max="50G"),
             "50G",
         )
         self.assertEqual(
-            BasisQueueAdapter._value_in_range("60000", value_min="1K", value_max="50G"),
+            value_in_range("60000", value_min="1K", value_max="50G"),
             "50G",
         )
         self.assertEqual(
-            BasisQueueAdapter._value_in_range(
-                "60000M", value_min="1K", value_max="70G"
-            ),
+            value_in_range("60000M", value_min="1K", value_max="70G"),
             "60000M",
         )
         self.assertEqual(
-            BasisQueueAdapter._value_in_range(60000, value_min="1K", value_max="70G"),
+            value_in_range(60000, value_min="1K", value_max="70G"),
             60000,
         )
         self.assertEqual(
-            BasisQueueAdapter._value_in_range(
-                90000 * 1024**2, value_min="1K", value_max="70G"
-            ),
+            value_in_range(90000 * 1024**2, value_min="1K", value_max="70G"),
             "70G",
         )
         self.assertEqual(
-            BasisQueueAdapter._value_in_range("90000", value_min="1K", value_max="70G"),
+            value_in_range("90000", value_min="1K", value_max="70G"),
             "70G",
         )
         self.assertEqual(
-            BasisQueueAdapter._value_in_range(
-                "60000M", value_min="60G", value_max="70G"
-            ),
+            value_in_range("60000M", value_min="60G", value_max="70G"),
             "60G",
         )

tests/test_sge.py

@@ -6,6 +6,7 @@
 import unittest
 import getpass
 from pysqa import QueueAdapter
+from pysqa.utils.validate import value_in_range
 
 try:
     import defusedxml.ElementTree as ETree
@@ -46,31 +47,23 @@ def test_ssh_delete_file_on_remote(self):
     def test_value_in_range(self):
         self.assertEqual(
             None,
-            self.sge._adapter._value_in_range(
-                value=None, value_min=None, value_max=None
-            ),
+            value_in_range(value=None, value_min=None, value_max=None),
         )
         self.assertEqual(
             1,
-            self.sge._adapter._value_in_range(value=None, value_min=1, value_max=None),
+            value_in_range(value=None, value_min=1, value_max=None),
         )
         self.assertEqual(
             1,
-            self.sge._adapter._value_in_range(value=None, value_min=None, value_max=1),
+            value_in_range(value=None, value_min=None, value_max=1),
         )
         self.assertEqual(
             1,
-            self.sge._adapter._value_in_range(value=1, value_min=None, value_max=None),
-        )
-        self.assertEqual(
-            1, self.sge._adapter._value_in_range(value=0, value_min=1, value_max=None)
-        )
-        self.assertEqual(
-            1, self.sge._adapter._value_in_range(value=2, value_min=None, value_max=1)
-        )
-        self.assertEqual(
-            1, self.sge._adapter._value_in_range(value=1, value_min=0, value_max=2)
+            value_in_range(value=1, value_min=None, value_max=None),
         )
+        self.assertEqual(1, value_in_range(value=0, value_min=1, value_max=None))
+        self.assertEqual(1, value_in_range(value=2, value_min=None, value_max=1))
+        self.assertEqual(1, value_in_range(value=1, value_min=0, value_max=2))
 
     def test_job_submission_template(self):
         self.assertRaises(