import operator
import contextlib
import copy
+import time
from test import test_support
self.cycle = self
+@contextlib.contextmanager
+def collect_in_thread(period=0.0001):
+ """
+ Ensure GC collections happen in a different thread, at a high frequency.
+ """
+ threading = test_support.import_module('threading')
+ please_stop = False
+
+ def collect():
+ while not please_stop:
+ time.sleep(period)
+ gc.collect()
+
+ with test_support.disable_gc():
+ old_interval = sys.getcheckinterval()
+ sys.setcheckinterval(20)
+ t = threading.Thread(target=collect)
+ t.start()
+ try:
+ yield
+ finally:
+ please_stop = True
+ t.join()
+ sys.setcheckinterval(old_interval)
+
+
class TestBase(unittest.TestCase):
def setUp(self):
self.assertEqual(len(d), 0)
self.assertEqual(count, 2)
+ def test_threaded_weak_valued_setdefault(self):
+ d = weakref.WeakValueDictionary()
+ with collect_in_thread():
+ for i in range(50000):
+ x = d.setdefault(10, RefCycle())
+ self.assertIsNot(x, None) # we never put None in there!
+ del x
+
+ def test_threaded_weak_valued_pop(self):
+ d = weakref.WeakValueDictionary()
+ with collect_in_thread():
+ for i in range(50000):
+ d[10] = RefCycle()
+ x = d.pop(10, 10)
+ self.assertIsNot(x, None) # we never put None in there!
+
+
from test import mapping_tests
class WeakValueDictionaryTestCase(mapping_tests.BasicTestMappingProtocol):
try:
o = self.data.pop(key)()
except KeyError:
+ o = None
+ if o is None:
if args:
return args[0]
- raise
- if o is None:
- raise KeyError, key
+ else:
+ raise KeyError, key
else:
return o
def setdefault(self, key, default=None):
try:
- wr = self.data[key]
+ o = self.data[key]()
except KeyError:
+ o = None
+ if o is None:
if self._pending_removals:
self._commit_removals()
self.data[key] = KeyedRef(default, self._remove, key)
return default
else:
- return wr()
+ return o
def update(*args, **kwargs):
if not args: