ยปCore Development>Code coverage>Lib/test/libregrtest/save_env.py

Python code coverage for Lib/test/libregrtest/save_env.py

#countcontent
1n/aimport builtins
2n/aimport locale
3n/aimport logging
4n/aimport os
5n/aimport shutil
6n/aimport sys
7n/aimport sysconfig
8n/aimport warnings
9n/afrom test import support
10n/atry:
11n/a import threading
12n/aexcept ImportError:
13n/a threading = None
14n/atry:
15n/a import _multiprocessing, multiprocessing.process
16n/aexcept ImportError:
17n/a multiprocessing = None
18n/a
19n/a
20n/a# Unit tests are supposed to leave the execution environment unchanged
21n/a# once they complete. But sometimes tests have bugs, especially when
22n/a# tests fail, and the changes to environment go on to mess up other
23n/a# tests. This can cause issues with buildbot stability, since tests
24n/a# are run in random order and so problems may appear to come and go.
25n/a# There are a few things we can save and restore to mitigate this, and
26n/a# the following context manager handles this task.
27n/a
28n/aclass saved_test_environment:
29n/a """Save bits of the test environment and restore them at block exit.
30n/a
31n/a with saved_test_environment(testname, verbose, quiet):
32n/a #stuff
33n/a
34n/a Unless quiet is True, a warning is printed to stderr if any of
35n/a the saved items was changed by the test. The attribute 'changed'
36n/a is initially False, but is set to True if a change is detected.
37n/a
38n/a If verbose is more than 1, the before and after state of changed
39n/a items is also printed.
40n/a """
41n/a
42n/a changed = False
43n/a
44n/a def __init__(self, testname, verbose=0, quiet=False, *, pgo=False):
45n/a self.testname = testname
46n/a self.verbose = verbose
47n/a self.quiet = quiet
48n/a self.pgo = pgo
49n/a
50n/a # To add things to save and restore, add a name XXX to the resources list
51n/a # and add corresponding get_XXX/restore_XXX functions. get_XXX should
52n/a # return the value to be saved and compared against a second call to the
53n/a # get function when test execution completes. restore_XXX should accept
54n/a # the saved value and restore the resource using it. It will be called if
55n/a # and only if a change in the value is detected.
56n/a #
57n/a # Note: XXX will have any '.' replaced with '_' characters when determining
58n/a # the corresponding method names.
59n/a
60n/a resources = ('sys.argv', 'cwd', 'sys.stdin', 'sys.stdout', 'sys.stderr',
61n/a 'os.environ', 'sys.path', 'sys.path_hooks', '__import__',
62n/a 'warnings.filters', 'asyncore.socket_map',
63n/a 'logging._handlers', 'logging._handlerList', 'sys.gettrace',
64n/a 'sys.warnoptions',
65n/a # multiprocessing.process._cleanup() may release ref
66n/a # to a thread, so check processes first.
67n/a 'multiprocessing.process._dangling', 'threading._dangling',
68n/a 'sysconfig._CONFIG_VARS', 'sysconfig._INSTALL_SCHEMES',
69n/a 'files', 'locale', 'warnings.showwarning',
70n/a 'shutil_archive_formats', 'shutil_unpack_formats',
71n/a )
72n/a
73n/a def get_sys_argv(self):
74n/a return id(sys.argv), sys.argv, sys.argv[:]
75n/a def restore_sys_argv(self, saved_argv):
76n/a sys.argv = saved_argv[1]
77n/a sys.argv[:] = saved_argv[2]
78n/a
79n/a def get_cwd(self):
80n/a return os.getcwd()
81n/a def restore_cwd(self, saved_cwd):
82n/a os.chdir(saved_cwd)
83n/a
84n/a def get_sys_stdout(self):
85n/a return sys.stdout
86n/a def restore_sys_stdout(self, saved_stdout):
87n/a sys.stdout = saved_stdout
88n/a
89n/a def get_sys_stderr(self):
90n/a return sys.stderr
91n/a def restore_sys_stderr(self, saved_stderr):
92n/a sys.stderr = saved_stderr
93n/a
94n/a def get_sys_stdin(self):
95n/a return sys.stdin
96n/a def restore_sys_stdin(self, saved_stdin):
97n/a sys.stdin = saved_stdin
98n/a
99n/a def get_os_environ(self):
100n/a return id(os.environ), os.environ, dict(os.environ)
101n/a def restore_os_environ(self, saved_environ):
102n/a os.environ = saved_environ[1]
103n/a os.environ.clear()
104n/a os.environ.update(saved_environ[2])
105n/a
106n/a def get_sys_path(self):
107n/a return id(sys.path), sys.path, sys.path[:]
108n/a def restore_sys_path(self, saved_path):
109n/a sys.path = saved_path[1]
110n/a sys.path[:] = saved_path[2]
111n/a
112n/a def get_sys_path_hooks(self):
113n/a return id(sys.path_hooks), sys.path_hooks, sys.path_hooks[:]
114n/a def restore_sys_path_hooks(self, saved_hooks):
115n/a sys.path_hooks = saved_hooks[1]
116n/a sys.path_hooks[:] = saved_hooks[2]
117n/a
118n/a def get_sys_gettrace(self):
119n/a return sys.gettrace()
120n/a def restore_sys_gettrace(self, trace_fxn):
121n/a sys.settrace(trace_fxn)
122n/a
123n/a def get___import__(self):
124n/a return builtins.__import__
125n/a def restore___import__(self, import_):
126n/a builtins.__import__ = import_
127n/a
128n/a def get_warnings_filters(self):
129n/a return id(warnings.filters), warnings.filters, warnings.filters[:]
130n/a def restore_warnings_filters(self, saved_filters):
131n/a warnings.filters = saved_filters[1]
132n/a warnings.filters[:] = saved_filters[2]
133n/a
134n/a def get_asyncore_socket_map(self):
135n/a asyncore = sys.modules.get('asyncore')
136n/a # XXX Making a copy keeps objects alive until __exit__ gets called.
137n/a return asyncore and asyncore.socket_map.copy() or {}
138n/a def restore_asyncore_socket_map(self, saved_map):
139n/a asyncore = sys.modules.get('asyncore')
140n/a if asyncore is not None:
141n/a asyncore.close_all(ignore_all=True)
142n/a asyncore.socket_map.update(saved_map)
143n/a
144n/a def get_shutil_archive_formats(self):
145n/a # we could call get_archives_formats() but that only returns the
146n/a # registry keys; we want to check the values too (the functions that
147n/a # are registered)
148n/a return shutil._ARCHIVE_FORMATS, shutil._ARCHIVE_FORMATS.copy()
149n/a def restore_shutil_archive_formats(self, saved):
150n/a shutil._ARCHIVE_FORMATS = saved[0]
151n/a shutil._ARCHIVE_FORMATS.clear()
152n/a shutil._ARCHIVE_FORMATS.update(saved[1])
153n/a
154n/a def get_shutil_unpack_formats(self):
155n/a return shutil._UNPACK_FORMATS, shutil._UNPACK_FORMATS.copy()
156n/a def restore_shutil_unpack_formats(self, saved):
157n/a shutil._UNPACK_FORMATS = saved[0]
158n/a shutil._UNPACK_FORMATS.clear()
159n/a shutil._UNPACK_FORMATS.update(saved[1])
160n/a
161n/a def get_logging__handlers(self):
162n/a # _handlers is a WeakValueDictionary
163n/a return id(logging._handlers), logging._handlers, logging._handlers.copy()
164n/a def restore_logging__handlers(self, saved_handlers):
165n/a # Can't easily revert the logging state
166n/a pass
167n/a
168n/a def get_logging__handlerList(self):
169n/a # _handlerList is a list of weakrefs to handlers
170n/a return id(logging._handlerList), logging._handlerList, logging._handlerList[:]
171n/a def restore_logging__handlerList(self, saved_handlerList):
172n/a # Can't easily revert the logging state
173n/a pass
174n/a
175n/a def get_sys_warnoptions(self):
176n/a return id(sys.warnoptions), sys.warnoptions, sys.warnoptions[:]
177n/a def restore_sys_warnoptions(self, saved_options):
178n/a sys.warnoptions = saved_options[1]
179n/a sys.warnoptions[:] = saved_options[2]
180n/a
181n/a # Controlling dangling references to Thread objects can make it easier
182n/a # to track reference leaks.
183n/a def get_threading__dangling(self):
184n/a if not threading:
185n/a return None
186n/a # This copies the weakrefs without making any strong reference
187n/a return threading._dangling.copy()
188n/a def restore_threading__dangling(self, saved):
189n/a if not threading:
190n/a return
191n/a threading._dangling.clear()
192n/a threading._dangling.update(saved)
193n/a
194n/a # Same for Process objects
195n/a def get_multiprocessing_process__dangling(self):
196n/a if not multiprocessing:
197n/a return None
198n/a # Unjoined process objects can survive after process exits
199n/a multiprocessing.process._cleanup()
200n/a # This copies the weakrefs without making any strong reference
201n/a return multiprocessing.process._dangling.copy()
202n/a def restore_multiprocessing_process__dangling(self, saved):
203n/a if not multiprocessing:
204n/a return
205n/a multiprocessing.process._dangling.clear()
206n/a multiprocessing.process._dangling.update(saved)
207n/a
208n/a def get_sysconfig__CONFIG_VARS(self):
209n/a # make sure the dict is initialized
210n/a sysconfig.get_config_var('prefix')
211n/a return (id(sysconfig._CONFIG_VARS), sysconfig._CONFIG_VARS,
212n/a dict(sysconfig._CONFIG_VARS))
213n/a def restore_sysconfig__CONFIG_VARS(self, saved):
214n/a sysconfig._CONFIG_VARS = saved[1]
215n/a sysconfig._CONFIG_VARS.clear()
216n/a sysconfig._CONFIG_VARS.update(saved[2])
217n/a
218n/a def get_sysconfig__INSTALL_SCHEMES(self):
219n/a return (id(sysconfig._INSTALL_SCHEMES), sysconfig._INSTALL_SCHEMES,
220n/a sysconfig._INSTALL_SCHEMES.copy())
221n/a def restore_sysconfig__INSTALL_SCHEMES(self, saved):
222n/a sysconfig._INSTALL_SCHEMES = saved[1]
223n/a sysconfig._INSTALL_SCHEMES.clear()
224n/a sysconfig._INSTALL_SCHEMES.update(saved[2])
225n/a
226n/a def get_files(self):
227n/a return sorted(fn + ('/' if os.path.isdir(fn) else '')
228n/a for fn in os.listdir())
229n/a def restore_files(self, saved_value):
230n/a fn = support.TESTFN
231n/a if fn not in saved_value and (fn + '/') not in saved_value:
232n/a if os.path.isfile(fn):
233n/a support.unlink(fn)
234n/a elif os.path.isdir(fn):
235n/a support.rmtree(fn)
236n/a
237n/a _lc = [getattr(locale, lc) for lc in dir(locale)
238n/a if lc.startswith('LC_')]
239n/a def get_locale(self):
240n/a pairings = []
241n/a for lc in self._lc:
242n/a try:
243n/a pairings.append((lc, locale.setlocale(lc, None)))
244n/a except (TypeError, ValueError):
245n/a continue
246n/a return pairings
247n/a def restore_locale(self, saved):
248n/a for lc, setting in saved:
249n/a locale.setlocale(lc, setting)
250n/a
251n/a def get_warnings_showwarning(self):
252n/a return warnings.showwarning
253n/a def restore_warnings_showwarning(self, fxn):
254n/a warnings.showwarning = fxn
255n/a
256n/a def resource_info(self):
257n/a for name in self.resources:
258n/a method_suffix = name.replace('.', '_')
259n/a get_name = 'get_' + method_suffix
260n/a restore_name = 'restore_' + method_suffix
261n/a yield name, getattr(self, get_name), getattr(self, restore_name)
262n/a
263n/a def __enter__(self):
264n/a self.saved_values = dict((name, get()) for name, get, restore
265n/a in self.resource_info())
266n/a return self
267n/a
268n/a def __exit__(self, exc_type, exc_val, exc_tb):
269n/a saved_values = self.saved_values
270n/a del self.saved_values
271n/a support.gc_collect() # Some resources use weak references
272n/a for name, get, restore in self.resource_info():
273n/a current = get()
274n/a original = saved_values.pop(name)
275n/a # Check for changes to the resource's value
276n/a if current != original:
277n/a self.changed = True
278n/a restore(original)
279n/a if not self.quiet and not self.pgo:
280n/a print(f"Warning -- {name} was modified by {self.testname}",
281n/a file=sys.stderr, flush=True)
282n/a if self.verbose > 1:
283n/a print(f" Before: {original}\n After: {current} ",
284n/a file=sys.stderr, flush=True)
285n/a return False