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

Python code coverage for Lib/test/test_threading_local.py

#countcontent
1n/aimport unittest
2n/afrom doctest import DocTestSuite
3n/afrom test import support
4n/aimport weakref
5n/aimport gc
6n/a
7n/a# Modules under test
8n/a_thread = support.import_module('_thread')
9n/athreading = support.import_module('threading')
10n/aimport _threading_local
11n/a
12n/a
13n/aclass Weak(object):
14n/a pass
15n/a
16n/adef target(local, weaklist):
17n/a weak = Weak()
18n/a local.weak = weak
19n/a weaklist.append(weakref.ref(weak))
20n/a
21n/a
22n/aclass BaseLocalTest:
23n/a
24n/a def test_local_refs(self):
25n/a self._local_refs(20)
26n/a self._local_refs(50)
27n/a self._local_refs(100)
28n/a
29n/a def _local_refs(self, n):
30n/a local = self._local()
31n/a weaklist = []
32n/a for i in range(n):
33n/a t = threading.Thread(target=target, args=(local, weaklist))
34n/a t.start()
35n/a t.join()
36n/a del t
37n/a
38n/a gc.collect()
39n/a self.assertEqual(len(weaklist), n)
40n/a
41n/a # XXX _threading_local keeps the local of the last stopped thread alive.
42n/a deadlist = [weak for weak in weaklist if weak() is None]
43n/a self.assertIn(len(deadlist), (n-1, n))
44n/a
45n/a # Assignment to the same thread local frees it sometimes (!)
46n/a local.someothervar = None
47n/a gc.collect()
48n/a deadlist = [weak for weak in weaklist if weak() is None]
49n/a self.assertIn(len(deadlist), (n-1, n), (n, len(deadlist)))
50n/a
51n/a def test_derived(self):
52n/a # Issue 3088: if there is a threads switch inside the __init__
53n/a # of a threading.local derived class, the per-thread dictionary
54n/a # is created but not correctly set on the object.
55n/a # The first member set may be bogus.
56n/a import time
57n/a class Local(self._local):
58n/a def __init__(self):
59n/a time.sleep(0.01)
60n/a local = Local()
61n/a
62n/a def f(i):
63n/a local.x = i
64n/a # Simply check that the variable is correctly set
65n/a self.assertEqual(local.x, i)
66n/a
67n/a with support.start_threads(threading.Thread(target=f, args=(i,))
68n/a for i in range(10)):
69n/a pass
70n/a
71n/a def test_derived_cycle_dealloc(self):
72n/a # http://bugs.python.org/issue6990
73n/a class Local(self._local):
74n/a pass
75n/a locals = None
76n/a passed = False
77n/a e1 = threading.Event()
78n/a e2 = threading.Event()
79n/a
80n/a def f():
81n/a nonlocal passed
82n/a # 1) Involve Local in a cycle
83n/a cycle = [Local()]
84n/a cycle.append(cycle)
85n/a cycle[0].foo = 'bar'
86n/a
87n/a # 2) GC the cycle (triggers threadmodule.c::local_clear
88n/a # before local_dealloc)
89n/a del cycle
90n/a gc.collect()
91n/a e1.set()
92n/a e2.wait()
93n/a
94n/a # 4) New Locals should be empty
95n/a passed = all(not hasattr(local, 'foo') for local in locals)
96n/a
97n/a t = threading.Thread(target=f)
98n/a t.start()
99n/a e1.wait()
100n/a
101n/a # 3) New Locals should recycle the original's address. Creating
102n/a # them in the thread overwrites the thread state and avoids the
103n/a # bug
104n/a locals = [Local() for i in range(10)]
105n/a e2.set()
106n/a t.join()
107n/a
108n/a self.assertTrue(passed)
109n/a
110n/a def test_arguments(self):
111n/a # Issue 1522237
112n/a class MyLocal(self._local):
113n/a def __init__(self, *args, **kwargs):
114n/a pass
115n/a
116n/a MyLocal(a=1)
117n/a MyLocal(1)
118n/a self.assertRaises(TypeError, self._local, a=1)
119n/a self.assertRaises(TypeError, self._local, 1)
120n/a
121n/a def _test_one_class(self, c):
122n/a self._failed = "No error message set or cleared."
123n/a obj = c()
124n/a e1 = threading.Event()
125n/a e2 = threading.Event()
126n/a
127n/a def f1():
128n/a obj.x = 'foo'
129n/a obj.y = 'bar'
130n/a del obj.y
131n/a e1.set()
132n/a e2.wait()
133n/a
134n/a def f2():
135n/a try:
136n/a foo = obj.x
137n/a except AttributeError:
138n/a # This is expected -- we haven't set obj.x in this thread yet!
139n/a self._failed = "" # passed
140n/a else:
141n/a self._failed = ('Incorrectly got value %r from class %r\n' %
142n/a (foo, c))
143n/a sys.stderr.write(self._failed)
144n/a
145n/a t1 = threading.Thread(target=f1)
146n/a t1.start()
147n/a e1.wait()
148n/a t2 = threading.Thread(target=f2)
149n/a t2.start()
150n/a t2.join()
151n/a # The test is done; just let t1 know it can exit, and wait for it.
152n/a e2.set()
153n/a t1.join()
154n/a
155n/a self.assertFalse(self._failed, self._failed)
156n/a
157n/a def test_threading_local(self):
158n/a self._test_one_class(self._local)
159n/a
160n/a def test_threading_local_subclass(self):
161n/a class LocalSubclass(self._local):
162n/a """To test that subclasses behave properly."""
163n/a self._test_one_class(LocalSubclass)
164n/a
165n/a def _test_dict_attribute(self, cls):
166n/a obj = cls()
167n/a obj.x = 5
168n/a self.assertEqual(obj.__dict__, {'x': 5})
169n/a with self.assertRaises(AttributeError):
170n/a obj.__dict__ = {}
171n/a with self.assertRaises(AttributeError):
172n/a del obj.__dict__
173n/a
174n/a def test_dict_attribute(self):
175n/a self._test_dict_attribute(self._local)
176n/a
177n/a def test_dict_attribute_subclass(self):
178n/a class LocalSubclass(self._local):
179n/a """To test that subclasses behave properly."""
180n/a self._test_dict_attribute(LocalSubclass)
181n/a
182n/a def test_cycle_collection(self):
183n/a class X:
184n/a pass
185n/a
186n/a x = X()
187n/a x.local = self._local()
188n/a x.local.x = x
189n/a wr = weakref.ref(x)
190n/a del x
191n/a gc.collect()
192n/a self.assertIsNone(wr())
193n/a
194n/a
195n/aclass ThreadLocalTest(unittest.TestCase, BaseLocalTest):
196n/a _local = _thread._local
197n/a
198n/aclass PyThreadingLocalTest(unittest.TestCase, BaseLocalTest):
199n/a _local = _threading_local.local
200n/a
201n/a
202n/adef test_main():
203n/a suite = unittest.TestSuite()
204n/a suite.addTest(DocTestSuite('_threading_local'))
205n/a suite.addTest(unittest.makeSuite(ThreadLocalTest))
206n/a suite.addTest(unittest.makeSuite(PyThreadingLocalTest))
207n/a
208n/a local_orig = _threading_local.local
209n/a def setUp(test):
210n/a _threading_local.local = _thread._local
211n/a def tearDown(test):
212n/a _threading_local.local = local_orig
213n/a suite.addTest(DocTestSuite('_threading_local',
214n/a setUp=setUp, tearDown=tearDown)
215n/a )
216n/a
217n/a support.run_unittest(suite)
218n/a
219n/aif __name__ == '__main__':
220n/a test_main()