first testst for partition_ceheck #106

test/test_modules/test_partition_check.py

+import logging
+
+import pytest
+import mock
+from src.run_modules.experiment_setup import ExperimentSetup
+from src.run_modules.partition_check import PartitionCheck
+from src.run_modules.run_environment import RunEnvironment
+from src.helpers import get_host
+from src.helpers import PyTestRegex
+
+
+class TestPartitionCheck:
+
+    # @pytest.fixture
+    # def obj_super_init(self):
+    #     obj = object.__new__(PartitionCheck)
+    #     super(PartitionCheck, obj).__init__()
+    #     obj.data_store.set("NAME1", 1, "general")
+    #     yield obj
+    #     RunEnvironment().__del__()
+
+    @pytest.fixture
+    @mock.patch("src.helpers.get_host", return_value="juwels")
+    @mock.patch("getpass.getuser", return_value="testUser")
+    @mock.patch("os.path.exists", return_value=False)
+    @mock.patch("os.makedirs", side_effect=None)
+    def obj_with_exp_setup_login(self, mock_host, mock_user,  mock_path, mock_check):
+        ExperimentSetup(parser_args={}, stations=['DEBW107', 'DEBY081', 'DEBW013', 'DEBW076', 'DEBW087', 'DEBW001'],
+                        statistics_per_var={'o3': 'dma8eu', 'temp': 'maximum'}, station_type="background")
+        pre = object.__new__(PartitionCheck)
+        super(PartitionCheck, pre).__init__()
+        yield pre
+        RunEnvironment().__del__()
+
+    @pytest.fixture
+    @mock.patch("src.helpers.get_host", return_value="jwtest")
+    @mock.patch("getpass.getuser", return_value="testUser")
+    @mock.patch("os.path.exists", return_value=False)
+    @mock.patch("os.makedirs", side_effect=None)
+    def obj_with_exp_setup_compute(self, mock_host, mock_user,  mock_path, mock_check):
+        ExperimentSetup(parser_args={}, stations=['DEBW107', 'DEBY081', 'DEBW013', 'DEBW076', 'DEBW087', 'DEBW001'],
+                        statistics_per_var={'o3': 'dma8eu', 'temp': 'maximum'}, station_type="background")
+        pre = object.__new__(PartitionCheck)
+        super(PartitionCheck, pre).__init__()
+        yield pre
+        RunEnvironment().__del__()
+
+    def test_init(self, caplog):
+        ExperimentSetup(parser_args={}, stations=['DEBW107', 'DEBY081', 'DEBW013', 'DEBW076', 'DEBW087'],
+                        statistics_per_var={'o3': 'dma8eu', 'temp': 'maximum'})
+        caplog.clear()
+        caplog.set_level(logging.INFO)
+        with PartitionCheck():
+            assert caplog.record_tuples[0] == ('root', 20, 'PartitionCheck started')
+
+        RunEnvironment().__del__()
+
+    @mock.patch("src.helpers.get_host", return_value="juwels")
+    @mock.patch("getpass.getuser", return_value="testUser")
+    @mock.patch("os.path.exists", return_value=False)
+    @mock.patch("os.makedirs", side_effect=None)
+    def test_run_login(self, mock_host, mock_user,  mock_path, mock_check, obj_with_exp_setup_login, caplog):
+
+        with pytest.raises(OSError) as e:
+            obj_with_exp_setup_login.__next__()._run()
+        assert "You are on a login node to download data. Use compute nodes and run again if you want to train and " \
+               "validate a model." == \
+               e.value.args[0]
+
+    @mock.patch("src.helpers.get_host", return_value="jwtest")
+    @mock.patch("getpass.getuser", return_value="testUser")
+    @mock.patch("os.path.exists", return_value=False)
+    @mock.patch("os.makedirs", side_effect=None)
+    def test_run_compute(self, mock_host, mock_user, mock_path, mock_check, obj_with_exp_setup_compute, caplog):
+
+        assert obj_with_exp_setup_compute.__next__()._run() is None