+++ /dev/null
-import unittest\r
-from doctest import DocTestSuite\r
-from test import test_support\r
-import weakref\r
-import gc\r
-\r
-# Modules under test\r
-_thread = test_support.import_module('thread')\r
-threading = test_support.import_module('threading')\r
-import _threading_local\r
-\r
-\r
-class Weak(object):\r
- pass\r
-\r
-def target(local, weaklist):\r
- weak = Weak()\r
- local.weak = weak\r
- weaklist.append(weakref.ref(weak))\r
-\r
-class BaseLocalTest:\r
-\r
- def test_local_refs(self):\r
- self._local_refs(20)\r
- self._local_refs(50)\r
- self._local_refs(100)\r
-\r
- def _local_refs(self, n):\r
- local = self._local()\r
- weaklist = []\r
- for i in range(n):\r
- t = threading.Thread(target=target, args=(local, weaklist))\r
- t.start()\r
- t.join()\r
- del t\r
-\r
- gc.collect()\r
- self.assertEqual(len(weaklist), n)\r
-\r
- # XXX _threading_local keeps the local of the last stopped thread alive.\r
- deadlist = [weak for weak in weaklist if weak() is None]\r
- self.assertIn(len(deadlist), (n-1, n))\r
-\r
- # Assignment to the same thread local frees it sometimes (!)\r
- local.someothervar = None\r
- gc.collect()\r
- deadlist = [weak for weak in weaklist if weak() is None]\r
- self.assertIn(len(deadlist), (n-1, n), (n, len(deadlist)))\r
-\r
- def test_derived(self):\r
- # Issue 3088: if there is a threads switch inside the __init__\r
- # of a threading.local derived class, the per-thread dictionary\r
- # is created but not correctly set on the object.\r
- # The first member set may be bogus.\r
- import time\r
- class Local(self._local):\r
- def __init__(self):\r
- time.sleep(0.01)\r
- local = Local()\r
-\r
- def f(i):\r
- local.x = i\r
- # Simply check that the variable is correctly set\r
- self.assertEqual(local.x, i)\r
-\r
- threads= []\r
- for i in range(10):\r
- t = threading.Thread(target=f, args=(i,))\r
- t.start()\r
- threads.append(t)\r
-\r
- for t in threads:\r
- t.join()\r
-\r
- def test_derived_cycle_dealloc(self):\r
- # http://bugs.python.org/issue6990\r
- class Local(self._local):\r
- pass\r
- locals = None\r
- passed = [False]\r
- e1 = threading.Event()\r
- e2 = threading.Event()\r
-\r
- def f():\r
- # 1) Involve Local in a cycle\r
- cycle = [Local()]\r
- cycle.append(cycle)\r
- cycle[0].foo = 'bar'\r
-\r
- # 2) GC the cycle (triggers threadmodule.c::local_clear\r
- # before local_dealloc)\r
- del cycle\r
- gc.collect()\r
- e1.set()\r
- e2.wait()\r
-\r
- # 4) New Locals should be empty\r
- passed[0] = all(not hasattr(local, 'foo') for local in locals)\r
-\r
- t = threading.Thread(target=f)\r
- t.start()\r
- e1.wait()\r
-\r
- # 3) New Locals should recycle the original's address. Creating\r
- # them in the thread overwrites the thread state and avoids the\r
- # bug\r
- locals = [Local() for i in range(10)]\r
- e2.set()\r
- t.join()\r
-\r
- self.assertTrue(passed[0])\r
-\r
- def test_arguments(self):\r
- # Issue 1522237\r
- from thread import _local as local\r
- from _threading_local import local as py_local\r
-\r
- for cls in (local, py_local):\r
- class MyLocal(cls):\r
- def __init__(self, *args, **kwargs):\r
- pass\r
-\r
- MyLocal(a=1)\r
- MyLocal(1)\r
- self.assertRaises(TypeError, cls, a=1)\r
- self.assertRaises(TypeError, cls, 1)\r
-\r
- def _test_one_class(self, c):\r
- self._failed = "No error message set or cleared."\r
- obj = c()\r
- e1 = threading.Event()\r
- e2 = threading.Event()\r
-\r
- def f1():\r
- obj.x = 'foo'\r
- obj.y = 'bar'\r
- del obj.y\r
- e1.set()\r
- e2.wait()\r
-\r
- def f2():\r
- try:\r
- foo = obj.x\r
- except AttributeError:\r
- # This is expected -- we haven't set obj.x in this thread yet!\r
- self._failed = "" # passed\r
- else:\r
- self._failed = ('Incorrectly got value %r from class %r\n' %\r
- (foo, c))\r
- sys.stderr.write(self._failed)\r
-\r
- t1 = threading.Thread(target=f1)\r
- t1.start()\r
- e1.wait()\r
- t2 = threading.Thread(target=f2)\r
- t2.start()\r
- t2.join()\r
- # The test is done; just let t1 know it can exit, and wait for it.\r
- e2.set()\r
- t1.join()\r
-\r
- self.assertFalse(self._failed, self._failed)\r
-\r
- def test_threading_local(self):\r
- self._test_one_class(self._local)\r
-\r
- def test_threading_local_subclass(self):\r
- class LocalSubclass(self._local):\r
- """To test that subclasses behave properly."""\r
- self._test_one_class(LocalSubclass)\r
-\r
- def _test_dict_attribute(self, cls):\r
- obj = cls()\r
- obj.x = 5\r
- self.assertEqual(obj.__dict__, {'x': 5})\r
- with self.assertRaises(AttributeError):\r
- obj.__dict__ = {}\r
- with self.assertRaises(AttributeError):\r
- del obj.__dict__\r
-\r
- def test_dict_attribute(self):\r
- self._test_dict_attribute(self._local)\r
-\r
- def test_dict_attribute_subclass(self):\r
- class LocalSubclass(self._local):\r
- """To test that subclasses behave properly."""\r
- self._test_dict_attribute(LocalSubclass)\r
-\r
-\r
-class ThreadLocalTest(unittest.TestCase, BaseLocalTest):\r
- _local = _thread._local\r
-\r
- # Fails for the pure Python implementation\r
- def test_cycle_collection(self):\r
- class X:\r
- pass\r
-\r
- x = X()\r
- x.local = self._local()\r
- x.local.x = x\r
- wr = weakref.ref(x)\r
- del x\r
- gc.collect()\r
- self.assertIs(wr(), None)\r
-\r
-class PyThreadingLocalTest(unittest.TestCase, BaseLocalTest):\r
- _local = _threading_local.local\r
-\r
-\r
-def test_main():\r
- suite = unittest.TestSuite()\r
- suite.addTest(DocTestSuite('_threading_local'))\r
- suite.addTest(unittest.makeSuite(ThreadLocalTest))\r
- suite.addTest(unittest.makeSuite(PyThreadingLocalTest))\r
-\r
- try:\r
- from thread import _local\r
- except ImportError:\r
- pass\r
- else:\r
- import _threading_local\r
- local_orig = _threading_local.local\r
- def setUp(test):\r
- _threading_local.local = _local\r
- def tearDown(test):\r
- _threading_local.local = local_orig\r
- suite.addTest(DocTestSuite('_threading_local',\r
- setUp=setUp, tearDown=tearDown)\r
- )\r
-\r
- test_support.run_unittest(suite)\r
-\r
-if __name__ == '__main__':\r
- test_main()\r
projects / mirror_edk2.git / blobdiff