Skip to content
Snippets Groups Projects
Select Git revision
  • 76e406ddad20449e1cc3d61bdc1aff67660b67bc
  • 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

conftest.py

Blame
  • conftest.py 2.36 KiB
    import os
    import re
    import shutil
    import pytest
    import mock
    
    
    def pytest_runtest_teardown(item, nextitem):
        """
        Teardown method to clean up folder creations during testing. This method is called after each test, but performs
        deletions only after an entire test class was executed.
        :param item: tested item
        :param nextitem: next item (could be None, if no following test is available)
        """
        if nextitem is None or item.cls != nextitem.cls:
            # clean up all TestExperiment and data folder that have been created during testing
            rel_path = os.path.relpath(item.fspath.dirname, os.path.abspath(__file__))
            path = os.path.dirname(__file__)
            for stage in filter(None, rel_path.replace("..", ".").split("/")):
                path = os.path.abspath(os.path.join(path, stage))
                execute_removing(path)
            execute_removing(os.path.dirname(__file__))
        else:
            pass  # nothing to do if next test is from same test class
    
    
    def execute_removing(path):
        list_dir = os.listdir(path)
        if "data" in list_dir and path != os.path.dirname(__file__):  # do not delete data folder in src
            shutil.rmtree(os.path.join(path, "data"), ignore_errors=True)
        # remove TestExperiment folders
        remove_files_from_regex(list_dir, path, re.compile(r"TestExperiment.*"))
        # remove all tracking json
        remove_files_from_regex(list_dir, path, re.compile(r"tracking_\d*\.json"))
        # remove all tracking pdf
        remove_files_from_regex(list_dir, path, re.compile(r"tracking_\d*\.pdf"))
        # remove all tracking json
        remove_files_from_regex(list_dir, path, re.compile(r"logging_\d*\.log"))
    
    
    def remove_files_from_regex(list_dir, path, regex):
        r = list(filter(regex.search, list_dir))
        if len(r) > 0:
            for e in r:
                del_path = os.path.join(path, e)
                try:
                    if os.path.isfile(del_path):
                        os.remove(del_path)
                    else:
                        shutil.rmtree(os.path.join(path, e), ignore_errors=True)
                except:
                    pass
    
    
    @pytest.fixture(scope="session", autouse=True)
    def default_session_fixture(request):
        """
        :type request: _pytest.python.SubRequest
        :return:
        """
        patched = mock.patch("multiprocessing.cpu_count", return_value=1)
        patched.__enter__()
    
        def unpatch():
            patched.__exit__()
    
        request.addfinalizer(unpatch)