ó §«Õ\c@sdZddd„ƒYZddlmZddlZddlmZddlm Z yddl m Z Wn*e k r•dd l mZd „Z nXd „Zd Zed Zd\ZZededeƒd„ƒZeeeƒƒZd„Zd„Zdejfd„ƒYZdS(s»The Python interpreter may switch between threads inbetween bytecode execution. Bytecode execution in fastcache may occur during: (1) Calls to make_key which will call the __hash__ methods of the args and (2) `PyDict_Get(Set)Item` calls rely on Python comparisons (i.e, __eq__) to determine if a match has been found A good test for threadsafety is then to cache a function which takes user defined Python objects that have __hash__ and __eq__ methods which live in Python land rather built-in land. The test should not only ensure that the correct result is acheived (and no segfaults) but also assess memory leaks. The thread switching interval can be altered using sys.setswitchinterval. t PythonIntcBs)eZdZd„Zd„Zd„ZRS(sD Wrapper for an integer with python versions of __eq__ and __hash__.cCs ||_dS(N(tvalue(tselftval((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyt__init__scCs t|jƒS(N(thashR(R((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyt__hash__scCs8t|tƒs(tdt|ƒƒ‚n|j|jkS(Ns"PythonInt cannot be compared to %s(t isinstanceRt TypeErrorttypeR(Rtother((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyt__eq__s(t__name__t __module__t__doc__RRR (((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyRs  iÿÿÿÿ(trandintN(t clru_cache(tThread(tsetswitchinterval(tsetcheckintervalcCstt|ƒƒS(N(Rtint(ti((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyt setinterval(scCs:x|D]}|jƒqWx|D]}|jƒq"WdS(N(tstarttjoin(tthreadstt((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyt run_threads,s  i-ii tmaxsizettypedcCsA|j}|dkr|Stt|dƒƒtt|dƒƒS(s$Terrible Fibonacci number generator.ii(RtfibR(tntv((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyR6s cCstxmt|ƒD]_}tttƒtkr5tjƒntttƒƒ}t|kr t dt|fƒ‚q q WdS(s" Run Fibonacci generator r times. sExpected %d, Got %dN( trangeRtRAND_MINtRAND_MAXRt cache_clearRtFIBtRESULTt ValueError(trRtres((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pytrun_fib_with_clear?s   cCsRxKt|ƒD]=}tttƒƒ}t|kr tdt|fƒ‚q q WdS(s" Run Fibonacci generator r times. sExpected %d, Got %dN(R!RRR%R&R'(R(RR)((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pytrun_fib_with_statsHs tTest_ThreadingcBs)eZdZd„Zd„Zd„ZRS(s# Threadsafety Tests for lru_cache. cCs tdƒd|_d|_dS(Ngíµ ÷Æ°>iiè(Rt numthreadstrepeat(R((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pytsetUpSs  cCsUgt|jƒD]!}tdtd|jfƒ^q}t|ƒ|jddƒdS(s/ randomly clear the cache during calls to fib. ttargettargsiN(R!R-RR*R.Rt assertEqual(Rt_R((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyttest_thread_random_cache_clearsXs4 cCs‡tjƒgt|jƒD]!}tdtd|jfƒ^q}t|ƒtjƒ\}}}}|j |t ƒ|j |t ƒdS(sN Run thread safety test to make sure the cache statistics are correct.R0R1N( RR$R!R-RR+R.Rt cache_infoR2t CACHE_SIZE(RR3RthitstmissesRtcurrsize((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyttest_thread_cache_infoas 4 (R R RR/R4R:(((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyR,Ps  ((ii (RRtrandomRtunittestt fastcacheRt lru_cachet threadingRtsysRRt ImportErrorRRR6R%R"R#tFalseRR&R*R+tTestCaseR,(((s:lib/python2.7/site-packages/fastcache/tests/test_thread.pyts&