Skip to content
Snippets Groups Projects
Select Git revision
  • 63d232d56a8a95bf633fadc38e9251507f00daa4
  • master default protected
  • enxhi_issue460_remove_TOAR-I_access
  • michael_issue459_preprocess_german_stations
  • sh_pollutants
  • develop protected
  • release_v2.4.0
  • michael_issue450_feat_load-ifs-data
  • lukas_issue457_feat_set-config-paths-as-parameter
  • lukas_issue454_feat_use-toar-statistics-api-v2
  • lukas_issue453_refac_advanced-retry-strategy
  • lukas_issue452_bug_update-proj-version
  • lukas_issue449_refac_load-era5-data-from-toar-db
  • lukas_issue451_feat_robust-apriori-estimate-for-short-timeseries
  • lukas_issue448_feat_load-model-from-path
  • lukas_issue447_feat_store-and-load-local-clim-apriori-data
  • lukas_issue445_feat_data-insight-plot-monthly-distribution
  • lukas_issue442_feat_bias-free-evaluation
  • lukas_issue444_feat_choose-interp-method-cams
  • 414-include-crps-analysis-and-other-ens-verif-methods-or-plots
  • lukas_issue384_feat_aqw-data-handler
  • v2.4.0 protected
  • v2.3.0 protected
  • v2.2.0 protected
  • v2.1.0 protected
  • Kleinert_etal_2022_initial_submission
  • v2.0.0 protected
  • v1.5.0 protected
  • v1.4.0 protected
  • v1.3.0 protected
  • v1.2.1 protected
  • v1.2.0 protected
  • v1.1.0 protected
  • IntelliO3-ts-v1.0_R1-submit
  • v1.0.0 protected
  • v0.12.2 protected
  • v0.12.1 protected
  • v0.12.0 protected
  • v0.11.0 protected
  • v0.10.0 protected
  • IntelliO3-ts-v1.0_initial-submit
41 results

test_partition_check.py

Blame
  • test_partition_check.py 3.34 KiB
    import logging
    
    import pytest
    import mock
    from mlair.run_modules.experiment_setup import ExperimentSetup
    from mlair.run_modules.partition_check import PartitionCheck
    from mlair.run_modules.run_environment import RunEnvironment
    
    
    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("socket.gethostname", 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):
            RunEnvironment().__del__()
            ExperimentSetup(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("socket.gethostname", return_value="hdfmlc01")
        @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):
            RunEnvironment().__del__()
            ExperimentSetup(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(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("socket.gethostname", 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("socket.gethostname", return_value="hdfmlc01")
        @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):
            obj = obj_with_exp_setup_compute.__next__()
            assert obj._run() is None