Issue #4180: The warnings registries are now reset when the filters are modified.

This commit is contained in:
Antoine Pitrou 2014-09-18 02:40:46 +02:00
parent 1b38bc65dd
commit cb0a006fd1
4 changed files with 101 additions and 9 deletions

View File

@ -92,6 +92,16 @@ class FilterTests(BaseTest):
self.assertRaises(UserWarning, self.module.warn,
"FilterTests.test_error")
def test_error_after_default(self):
with original_warnings.catch_warnings(module=self.module) as w:
self.module.resetwarnings()
message = "FilterTests.test_ignore_after_default"
def f():
self.module.warn(message, UserWarning)
f()
self.module.filterwarnings("error", category=UserWarning)
self.assertRaises(UserWarning, f)
def test_ignore(self):
with original_warnings.catch_warnings(record=True,
module=self.module) as w:
@ -100,6 +110,19 @@ class FilterTests(BaseTest):
self.module.warn("FilterTests.test_ignore", UserWarning)
self.assertEqual(len(w), 0)
def test_ignore_after_default(self):
with original_warnings.catch_warnings(record=True,
module=self.module) as w:
self.module.resetwarnings()
message = "FilterTests.test_ignore_after_default"
def f():
self.module.warn(message, UserWarning)
f()
self.module.filterwarnings("ignore", category=UserWarning)
f()
f()
self.assertEqual(len(w), 1)
def test_always(self):
with original_warnings.catch_warnings(record=True,
module=self.module) as w:
@ -111,6 +134,26 @@ class FilterTests(BaseTest):
self.module.warn(message, UserWarning)
self.assertTrue(w[-1].message, message)
def test_always_after_default(self):
with original_warnings.catch_warnings(record=True,
module=self.module) as w:
self.module.resetwarnings()
message = "FilterTests.test_always_after_ignore"
def f():
self.module.warn(message, UserWarning)
f()
self.assertEqual(len(w), 1)
self.assertEqual(w[-1].message.args[0], message)
f()
self.assertEqual(len(w), 1)
self.module.filterwarnings("always", category=UserWarning)
f()
self.assertEqual(len(w), 2)
self.assertEqual(w[-1].message.args[0], message)
f()
self.assertEqual(len(w), 3)
self.assertEqual(w[-1].message.args[0], message)
def test_default(self):
with original_warnings.catch_warnings(record=True,
module=self.module) as w:
@ -506,7 +549,9 @@ class _WarningsTests(BaseTest, unittest.TestCase):
registry=registry)
self.assertEqual(w[-1].message, message)
self.assertEqual(len(w), 1)
self.assertEqual(len(registry), 1)
# One actual registry key plus the "version" key
self.assertEqual(len(registry), 2)
self.assertIn("version", registry)
del w[:]
# Test removal.
del self.module.defaultaction
@ -516,7 +561,7 @@ class _WarningsTests(BaseTest, unittest.TestCase):
registry=registry)
self.assertEqual(w[-1].message, message)
self.assertEqual(len(w), 1)
self.assertEqual(len(registry), 1)
self.assertEqual(len(registry), 2)
del w[:]
# Test setting.
self.module.defaultaction = "ignore"

View File

@ -53,6 +53,7 @@ def filterwarnings(action, message="", category=Warning, module="", lineno=0,
filters.append(item)
else:
filters.insert(0, item)
_filters_mutated()
def simplefilter(action, category=Warning, lineno=0, append=False):
"""Insert a simple entry into the list of warnings filters (at the front).
@ -73,10 +74,12 @@ def simplefilter(action, category=Warning, lineno=0, append=False):
filters.append(item)
else:
filters.insert(0, item)
_filters_mutated()
def resetwarnings():
"""Clear the list of warning filters, so that no filters are active."""
filters[:] = []
_filters_mutated()
class _OptionError(Exception):
"""Exception used by option processing helpers."""
@ -204,6 +207,9 @@ def warn_explicit(message, category, filename, lineno,
module = module[:-3] # XXX What about leading pathname?
if registry is None:
registry = {}
if registry.get('version', 0) != _filters_version:
registry.clear()
registry['version'] = _filters_version
if isinstance(message, Warning):
text = str(message)
category = message.__class__
@ -329,6 +335,7 @@ class catch_warnings(object):
self._entered = True
self._filters = self._module.filters
self._module.filters = self._filters[:]
self._module._filters_mutated()
self._showwarning = self._module.showwarning
if self._record:
log = []
@ -343,6 +350,7 @@ class catch_warnings(object):
if not self._entered:
raise RuntimeError("Cannot exit %r without entering first" % self)
self._module.filters = self._filters
self._module._filters_mutated()
self._module.showwarning = self._showwarning
@ -357,15 +365,22 @@ class catch_warnings(object):
_warnings_defaults = False
try:
from _warnings import (filters, _defaultaction, _onceregistry,
warn, warn_explicit)
warn, warn_explicit, _filters_mutated)
defaultaction = _defaultaction
onceregistry = _onceregistry
_warnings_defaults = True
except ImportError:
filters = []
defaultaction = "default"
onceregistry = {}
_filters_version = 1
def _filters_mutated():
global _filters_version
_filters_version += 1
# Module initialization
_processoptions(sys.warnoptions)

View File

@ -32,6 +32,9 @@ Core and Builtins
Library
-------
- Issue #4180: The warnings registries are now reset when the filters
are modified.
- Issue #22419: Limit the length of incoming HTTP request in wsgiref server to
65536 bytes and send a 414 error code for higher lengths. Patch contributed
by Devin Cook.

View File

@ -12,6 +12,7 @@ MODULE_NAME " provides basic warning filtering support.\n"
static PyObject *_filters; /* List */
static PyObject *_once_registry; /* Dict */
static PyObject *_default_action; /* String */
static long _filters_version;
_Py_IDENTIFIER(argv);
_Py_IDENTIFIER(stderr);
@ -178,17 +179,34 @@ get_filter(PyObject *category, PyObject *text, Py_ssize_t lineno,
static int
already_warned(PyObject *registry, PyObject *key, int should_set)
{
PyObject *already_warned;
PyObject *version_obj, *already_warned;
_Py_IDENTIFIER(version);
if (key == NULL)
return -1;
version_obj = _PyDict_GetItemId(registry, &PyId_version);
if (version_obj == NULL
|| !PyLong_CheckExact(version_obj)
|| PyLong_AsLong(version_obj) != _filters_version) {
PyDict_Clear(registry);
version_obj = PyLong_FromLong(_filters_version);
if (version_obj == NULL)
return -1;
if (_PyDict_SetItemId(registry, &PyId_version, version_obj) < 0) {
Py_DECREF(version_obj);
return -1;
}
Py_DECREF(version_obj);
}
else {
already_warned = PyDict_GetItem(registry, key);
if (already_warned != NULL) {
int rc = PyObject_IsTrue(already_warned);
if (rc != 0)
return rc;
}
}
/* This warning wasn't found in the registry, set it. */
if (should_set)
@ -750,6 +768,13 @@ warnings_warn_explicit(PyObject *self, PyObject *args, PyObject *kwds)
registry, NULL);
}
static PyObject *
warnings_filters_mutated(PyObject *self, PyObject *args)
{
_filters_version++;
Py_RETURN_NONE;
}
/* Function to issue a warning message; may raise an exception. */
@ -917,6 +942,8 @@ static PyMethodDef warnings_functions[] = {
warn_doc},
{"warn_explicit", (PyCFunction)warnings_warn_explicit,
METH_VARARGS | METH_KEYWORDS, warn_explicit_doc},
{"_filters_mutated", (PyCFunction)warnings_filters_mutated, METH_NOARGS,
NULL},
/* XXX(brett.cannon): add showwarning? */
/* XXX(brett.cannon): Reasonable to add formatwarning? */
{NULL, NULL} /* sentinel */
@ -1069,5 +1096,7 @@ _PyWarnings_Init(void)
Py_INCREF(_default_action);
if (PyModule_AddObject(m, "_defaultaction", _default_action) < 0)
return NULL;
_filters_version = 0;
return m;
}