Skip to content
Snippets Groups Projects
Select Git revision
  • 5355e9d49fb04aeab631ff6ce04904475c97178d
  • main default protected
  • airflow-2.7.0 protected
  • airflow253 protected
  • air251
  • test_docker_op
  • airflow225
  • mptest
  • https-deployment
  • datacat_integration protected
  • datacatalog-integration
  • stable-2.2.2 protected
  • stable-2.2.1 protected
  • stable-2.2.0 protected
  • stable-2.1.4 protected
  • stable-2.1.3 protected
  • stable-2.1.2 protected
  • stable-2.1.1 protected
  • stable-2.1.0 protected
  • stable-2.0.2 protected
  • stable-2.0.1 protected
  • stable-2.0.0 protected
  • stable-1.0.1 protected
  • stable-1.0 protected
  • stable-0.1 protected
25 results

taskflow.py

Blame
  • conftest.py 2.00 KiB
    import os
    import re
    import shutil
    
    
    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