unit tests are supposed to leave the execution environment unchanged once they complete but sometimes tests have bugs especially when tests fail and the changes to environment go on to mess up other tests this can cause issues with buildbot stability since tests are run in random order and so problems may appear to come and go there are a few things we can save and restore to mitigate this and the following context manager handles this task save bits of the test environment and restore them at block exit with savedtestenvironmenttestname verbose quiet stuff unless quiet is true a warning is printed to stderr if any of the saved items was changed by the test the support environmentaltered attribute is set to true if a change is detected if verbose is more than 1 the before and after state of changed items is also printed to add things to save and restore add a name xxx to the resources list and add corresponding getxxxrestorexxx functions getxxx should return the value to be saved and compared against a second call to the get function when test execution completes restorexxx should accept the saved value and restore the resource using it it will be called if and only if a change in the value is detected note xxx will have any replaced with characters when determining the corresponding method names multiprocessing process cleanup may release ref to a thread so check processes first function for restore methods function for get methods xxx making a copy keeps objects alive until exit gets called we could call getarchivesformats but that only returns the registry keys we want to check the values too the functions that are registered handlers is a weakvaluedictionary can t easily revert the logging state handlerlist is a list of weakrefs to handlers can t easily revert the logging state controlling dangling references to thread objects can make it easier to track reference leaks this copies the weakrefs without making any strong reference same for process objects unjoined process objects can survive after process exits this copies the weakrefs without making any strong reference make sure the dict is initialized xxx maybe add an allowlist here some resources use weak references check for changes to the resource s value unit tests are supposed to leave the execution environment unchanged once they complete but sometimes tests have bugs especially when tests fail and the changes to environment go on to mess up other tests this can cause issues with buildbot stability since tests are run in random order and so problems may appear to come and go there are a few things we can save and restore to mitigate this and the following context manager handles this task save bits of the test environment and restore them at block exit with saved_test_environment test_name verbose quiet stuff unless quiet is true a warning is printed to stderr if any of the saved items was changed by the test the support environment_altered attribute is set to true if a change is detected if verbose is more than 1 the before and after state of changed items is also printed to add things to save and restore add a name xxx to the resources list and add corresponding get_xxx restore_xxx functions get_xxx should return the value to be saved and compared against a second call to the get function when test execution completes restore_xxx should accept the saved value and restore the resource using it it will be called if and only if a change in the value is detected note xxx will have any replaced with _ characters when determining the corresponding method names multiprocessing process _cleanup may release ref to a thread so check processes first function for restore methods function for get methods xxx making a copy keeps objects alive until __exit__ gets called we could call get_archives_formats but that only returns the registry keys we want to check the values too the functions that are registered _handlers is a weakvaluedictionary can t easily revert the logging state _handlerlist is a list of weakrefs to handlers can t easily revert the logging state controlling dangling references to thread objects can make it easier to track reference leaks this copies the weakrefs without making any strong reference same for process objects unjoined process objects can survive after process exits this copies the weakrefs without making any strong reference make sure the dict is initialized xxx maybe add an allow list here some resources use weak references check for changes to the resource s value
import builtins
import locale
import os
import sys
import threading

from test import support
from test.support import os_helper

from .utils import print_warning


class SkipTestEnvironment(Exception):
    pass










class saved_test_environment:
    

    def __init__(self, test_name, verbose, quiet, *, pgo):
        self.test_name = test_name
        self.verbose = verbose
        self.quiet = quiet
        self.pgo = pgo

    
    
    
    
    
    
    
    
    

    resources = ('sys.argv', 'cwd', 'sys.stdin', 'sys.stdout', 'sys.stderr',
                 'os.environ', 'sys.path', 'sys.path_hooks', '__import__',
                 'warnings.filters', 'asyncore.socket_map',
                 'logging._handlers', 'logging._handlerList', 'sys.gettrace',
                 'sys.warnoptions',
                 
                 
                 'multiprocessing.process._dangling', 'threading._dangling',
                 'sysconfig._CONFIG_VARS', 'sysconfig._INSTALL_SCHEMES',
                 'files', 'locale', 'warnings.showwarning',
                 'shutil_archive_formats', 'shutil_unpack_formats',
                 'asyncio.events._event_loop_policy',
                 'urllib.requests._url_tempfiles', 'urllib.requests._opener',
                )

    def get_module(self, name):
        
        return sys.modules[name]

    def try_get_module(self, name):
        
        try:
            return self.get_module(name)
        except KeyError:
            raise SkipTestEnvironment

    def get_urllib_requests__url_tempfiles(self):
        urllib_request = self.try_get_module('urllib.request')
        return list(urllib_request._url_tempfiles)
    def restore_urllib_requests__url_tempfiles(self, tempfiles):
        for filename in tempfiles:
            os_helper.unlink(filename)

    def get_urllib_requests__opener(self):
        urllib_request = self.try_get_module('urllib.request')
        return urllib_request._opener
    def restore_urllib_requests__opener(self, opener):
        urllib_request = self.get_module('urllib.request')
        urllib_request._opener = opener

    def get_asyncio_events__event_loop_policy(self):
        self.try_get_module('asyncio')
        return support.maybe_get_event_loop_policy()
    def restore_asyncio_events__event_loop_policy(self, policy):
        asyncio = self.get_module('asyncio')
        asyncio.set_event_loop_policy(policy)

    def get_sys_argv(self):
        return id(sys.argv), sys.argv, sys.argv[:]
    def restore_sys_argv(self, saved_argv):
        sys.argv = saved_argv[1]
        sys.argv[:] = saved_argv[2]

    def get_cwd(self):
        return os.getcwd()
    def restore_cwd(self, saved_cwd):
        os.chdir(saved_cwd)

    def get_sys_stdout(self):
        return sys.stdout
    def restore_sys_stdout(self, saved_stdout):
        sys.stdout = saved_stdout

    def get_sys_stderr(self):
        return sys.stderr
    def restore_sys_stderr(self, saved_stderr):
        sys.stderr = saved_stderr

    def get_sys_stdin(self):
        return sys.stdin
    def restore_sys_stdin(self, saved_stdin):
        sys.stdin = saved_stdin

    def get_os_environ(self):
        return id(os.environ), os.environ, dict(os.environ)
    def restore_os_environ(self, saved_environ):
        os.environ = saved_environ[1]
        os.environ.clear()
        os.environ.update(saved_environ[2])

    def get_sys_path(self):
        return id(sys.path), sys.path, sys.path[:]
    def restore_sys_path(self, saved_path):
        sys.path = saved_path[1]
        sys.path[:] = saved_path[2]

    def get_sys_path_hooks(self):
        return id(sys.path_hooks), sys.path_hooks, sys.path_hooks[:]
    def restore_sys_path_hooks(self, saved_hooks):
        sys.path_hooks = saved_hooks[1]
        sys.path_hooks[:] = saved_hooks[2]

    def get_sys_gettrace(self):
        return sys.gettrace()
    def restore_sys_gettrace(self, trace_fxn):
        sys.settrace(trace_fxn)

    def get___import__(self):
        return builtins.__import__
    def restore___import__(self, import_):
        builtins.__import__ = import_

    def get_warnings_filters(self):
        warnings = self.try_get_module('warnings')
        return id(warnings.filters), warnings.filters, warnings.filters[:]
    def restore_warnings_filters(self, saved_filters):
        warnings = self.get_module('warnings')
        warnings.filters = saved_filters[1]
        warnings.filters[:] = saved_filters[2]

    def get_asyncore_socket_map(self):
        asyncore = sys.modules.get('test.support.asyncore')
        
        return asyncore and asyncore.socket_map.copy() or {}
    def restore_asyncore_socket_map(self, saved_map):
        asyncore = sys.modules.get('test.support.asyncore')
        if asyncore is not None:
            asyncore.close_all(ignore_all=True)
            asyncore.socket_map.update(saved_map)

    def get_shutil_archive_formats(self):
        shutil = self.try_get_module('shutil')
        
        
        
        return shutil._ARCHIVE_FORMATS, shutil._ARCHIVE_FORMATS.copy()
    def restore_shutil_archive_formats(self, saved):
        shutil = self.get_module('shutil')
        shutil._ARCHIVE_FORMATS = saved[0]
        shutil._ARCHIVE_FORMATS.clear()
        shutil._ARCHIVE_FORMATS.update(saved[1])

    def get_shutil_unpack_formats(self):
        shutil = self.try_get_module('shutil')
        return shutil._UNPACK_FORMATS, shutil._UNPACK_FORMATS.copy()
    def restore_shutil_unpack_formats(self, saved):
        shutil = self.get_module('shutil')
        shutil._UNPACK_FORMATS = saved[0]
        shutil._UNPACK_FORMATS.clear()
        shutil._UNPACK_FORMATS.update(saved[1])

    def get_logging__handlers(self):
        logging = self.try_get_module('logging')
        
        return id(logging._handlers), logging._handlers, logging._handlers.copy()
    def restore_logging__handlers(self, saved_handlers):
        
        pass

    def get_logging__handlerList(self):
        logging = self.try_get_module('logging')
        
        return id(logging._handlerList), logging._handlerList, logging._handlerList[:]
    def restore_logging__handlerList(self, saved_handlerList):
        
        pass

    def get_sys_warnoptions(self):
        return id(sys.warnoptions), sys.warnoptions, sys.warnoptions[:]
    def restore_sys_warnoptions(self, saved_options):
        sys.warnoptions = saved_options[1]
        sys.warnoptions[:] = saved_options[2]

    
    
    def get_threading__dangling(self):
        
        return threading._dangling.copy()
    def restore_threading__dangling(self, saved):
        threading._dangling.clear()
        threading._dangling.update(saved)

    
    def get_multiprocessing_process__dangling(self):
        multiprocessing_process = self.try_get_module('multiprocessing.process')
        
        multiprocessing_process._cleanup()
        
        return multiprocessing_process._dangling.copy()
    def restore_multiprocessing_process__dangling(self, saved):
        multiprocessing_process = self.get_module('multiprocessing.process')
        multiprocessing_process._dangling.clear()
        multiprocessing_process._dangling.update(saved)

    def get_sysconfig__CONFIG_VARS(self):
        
        sysconfig = self.try_get_module('sysconfig')
        sysconfig.get_config_var('prefix')
        return (id(sysconfig._CONFIG_VARS), sysconfig._CONFIG_VARS,
                dict(sysconfig._CONFIG_VARS))
    def restore_sysconfig__CONFIG_VARS(self, saved):
        sysconfig = self.get_module('sysconfig')
        sysconfig._CONFIG_VARS = saved[1]
        sysconfig._CONFIG_VARS.clear()
        sysconfig._CONFIG_VARS.update(saved[2])

    def get_sysconfig__INSTALL_SCHEMES(self):
        sysconfig = self.try_get_module('sysconfig')
        return (id(sysconfig._INSTALL_SCHEMES), sysconfig._INSTALL_SCHEMES,
                sysconfig._INSTALL_SCHEMES.copy())
    def restore_sysconfig__INSTALL_SCHEMES(self, saved):
        sysconfig = self.get_module('sysconfig')
        sysconfig._INSTALL_SCHEMES = saved[1]
        sysconfig._INSTALL_SCHEMES.clear()
        sysconfig._INSTALL_SCHEMES.update(saved[2])

    def get_files(self):
        
        return sorted(fn + ('/' if os.path.isdir(fn) else '')
                      for fn in os.listdir()
                      if not fn.startswith(".hypothesis"))
    def restore_files(self, saved_value):
        fn = os_helper.TESTFN
        if fn not in saved_value and (fn + '/') not in saved_value:
            if os.path.isfile(fn):
                os_helper.unlink(fn)
            elif os.path.isdir(fn):
                os_helper.rmtree(fn)

    _lc = [getattr(locale, lc) for lc in dir(locale)
           if lc.startswith('LC_')]
    def get_locale(self):
        pairings = []
        for lc in self._lc:
            try:
                pairings.append((lc, locale.setlocale(lc, None)))
            except (TypeError, ValueError):
                continue
        return pairings
    def restore_locale(self, saved):
        for lc, setting in saved:
            locale.setlocale(lc, setting)

    def get_warnings_showwarning(self):
        warnings = self.try_get_module('warnings')
        return warnings.showwarning
    def restore_warnings_showwarning(self, fxn):
        warnings = self.get_module('warnings')
        warnings.showwarning = fxn

    def resource_info(self):
        for name in self.resources:
            method_suffix = name.replace('.', '_')
            get_name = 'get_' + method_suffix
            restore_name = 'restore_' + method_suffix
            yield name, getattr(self, get_name), getattr(self, restore_name)

    def __enter__(self):
        self.saved_values = []
        for name, get, restore in self.resource_info():
            try:
                original = get()
            except SkipTestEnvironment:
                continue

            self.saved_values.append((name, get, restore, original))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        saved_values = self.saved_values
        self.saved_values = None

        
        support.gc_collect()

        for name, get, restore, original in saved_values:
            current = get()
            
            if current != original:
                support.environment_altered = True
                restore(original)
                if not self.quiet and not self.pgo:
                    print_warning(
                        f"{name} was modified by {self.test_name}\n"
                        f"  Before: {original}\n"
                        f"  After:  {current} ")
        return False