from __future__ import absolute_import
from celery.five import range
from celery.utils.dispatch.saferef import safe_ref
from celery.tests.case import Case
class Class1(object):
def x(self):
pass
def fun(obj):
pass
class Class2(object):
def __call__(self, obj):
pass
class SaferefTests(Case):
def setUp(self):
ts = []
ss = []
for x in range(5000):
t = Class1()
ts.append(t)
s = safe_ref(t.x, self._closure)
ss.append(s)
ts.append(fun)
ss.append(safe_ref(fun, self._closure))
for x in range(30):
t = Class2()
ts.append(t)
s = safe_ref(t, self._closure)
ss.append(s)
self.ts = ts
self.ss = ss
self.closureCount = 0
def tearDown(self):
del self.ts
del self.ss
def test_in(self):
"""test_in
Test the "in" operator for safe references (cmp)
"""
for t in self.ts[:50]:
self.assertTrue(safe_ref(t.x) in self.ss)
def test_valid(self):
"""test_value
Test that the references are valid (return instance methods)
"""
for s in self.ss:
self.assertTrue(s())
def test_shortcircuit(self):
"""test_shortcircuit
Test that creation short-circuits to reuse existing references
"""
sd = {}
for s in self.ss:
sd[s] = 1
for t in self.ts:
if hasattr(t, 'x'):
self.assertIn(safe_ref(t.x), sd)
else:
self.assertIn(safe_ref(t), sd)
def test_representation(self):
"""test_representation
Test that the reference object's representation works
XXX Doesn't currently check the results, just that no error
is raised
"""
repr(self.ss[-1])
def _closure(self, ref):
"""Dumb utility mechanism to increment deletion counter"""
self.closureCount += 1