Refactor tests

tests/test_basic.py

@@ -0,0 +1,70 @@
+# coding: utf-8
+# Copyright (c) Jan Janssen
+
+import os
+import unittest
+from pysqa import QueueAdapter
+from pysqa.utils.basic import BasisQueueAdapter
+
+__author__ = "Jan Janssen"
+__copyright__ = "Copyright 2019, Jan Janssen"
+__version__ = "0.0.1"
+__maintainer__ = "Jan Janssen"
+__email__ = "[email protected]"
+__status__ = "production"
+__date__ = "Feb 9, 2019"
+
+
+class TestRunmode(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.path = os.path.dirname(os.path.abspath(__file__))
+
+    def test_missing_config(self):
+        self.assertRaises(
+            ValueError, QueueAdapter, directory=os.path.join(self.path, "config/error")
+        )
+
+    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(
+            BasisQueueAdapter._value_in_range("1035", value_min="1K"), "1035"
+        )
+        self.assertEqual(
+            BasisQueueAdapter._value_in_range(
+                "60000M", value_min="1K", value_max="50G"
+            ),
+            "50G",
+        )
+        self.assertEqual(
+            BasisQueueAdapter._value_in_range("60000", value_min="1K", value_max="50G"),
+            "50G",
+        )
+        self.assertEqual(
+            BasisQueueAdapter._value_in_range(
+                "60000M", value_min="1K", value_max="70G"
+            ),
+            "60000M",
+        )
+        self.assertEqual(
+            BasisQueueAdapter._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"
+            ),
+            "70G",
+        )
+        self.assertEqual(
+            BasisQueueAdapter._value_in_range("90000", value_min="1K", value_max="70G"),
+            "70G",
+        )
+        self.assertEqual(
+            BasisQueueAdapter._value_in_range(
+                "60000M", value_min="60G", value_max="70G"
+            ),
+            "60G",
+        )

tests/test_gent.py

@@ -0,0 +1,51 @@
+# coding: utf-8
+# Copyright (c) Jan Janssen
+
+import os
+import unittest
+from pysqa import QueueAdapter
+
+__author__ = "Jan Janssen"
+__copyright__ = "Copyright 2019, Jan Janssen"
+__version__ = "0.0.1"
+__maintainer__ = "Jan Janssen"
+__email__ = "[email protected]"
+__status__ = "production"
+__date__ = "Feb 9, 2019"
+
+
+class TestRunmode(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.path = os.path.dirname(os.path.abspath(__file__))
+        cls.gent = QueueAdapter(directory=os.path.join(cls.path, "config/gent"))
+
+    def test_config(self):
+        self.assertEqual(self.gent.config["queue_type"], "GENT")
+        self.assertEqual(self.gent.config["queue_primary"], "slurm")
+
+    def test_list_clusters(self):
+        self.assertEqual(self.gent.list_clusters(), ['default'])
+
+    def test_ssh_delete_file_on_remote(self):
+        self.assertEqual(self.gent.ssh_delete_file_on_remote, True)
+
+    def test_interfaces(self):
+        self.assertEqual(
+            self.gent._adapter._commands.get_queue_status_command,
+            ["squeue", "--format", "%A|%u|%t|%.15j|%Z", "--noheader"],
+        )
+
+    def test__list_command_to_be_executed(self):
+        with self.subTest("gent"):
+            self.assertEqual(
+                self.gent._adapter._list_command_to_be_executed(None, "here"),
+                ["sbatch", "--parsable", "here"],
+            )
+        with self.subTest("gent with dependency"):
+            self.assertRaises(
+                NotImplementedError,
+                self.gent._adapter._list_command_to_be_executed,
+                [],
+                "here",
+            )

tests/test_lsf.py

@@ -0,0 +1,64 @@
+# coding: utf-8
+# Copyright (c) Jan Janssen
+
+import os
+import unittest
+from pysqa import QueueAdapter
+
+__author__ = "Jan Janssen"
+__copyright__ = "Copyright 2019, Jan Janssen"
+__version__ = "0.0.1"
+__maintainer__ = "Jan Janssen"
+__email__ = "[email protected]"
+__status__ = "production"
+__date__ = "Feb 9, 2019"
+
+
+class TestRunmode(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.path = os.path.dirname(os.path.abspath(__file__))
+        cls.lsf = QueueAdapter(directory=os.path.join(cls.path, "config/lsf"))
+
+    def test_config(self):
+        self.assertEqual(self.lsf.config["queue_type"], "LSF")
+        self.assertEqual(self.lsf.config["queue_primary"], "lsf")
+
+    def test_list_clusters(self):
+        self.assertEqual(self.lsf.list_clusters(), ['default'])
+
+    def test_ssh_delete_file_on_remote(self):
+        self.assertEqual(self.lsf.ssh_delete_file_on_remote, True)
+
+    def test_job_submission_template(self):
+        template = (
+            "#!/bin/bash\n#BSUB -q queue\n#BSUB -J job.py\n#BSUB -o time.out\n#BSUB -n 10\n#BSUB -cwd .\n"
+            "#BSUB -e error.out\n#BSUB -W 259200\n\npython test.py"
+        )
+        self.assertEqual(
+            self.lsf._adapter._job_submission_template(command="python test.py"),
+            template,
+        )
+
+    def test_interfaces(self):
+        self.assertEqual(
+            self.lsf._adapter._commands.submit_job_command, ["bsub", "-terse"]
+        )
+        self.assertEqual(self.lsf._adapter._commands.delete_job_command, ["bkill"])
+        self.assertEqual(
+            self.lsf._adapter._commands.get_queue_status_command, ["qstat", "-x"]
+        )
+
+    def test__list_command_to_be_executed(self):
+        with self.subTest("lsf"):
+            self.assertEqual(
+                self.lsf._adapter._list_command_to_be_executed(None, "here"),
+                ["bsub", "-terse", "here"],
+            )
+        with self.subTest("lsf with dependency"):
+            self.assertRaises(
+                NotImplementedError,
+                self.lsf._adapter._list_command_to_be_executed,
+                [],
+                "here",
+            )

tests/test_moab.py

@@ -0,0 +1,54 @@
+# coding: utf-8
+# Copyright (c) Jan Janssen
+
+import os
+import unittest
+from pysqa import QueueAdapter
+
+__author__ = "Jan Janssen"
+__copyright__ = "Copyright 2019, Jan Janssen"
+__version__ = "0.0.1"
+__maintainer__ = "Jan Janssen"
+__email__ = "[email protected]"
+__status__ = "production"
+__date__ = "Feb 9, 2019"
+
+
+class TestRunmode(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.path = os.path.dirname(os.path.abspath(__file__))
+        cls.moab = QueueAdapter(directory=os.path.join(cls.path, "config/moab"))
+
+    def test_config(self):
+        self.assertEqual(self.moab.config["queue_type"], "MOAB")
+        self.assertEqual(self.moab.config["queue_primary"], "moab")
+
+    def test_list_clusters(self):
+        self.assertEqual(self.moab.list_clusters(), ['default'])
+
+    def test_ssh_delete_file_on_remote(self):
+        self.assertEqual(self.moab.ssh_delete_file_on_remote, True)
+
+    def test_interfaces(self):
+        self.assertEqual(self.moab._adapter._commands.submit_job_command, ["msub"])
+        self.assertEqual(
+            self.moab._adapter._commands.delete_job_command, ["mjobctl", "-c"]
+        )
+        self.assertEqual(
+            self.moab._adapter._commands.get_queue_status_command, ["mdiag", "-x"]
+        )
+
+    def test__list_command_to_be_executed(self):
+        with self.subTest("moab with dependency"):
+            self.assertRaises(
+                NotImplementedError,
+                self.moab._adapter._list_command_to_be_executed,
+                [],
+                "here",
+            )
+        with self.subTest("moab"):
+            self.assertEqual(
+                self.moab._adapter._list_command_to_be_executed(None, "here"),
+                ["msub", "here"],
+            )

tests/test_multi.py

@@ -0,0 +1,28 @@
+# coding: utf-8
+# Copyright (c) Jan Janssen
+
+import os
+import unittest
+from pysqa import QueueAdapter
+
+__author__ = "Jan Janssen"
+__copyright__ = "Copyright 2019, Jan Janssen"
+__version__ = "0.0.1"
+__maintainer__ = "Jan Janssen"
+__email__ = "[email protected]"
+__status__ = "production"
+__date__ = "Feb 9, 2019"
+
+
+class TestRunmode(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.path = os.path.dirname(os.path.abspath(__file__))
+        cls.multi = QueueAdapter(directory=os.path.join(cls.path, "config/multicluster"))
+
+    def test_config(self):
+        self.assertEqual(self.multi.config["queue_type"], "SLURM")
+        self.assertEqual(self.multi.config["queue_primary"], "slurm")
+
+    def test_list_clusters(self):
+        self.assertEqual(self.multi.list_clusters(), ['local_slurm', 'remote_slurm'])