unit tests for cache framework uses whatever cache backend is set in the test settings file functionsclasses for complex data type tests the dummy cache backend doesn t really behave like a test backend so it has its own test case dummy cache can t do touch self assertiscache touchwhatever false def testdatatypesself all data types are ignored equally by the dummy cache tests string this is a string int 42 bool true list 1 2 3 4 tuple 1 2 3 4 dict a 1 b 2 function f class c for key value in tests items with self subtestkeykey cache setkey value self assertisnonecache getkey def testexpirationself expiration has no effect on the dummy cache cache setexpire1 very quickly 1 cache setexpire2 very quickly 1 cache setexpire3 very quickly 1 time sleep2 self assertisnonecache getexpire1 self assertiscache addexpire2 newvalue true self assertisnonecache getexpire2 self assertiscache haskeyexpire3 false def testunicodeself unicode values are ignored by the dummy cache stuff ascii asciivalue unicodeascii itrntinliztin1 itrntinliztin itrntinliztin2 ascii2 x 1 for key value in stuff items with self subtestkeykey cache setkey value self assertisnonecache getkey def testsetmanyself setmany does nothing for the dummy cache backend self assertequalcache setmanya 1 b 2 self assertequalcache setmanya 1 b 2 timeout2 version1 def testsetmanyinvalidkeyself msg keyerrorswithmemcachedmsg 1 key with spaces with self assertwarnsmessagecachekeywarning msg cache setmanykey with spaces foo def testdeletemanyself deletemany does nothing for the dummy cache backend cache deletemanya b def testdeletemanyinvalidkeyself msg keyerrorswithmemcachedmsg 1 key with spaces with self assertwarnsmessagecachekeywarning msg cache deletemanykey with spaces def testclearself clear does nothing for the dummy cache backend cache clear def testincrversionself dummy cache versions can t be incremented cache setanswer 42 with self assertraisesvalueerror cache incrversionanswer with self assertraisesvalueerror cache incrversiondoesnotexist def testdecrversionself dummy cache versions can t be decremented cache setanswer 42 with self assertraisesvalueerror cache decrversionanswer with self assertraisesvalueerror cache decrversiondoesnotexist def testgetorsetself self assertequalcache getorsetmykey default default self assertisnonecache getorsetmykey none def testgetorsetcallableself def mycallable return default self assertequalcache getorsetmykey mycallable default self assertequalcache getorsetmykey mycallable default def customkeyfunckey keyprefix version a customized cache key function return custom joinkeyprefix strversion key cachessettingbase default prefix keyprefix cacheprefix formatos getpid v2 version 2 customkey keyfunction customkeyfunc customkey2 keyfunction cache tests customkeyfunc cull options maxentries 30 zerocull options cullfrequency 0 maxentries 30 def cachessettingfortestsbasenone excludenone params base is used to pull in the memcached config from the original settings exclude is a set of cache names denoting which cachessettingbase keys should be omitted params are test specific overrides and cachessettingsbase is the base config for the tests this results in the following search order params cachessettingbase base base base or exclude exclude or set setting k base copy for k in cachessettingbase if k not in exclude for key cacheparams in setting items cacheparams updatecachessettingbasekey cacheparams updateparams return setting class basecachetests a common set of tests to apply to all cache backends factory requestfactory some clients raise custom exceptions when incr or decr are called with a noninteger value incrdecrtypeerror typeerror def teardownself cache clear def testsimpleself simple cache setget works cache setkey value self assertequalcache getkey value def testdefaultusedwhennoneissetself a key can be added to a cache test for same cache key conflicts between shared backend should not be set in the prefixed cache nonexistent cache keys return as nonedefault self assertisnonecache getdoesnotexist self assertequalcache getdoesnotexist bang bang def testgetmanyself multiple cache keys can be returned using getmany cache setmanya a b b c c d d self assertequal cache getmanya c d a a c c d d self assertequalcache getmanya b e a a b b self assertequalcache getmanyitera b e a a b b cache setmanyx none y 1 self assertequalcache getmanyx y x none y 1 def testdeleteself cache keys can be deleted cache setmanykey1 spam key2 eggs self assertequalcache getkey1 spam self assertiscache deletekey1 true self assertisnonecache getkey1 self assertequalcache getkey2 eggs def testdeletenonexistentself self assertiscache deletenonexistentkey false def testhaskeyself the cache can be inspected for cache keys cache sethello1 goodbye1 self assertiscache haskeyhello1 true self assertiscache haskeygoodbye1 false cache setnoexpiry here none self assertiscache haskeynoexpiry true cache setnull none self assertiscache haskeynull true def testinself the in operator can be used to inspect cache contents cache sethello2 goodbye2 self assertinhello2 cache self assertnotingoodbye2 cache cache setnull none self assertinnull cache def testincrself cache values can be incremented cache setanswer 41 self assertequalcache incranswer 42 self assertequalcache getanswer 42 self assertequalcache incranswer 10 52 self assertequalcache getanswer 52 self assertequalcache incranswer 10 42 with self assertraisesvalueerror cache incrdoesnotexist with self assertraisesvalueerror cache incrdoesnotexist 1 cache setnull none with self assertraisesself incrdecrtypeerror cache incrnull def testdecrself cache values can be decremented cache setanswer 43 self assertequalcache decranswer 42 self assertequalcache getanswer 42 self assertequalcache decranswer 10 32 self assertequalcache getanswer 32 self assertequalcache decranswer 10 42 with self assertraisesvalueerror cache decrdoesnotexist with self assertraisesvalueerror cache incrdoesnotexist 1 cache setnull none with self assertraisesself incrdecrtypeerror cache decrnull def testcloseself self asserttruehasattrcache close cache close def testdatatypesself many different data types can be cached tests string this is a string int 42 bool true list 1 2 3 4 tuple 1 2 3 4 dict a 1 b 2 function f class c for key value in tests items with self subtestkeykey cache setkey value self assertequalcache getkey value def testcachereadformodelinstanceself don t want fields with callable as default to be called on cache read expensivecalculation numruns 0 poll objects all delete mypoll poll objects createquestionwell self assertequalpoll objects count 1 pubdate mypoll pubdate cache setquestion mypoll cachedpoll cache getquestion self assertequalcachedpoll pubdate pubdate we only want the default expensive calculation run once self assertequalexpensivecalculation numruns 1 def testcachewriteformodelinstancewithdeferredself don t want fields with callable as default to be called on cache write expensivecalculation numruns 0 poll objects all delete poll objects createquestionwhat self assertequalexpensivecalculation numruns 1 deferqs poll objects deferquestion self assertequaldeferqs count 1 self assertequalexpensivecalculation numruns 1 cache setdeferredqueryset deferqs cache set should not reevaluate default functions self assertequalexpensivecalculation numruns 1 def testcachereadformodelinstancewithdeferredself don t want fields with callable as default to be called on cache read expensivecalculation numruns 0 poll objects all delete poll objects createquestionwhat self assertequalexpensivecalculation numruns 1 deferqs poll objects deferquestion self assertequaldeferqs count 1 cache setdeferredqueryset deferqs self assertequalexpensivecalculation numruns 1 runsbeforecacheread expensivecalculation numruns cache getdeferredqueryset we only want the default expensive calculation run on creation and set self assertequalexpensivecalculation numruns runsbeforecacheread def testexpirationself cache values can be set to expire cache setexpire1 very quickly 1 cache setexpire2 very quickly 1 cache setexpire3 very quickly 1 time sleep2 self assertisnonecache getexpire1 self assertiscache addexpire2 newvalue true self assertequalcache getexpire2 newvalue self assertiscache haskeyexpire3 false def testtouchself cache touch updates the timeout cache setexpire1 very quickly timeout1 self assertiscache touchexpire1 timeout4 true time sleep2 self assertiscache haskeyexpire1 true time sleep3 self assertiscache haskeyexpire1 false cache touch works without the timeout argument cache setexpire1 very quickly timeout1 self assertiscache touchexpire1 true time sleep2 self assertiscache haskeyexpire1 true self assertiscache touchnonexistent false def testunicodeself unicode values can be cached stuff ascii asciivalue unicodeascii itrntinliztin1 itrntinliztin itrntinliztin2 ascii2 x 1 test set for key value in stuff items with self subtestkeykey cache setkey value self assertequalcache getkey value test add for key value in stuff items with self subtestkeykey self assertiscache deletekey true self assertiscache addkey value true self assertequalcache getkey value test setmany for key value in stuff items self assertiscache deletekey true cache setmanystuff for key value in stuff items with self subtestkeykey self assertequalcache getkey value def testbinarystringself binary strings should be cacheable from zlib import compress decompress value valuetobecompressed compressedvalue compressvalue encode test set cache setbinary1 compressedvalue compressedresult cache getbinary1 self assertequalcompressedvalue compressedresult self assertequalvalue decompresscompressedresult decode test add self assertiscache addbinary1add compressedvalue true compressedresult cache getbinary1add self assertequalcompressedvalue compressedresult self assertequalvalue decompresscompressedresult decode test setmany cache setmanybinary1setmany compressedvalue compressedresult cache getbinary1setmany self assertequalcompressedvalue compressedresult self assertequalvalue decompresscompressedresult decode def testsetmanyself multiple keys can be set using setmany cache setmanykey1 spam key2 eggs self assertequalcache getkey1 spam self assertequalcache getkey2 eggs def testsetmanyreturnsemptylistonsuccessself setmany takes a second timeout parameter multiple keys can be deleted using deletemany the cache can be emptied using clear follow memcached s convention where a timeout greater than 30 days is treated as an absolute expiration timestamp instead of a relative offset 12399 passing in none into timeout results in a value that is cached forever passing in zero into timeout results in a value that is not cached make sure a timeout given as a float doesn t crash anything create initial cache key entries this will overflow the cache causing a cull count how many keys are left in the cache force cull to delete on first cached record all the builtin backends should warn except memcached that should error on keys that would be refused by memcached this encourages portable caching code without making it too difficult to use production backends with more liberal key rules refs 6447 mimic custom makekey method being defined since the default will never show the below warnings memcached doesn t allow whitespace or control characters in keys memcached limits key length to 250 custom makekey that adds a version to the key and exceeds the limit set using default version 1 set default version 1 but manually override version 2 v2 set using default version 2 v2 set default version 2 but manually override version 1 add default version 1 but manually override version 2 v2 add using default version 2 v2 add default version 2 but manually override version 1 haskey set using default version 1 set default version 1 but manually override version 2 v2 set using default version 2 v2 set default version 2 but manually override version 1 two caches with different key functions aren t visible to each other shouldn t fail silently if trying to cache an unpicklable type previous getorset stores none in the cache previous getorset stores none in the cache simulate cache add failing to add a value in that case the default value should be returned spaces are used in the table name to ensure quotingescaping is working the super calls needs to happen first for the settings override the super call needs to happen first because it uses the database force cull to delete on first cached record column names are quoted the rowcount attribute should not be checked on a closed cursor use another table name to avoid the table already exists message delete and recreate cache table with legacy behavior explicitly specifying the table name column names are quoted a router that puts the cache table on the other database def dbforreadself model hints if model meta applabel djangocache return other return none def dbforwriteself model hints if model meta applabel djangocache return other return none def allowmigrateself db applabel hints if applabel djangocache return db other return none overridesettings caches default backend django core cache backends db databasecache location mycachetable class createcachetablefordbcacheteststestcase databases default other overridesettingsdatabaseroutersdbcacherouter def testcreatecachetableobservesdatabaserouterself cache table should not be created on default with self assertnumqueries0 usingdefault management callcommandcreatecachetable databasedefault verbosity0 cache table should be created on other queries 1 check table doesn t already exist 2 create savepoint if transactional ddl is supported 3 create the table 4 create the index 5 release savepoint if transactional ddl is supported num 5 if connectionsother features canrollbackddl else 3 with self assertnumqueriesnum usingother management callcommandcreatecachetable databaseother verbosity0 class picklingsideeffect def initself cache self cache cache self locked false def getstateself self locked self cache lock locked return limitlocmementries overridesettings cachescachessettingfortests backenddjango core cache backends locmem locmemcache optionsmaxentries 9 overridesettings cachescachessettingfortests backenddjango core cache backends locmem locmemcache class locmemcachetestsbasecachetests testcase def setupself super setup locmem requires a hack to make the other caches share a data store with the normal cache cachesprefix cache cache cache cachesprefix expireinfo cache expireinfo cachesv2 cache cache cache cachesv2 expireinfo cache expireinfo cachescustomkey cache cache cache cachescustomkey expireinfo cache expireinfo cachescustomkey2 cache cache cache cachescustomkey2 expireinfo cache expireinfo overridesettings caches default backend django core cache backends locmem locmemcache other backend django core cache backends locmem locmemcache location other def testmultiplecachesself multiple locmem caches are isolated cache setvalue 42 self assertequalcachesdefault getvalue 42 self assertisnonecachesother getvalue def testlockingonpickleself incrdecr does not modify expiry time matches memcached behavior key value key cache makekeykey cache setkey 1 timeoutcache defaulttimeout 10 expire cache expireinfokey self assertequalcache incrkey 2 self assertequalexpire cache expireinfokey self assertequalcache decrkey 1 self assertequalexpire cache expireinfokey limitlocmementries def testlrugetself set moves cache keys for key in range9 cache setkey key timeoutnone for key in range3 9 cache setkey key timeoutnone cache set9 9 timeoutnone for key in range3 10 self assertequalcache getkey key for key in range3 self assertisnonecache getkey limitlocmementries def testlruincrself memcached and redis backends aren t guaranteed to be available to check the backends the test settings file will need to contain at least one cache backend setting that points at your cache server the memcached backends don t support cullrelated options like maxentries the redis backend does not support cullrelated options like maxentries by default it s assumed that the client doesn t clean up connections properly in which case the backend must do so after each request while other backends merely warn memcached should raise for an invalid key makekey adds a version to the key and exceeds the limit regression test for 22845 regression test for 22845 606024365 1 year by default memcached allows objects up to 1mb for the cachedb session backend to always use the current session memcached needs to delete the old key if it fails to set most clients e g pymemcache or pylibmc raise when the value is too large this test is primarily checking that the key was deleted so the returnexception behavior for the set itself is not important smallvalue should be deleted or set if configured to accept larger values for clients that don t manage their connections properly the connection is closed when the request is complete libmemcached manages its own connections specific test cases for the filebased cache caches location cannot be modified through overridesettings modifysettings hence settings are manipulated directly here and the settingchanged signal is triggered manually call parent first as cache clear may recreate cache base directory returns the default instead of erroring tests for the ability to mixin a custom validatekey method to a custom cache backend that otherwise inherits from a builtin backend and override the default key validation refs 6447 this key is both longer than 250 characters and has spaces settings having cache arguments with a timeoutnone create caches that will set nonexpiring keys the 5 minute 300 seconds default expiration time for keys is defined in the implementation of the initializer method of the basecache type the default expiration time of a cache key is 5 minutes this value is defined in django core cache backends base basecache init caches that have the timeout parameter undefined in the default settings will use the default 5 minute timeout memory caches that have the timeout parameter set to none in the default settings with have none as the default timeout this means no timeout memory caches that have the timeout parameter unset will set cache keys having the default 5 minute timeout memory caches that have the timeout parameter set to none will set a non expiring key by default testcase for django utils cache functions host www example com path cachetest factory requestfactoryheadershost host def teardownself cache clear def getrequestcacheself methodget querystringnone updatecachenone request self getrequest self host self path method querystringquerystring request cacheupdatecache updatecache if updatecache else true return request def testpatchvaryheadersself headers initial vary new headers resulting vary none acceptencoding acceptencoding acceptencoding acceptencoding acceptencoding acceptencoding acceptencoding acceptencoding cookie acceptencoding cookie acceptencoding cookie acceptencoding acceptencoding cookie acceptencoding cookie acceptencoding acceptencoding cookie cookie acceptencoding none acceptencoding cookie acceptencoding cookie cookie acceptencoding acceptencoding cookie cookie acceptencoding cookie acceptencoding acceptencoding cookie cookie acceptencoding acceptlanguage cookie acceptlanguage cookie for initialvary newheaders resultingvary in headers with self subtestinitialvaryinitialvary newheadersnewheaders response httpresponse if initialvary is not none response headersvary initialvary patchvaryheadersresponse newheaders self assertequalresponse headersvary resultingvary def testgetcachekeyself request self factory getself path response httpresponse expect none if no headers have been set yet self assertisnonegetcachekeyrequest set headers to an empty list learncachekeyrequest response self assertequal getcachekeyrequest views decorators cache cachepage settingsprefix get 18a03f9c9649f7d684af5db3524f5c99 d41d8cd98f00b204e9800998ecf8427e a specified keyprefix is taken into account keyprefix localprefix learncachekeyrequest response keyprefixkeyprefix self assertequal getcachekeyrequest keyprefixkeyprefix views decorators cache cachepage localprefix get 18a03f9c9649f7d684af5db3524f5c99 d41d8cd98f00b204e9800998ecf8427e def testgetcachekeywithqueryself request self factory getself path test 1 response httpresponse expect none if no headers have been set yet self assertisnonegetcachekeyrequest set headers to an empty list learncachekeyrequest response the querystring is taken into account self assertequal getcachekeyrequest views decorators cache cachepage settingsprefix get beaf87a9a99ee81c673ea2d67ccbec2a d41d8cd98f00b204e9800998ecf8427e def testcachekeyvariesbyurlself request1 self factory getself path headershost sub1 example com learncachekeyrequest1 httpresponse request2 self factory getself path headershost sub2 example com learncachekeyrequest2 httpresponse self assertnotequalgetcachekeyrequest1 getcachekeyrequest2 def testlearncachekeyself request self factory headself path response httpresponse response headersvary pony make sure that the vary header is added to the key hash learncachekeyrequest response self assertequal getcachekeyrequest views decorators cache cachepage settingsprefix get 18a03f9c9649f7d684af5db3524f5c99 d41d8cd98f00b204e9800998ecf8427e def testpatchcachecontrolself tests initial cachecontrol kwargs to patchcachecontrol expected cachecontrol parts none private true private private true private nocache nocache setcookie nocachesetcookie nocache setcookie nocachesetcookie nocachesetcookie nocache true nocache nocachesetcookie nocachelink nocache true nocache nocachesetcookie nocache link nocachesetcookie nocachelink nocachesetcookie nocachelink nocache custom nocachesetcookie nocachelink nocachecustom test whether privatepublic attributes are mutually exclusive private private true private private public true public public public true public public private true private mustrevalidate maxage60 private public true mustrevalidate maxage60 public mustrevalidate maxage60 public private true mustrevalidate maxage60 private mustrevalidate maxage60 public true mustrevalidate maxage60 public ccdelimre re compilers s for initialcc newheaders expectedcc in tests with self subtestinitialccinitialcc newheadersnewheaders response httpresponse if initialcc is not none response headerscachecontrol initialcc patchcachecontrolresponse newheaders parts setccdelimre splitresponse headerscachecontrol self assertequalparts expectedcc overridesettings caches default backend django core cache backends locmem locmemcache keyprefix cacheprefix class prefixedcacheutilscacheutils pass overridesettings cachemiddlewareseconds60 cachemiddlewarekeyprefixtest caches default backend django core cache backends locmem locmemcache class cacheheadtestsimpletestcase path cachetest factory requestfactory def teardownself cache clear def setcacheself request msg return updatecachemiddlewarelambda req httpresponsemsgrequest def testheadcachescorrectlyself testcontent test content request self factory headself path request cacheupdatecache true self setcacherequest testcontent request self factory headself path request cacheupdatecache true getcachedata fetchfromcachemiddlewareemptyresponse processrequest request self assertisnotnonegetcachedata self assertequaltestcontent encode getcachedata content def testheadwithcachedgetself testcontent test content request self factory getself path request cacheupdatecache true self setcacherequest testcontent request self factory headself path getcachedata fetchfromcachemiddlewareemptyresponse processrequest request self assertisnotnonegetcachedata self assertequaltestcontent encode getcachedata content overridesettings cachemiddlewarekeyprefixsettingsprefix caches default backend django core cache backends locmem locmemcache languages en english es spanish class cachei18ntestsimpletestcase path cachetest factory requestfactory def teardownself cache clear overridesettingsusei18ntrue usetzfalse def testcachekeyi18ntranslationself request self factory getself path lang translation getlanguage response httpresponse key learncachekeyrequest response self assertin lang key cache keys should include the language name when translation is active key2 getcachekeyrequest self assertequalkey key2 def checkacceptlanguagevaryself acceptlanguage vary referencekey request self factory getself path request metahttpacceptlanguage acceptlanguage request metahttpacceptencoding gzip q1 0 identity q0 5 q0 response httpresponse response headersvary vary key learncachekeyrequest response key2 getcachekeyrequest self assertequalkey referencekey self assertequalkey2 referencekey overridesettingsusei18ntrue usetzfalse def testcachekeyi18ntranslationacceptlanguageself lang translation getlanguage self assertequallang en request self factory getself path request metahttpacceptencoding gzip q1 0 identity q0 5 q0 response httpresponse response headersvary acceptencoding key learncachekeyrequest response self assertin lang key cache keys should include the language name when translation is active self checkacceptlanguagevary enus cookie acceptlanguage acceptencoding key self checkacceptlanguagevary enus cookie acceptencoding acceptlanguage key self checkacceptlanguagevary enus en q0 8 acceptencoding acceptlanguage cookie key self checkacceptlanguagevary enus en q0 8 ko q0 6 acceptlanguage cookie acceptencoding key self checkacceptlanguagevary kokr ko q0 8 enus q0 5 en q0 3 acceptencoding cookie acceptlanguage key self checkacceptlanguagevary kokr ko q0 8 enus q0 6 en q0 4 acceptlanguage acceptencoding cookie key self checkacceptlanguagevary ko q1 0 en q0 5 cookie acceptlanguage acceptencoding key self checkacceptlanguagevary ko en cookie acceptencoding acceptlanguage key self checkacceptlanguagevary kokr enus acceptencoding acceptlanguage cookie key overridesettingsusei18nfalse usetztrue def testcachekeyi18ntimezoneself request self factory getself path tz timezone getcurrenttimezonename response httpresponse key learncachekeyrequest response self assertin tz key cache keys should include the time zone name when time zones are active key2 getcachekeyrequest self assertequalkey key2 overridesettingsusei18nfalse def testcachekeynoi18nself request self factory getself path lang translation getlanguage tz timezone getcurrenttimezonename response httpresponse key learncachekeyrequest response self assertnotin lang key cache keys shouldn t include the language name when i18n isn t active self assertnotin tz key cache keys shouldn t include the time zone name when i18n isn t active overridesettings cachemiddlewarekeyprefixtest cachemiddlewareseconds60 usei18ntrue def testmiddlewareself def setcacherequest lang msg def getresponsereq return httpresponsemsg translation activatelang return updatecachemiddlewaregetresponserequest cache with non empty request get request self factory getself path foo bar other true request cacheupdatecache true getcachedata fetchfromcachemiddlewareemptyresponse processrequest request first access cache must return none self assertisnonegetcachedata content check for cache with querystring def getresponsereq return httpresponsecontent updatecachemiddlewaregetresponserequest getcachedata fetchfromcachemiddlewareemptyresponse processrequest request cache must return content self assertisnotnonegetcachedata self assertequalgetcachedata content content encode different querystring cache must be empty request self factory getself path foo bar somethingelse true request cacheupdatecache true getcachedata fetchfromcachemiddlewareemptyresponse processrequest request self assertisnonegetcachedata i18n tests enmessage hello world esmessage hola mundo request self factory getself path request cacheupdatecache true setcacherequest en enmessage getcachedata fetchfromcachemiddlewareemptyresponse processrequest request the cache can be recovered self assertisnotnonegetcachedata self assertequalgetcachedata content enmessage encode change the session language and set content request self factory getself path request cacheupdatecache true setcacherequest es esmessage change again the language translation activateen retrieve the content from cache getcachedata fetchfromcachemiddlewareemptyresponse processrequest request self assertequalgetcachedata content enmessage encode change again the language translation activatees getcachedata fetchfromcachemiddlewareemptyresponse processrequest request self assertequalgetcachedata content esmessage encode reset the language translation deactivate overridesettings cachemiddlewarekeyprefixtest cachemiddlewareseconds60 def testmiddlewaredoesntcachestreamingresponseself request self factory getself path getcachedata fetchfromcachemiddlewareemptyresponse processrequest request self assertisnonegetcachedata def getstreamresponsereq return streaminghttpresponsecheck for cache with streaming content updatecachemiddlewaregetstreamresponserequest getcachedata fetchfromcachemiddlewareemptyresponse processrequest request self assertisnonegetcachedata overridesettings caches default backend django core cache backends locmem locmemcache keyprefix cacheprefix class prefixedcachei18ntestcachei18ntest pass def helloworldviewrequest value return httpresponsehello world s value def csrfviewrequest return httpresponsecsrfrequestcsrftoken overridesettings cachemiddlewarealiasother cachemiddlewarekeyprefixmiddlewareprefix cachemiddlewareseconds30 caches default backend django core cache backends locmem locmemcache other backend django core cache backends locmem locmemcache location other timeout 1 class cachemiddlewaretestsimpletestcase factory requestfactory def setupself self defaultcache cachesdefault self othercache cachesother def teardownself self defaultcache clear self othercache clear super teardown def testconstructorself if only one argument is passed in construction it s being used as middleware middleware cachemiddlewareemptyresponse now test object attributes against values defined in setup above self assertequalmiddleware cachetimeout 30 self assertequalmiddleware keyprefix middlewareprefix self assertequalmiddleware cachealias other self assertequalmiddleware cache self othercache if more arguments are being passed in construction it s being used as a decorator first test with defaults asviewdecorator cachemiddleware emptyresponse cachealiasnone keyprefixnone self assertequal asviewdecorator cachetimeout 30 timeout value for default cache i e 30 self assertequalasviewdecorator keyprefix value of defaultcachealias from django core cache self assertequalasviewdecorator cachealias default self assertequalasviewdecorator cache self defaultcache next test with custom values asviewdecoratorwithcustom cachemiddleware helloworldview cachetimeout60 cachealiasother keyprefixfoo self assertequalasviewdecoratorwithcustom cachetimeout 60 self assertequalasviewdecoratorwithcustom keyprefix foo self assertequalasviewdecoratorwithcustom cachealias other self assertequalasviewdecoratorwithcustom cache self othercache def testupdatecachemiddlewareconstructorself middleware updatecachemiddlewareemptyresponse self assertequalmiddleware cachetimeout 30 self assertisnonemiddleware pagetimeout self assertequalmiddleware keyprefix middlewareprefix self assertequalmiddleware cachealias other self assertequalmiddleware cache self othercache def testfetchcachemiddlewareconstructorself middleware fetchfromcachemiddlewareemptyresponse self assertequalmiddleware keyprefix middlewareprefix self assertequalmiddleware cachealias other self assertequalmiddleware cache self othercache def testmiddlewareself middleware cachemiddlewarehelloworldview prefixmiddleware cachemiddlewarehelloworldview keyprefixprefix1 timeoutmiddleware cachemiddlewarehelloworldview cachetimeout1 request self factory getview put the request through the request middleware result middleware processrequestrequest self assertisnoneresult response helloworldviewrequest 1 now put the response through the response middleware response middleware processresponserequest response repeating the request should result in a cache hit result middleware processrequestrequest self assertisnotnoneresult self assertequalresult content bhello world 1 the same request through a different middleware won t hit result prefixmiddleware processrequestrequest self assertisnoneresult the same request with a timeout will hit result timeoutmiddleware processrequestrequest self assertisnotnoneresult self assertequalresult content bhello world 1 def testviewdecoratorself decorate the same view with different cache decorators defaultview cachepage3helloworldview defaultwithprefixview cachepage3 keyprefixprefix1helloworldview explicitdefaultview cachepage3 cachedefaulthelloworldview explicitdefaultwithprefixview cachepage 3 cachedefault keyprefixprefix1 helloworldview otherview cachepage1 cacheotherhelloworldview otherwithprefixview cachepage1 cacheother keyprefixprefix2 helloworldview request self factory getview request the view once response defaultviewrequest 1 self assertequalresponse content bhello world 1 request again hit the cache response defaultviewrequest 2 self assertequalresponse content bhello world 1 requesting the same view with the explicit cache should yield the same result response explicitdefaultviewrequest 3 self assertequalresponse content bhello world 1 requesting with a prefix will hit a different cache key response explicitdefaultwithprefixviewrequest 4 self assertequalresponse content bhello world 4 hitting the same view again gives a cache hit response explicitdefaultwithprefixviewrequest 5 self assertequalresponse content bhello world 4 and going back to the implicit cache will hit the same cache response defaultwithprefixviewrequest 6 self assertequalresponse content bhello world 4 requesting from an alternate cache won t hit cache response otherviewrequest 7 self assertequalresponse content bhello world 7 but a repeated hit will hit cache response otherviewrequest 8 self assertequalresponse content bhello world 7 and prefixing the alternate cache yields yet another cache entry response otherwithprefixviewrequest 9 self assertequalresponse content bhello world 9 but if we wait a couple of seconds time sleep2 the default cache will still hit cachesdefault response defaultviewrequest 11 self assertequalresponse content bhello world 1 the default cache with a prefix will still hit response defaultwithprefixviewrequest 12 self assertequalresponse content bhello world 4 the explicit default cache will still hit response explicitdefaultviewrequest 13 self assertequalresponse content bhello world 1 the explicit default cache with a prefix will still hit response explicitdefaultwithprefixviewrequest 14 self assertequalresponse content bhello world 4 but a rapidly expiring cache won t hit response otherviewrequest 15 self assertequalresponse content bhello world 15 even if it has a prefix response otherwithprefixviewrequest 16 self assertequalresponse content bhello world 16 def testcachepagetimeoutself page timeout takes precedence over the maxage section of the cachecontrol tests 1 3 maxage pagetimeout 3 1 maxage pagetimeout for maxage pagetimeout in tests with self subtestmaxagemaxage pagetimeoutpagetimeout view cachepagetimeoutpagetimeout cachecontrolmaxagemaxagehelloworldview request self factory getview response viewrequest 1 self assertequalresponse content bhello world 1 time sleep1 response viewrequest 2 self assertequal response content bhello world 1 if pagetimeout maxage else bhello world 2 cache clear def testcachedcontrolprivatenotcachedself django must prevent caching of responses that set a userspecific and maybe security sensitive cookie in response to a cookieless request inserting a csrf cookie in a cookieless request prevented caching the view shouldn t be cached on the second call the cache instance is different for each thread threadcaches middleware cachemiddlewareemptyresponse def runner threadcaches appendmiddleware cache for in range2 thread threading threadtargetrunner thread start thread join self assertisnotthreadcaches0 threadcaches1 overridesettings cachemiddlewarekeyprefixsettingsprefix cachemiddlewareseconds1 caches default backend django core cache backends locmem locmemcache usei18nfalse class testwithtemplateresponsesimpletestcase path cachetest factory requestfactory def teardownself cache clear def testpatchvaryheadersself headers initial vary new headers resulting vary none acceptencoding acceptencoding acceptencoding acceptencoding acceptencoding acceptencoding acceptencoding acceptencoding cookie acceptencoding cookie acceptencoding cookie acceptencoding acceptencoding cookie acceptencoding cookie acceptencoding acceptencoding cookie cookie acceptencoding none acceptencoding cookie acceptencoding cookie cookie acceptencoding acceptencoding cookie cookie acceptencoding cookie acceptencoding acceptencoding cookie cookie acceptencoding for initialvary newheaders resultingvary in headers with self subtestinitialvaryinitialvary newheadersnewheaders template enginesdjango fromstringthis is a test response templateresponsehttprequest template if initialvary is not none response headersvary initialvary patchvaryheadersresponse newheaders self assertequalresponse headersvary resultingvary def testgetcachekeyself request self factory getself path template enginesdjango fromstringthis is a test response templateresponsehttprequest template keyprefix localprefix expect none if no headers have been set yet self assertisnonegetcachekeyrequest set headers to an empty list learncachekeyrequest response self assertequal getcachekeyrequest views decorators cache cachepage settingsprefix get 58a0a05c8a5620f813686ff969c26853 d41d8cd98f00b204e9800998ecf8427e a specified keyprefix is taken into account learncachekeyrequest response keyprefixkeyprefix self assertequal getcachekeyrequest keyprefixkeyprefix views decorators cache cachepage localprefix get 58a0a05c8a5620f813686ff969c26853 d41d8cd98f00b204e9800998ecf8427e def testgetcachekeywithqueryself request self factory getself path test 1 template enginesdjango fromstringthis is a test response templateresponsehttprequest template expect none if no headers have been set yet self assertisnonegetcachekeyrequest set headers to an empty list learncachekeyrequest response the querystring is taken into account self assertequal getcachekeyrequest views decorators cache cachepage settingsprefix get 0f1c2d56633c943073c4569d9a9502fe d41d8cd98f00b204e9800998ecf8427e class testmaketemplatefragmentkeysimpletestcase def testwithoutvaryonself key maketemplatefragmentkeya fragment self assertequal key template cache a fragment d41d8cd98f00b204e9800998ecf8427e def testwithonevaryonself key maketemplatefragmentkeyfoo abc self assertequalkey template cache foo 493e283d571a73056196f1a68efd0f66 def testwithmanyvaryonself key maketemplatefragmentkeybar abc def self assertequalkey template cache bar 17c1a507a0cb58384f4c639067a93520 def testproperescapingself key maketemplatefragmentkeyspam abc def self assertequalkey template cache spam 06c8ae8e8c430b69fb0a6443504153dc def testwithintsvaryonself key maketemplatefragmentkeyfoo 1 2 3 4 5 self assertequalkey template cache foo 7ae8fd2e0d25d651c683bdeebdb29461 def testwithunicodevaryonself key maketemplatefragmentkeyfoo 42 self assertequalkey template cache foo 7ced1c94e543668590ba39b3c08b0237 def testlongvaryonself key maketemplatefragmentkeyfoo x 10000 self assertequalkey template cache foo 3670b349b5124aa56bdb50678b02b23a class cachehandlertestsimpletestcase def testsameinstanceself cache1 cachesdefault cache2 cachesdefault self assertiscache1 cache2 def testperthreadself c def runner c appendcachesdefault for x in range2 t threading threadtargetrunner t start t join self assertisnotc0 c1 def testnonexistentaliasself msg the connection nonexistent doesn t exist with self assertraisesmessageinvalidcachebackenderror msg cachesnonexistent def testnonexistentbackendself testcaches cachehandler invalidbackend backend django nonexistent nonexistentbackend msg could not find backend django nonexistent nonexistentbackend no module named django nonexistent with self assertraisesmessageinvalidcachebackenderror msg testcachesinvalidbackend def testallself testcaches cachehandler cache1 backend django core cache backends dummy dummycache cache2 backend django core cache backends dummy dummycache self assertequaltestcaches allinitializedonlytrue cache1 testcachescache1 self assertequaltestcaches allinitializedonlytrue cache1 self assertequallentestcaches all 2 all initializes all caches self assertequallentestcaches allinitializedonlytrue 2 self assertequaltestcaches all testcaches allinitializedonlytrue unit tests for cache framework uses whatever cache backend is set in the test settings file functions classes for complex data type tests the dummy cache backend doesn t really behave like a test backend so it has its own test case dummy cache can t do touch base is used to pull in the memcached config from the original settings exclude is a set of cache names denoting which _caches_setting_base keys should be omitted params are test specific overrides and _caches_settings_base is the base config for the tests this results in the following search order params _caches_setting_base base a common set of tests to apply to all cache backends some clients raise custom exceptions when incr or decr are called with a non integer value simple cache set get works if none is cached get returns it instead of the default a key can be added to a cache test for same cache key conflicts between shared backend should not be set in the prefixed cache nonexistent cache keys return as none default multiple cache keys can be returned using get_many cache keys can be deleted the cache can be inspected for cache keys the in operator can be used to inspect cache contents cache values can be incremented cache values can be decremented many different data types can be cached don t want fields with callable as default to be called on cache read we only want the default expensive calculation run once don t want fields with callable as default to be called on cache write cache set should not re evaluate default functions don t want fields with callable as default to be called on cache read we only want the default expensive calculation run on creation and set cache values can be set to expire cache touch updates the timeout cache touch works without the timeout argument unicode values can be cached test set test add test set_many binary strings should be cacheable test set test add test set_many multiple keys can be set using set_many set_many returns an empty list when all keys are inserted set_many takes a second timeout parameter multiple keys can be deleted using delete_many the cache can be emptied using clear follow memcached s convention where a timeout greater than 30 days is treated as an absolute expiration timestamp instead of a relative offset 12399 30 days 1 second passing in none into timeout results in a value that is cached forever passing in zero into timeout results in a value that is not cached make sure a timeout given as a float doesn t crash anything create initial cache key entries this will overflow the cache causing a cull count how many keys are left in the cache force _cull to delete on first cached record all the builtin backends should warn except memcached that should error on keys that would be refused by memcached this encourages portable caching code without making it too difficult to use production backends with more liberal key rules refs 6447 mimic custom make_key method being defined since the default will never show the below warnings memcached doesn t allow whitespace or control characters in keys memcached limits key length to 250 custom make_key that adds a version to the key and exceeds the limit set using default version 1 set default version 1 but manually override version 2 v2 set using default version 2 v2 set default version 2 but manually override version 1 add default version 1 but manually override version 2 v2 add using default version 2 v2 add default version 2 but manually override version 1 has_key set using default version 1 set default version 1 but manually override version 2 v2 set using default version 2 v2 set default version 2 but manually override version 1 two caches with different key functions aren t visible to each other shouldn t fail silently if trying to cache an unpicklable type previous get_or_set stores none in the cache previous get_or_set stores none in the cache simulate cache add failing to add a value in that case the default value should be returned spaces are used in the table name to ensure quoting escaping is working the super calls needs to happen first for the settings override the super call needs to happen first because it uses the database force _cull to delete on first cached record column names are quoted the rowcount attribute should not be checked on a closed cursor use another table name to avoid the table already exists message delete and recreate cache table with legacy behavior explicitly specifying the table name column names are quoted a router that puts the cache table on the other database cache table should not be created on default cache table should be created on other queries 1 check table doesn t already exist 2 create savepoint if transactional ddl is supported 3 create the table 4 create the index 5 release savepoint if transactional ddl is supported locmem requires a hack to make the other caches share a data store with the normal cache 20613 18541 ensures pickling is done outside of the lock incr decr does not modify expiry time matches memcached behavior get moves cache keys set moves cache keys incr moves cache keys memcached and redis backends aren t guaranteed to be available to check the backends the test settings file will need to contain at least one cache backend setting that points at your cache server the memcached backends don t support cull related options like max_entries the redis backend does not support cull related options like max_entries by default it s assumed that the client doesn t clean up connections properly in which case the backend must do so after each request while other backends merely warn memcached should raise for an invalid key make_key adds a version to the key and exceeds the limit regression test for 22845 regression test for 22845 60 60 24 365 1 year by default memcached allows objects up to 1mb for the cache_db session backend to always use the current session memcached needs to delete the old key if it fails to set most clients e g pymemcache or pylibmc raise when the value is too large this test is primarily checking that the key was deleted so the return exception behavior for the set itself is not important small_value should be deleted or set if configured to accept larger values for clients that don t manage their connections properly the connection is closed when the request is complete libmemcached manages its own connections specific test cases for the file based cache caches location cannot be modified through override_settings modify_settings hence settings are manipulated directly here and the setting_changed signal is triggered manually call parent first as cache clear may recreate cache base directory returns the default instead of erroring tests for the ability to mixin a custom validate_key method to a custom cache backend that otherwise inherits from a builtin backend and override the default key validation refs 6447 this key is both longer than 250 characters and has spaces settings having cache arguments with a timeout none create caches that will set non expiring keys the 5 minute 300 seconds default expiration time for keys is defined in the implementation of the initializer method of the basecache type the default expiration time of a cache key is 5 minutes this value is defined in django core cache backends base basecache __init__ caches that have the timeout parameter undefined in the default settings will use the default 5 minute timeout memory caches that have the timeout parameter set to none in the default settings with have none as the default timeout this means no timeout memory caches that have the timeout parameter unset will set cache keys having the default 5 minute timeout memory caches that have the timeout parameter set to none will set a non expiring key by default testcase for django utils cache functions initial vary new headers resulting vary expect none if no headers have been set yet set headers to an empty list a specified key_prefix is taken into account expect none if no headers have been set yet set headers to an empty list the querystring is taken into account get_cache_key keys differ by fully qualified url instead of path make sure that the vary header is added to the key hash initial cache control kwargs to patch_cache_control expected cache control parts no cache test whether private public attributes are mutually exclusive cache with non empty request get first access cache must return none cache must return content different query_string cache must be empty i18n tests the cache can be recovered change the session language and set content change again the language retrieve the content from cache change again the language reset the language the constructor is correctly distinguishing between usage of cachemiddleware as middleware vs usage of cachemiddleware as view decorator and setting attributes appropriately if only one argument is passed in construction it s being used as middleware now test object attributes against values defined in setup above if more arguments are being passed in construction it s being used as a decorator first test with defaults timeout value for default cache i e 30 value of default_cache_alias from django core cache next test with custom values put the request through the request middleware now put the response through the response middleware repeating the request should result in a cache hit the same request through a different middleware won t hit the same request with a timeout _will_ hit decorate the same view with different cache decorators request the view once request again hit the cache requesting the same view with the explicit cache should yield the same result requesting with a prefix will hit a different cache key hitting the same view again gives a cache hit and going back to the implicit cache will hit the same cache requesting from an alternate cache won t hit cache but a repeated hit will hit cache and prefixing the alternate cache yields yet another cache entry but if we wait a couple of seconds the default cache will still hit the default cache with a prefix will still hit the explicit default cache will still hit the explicit default cache with a prefix will still hit but a rapidly expiring cache won t hit even if it has a prefix page timeout takes precedence over the max age section of the cache control max_age page_timeout max_age page_timeout responses with cache control private are not cached django must prevent caching of responses that set a user specific and maybe security sensitive cookie in response to a cookie less request inserting a csrf cookie in a cookie less request prevented caching the view shouldn t be cached on the second call the cache instance is different for each thread tests various headers w templateresponse most are probably redundant since they manipulate the same object anyway but the etag header is special because it relies on the content being complete which is not necessarily always the case with a templateresponse initial vary new headers resulting vary expect none if no headers have been set yet set headers to an empty list a specified key_prefix is taken into account expect none if no headers have been set yet set headers to an empty list the querystring is taken into account attempting to retrieve the same alias should yield the same instance requesting the same alias from separate threads should yield separate instances all initializes all caches
import copy
import io
import os
import pickle
import re
import shutil
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path
from unittest import mock, skipIf

from django.conf import settings
from django.core import management, signals
from django.core.cache import (
    DEFAULT_CACHE_ALIAS,
    CacheHandler,
    CacheKeyWarning,
    InvalidCacheKey,
    cache,
    caches,
)
from django.core.cache.backends.base import InvalidCacheBackendError
from django.core.cache.backends.redis import RedisCacheClient
from django.core.cache.utils import make_template_fragment_key
from django.db import close_old_connections, connection, connections
from django.db.backends.utils import CursorWrapper
from django.http import (
    HttpRequest,
    HttpResponse,
    HttpResponseNotModified,
    StreamingHttpResponse,
)
from django.middleware.cache import (
    CacheMiddleware,
    FetchFromCacheMiddleware,
    UpdateCacheMiddleware,
)
from django.middleware.csrf import CsrfViewMiddleware
from django.template import engines
from django.template.context_processors import csrf
from django.template.response import TemplateResponse
from django.test import (
    RequestFactory,
    SimpleTestCase,
    TestCase,
    TransactionTestCase,
    override_settings,
)
from django.test.signals import setting_changed
from django.test.utils import CaptureQueriesContext
from django.utils import timezone, translation
from django.utils.cache import (
    get_cache_key,
    learn_cache_key,
    patch_cache_control,
    patch_vary_headers,
)
from django.views.decorators.cache import cache_control, cache_page

from .models import Poll, expensive_calculation



def f():
    return 42


class C:
    def m(n):
        return 24


class Unpicklable:
    def __getstate__(self):
        raise pickle.PickleError()


def empty_response(request):
    return HttpResponse()


KEY_ERRORS_WITH_MEMCACHED_MSG = (
    "Cache key contains characters that will cause errors if used with memcached: %r"
)


@override_settings(
    CACHES={
        "default": {
            "BACKEND": "django.core.cache.backends.dummy.DummyCache",
        }
    }
)
class DummyCacheTests(SimpleTestCase):
    
    

    def test_simple(self):
        "Dummy cache backend ignores cache set calls"
        cache.set("key", "value")
        self.assertIsNone(cache.get("key"))

    def test_add(self):
        "Add doesn't do anything in dummy cache backend"
        self.assertIs(cache.add("addkey1", "value"), True)
        self.assertIs(cache.add("addkey1", "newvalue"), True)
        self.assertIsNone(cache.get("addkey1"))

    def test_non_existent(self):
        "Nonexistent keys aren't found in the dummy cache backend"
        self.assertIsNone(cache.get("does_not_exist"))
        self.assertEqual(cache.get("does_not_exist", "bang!"), "bang!")

    def test_get_many(self):
        "get_many returns nothing for the dummy cache backend"
        cache.set_many({"a": "a", "b": "b", "c": "c", "d": "d"})
        self.assertEqual(cache.get_many(["a", "c", "d"]), {})
        self.assertEqual(cache.get_many(["a", "b", "e"]), {})

    def test_get_many_invalid_key(self):
        msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ":1:key with spaces"
        with self.assertWarnsMessage(CacheKeyWarning, msg):
            cache.get_many(["key with spaces"])

    def test_delete(self):
        "Cache deletion is transparently ignored on the dummy cache backend"
        cache.set_many({"key1": "spam", "key2": "eggs"})
        self.assertIsNone(cache.get("key1"))
        self.assertIs(cache.delete("key1"), False)
        self.assertIsNone(cache.get("key1"))
        self.assertIsNone(cache.get("key2"))

    def test_has_key(self):
        "The has_key method doesn't ever return True for the dummy cache backend"
        cache.set("hello1", "goodbye1")
        self.assertIs(cache.has_key("hello1"), False)
        self.assertIs(cache.has_key("goodbye1"), False)

    def test_in(self):
        "The in operator doesn't ever return True for the dummy cache backend"
        cache.set("hello2", "goodbye2")
        self.assertNotIn("hello2", cache)
        self.assertNotIn("goodbye2", cache)

    def test_incr(self):
        "Dummy cache values can't be incremented"
        cache.set("answer", 42)
        with self.assertRaises(ValueError):
            cache.incr("answer")
        with self.assertRaises(ValueError):
            cache.incr("does_not_exist")
        with self.assertRaises(ValueError):
            cache.incr("does_not_exist", -1)

    def test_decr(self):
        "Dummy cache values can't be decremented"
        cache.set("answer", 42)
        with self.assertRaises(ValueError):
            cache.decr("answer")
        with self.assertRaises(ValueError):
            cache.decr("does_not_exist")
        with self.assertRaises(ValueError):
            cache.decr("does_not_exist", -1)

    def test_touch(self):
        
        self.assertIs(cache.touch("whatever"), False)

    def test_data_types(self):
        "All data types are ignored equally by the dummy cache"
        tests = {
            "string": "this is a string",
            "int": 42,
            "bool": True,
            "list": [1, 2, 3, 4],
            "tuple": (1, 2, 3, 4),
            "dict": {"A": 1, "B": 2},
            "function": f,
            "class": C,
        }
        for key, value in tests.items():
            with self.subTest(key=key):
                cache.set(key, value)
                self.assertIsNone(cache.get(key))

    def test_expiration(self):
        "Expiration has no effect on the dummy cache"
        cache.set("expire1", "very quickly", 1)
        cache.set("expire2", "very quickly", 1)
        cache.set("expire3", "very quickly", 1)

        time.sleep(2)
        self.assertIsNone(cache.get("expire1"))

        self.assertIs(cache.add("expire2", "newvalue"), True)
        self.assertIsNone(cache.get("expire2"))
        self.assertIs(cache.has_key("expire3"), False)

    def test_unicode(self):
        "Unicode values are ignored by the dummy cache"
        stuff = {
            "ascii": "ascii_value",
            "unicode_ascii": "Iñtërnâtiônàlizætiøn1",
            "Iñtërnâtiônàlizætiøn": "Iñtërnâtiônàlizætiøn2",
            "ascii2": {"x": 1},
        }
        for key, value in stuff.items():
            with self.subTest(key=key):
                cache.set(key, value)
                self.assertIsNone(cache.get(key))

    def test_set_many(self):
        "set_many does nothing for the dummy cache backend"
        self.assertEqual(cache.set_many({"a": 1, "b": 2}), [])
        self.assertEqual(cache.set_many({"a": 1, "b": 2}, timeout=2, version="1"), [])

    def test_set_many_invalid_key(self):
        msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ":1:key with spaces"
        with self.assertWarnsMessage(CacheKeyWarning, msg):
            cache.set_many({"key with spaces": "foo"})

    def test_delete_many(self):
        "delete_many does nothing for the dummy cache backend"
        cache.delete_many(["a", "b"])

    def test_delete_many_invalid_key(self):
        msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ":1:key with spaces"
        with self.assertWarnsMessage(CacheKeyWarning, msg):
            cache.delete_many(["key with spaces"])

    def test_clear(self):
        "clear does nothing for the dummy cache backend"
        cache.clear()

    def test_incr_version(self):
        "Dummy cache versions can't be incremented"
        cache.set("answer", 42)
        with self.assertRaises(ValueError):
            cache.incr_version("answer")
        with self.assertRaises(ValueError):
            cache.incr_version("does_not_exist")

    def test_decr_version(self):
        "Dummy cache versions can't be decremented"
        cache.set("answer", 42)
        with self.assertRaises(ValueError):
            cache.decr_version("answer")
        with self.assertRaises(ValueError):
            cache.decr_version("does_not_exist")

    def test_get_or_set(self):
        self.assertEqual(cache.get_or_set("mykey", "default"), "default")
        self.assertIsNone(cache.get_or_set("mykey", None))

    def test_get_or_set_callable(self):
        def my_callable():
            return "default"

        self.assertEqual(cache.get_or_set("mykey", my_callable), "default")
        self.assertEqual(cache.get_or_set("mykey", my_callable()), "default")


def custom_key_func(key, key_prefix, version):
    "A customized cache key function"
    return "CUSTOM-" + "-".join([key_prefix, str(version), key])


_caches_setting_base = {
    "default": {},
    "prefix": {"KEY_PREFIX": "cacheprefix{}".format(os.getpid())},
    "v2": {"VERSION": 2},
    "custom_key": {"KEY_FUNCTION": custom_key_func},
    "custom_key2": {"KEY_FUNCTION": "cache.tests.custom_key_func"},
    "cull": {"OPTIONS": {"MAX_ENTRIES": 30}},
    "zero_cull": {"OPTIONS": {"CULL_FREQUENCY": 0, "MAX_ENTRIES": 30}},
}


def caches_setting_for_tests(base=None, exclude=None, **params):
    
    
    
    
    
    
    
    base = base or {}
    exclude = exclude or set()
    setting = {k: base.copy() for k in _caches_setting_base if k not in exclude}
    for key, cache_params in setting.items():
        cache_params.update(_caches_setting_base[key])
        cache_params.update(params)
    return setting


class BaseCacheTests:
    
    factory = RequestFactory()

    
    
    incr_decr_type_error = TypeError

    def tearDown(self):
        cache.clear()

    def test_simple(self):
        
        cache.set("key", "value")
        self.assertEqual(cache.get("key"), "value")

    def test_default_used_when_none_is_set(self):
        
        cache.set("key_default_none", None)
        self.assertIsNone(cache.get("key_default_none", default="default"))

    def test_add(self):
        
        self.assertIs(cache.add("addkey1", "value"), True)
        self.assertIs(cache.add("addkey1", "newvalue"), False)
        self.assertEqual(cache.get("addkey1"), "value")

    def test_prefix(self):
        
        cache.set("somekey", "value")

        
        self.assertIs(caches["prefix"].has_key("somekey"), False)

        caches["prefix"].set("somekey", "value2")

        self.assertEqual(cache.get("somekey"), "value")
        self.assertEqual(caches["prefix"].get("somekey"), "value2")

    def test_non_existent(self):
        
        self.assertIsNone(cache.get("does_not_exist"))
        self.assertEqual(cache.get("does_not_exist", "bang!"), "bang!")

    def test_get_many(self):
        
        cache.set_many({"a": "a", "b": "b", "c": "c", "d": "d"})
        self.assertEqual(
            cache.get_many(["a", "c", "d"]), {"a": "a", "c": "c", "d": "d"}
        )
        self.assertEqual(cache.get_many(["a", "b", "e"]), {"a": "a", "b": "b"})
        self.assertEqual(cache.get_many(iter(["a", "b", "e"])), {"a": "a", "b": "b"})
        cache.set_many({"x": None, "y": 1})
        self.assertEqual(cache.get_many(["x", "y"]), {"x": None, "y": 1})

    def test_delete(self):
        
        cache.set_many({"key1": "spam", "key2": "eggs"})
        self.assertEqual(cache.get("key1"), "spam")
        self.assertIs(cache.delete("key1"), True)
        self.assertIsNone(cache.get("key1"))
        self.assertEqual(cache.get("key2"), "eggs")

    def test_delete_nonexistent(self):
        self.assertIs(cache.delete("nonexistent_key"), False)

    def test_has_key(self):
        
        cache.set("hello1", "goodbye1")
        self.assertIs(cache.has_key("hello1"), True)
        self.assertIs(cache.has_key("goodbye1"), False)
        cache.set("no_expiry", "here", None)
        self.assertIs(cache.has_key("no_expiry"), True)
        cache.set("null", None)
        self.assertIs(cache.has_key("null"), True)

    def test_in(self):
        
        cache.set("hello2", "goodbye2")
        self.assertIn("hello2", cache)
        self.assertNotIn("goodbye2", cache)
        cache.set("null", None)
        self.assertIn("null", cache)

    def test_incr(self):
        
        cache.set("answer", 41)
        self.assertEqual(cache.incr("answer"), 42)
        self.assertEqual(cache.get("answer"), 42)
        self.assertEqual(cache.incr("answer", 10), 52)
        self.assertEqual(cache.get("answer"), 52)
        self.assertEqual(cache.incr("answer", -10), 42)
        with self.assertRaises(ValueError):
            cache.incr("does_not_exist")
        with self.assertRaises(ValueError):
            cache.incr("does_not_exist", -1)
        cache.set("null", None)
        with self.assertRaises(self.incr_decr_type_error):
            cache.incr("null")

    def test_decr(self):
        
        cache.set("answer", 43)
        self.assertEqual(cache.decr("answer"), 42)
        self.assertEqual(cache.get("answer"), 42)
        self.assertEqual(cache.decr("answer", 10), 32)
        self.assertEqual(cache.get("answer"), 32)
        self.assertEqual(cache.decr("answer", -10), 42)
        with self.assertRaises(ValueError):
            cache.decr("does_not_exist")
        with self.assertRaises(ValueError):
            cache.incr("does_not_exist", -1)
        cache.set("null", None)
        with self.assertRaises(self.incr_decr_type_error):
            cache.decr("null")

    def test_close(self):
        self.assertTrue(hasattr(cache, "close"))
        cache.close()

    def test_data_types(self):
        
        tests = {
            "string": "this is a string",
            "int": 42,
            "bool": True,
            "list": [1, 2, 3, 4],
            "tuple": (1, 2, 3, 4),
            "dict": {"A": 1, "B": 2},
            "function": f,
            "class": C,
        }
        for key, value in tests.items():
            with self.subTest(key=key):
                cache.set(key, value)
                self.assertEqual(cache.get(key), value)

    def test_cache_read_for_model_instance(self):
        
        expensive_calculation.num_runs = 0
        Poll.objects.all().delete()
        my_poll = Poll.objects.create(question="Well?")
        self.assertEqual(Poll.objects.count(), 1)
        pub_date = my_poll.pub_date
        cache.set("question", my_poll)
        cached_poll = cache.get("question")
        self.assertEqual(cached_poll.pub_date, pub_date)
        
        self.assertEqual(expensive_calculation.num_runs, 1)

    def test_cache_write_for_model_instance_with_deferred(self):
        
        expensive_calculation.num_runs = 0
        Poll.objects.all().delete()
        Poll.objects.create(question="What?")
        self.assertEqual(expensive_calculation.num_runs, 1)
        defer_qs = Poll.objects.defer("question")
        self.assertEqual(defer_qs.count(), 1)
        self.assertEqual(expensive_calculation.num_runs, 1)
        cache.set("deferred_queryset", defer_qs)
        
        self.assertEqual(expensive_calculation.num_runs, 1)

    def test_cache_read_for_model_instance_with_deferred(self):
        
        expensive_calculation.num_runs = 0
        Poll.objects.all().delete()
        Poll.objects.create(question="What?")
        self.assertEqual(expensive_calculation.num_runs, 1)
        defer_qs = Poll.objects.defer("question")
        self.assertEqual(defer_qs.count(), 1)
        cache.set("deferred_queryset", defer_qs)
        self.assertEqual(expensive_calculation.num_runs, 1)
        runs_before_cache_read = expensive_calculation.num_runs
        cache.get("deferred_queryset")
        
        self.assertEqual(expensive_calculation.num_runs, runs_before_cache_read)

    def test_expiration(self):
        
        cache.set("expire1", "very quickly", 1)
        cache.set("expire2", "very quickly", 1)
        cache.set("expire3", "very quickly", 1)

        time.sleep(2)
        self.assertIsNone(cache.get("expire1"))

        self.assertIs(cache.add("expire2", "newvalue"), True)
        self.assertEqual(cache.get("expire2"), "newvalue")
        self.assertIs(cache.has_key("expire3"), False)

    def test_touch(self):
        
        cache.set("expire1", "very quickly", timeout=1)
        self.assertIs(cache.touch("expire1", timeout=4), True)
        time.sleep(2)
        self.assertIs(cache.has_key("expire1"), True)
        time.sleep(3)
        self.assertIs(cache.has_key("expire1"), False)
        
        cache.set("expire1", "very quickly", timeout=1)
        self.assertIs(cache.touch("expire1"), True)
        time.sleep(2)
        self.assertIs(cache.has_key("expire1"), True)

        self.assertIs(cache.touch("nonexistent"), False)

    def test_unicode(self):
        
        stuff = {
            "ascii": "ascii_value",
            "unicode_ascii": "Iñtërnâtiônàlizætiøn1",
            "Iñtërnâtiônàlizætiøn": "Iñtërnâtiônàlizætiøn2",
            "ascii2": {"x": 1},
        }
        
        for key, value in stuff.items():
            with self.subTest(key=key):
                cache.set(key, value)
                self.assertEqual(cache.get(key), value)

        
        for key, value in stuff.items():
            with self.subTest(key=key):
                self.assertIs(cache.delete(key), True)
                self.assertIs(cache.add(key, value), True)
                self.assertEqual(cache.get(key), value)

        
        for key, value in stuff.items():
            self.assertIs(cache.delete(key), True)
        cache.set_many(stuff)
        for key, value in stuff.items():
            with self.subTest(key=key):
                self.assertEqual(cache.get(key), value)

    def test_binary_string(self):
        
        from zlib import compress, decompress

        value = "value_to_be_compressed"
        compressed_value = compress(value.encode())

        
        cache.set("binary1", compressed_value)
        compressed_result = cache.get("binary1")
        self.assertEqual(compressed_value, compressed_result)
        self.assertEqual(value, decompress(compressed_result).decode())

        
        self.assertIs(cache.add("binary1-add", compressed_value), True)
        compressed_result = cache.get("binary1-add")
        self.assertEqual(compressed_value, compressed_result)
        self.assertEqual(value, decompress(compressed_result).decode())

        
        cache.set_many({"binary1-set_many": compressed_value})
        compressed_result = cache.get("binary1-set_many")
        self.assertEqual(compressed_value, compressed_result)
        self.assertEqual(value, decompress(compressed_result).decode())

    def test_set_many(self):
        
        cache.set_many({"key1": "spam", "key2": "eggs"})
        self.assertEqual(cache.get("key1"), "spam")
        self.assertEqual(cache.get("key2"), "eggs")

    def test_set_many_returns_empty_list_on_success(self):
        
        failing_keys = cache.set_many({"key1": "spam", "key2": "eggs"})
        self.assertEqual(failing_keys, [])

    def test_set_many_expiration(self):
        
        cache.set_many({"key1": "spam", "key2": "eggs"}, 1)
        time.sleep(2)
        self.assertIsNone(cache.get("key1"))
        self.assertIsNone(cache.get("key2"))

    def test_set_many_empty_data(self):
        self.assertEqual(cache.set_many({}), [])

    def test_delete_many(self):
        
        cache.set_many({"key1": "spam", "key2": "eggs", "key3": "ham"})
        cache.delete_many(["key1", "key2"])
        self.assertIsNone(cache.get("key1"))
        self.assertIsNone(cache.get("key2"))
        self.assertEqual(cache.get("key3"), "ham")

    def test_delete_many_no_keys(self):
        self.assertIsNone(cache.delete_many([]))

    def test_clear(self):
        
        cache.set_many({"key1": "spam", "key2": "eggs"})
        cache.clear()
        self.assertIsNone(cache.get("key1"))
        self.assertIsNone(cache.get("key2"))

    def test_long_timeout(self):
        
        cache.set("key1", "eggs", 60 * 60 * 24 * 30 + 1)  
        self.assertEqual(cache.get("key1"), "eggs")

        self.assertIs(cache.add("key2", "ham", 60 * 60 * 24 * 30 + 1), True)
        self.assertEqual(cache.get("key2"), "ham")

        cache.set_many(
            {"key3": "sausage", "key4": "lobster bisque"}, 60 * 60 * 24 * 30 + 1
        )
        self.assertEqual(cache.get("key3"), "sausage")
        self.assertEqual(cache.get("key4"), "lobster bisque")

    def test_forever_timeout(self):
        
        cache.set("key1", "eggs", None)
        self.assertEqual(cache.get("key1"), "eggs")

        self.assertIs(cache.add("key2", "ham", None), True)
        self.assertEqual(cache.get("key2"), "ham")
        self.assertIs(cache.add("key1", "new eggs", None), False)
        self.assertEqual(cache.get("key1"), "eggs")

        cache.set_many({"key3": "sausage", "key4": "lobster bisque"}, None)
        self.assertEqual(cache.get("key3"), "sausage")
        self.assertEqual(cache.get("key4"), "lobster bisque")

        cache.set("key5", "belgian fries", timeout=1)
        self.assertIs(cache.touch("key5", timeout=None), True)
        time.sleep(2)
        self.assertEqual(cache.get("key5"), "belgian fries")

    def test_zero_timeout(self):
        
        cache.set("key1", "eggs", 0)
        self.assertIsNone(cache.get("key1"))

        self.assertIs(cache.add("key2", "ham", 0), True)
        self.assertIsNone(cache.get("key2"))

        cache.set_many({"key3": "sausage", "key4": "lobster bisque"}, 0)
        self.assertIsNone(cache.get("key3"))
        self.assertIsNone(cache.get("key4"))

        cache.set("key5", "belgian fries", timeout=5)
        self.assertIs(cache.touch("key5", timeout=0), True)
        self.assertIsNone(cache.get("key5"))

    def test_float_timeout(self):
        
        cache.set("key1", "spam", 100.2)
        self.assertEqual(cache.get("key1"), "spam")

    def _perform_cull_test(self, cull_cache_name, initial_count, final_count):
        try:
            cull_cache = caches[cull_cache_name]
        except InvalidCacheBackendError:
            self.skipTest("Culling isn't implemented.")

        
        
        for i in range(1, initial_count):
            cull_cache.set("cull%d" % i, "value", 1000)
        count = 0
        
        for i in range(1, initial_count):
            if cull_cache.has_key("cull%d" % i):
                count += 1
        self.assertEqual(count, final_count)

    def test_cull(self):
        self._perform_cull_test("cull", 50, 29)

    def test_zero_cull(self):
        self._perform_cull_test("zero_cull", 50, 19)

    def test_cull_delete_when_store_empty(self):
        try:
            cull_cache = caches["cull"]
        except InvalidCacheBackendError:
            self.skipTest("Culling isn't implemented.")
        old_max_entries = cull_cache._max_entries
        
        cull_cache._max_entries = -1
        try:
            cull_cache.set("force_cull_delete", "value", 1000)
            self.assertIs(cull_cache.has_key("force_cull_delete"), True)
        finally:
            cull_cache._max_entries = old_max_entries

    def _perform_invalid_key_test(self, key, expected_warning, key_func=None):
        

        
        
        def func(key, *args):
            return key

        old_func = cache.key_func
        cache.key_func = key_func or func

        tests = [
            ("add", [key, 1]),
            ("get", [key]),
            ("set", [key, 1]),
            ("incr", [key]),
            ("decr", [key]),
            ("touch", [key]),
            ("delete", [key]),
            ("get_many", [[key, "b"]]),
            ("set_many", [{key: 1, "b": 2}]),
            ("delete_many", [[key, "b"]]),
        ]
        try:
            for operation, args in tests:
                with self.subTest(operation=operation):
                    with self.assertWarns(CacheKeyWarning) as cm:
                        getattr(cache, operation)(*args)
                    self.assertEqual(str(cm.warning), expected_warning)
        finally:
            cache.key_func = old_func

    def test_invalid_key_characters(self):
        
        key = "key with spaces and 清"
        self._perform_invalid_key_test(key, KEY_ERRORS_WITH_MEMCACHED_MSG % key)

    def test_invalid_key_length(self):
        
        key = ("a" * 250) + "清"
        expected_warning = (
            "Cache key will cause errors if used with memcached: "
            "%r (longer than %s)" % (key, 250)
        )
        self._perform_invalid_key_test(key, expected_warning)

    def test_invalid_with_version_key_length(self):
        
        
        def key_func(key, *args):
            return key + ":1"

        key = "a" * 249
        expected_warning = (
            "Cache key will cause errors if used with memcached: "
            "%r (longer than %s)" % (key_func(key), 250)
        )
        self._perform_invalid_key_test(key, expected_warning, key_func=key_func)

    def test_cache_versioning_get_set(self):
        
        cache.set("answer1", 42)
        self.assertEqual(cache.get("answer1"), 42)
        self.assertEqual(cache.get("answer1", version=1), 42)
        self.assertIsNone(cache.get("answer1", version=2))

        self.assertIsNone(caches["v2"].get("answer1"))
        self.assertEqual(caches["v2"].get("answer1", version=1), 42)
        self.assertIsNone(caches["v2"].get("answer1", version=2))

        
        cache.set("answer2", 42, version=2)
        self.assertIsNone(cache.get("answer2"))
        self.assertIsNone(cache.get("answer2", version=1))
        self.assertEqual(cache.get("answer2", version=2), 42)

        self.assertEqual(caches["v2"].get("answer2"), 42)
        self.assertIsNone(caches["v2"].get("answer2", version=1))
        self.assertEqual(caches["v2"].get("answer2", version=2), 42)

        
        caches["v2"].set("answer3", 42)
        self.assertIsNone(cache.get("answer3"))
        self.assertIsNone(cache.get("answer3", version=1))
        self.assertEqual(cache.get("answer3", version=2), 42)

        self.assertEqual(caches["v2"].get("answer3"), 42)
        self.assertIsNone(caches["v2"].get("answer3", version=1))
        self.assertEqual(caches["v2"].get("answer3", version=2), 42)

        
        caches["v2"].set("answer4", 42, version=1)
        self.assertEqual(cache.get("answer4"), 42)
        self.assertEqual(cache.get("answer4", version=1), 42)
        self.assertIsNone(cache.get("answer4", version=2))

        self.assertIsNone(caches["v2"].get("answer4"))
        self.assertEqual(caches["v2"].get("answer4", version=1), 42)
        self.assertIsNone(caches["v2"].get("answer4", version=2))

    def test_cache_versioning_add(self):
        
        self.assertIs(cache.add("answer1", 42, version=2), True)
        self.assertIsNone(cache.get("answer1", version=1))
        self.assertEqual(cache.get("answer1", version=2), 42)

        self.assertIs(cache.add("answer1", 37, version=2), False)
        self.assertIsNone(cache.get("answer1", version=1))
        self.assertEqual(cache.get("answer1", version=2), 42)

        self.assertIs(cache.add("answer1", 37, version=1), True)
        self.assertEqual(cache.get("answer1", version=1), 37)
        self.assertEqual(cache.get("answer1", version=2), 42)

        
        self.assertIs(caches["v2"].add("answer2", 42), True)
        self.assertIsNone(cache.get("answer2", version=1))
        self.assertEqual(cache.get("answer2", version=2), 42)

        self.assertIs(caches["v2"].add("answer2", 37), False)
        self.assertIsNone(cache.get("answer2", version=1))
        self.assertEqual(cache.get("answer2", version=2), 42)

        self.assertIs(caches["v2"].add("answer2", 37, version=1), True)
        self.assertEqual(cache.get("answer2", version=1), 37)
        self.assertEqual(cache.get("answer2", version=2), 42)

        
        self.assertIs(caches["v2"].add("answer3", 42, version=1), True)
        self.assertEqual(cache.get("answer3", version=1), 42)
        self.assertIsNone(cache.get("answer3", version=2))

        self.assertIs(caches["v2"].add("answer3", 37, version=1), False)
        self.assertEqual(cache.get("answer3", version=1), 42)
        self.assertIsNone(cache.get("answer3", version=2))

        self.assertIs(caches["v2"].add("answer3", 37), True)
        self.assertEqual(cache.get("answer3", version=1), 42)
        self.assertEqual(cache.get("answer3", version=2), 37)

    def test_cache_versioning_has_key(self):
        cache.set("answer1", 42)

        
        self.assertIs(cache.has_key("answer1"), True)
        self.assertIs(cache.has_key("answer1", version=1), True)
        self.assertIs(cache.has_key("answer1", version=2), False)

        self.assertIs(caches["v2"].has_key("answer1"), False)
        self.assertIs(caches["v2"].has_key("answer1", version=1), True)
        self.assertIs(caches["v2"].has_key("answer1", version=2), False)

    def test_cache_versioning_delete(self):
        cache.set("answer1", 37, version=1)
        cache.set("answer1", 42, version=2)
        self.assertIs(cache.delete("answer1"), True)
        self.assertIsNone(cache.get("answer1", version=1))
        self.assertEqual(cache.get("answer1", version=2), 42)

        cache.set("answer2", 37, version=1)
        cache.set("answer2", 42, version=2)
        self.assertIs(cache.delete("answer2", version=2), True)
        self.assertEqual(cache.get("answer2", version=1), 37)
        self.assertIsNone(cache.get("answer2", version=2))

        cache.set("answer3", 37, version=1)
        cache.set("answer3", 42, version=2)
        self.assertIs(caches["v2"].delete("answer3"), True)
        self.assertEqual(cache.get("answer3", version=1), 37)
        self.assertIsNone(cache.get("answer3", version=2))

        cache.set("answer4", 37, version=1)
        cache.set("answer4", 42, version=2)
        self.assertIs(caches["v2"].delete("answer4", version=1), True)
        self.assertIsNone(cache.get("answer4", version=1))
        self.assertEqual(cache.get("answer4", version=2), 42)

    def test_cache_versioning_incr_decr(self):
        cache.set("answer1", 37, version=1)
        cache.set("answer1", 42, version=2)
        self.assertEqual(cache.incr("answer1"), 38)
        self.assertEqual(cache.get("answer1", version=1), 38)
        self.assertEqual(cache.get("answer1", version=2), 42)
        self.assertEqual(cache.decr("answer1"), 37)
        self.assertEqual(cache.get("answer1", version=1), 37)
        self.assertEqual(cache.get("answer1", version=2), 42)

        cache.set("answer2", 37, version=1)
        cache.set("answer2", 42, version=2)
        self.assertEqual(cache.incr("answer2", version=2), 43)
        self.assertEqual(cache.get("answer2", version=1), 37)
        self.assertEqual(cache.get("answer2", version=2), 43)
        self.assertEqual(cache.decr("answer2", version=2), 42)
        self.assertEqual(cache.get("answer2", version=1), 37)
        self.assertEqual(cache.get("answer2", version=2), 42)

        cache.set("answer3", 37, version=1)
        cache.set("answer3", 42, version=2)
        self.assertEqual(caches["v2"].incr("answer3"), 43)
        self.assertEqual(cache.get("answer3", version=1), 37)
        self.assertEqual(cache.get("answer3", version=2), 43)
        self.assertEqual(caches["v2"].decr("answer3"), 42)
        self.assertEqual(cache.get("answer3", version=1), 37)
        self.assertEqual(cache.get("answer3", version=2), 42)

        cache.set("answer4", 37, version=1)
        cache.set("answer4", 42, version=2)
        self.assertEqual(caches["v2"].incr("answer4", version=1), 38)
        self.assertEqual(cache.get("answer4", version=1), 38)
        self.assertEqual(cache.get("answer4", version=2), 42)
        self.assertEqual(caches["v2"].decr("answer4", version=1), 37)
        self.assertEqual(cache.get("answer4", version=1), 37)
        self.assertEqual(cache.get("answer4", version=2), 42)

    def test_cache_versioning_get_set_many(self):
        
        cache.set_many({"ford1": 37, "arthur1": 42})
        self.assertEqual(
            cache.get_many(["ford1", "arthur1"]), {"ford1": 37, "arthur1": 42}
        )
        self.assertEqual(
            cache.get_many(["ford1", "arthur1"], version=1),
            {"ford1": 37, "arthur1": 42},
        )
        self.assertEqual(cache.get_many(["ford1", "arthur1"], version=2), {})

        self.assertEqual(caches["v2"].get_many(["ford1", "arthur1"]), {})
        self.assertEqual(
            caches["v2"].get_many(["ford1", "arthur1"], version=1),
            {"ford1": 37, "arthur1": 42},
        )
        self.assertEqual(caches["v2"].get_many(["ford1", "arthur1"], version=2), {})

        
        cache.set_many({"ford2": 37, "arthur2": 42}, version=2)
        self.assertEqual(cache.get_many(["ford2", "arthur2"]), {})
        self.assertEqual(cache.get_many(["ford2", "arthur2"], version=1), {})
        self.assertEqual(
            cache.get_many(["ford2", "arthur2"], version=2),
            {"ford2": 37, "arthur2": 42},
        )

        self.assertEqual(
            caches["v2"].get_many(["ford2", "arthur2"]), {"ford2": 37, "arthur2": 42}
        )
        self.assertEqual(caches["v2"].get_many(["ford2", "arthur2"], version=1), {})
        self.assertEqual(
            caches["v2"].get_many(["ford2", "arthur2"], version=2),
            {"ford2": 37, "arthur2": 42},
        )

        
        caches["v2"].set_many({"ford3": 37, "arthur3": 42})
        self.assertEqual(cache.get_many(["ford3", "arthur3"]), {})
        self.assertEqual(cache.get_many(["ford3", "arthur3"], version=1), {})
        self.assertEqual(
            cache.get_many(["ford3", "arthur3"], version=2),
            {"ford3": 37, "arthur3": 42},
        )

        self.assertEqual(
            caches["v2"].get_many(["ford3", "arthur3"]), {"ford3": 37, "arthur3": 42}
        )
        self.assertEqual(caches["v2"].get_many(["ford3", "arthur3"], version=1), {})
        self.assertEqual(
            caches["v2"].get_many(["ford3", "arthur3"], version=2),
            {"ford3": 37, "arthur3": 42},
        )

        
        caches["v2"].set_many({"ford4": 37, "arthur4": 42}, version=1)
        self.assertEqual(
            cache.get_many(["ford4", "arthur4"]), {"ford4": 37, "arthur4": 42}
        )
        self.assertEqual(
            cache.get_many(["ford4", "arthur4"], version=1),
            {"ford4": 37, "arthur4": 42},
        )
        self.assertEqual(cache.get_many(["ford4", "arthur4"], version=2), {})

        self.assertEqual(caches["v2"].get_many(["ford4", "arthur4"]), {})
        self.assertEqual(
            caches["v2"].get_many(["ford4", "arthur4"], version=1),
            {"ford4": 37, "arthur4": 42},
        )
        self.assertEqual(caches["v2"].get_many(["ford4", "arthur4"], version=2), {})

    def test_incr_version(self):
        cache.set("answer", 42, version=2)
        self.assertIsNone(cache.get("answer"))
        self.assertIsNone(cache.get("answer", version=1))
        self.assertEqual(cache.get("answer", version=2), 42)
        self.assertIsNone(cache.get("answer", version=3))

        self.assertEqual(cache.incr_version("answer", version=2), 3)
        self.assertIsNone(cache.get("answer"))
        self.assertIsNone(cache.get("answer", version=1))
        self.assertIsNone(cache.get("answer", version=2))
        self.assertEqual(cache.get("answer", version=3), 42)

        caches["v2"].set("answer2", 42)
        self.assertEqual(caches["v2"].get("answer2"), 42)
        self.assertIsNone(caches["v2"].get("answer2", version=1))
        self.assertEqual(caches["v2"].get("answer2", version=2), 42)
        self.assertIsNone(caches["v2"].get("answer2", version=3))

        self.assertEqual(caches["v2"].incr_version("answer2"), 3)
        self.assertIsNone(caches["v2"].get("answer2"))
        self.assertIsNone(caches["v2"].get("answer2", version=1))
        self.assertIsNone(caches["v2"].get("answer2", version=2))
        self.assertEqual(caches["v2"].get("answer2", version=3), 42)

        with self.assertRaises(ValueError):
            cache.incr_version("does_not_exist")

        cache.set("null", None)
        self.assertEqual(cache.incr_version("null"), 2)

    def test_decr_version(self):
        cache.set("answer", 42, version=2)
        self.assertIsNone(cache.get("answer"))
        self.assertIsNone(cache.get("answer", version=1))
        self.assertEqual(cache.get("answer", version=2), 42)

        self.assertEqual(cache.decr_version("answer", version=2), 1)
        self.assertEqual(cache.get("answer"), 42)
        self.assertEqual(cache.get("answer", version=1), 42)
        self.assertIsNone(cache.get("answer", version=2))

        caches["v2"].set("answer2", 42)
        self.assertEqual(caches["v2"].get("answer2"), 42)
        self.assertIsNone(caches["v2"].get("answer2", version=1))
        self.assertEqual(caches["v2"].get("answer2", version=2), 42)

        self.assertEqual(caches["v2"].decr_version("answer2"), 1)
        self.assertIsNone(caches["v2"].get("answer2"))
        self.assertEqual(caches["v2"].get("answer2", version=1), 42)
        self.assertIsNone(caches["v2"].get("answer2", version=2))

        with self.assertRaises(ValueError):
            cache.decr_version("does_not_exist", version=2)

        cache.set("null", None, version=2)
        self.assertEqual(cache.decr_version("null", version=2), 1)

    def test_custom_key_func(self):
        
        cache.set("answer1", 42)
        self.assertEqual(cache.get("answer1"), 42)
        self.assertIsNone(caches["custom_key"].get("answer1"))
        self.assertIsNone(caches["custom_key2"].get("answer1"))

        caches["custom_key"].set("answer2", 42)
        self.assertIsNone(cache.get("answer2"))
        self.assertEqual(caches["custom_key"].get("answer2"), 42)
        self.assertEqual(caches["custom_key2"].get("answer2"), 42)

    @override_settings(CACHE_MIDDLEWARE_ALIAS=DEFAULT_CACHE_ALIAS)
    def test_cache_write_unpicklable_object(self):
        fetch_middleware = FetchFromCacheMiddleware(empty_response)

        request = self.factory.get("/cache/test")
        request._cache_update_cache = True
        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        self.assertIsNone(get_cache_data)

        content = "Testing cookie serialization."

        def get_response(req):
            response = HttpResponse(content)
            response.set_cookie("foo", "bar")
            return response

        update_middleware = UpdateCacheMiddleware(get_response)
        response = update_middleware(request)

        get_cache_data = fetch_middleware.process_request(request)
        self.assertIsNotNone(get_cache_data)
        self.assertEqual(get_cache_data.content, content.encode())
        self.assertEqual(get_cache_data.cookies, response.cookies)

        UpdateCacheMiddleware(lambda req: get_cache_data)(request)
        get_cache_data = fetch_middleware.process_request(request)
        self.assertIsNotNone(get_cache_data)
        self.assertEqual(get_cache_data.content, content.encode())
        self.assertEqual(get_cache_data.cookies, response.cookies)

    def test_add_fail_on_pickleerror(self):
        
        with self.assertRaises(pickle.PickleError):
            cache.add("unpicklable", Unpicklable())

    def test_set_fail_on_pickleerror(self):
        with self.assertRaises(pickle.PickleError):
            cache.set("unpicklable", Unpicklable())

    def test_get_or_set(self):
        self.assertIsNone(cache.get("projector"))
        self.assertEqual(cache.get_or_set("projector", 42), 42)
        self.assertEqual(cache.get("projector"), 42)
        self.assertIsNone(cache.get_or_set("null", None))
        
        self.assertIsNone(cache.get("null", "default"))

    def test_get_or_set_callable(self):
        def my_callable():
            return "value"

        self.assertEqual(cache.get_or_set("mykey", my_callable), "value")
        self.assertEqual(cache.get_or_set("mykey", my_callable()), "value")

        self.assertIsNone(cache.get_or_set("null", lambda: None))
        
        self.assertIsNone(cache.get("null", "default"))

    def test_get_or_set_version(self):
        msg = "get_or_set() missing 1 required positional argument: 'default'"
        self.assertEqual(cache.get_or_set("brian", 1979, version=2), 1979)
        with self.assertRaisesMessage(TypeError, msg):
            cache.get_or_set("brian")
        with self.assertRaisesMessage(TypeError, msg):
            cache.get_or_set("brian", version=1)
        self.assertIsNone(cache.get("brian", version=1))
        self.assertEqual(cache.get_or_set("brian", 42, version=1), 42)
        self.assertEqual(cache.get_or_set("brian", 1979, version=2), 1979)
        self.assertIsNone(cache.get("brian", version=3))

    def test_get_or_set_racing(self):
        with mock.patch(
            "%s.%s" % (settings.CACHES["default"]["BACKEND"], "add")
        ) as cache_add:
            
            
            cache_add.return_value = False
            self.assertEqual(cache.get_or_set("key", "default"), "default")


@override_settings(
    CACHES=caches_setting_for_tests(
        BACKEND="django.core.cache.backends.db.DatabaseCache",
        
        LOCATION="test cache table",
    )
)
class DBCacheTests(BaseCacheTests, TransactionTestCase):
    available_apps = ["cache"]

    def setUp(self):
        
        super().setUp()
        self.create_table()

    def tearDown(self):
        
        super().tearDown()
        self.drop_table()

    def create_table(self):
        management.call_command("createcachetable", verbosity=0)

    def drop_table(self):
        with connection.cursor() as cursor:
            table_name = connection.ops.quote_name("test cache table")
            cursor.execute("DROP TABLE %s" % table_name)

    def test_get_many_num_queries(self):
        cache.set_many({"a": 1, "b": 2})
        cache.set("expired", "expired", 0.01)
        with self.assertNumQueries(1):
            self.assertEqual(cache.get_many(["a", "b"]), {"a": 1, "b": 2})
        time.sleep(0.02)
        with self.assertNumQueries(2):
            self.assertEqual(cache.get_many(["a", "b", "expired"]), {"a": 1, "b": 2})

    def test_delete_many_num_queries(self):
        cache.set_many({"a": 1, "b": 2, "c": 3})
        with self.assertNumQueries(1):
            cache.delete_many(["a", "b", "c"])

    def test_cull_queries(self):
        old_max_entries = cache._max_entries
        
        cache._max_entries = -1
        with CaptureQueriesContext(connection) as captured_queries:
            try:
                cache.set("force_cull", "value", 1000)
            finally:
                cache._max_entries = old_max_entries
        num_count_queries = sum("COUNT" in query["sql"] for query in captured_queries)
        self.assertEqual(num_count_queries, 1)
        
        for query in captured_queries:
            sql = query["sql"]
            if "expires" in sql:
                self.assertIn(connection.ops.quote_name("expires"), sql)
            if "cache_key" in sql:
                self.assertIn(connection.ops.quote_name("cache_key"), sql)

    def test_delete_cursor_rowcount(self):
        

        class MockedCursorWrapper(CursorWrapper):
            is_closed = False

            def close(self):
                self.cursor.close()
                self.is_closed = True

            @property
            def rowcount(self):
                if self.is_closed:
                    raise Exception("Cursor is closed.")
                return self.cursor.rowcount

        cache.set_many({"a": 1, "b": 2})
        with mock.patch("django.db.backends.utils.CursorWrapper", MockedCursorWrapper):
            self.assertIs(cache.delete("a"), True)

    def test_zero_cull(self):
        self._perform_cull_test("zero_cull", 50, 18)

    def test_second_call_doesnt_crash(self):
        out = io.StringIO()
        management.call_command("createcachetable", stdout=out)
        self.assertEqual(
            out.getvalue(),
            "Cache table 'test cache table' already exists.\n" * len(settings.CACHES),
        )

    @override_settings(
        CACHES=caches_setting_for_tests(
            BACKEND="django.core.cache.backends.db.DatabaseCache",
            
            LOCATION="createcachetable_dry_run_mode",
        )
    )
    def test_createcachetable_dry_run_mode(self):
        out = io.StringIO()
        management.call_command("createcachetable", dry_run=True, stdout=out)
        output = out.getvalue()
        self.assertTrue(output.startswith("CREATE TABLE"))

    def test_createcachetable_with_table_argument(self):
        
        self.drop_table()
        out = io.StringIO()
        management.call_command(
            "createcachetable",
            "test cache table",
            verbosity=2,
            stdout=out,
        )
        self.assertEqual(out.getvalue(), "Cache table 'test cache table' created.\n")

    def test_has_key_query_columns_quoted(self):
        with CaptureQueriesContext(connection) as captured_queries:
            cache.has_key("key")
        self.assertEqual(len(captured_queries), 1)
        sql = captured_queries[0]["sql"]
        
        self.assertIn(connection.ops.quote_name("expires"), sql)
        self.assertIn(connection.ops.quote_name("cache_key"), sql)


@override_settings(USE_TZ=True)
class DBCacheWithTimeZoneTests(DBCacheTests):
    pass


class DBCacheRouter:
    

    def db_for_read(self, model, **hints):
        if model._meta.app_label == "django_cache":
            return "other"
        return None

    def db_for_write(self, model, **hints):
        if model._meta.app_label == "django_cache":
            return "other"
        return None

    def allow_migrate(self, db, app_label, **hints):
        if app_label == "django_cache":
            return db == "other"
        return None


@override_settings(
    CACHES={
        "default": {
            "BACKEND": "django.core.cache.backends.db.DatabaseCache",
            "LOCATION": "my_cache_table",
        },
    },
)
class CreateCacheTableForDBCacheTests(TestCase):
    databases = {"default", "other"}

    @override_settings(DATABASE_ROUTERS=[DBCacheRouter()])
    def test_createcachetable_observes_database_router(self):
        
        with self.assertNumQueries(0, using="default"):
            management.call_command("createcachetable", database="default", verbosity=0)
        
        
        
        
        
        
        
        num = 5 if connections["other"].features.can_rollback_ddl else 3
        with self.assertNumQueries(num, using="other"):
            management.call_command("createcachetable", database="other", verbosity=0)


class PicklingSideEffect:
    def __init__(self, cache):
        self.cache = cache
        self.locked = False

    def __getstate__(self):
        self.locked = self.cache._lock.locked()
        return {}


limit_locmem_entries = override_settings(
    CACHES=caches_setting_for_tests(
        BACKEND="django.core.cache.backends.locmem.LocMemCache",
        OPTIONS={"MAX_ENTRIES": 9},
    )
)


@override_settings(
    CACHES=caches_setting_for_tests(
        BACKEND="django.core.cache.backends.locmem.LocMemCache",
    )
)
class LocMemCacheTests(BaseCacheTests, TestCase):
    def setUp(self):
        super().setUp()

        
        
        caches["prefix"]._cache = cache._cache
        caches["prefix"]._expire_info = cache._expire_info

        caches["v2"]._cache = cache._cache
        caches["v2"]._expire_info = cache._expire_info

        caches["custom_key"]._cache = cache._cache
        caches["custom_key"]._expire_info = cache._expire_info

        caches["custom_key2"]._cache = cache._cache
        caches["custom_key2"]._expire_info = cache._expire_info

    @override_settings(
        CACHES={
            "default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"},
            "other": {
                "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
                "LOCATION": "other",
            },
        }
    )
    def test_multiple_caches(self):
        "Multiple locmem caches are isolated"
        cache.set("value", 42)
        self.assertEqual(caches["default"].get("value"), 42)
        self.assertIsNone(caches["other"].get("value"))

    def test_locking_on_pickle(self):
        
        bad_obj = PicklingSideEffect(cache)
        cache.set("set", bad_obj)
        self.assertFalse(bad_obj.locked, "Cache was locked during pickling")

        self.assertIs(cache.add("add", bad_obj), True)
        self.assertFalse(bad_obj.locked, "Cache was locked during pickling")

    def test_incr_decr_timeout(self):
        
        key = "value"
        _key = cache.make_key(key)
        cache.set(key, 1, timeout=cache.default_timeout * 10)
        expire = cache._expire_info[_key]
        self.assertEqual(cache.incr(key), 2)
        self.assertEqual(expire, cache._expire_info[_key])
        self.assertEqual(cache.decr(key), 1)
        self.assertEqual(expire, cache._expire_info[_key])

    @limit_locmem_entries
    def test_lru_get(self):
        
        for key in range(9):
            cache.set(key, key, timeout=None)
        for key in range(6):
            self.assertEqual(cache.get(key), key)
        cache.set(9, 9, timeout=None)
        for key in range(6):
            self.assertEqual(cache.get(key), key)
        for key in range(6, 9):
            self.assertIsNone(cache.get(key))
        self.assertEqual(cache.get(9), 9)

    @limit_locmem_entries
    def test_lru_set(self):
        
        for key in range(9):
            cache.set(key, key, timeout=None)
        for key in range(3, 9):
            cache.set(key, key, timeout=None)
        cache.set(9, 9, timeout=None)
        for key in range(3, 10):
            self.assertEqual(cache.get(key), key)
        for key in range(3):
            self.assertIsNone(cache.get(key))

    @limit_locmem_entries
    def test_lru_incr(self):
        
        for key in range(9):
            cache.set(key, key, timeout=None)
        for key in range(6):
            self.assertEqual(cache.incr(key), key + 1)
        cache.set(9, 9, timeout=None)
        for key in range(6):
            self.assertEqual(cache.get(key), key + 1)
        for key in range(6, 9):
            self.assertIsNone(cache.get(key))
        self.assertEqual(cache.get(9), 9)





configured_caches = {}
for _cache_params in settings.CACHES.values():
    configured_caches[_cache_params["BACKEND"]] = _cache_params

PyLibMCCache_params = configured_caches.get(
    "django.core.cache.backends.memcached.PyLibMCCache"
)
PyMemcacheCache_params = configured_caches.get(
    "django.core.cache.backends.memcached.PyMemcacheCache"
)


memcached_excluded_caches = {"cull", "zero_cull"}

RedisCache_params = configured_caches.get("django.core.cache.backends.redis.RedisCache")


redis_excluded_caches = {"cull", "zero_cull"}


class BaseMemcachedTests(BaseCacheTests):
    
    
    should_disconnect_on_close = True

    def test_location_multiple_servers(self):
        locations = [
            ["server1.tld", "server2:11211"],
            "server1.tld;server2:11211",
            "server1.tld,server2:11211",
        ]
        for location in locations:
            with self.subTest(location=location):
                params = {"BACKEND": self.base_params["BACKEND"], "LOCATION": location}
                with self.settings(CACHES={"default": params}):
                    self.assertEqual(cache._servers, ["server1.tld", "server2:11211"])

    def _perform_invalid_key_test(self, key, expected_warning):
        
        msg = expected_warning.replace(key, cache.make_key(key))
        tests = [
            ("add", [key, 1]),
            ("get", [key]),
            ("set", [key, 1]),
            ("incr", [key]),
            ("decr", [key]),
            ("touch", [key]),
            ("delete", [key]),
            ("get_many", [[key, "b"]]),
            ("set_many", [{key: 1, "b": 2}]),
            ("delete_many", [[key, "b"]]),
        ]
        for operation, args in tests:
            with self.subTest(operation=operation):
                with self.assertRaises(InvalidCacheKey) as cm:
                    getattr(cache, operation)(*args)
                self.assertEqual(str(cm.exception), msg)

    def test_invalid_with_version_key_length(self):
        
        key = "a" * 248
        expected_warning = (
            "Cache key will cause errors if used with memcached: "
            "%r (longer than %s)" % (key, 250)
        )
        self._perform_invalid_key_test(key, expected_warning)

    def test_default_never_expiring_timeout(self):
        
        with self.settings(
            CACHES=caches_setting_for_tests(
                base=self.base_params, exclude=memcached_excluded_caches, TIMEOUT=None
            )
        ):
            cache.set("infinite_foo", "bar")
            self.assertEqual(cache.get("infinite_foo"), "bar")

    def test_default_far_future_timeout(self):
        
        with self.settings(
            CACHES=caches_setting_for_tests(
                base=self.base_params,
                exclude=memcached_excluded_caches,
                
                TIMEOUT=31536000,
            )
        ):
            cache.set("future_foo", "bar")
            self.assertEqual(cache.get("future_foo"), "bar")

    def test_memcached_deletes_key_on_failed_set(self):
        
        
        
        max_value_length = 2**20

        cache.set("small_value", "a")
        self.assertEqual(cache.get("small_value"), "a")

        large_value = "a" * (max_value_length + 1)
        try:
            cache.set("small_value", large_value)
        except Exception:
            
            
            
            
            pass
        
        value = cache.get("small_value")
        self.assertTrue(value is None or value == large_value)

    def test_close(self):
        
        
        signals.request_finished.disconnect(close_old_connections)
        try:
            with mock.patch.object(
                cache._class, "disconnect_all", autospec=True
            ) as mock_disconnect:
                signals.request_finished.send(self.__class__)
                self.assertIs(mock_disconnect.called, self.should_disconnect_on_close)
        finally:
            signals.request_finished.connect(close_old_connections)

    def test_set_many_returns_failing_keys(self):
        def fail_set_multi(mapping, *args, **kwargs):
            return mapping.keys()

        with mock.patch.object(cache._class, "set_multi", side_effect=fail_set_multi):
            failing_keys = cache.set_many({"key": "value"})
            self.assertEqual(failing_keys, ["key"])


@unittest.skipUnless(PyLibMCCache_params, "PyLibMCCache backend not configured")
@override_settings(
    CACHES=caches_setting_for_tests(
        base=PyLibMCCache_params,
        exclude=memcached_excluded_caches,
    )
)
class PyLibMCCacheTests(BaseMemcachedTests, TestCase):
    base_params = PyLibMCCache_params
    
    should_disconnect_on_close = False

    @property
    def incr_decr_type_error(self):
        return cache._lib.ClientError

    @override_settings(
        CACHES=caches_setting_for_tests(
            base=PyLibMCCache_params,
            exclude=memcached_excluded_caches,
            OPTIONS={
                "binary": True,
                "behaviors": {"tcp_nodelay": True},
            },
        )
    )
    def test_pylibmc_options(self):
        self.assertTrue(cache._cache.binary)
        self.assertEqual(cache._cache.behaviors["tcp_nodelay"], int(True))

    def test_pylibmc_client_servers(self):
        backend = self.base_params["BACKEND"]
        tests = [
            ("unix:/run/memcached/socket", "/run/memcached/socket"),
            ("/run/memcached/socket", "/run/memcached/socket"),
            ("localhost", "localhost"),
            ("localhost:11211", "localhost:11211"),
            ("[::1]", "[::1]"),
            ("[::1]:11211", "[::1]:11211"),
            ("127.0.0.1", "127.0.0.1"),
            ("127.0.0.1:11211", "127.0.0.1:11211"),
        ]
        for location, expected in tests:
            settings = {"default": {"BACKEND": backend, "LOCATION": location}}
            with self.subTest(location), self.settings(CACHES=settings):
                self.assertEqual(cache.client_servers, [expected])


@unittest.skipUnless(PyMemcacheCache_params, "PyMemcacheCache backend not configured")
@override_settings(
    CACHES=caches_setting_for_tests(
        base=PyMemcacheCache_params,
        exclude=memcached_excluded_caches,
    )
)
class PyMemcacheCacheTests(BaseMemcachedTests, TestCase):
    base_params = PyMemcacheCache_params

    @property
    def incr_decr_type_error(self):
        return cache._lib.exceptions.MemcacheClientError

    def test_pymemcache_highest_pickle_version(self):
        self.assertEqual(
            cache._cache.default_kwargs["serde"]._serialize_func.keywords[
                "pickle_version"
            ],
            pickle.HIGHEST_PROTOCOL,
        )
        for cache_key in settings.CACHES:
            for client_key, client in caches[cache_key]._cache.clients.items():
                with self.subTest(cache_key=cache_key, server=client_key):
                    self.assertEqual(
                        client.serde._serialize_func.keywords["pickle_version"],
                        pickle.HIGHEST_PROTOCOL,
                    )

    @override_settings(
        CACHES=caches_setting_for_tests(
            base=PyMemcacheCache_params,
            exclude=memcached_excluded_caches,
            OPTIONS={"no_delay": True},
        )
    )
    def test_pymemcache_options(self):
        self.assertIs(cache._cache.default_kwargs["no_delay"], True)


@override_settings(
    CACHES=caches_setting_for_tests(
        BACKEND="django.core.cache.backends.filebased.FileBasedCache",
    )
)
class FileBasedCacheTests(BaseCacheTests, TestCase):
    

    def setUp(self):
        super().setUp()
        self.dirname = self.mkdtemp()
        
        
        
        for cache_params in settings.CACHES.values():
            cache_params["LOCATION"] = self.dirname
        setting_changed.send(self.__class__, setting="CACHES", enter=False)

    def tearDown(self):
        super().tearDown()
        
        shutil.rmtree(self.dirname)

    def mkdtemp(self):
        return tempfile.mkdtemp()

    def test_ignores_non_cache_files(self):
        fname = os.path.join(self.dirname, "not-a-cache-file")
        with open(fname, "w"):
            os.utime(fname, None)
        cache.clear()
        self.assertTrue(
            os.path.exists(fname), "Expected cache.clear to ignore non cache files"
        )
        os.remove(fname)

    def test_clear_does_not_remove_cache_dir(self):
        cache.clear()
        self.assertTrue(
            os.path.exists(self.dirname), "Expected cache.clear to keep the cache dir"
        )

    def test_creates_cache_dir_if_nonexistent(self):
        os.rmdir(self.dirname)
        cache.set("foo", "bar")
        self.assertTrue(os.path.exists(self.dirname))

    def test_get_ignores_enoent(self):
        cache.set("foo", "bar")
        os.unlink(cache._key_to_file("foo"))
        
        self.assertEqual(cache.get("foo", "baz"), "baz")

    @skipIf(
        sys.platform == "win32",
        "Windows only partially supports umasks and chmod.",
    )
    def test_cache_dir_permissions(self):
        os.rmdir(self.dirname)
        dir_path = Path(self.dirname) / "nested" / "filebasedcache"
        for cache_params in settings.CACHES.values():
            cache_params["LOCATION"] = dir_path
        setting_changed.send(self.__class__, setting="CACHES", enter=False)
        cache.set("foo", "bar")
        self.assertIs(dir_path.exists(), True)
        tests = [
            dir_path,
            dir_path.parent,
            dir_path.parent.parent,
        ]
        for directory in tests:
            with self.subTest(directory=directory):
                dir_mode = directory.stat().st_mode & 0o777
                self.assertEqual(dir_mode, 0o700)

    def test_get_does_not_ignore_non_filenotfound_exceptions(self):
        with mock.patch("builtins.open", side_effect=OSError):
            with self.assertRaises(OSError):
                cache.get("foo")

    def test_empty_cache_file_considered_expired(self):
        cache_file = cache._key_to_file("foo")
        with open(cache_file, "wb") as fh:
            fh.write(b"")
        with open(cache_file, "rb") as fh:
            self.assertIs(cache._is_expired(fh), True)

    def test_has_key_race_handling(self):
        self.assertIs(cache.add("key", "value"), True)
        with mock.patch("builtins.open", side_effect=FileNotFoundError) as mocked_open:
            self.assertIs(cache.has_key("key"), False)
            mocked_open.assert_called_once()


@unittest.skipUnless(RedisCache_params, "Redis backend not configured")
@override_settings(
    CACHES=caches_setting_for_tests(
        base=RedisCache_params,
        exclude=redis_excluded_caches,
    )
)
class RedisCacheTests(BaseCacheTests, TestCase):
    def setUp(self):
        import redis

        super().setUp()
        self.lib = redis

    @property
    def incr_decr_type_error(self):
        return self.lib.ResponseError

    def test_incr_write_connection(self):
        cache.set("number", 42)
        with mock.patch(
            "django.core.cache.backends.redis.RedisCacheClient.get_client"
        ) as mocked_get_client:
            cache.incr("number")
            self.assertEqual(mocked_get_client.call_args.kwargs, {"write": True})

    def test_cache_client_class(self):
        self.assertIs(cache._class, RedisCacheClient)
        self.assertIsInstance(cache._cache, RedisCacheClient)

    def test_get_backend_timeout_method(self):
        positive_timeout = 10
        positive_backend_timeout = cache.get_backend_timeout(positive_timeout)
        self.assertEqual(positive_backend_timeout, positive_timeout)

        negative_timeout = -5
        negative_backend_timeout = cache.get_backend_timeout(negative_timeout)
        self.assertEqual(negative_backend_timeout, 0)

        none_timeout = None
        none_backend_timeout = cache.get_backend_timeout(none_timeout)
        self.assertIsNone(none_backend_timeout)

    def test_get_connection_pool_index(self):
        pool_index = cache._cache._get_connection_pool_index(write=True)
        self.assertEqual(pool_index, 0)
        pool_index = cache._cache._get_connection_pool_index(write=False)
        if len(cache._cache._servers) == 1:
            self.assertEqual(pool_index, 0)
        else:
            self.assertGreater(pool_index, 0)
            self.assertLess(pool_index, len(cache._cache._servers))

    def test_get_connection_pool(self):
        pool = cache._cache._get_connection_pool(write=True)
        self.assertIsInstance(pool, self.lib.ConnectionPool)

        pool = cache._cache._get_connection_pool(write=False)
        self.assertIsInstance(pool, self.lib.ConnectionPool)

    def test_get_client(self):
        self.assertIsInstance(cache._cache.get_client(), self.lib.Redis)

    def test_serializer_dumps(self):
        self.assertEqual(cache._cache._serializer.dumps(123), 123)
        self.assertIsInstance(cache._cache._serializer.dumps(True), bytes)
        self.assertIsInstance(cache._cache._serializer.dumps("abc"), bytes)

    @override_settings(
        CACHES=caches_setting_for_tests(
            base=RedisCache_params,
            exclude=redis_excluded_caches,
            OPTIONS={
                "db": 5,
                "socket_timeout": 0.1,
                "retry_on_timeout": True,
            },
        )
    )
    def test_redis_pool_options(self):
        pool = cache._cache._get_connection_pool(write=False)
        self.assertEqual(pool.connection_kwargs["db"], 5)
        self.assertEqual(pool.connection_kwargs["socket_timeout"], 0.1)
        self.assertIs(pool.connection_kwargs["retry_on_timeout"], True)


class FileBasedCachePathLibTests(FileBasedCacheTests):
    def mkdtemp(self):
        tmp_dir = super().mkdtemp()
        return Path(tmp_dir)


@override_settings(
    CACHES={
        "default": {
            "BACKEND": "cache.liberal_backend.CacheClass",
        },
    }
)
class CustomCacheKeyValidationTests(SimpleTestCase):
    

    def test_custom_key_validation(self):
        
        key = "some key with spaces" * 15
        val = "a value"
        cache.set(key, val)
        self.assertEqual(cache.get(key), val)


@override_settings(
    CACHES={
        "default": {
            "BACKEND": "cache.closeable_cache.CacheClass",
        }
    }
)
class CacheClosingTests(SimpleTestCase):
    def test_close(self):
        self.assertFalse(cache.closed)
        signals.request_finished.send(self.__class__)
        self.assertTrue(cache.closed)

    def test_close_only_initialized(self):
        with self.settings(
            CACHES={
                "cache_1": {
                    "BACKEND": "cache.closeable_cache.CacheClass",
                },
                "cache_2": {
                    "BACKEND": "cache.closeable_cache.CacheClass",
                },
            }
        ):
            self.assertEqual(caches.all(initialized_only=True), [])
            signals.request_finished.send(self.__class__)
            self.assertEqual(caches.all(initialized_only=True), [])


DEFAULT_MEMORY_CACHES_SETTINGS = {
    "default": {
        "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
        "LOCATION": "unique-snowflake",
    }
}
NEVER_EXPIRING_CACHES_SETTINGS = copy.deepcopy(DEFAULT_MEMORY_CACHES_SETTINGS)
NEVER_EXPIRING_CACHES_SETTINGS["default"]["TIMEOUT"] = None


class DefaultNonExpiringCacheKeyTests(SimpleTestCase):
    

    def setUp(self):
        
        
        
        self.DEFAULT_TIMEOUT = caches[DEFAULT_CACHE_ALIAS].default_timeout

    def tearDown(self):
        del self.DEFAULT_TIMEOUT

    def test_default_expiration_time_for_keys_is_5_minutes(self):
        
        self.assertEqual(300, self.DEFAULT_TIMEOUT)

    def test_caches_with_unset_timeout_has_correct_default_timeout(self):
        
        cache = caches[DEFAULT_CACHE_ALIAS]
        self.assertEqual(self.DEFAULT_TIMEOUT, cache.default_timeout)

    @override_settings(CACHES=NEVER_EXPIRING_CACHES_SETTINGS)
    def test_caches_set_with_timeout_as_none_has_correct_default_timeout(self):
        
        cache = caches[DEFAULT_CACHE_ALIAS]
        self.assertIsNone(cache.default_timeout)
        self.assertIsNone(cache.get_backend_timeout())

    @override_settings(CACHES=DEFAULT_MEMORY_CACHES_SETTINGS)
    def test_caches_with_unset_timeout_set_expiring_key(self):
        
        key = "my-key"
        value = "my-value"
        cache = caches[DEFAULT_CACHE_ALIAS]
        cache.set(key, value)
        cache_key = cache.make_key(key)
        self.assertIsNotNone(cache._expire_info[cache_key])

    @override_settings(CACHES=NEVER_EXPIRING_CACHES_SETTINGS)
    def test_caches_set_with_timeout_as_none_set_non_expiring_key(self):
        
        key = "another-key"
        value = "another-value"
        cache = caches[DEFAULT_CACHE_ALIAS]
        cache.set(key, value)
        cache_key = cache.make_key(key)
        self.assertIsNone(cache._expire_info[cache_key])


@override_settings(
    CACHE_MIDDLEWARE_KEY_PREFIX="settingsprefix",
    CACHE_MIDDLEWARE_SECONDS=1,
    CACHES={
        "default": {
            "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
        },
    },
    USE_I18N=False,
    ALLOWED_HOSTS=[".example.com"],
)
class CacheUtils(SimpleTestCase):
    

    host = "www.example.com"
    path = "/cache/test/"
    factory = RequestFactory(headers={"host": host})

    def tearDown(self):
        cache.clear()

    def _get_request_cache(self, method="GET", query_string=None, update_cache=None):
        request = self._get_request(
            self.host, self.path, method, query_string=query_string
        )
        request._cache_update_cache = update_cache if update_cache else True
        return request

    def test_patch_vary_headers(self):
        headers = (
            
            (None, ("Accept-Encoding",), "Accept-Encoding"),
            ("Accept-Encoding", ("accept-encoding",), "Accept-Encoding"),
            ("Accept-Encoding", ("ACCEPT-ENCODING",), "Accept-Encoding"),
            ("Cookie", ("Accept-Encoding",), "Cookie, Accept-Encoding"),
            (
                "Cookie, Accept-Encoding",
                ("Accept-Encoding",),
                "Cookie, Accept-Encoding",
            ),
            (
                "Cookie, Accept-Encoding",
                ("Accept-Encoding", "cookie"),
                "Cookie, Accept-Encoding",
            ),
            (None, ("Accept-Encoding", "COOKIE"), "Accept-Encoding, COOKIE"),
            (
                "Cookie,     Accept-Encoding",
                ("Accept-Encoding", "cookie"),
                "Cookie, Accept-Encoding",
            ),
            (
                "Cookie    ,     Accept-Encoding",
                ("Accept-Encoding", "cookie"),
                "Cookie, Accept-Encoding",
            ),
            ("*", ("Accept-Language", "Cookie"), "*"),
            ("Accept-Language, Cookie", ("*",), "*"),
        )
        for initial_vary, newheaders, resulting_vary in headers:
            with self.subTest(initial_vary=initial_vary, newheaders=newheaders):
                response = HttpResponse()
                if initial_vary is not None:
                    response.headers["Vary"] = initial_vary
                patch_vary_headers(response, newheaders)
                self.assertEqual(response.headers["Vary"], resulting_vary)

    def test_get_cache_key(self):
        request = self.factory.get(self.path)
        response = HttpResponse()
        
        self.assertIsNone(get_cache_key(request))
        
        learn_cache_key(request, response)

        self.assertEqual(
            get_cache_key(request),
            "views.decorators.cache.cache_page.settingsprefix.GET."
            "18a03f9c9649f7d684af5db3524f5c99.d41d8cd98f00b204e9800998ecf8427e",
        )
        
        key_prefix = "localprefix"
        learn_cache_key(request, response, key_prefix=key_prefix)
        self.assertEqual(
            get_cache_key(request, key_prefix=key_prefix),
            "views.decorators.cache.cache_page.localprefix.GET."
            "18a03f9c9649f7d684af5db3524f5c99.d41d8cd98f00b204e9800998ecf8427e",
        )

    def test_get_cache_key_with_query(self):
        request = self.factory.get(self.path, {"test": 1})
        response = HttpResponse()
        
        self.assertIsNone(get_cache_key(request))
        
        learn_cache_key(request, response)
        
        self.assertEqual(
            get_cache_key(request),
            "views.decorators.cache.cache_page.settingsprefix.GET."
            "beaf87a9a99ee81c673ea2d67ccbec2a.d41d8cd98f00b204e9800998ecf8427e",
        )

    def test_cache_key_varies_by_url(self):
        
        request1 = self.factory.get(self.path, headers={"host": "sub-1.example.com"})
        learn_cache_key(request1, HttpResponse())
        request2 = self.factory.get(self.path, headers={"host": "sub-2.example.com"})
        learn_cache_key(request2, HttpResponse())
        self.assertNotEqual(get_cache_key(request1), get_cache_key(request2))

    def test_learn_cache_key(self):
        request = self.factory.head(self.path)
        response = HttpResponse()
        response.headers["Vary"] = "Pony"
        
        learn_cache_key(request, response)

        self.assertEqual(
            get_cache_key(request),
            "views.decorators.cache.cache_page.settingsprefix.GET."
            "18a03f9c9649f7d684af5db3524f5c99.d41d8cd98f00b204e9800998ecf8427e",
        )

    def test_patch_cache_control(self):
        tests = (
            
            
            (None, {"private": True}, {"private"}),
            ("", {"private": True}, {"private"}),
            
            ("", {"no_cache": "Set-Cookie"}, {"no-cache=Set-Cookie"}),
            ("", {"no-cache": "Set-Cookie"}, {"no-cache=Set-Cookie"}),
            ("no-cache=Set-Cookie", {"no_cache": True}, {"no-cache"}),
            ("no-cache=Set-Cookie,no-cache=Link", {"no_cache": True}, {"no-cache"}),
            (
                "no-cache=Set-Cookie",
                {"no_cache": "Link"},
                {"no-cache=Set-Cookie", "no-cache=Link"},
            ),
            (
                "no-cache=Set-Cookie,no-cache=Link",
                {"no_cache": "Custom"},
                {"no-cache=Set-Cookie", "no-cache=Link", "no-cache=Custom"},
            ),
            
            ("private", {"private": True}, {"private"}),
            ("private", {"public": True}, {"public"}),
            ("public", {"public": True}, {"public"}),
            ("public", {"private": True}, {"private"}),
            (
                "must-revalidate,max-age=60,private",
                {"public": True},
                {"must-revalidate", "max-age=60", "public"},
            ),
            (
                "must-revalidate,max-age=60,public",
                {"private": True},
                {"must-revalidate", "max-age=60", "private"},
            ),
            (
                "must-revalidate,max-age=60",
                {"public": True},
                {"must-revalidate", "max-age=60", "public"},
            ),
        )

        cc_delim_re = re.compile(r"\s*,\s*")

        for initial_cc, newheaders, expected_cc in tests:
            with self.subTest(initial_cc=initial_cc, newheaders=newheaders):
                response = HttpResponse()
                if initial_cc is not None:
                    response.headers["Cache-Control"] = initial_cc
                patch_cache_control(response, **newheaders)
                parts = set(cc_delim_re.split(response.headers["Cache-Control"]))
                self.assertEqual(parts, expected_cc)


@override_settings(
    CACHES={
        "default": {
            "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
            "KEY_PREFIX": "cacheprefix",
        },
    },
)
class PrefixedCacheUtils(CacheUtils):
    pass


@override_settings(
    CACHE_MIDDLEWARE_SECONDS=60,
    CACHE_MIDDLEWARE_KEY_PREFIX="test",
    CACHES={
        "default": {
            "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
        },
    },
)
class CacheHEADTest(SimpleTestCase):
    path = "/cache/test/"
    factory = RequestFactory()

    def tearDown(self):
        cache.clear()

    def _set_cache(self, request, msg):
        return UpdateCacheMiddleware(lambda req: HttpResponse(msg))(request)

    def test_head_caches_correctly(self):
        test_content = "test content"

        request = self.factory.head(self.path)
        request._cache_update_cache = True
        self._set_cache(request, test_content)

        request = self.factory.head(self.path)
        request._cache_update_cache = True
        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        self.assertIsNotNone(get_cache_data)
        self.assertEqual(test_content.encode(), get_cache_data.content)

    def test_head_with_cached_get(self):
        test_content = "test content"

        request = self.factory.get(self.path)
        request._cache_update_cache = True
        self._set_cache(request, test_content)

        request = self.factory.head(self.path)
        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        self.assertIsNotNone(get_cache_data)
        self.assertEqual(test_content.encode(), get_cache_data.content)


@override_settings(
    CACHE_MIDDLEWARE_KEY_PREFIX="settingsprefix",
    CACHES={
        "default": {
            "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
        },
    },
    LANGUAGES=[
        ("en", "English"),
        ("es", "Spanish"),
    ],
)
class CacheI18nTest(SimpleTestCase):
    path = "/cache/test/"
    factory = RequestFactory()

    def tearDown(self):
        cache.clear()

    @override_settings(USE_I18N=True, USE_TZ=False)
    def test_cache_key_i18n_translation(self):
        request = self.factory.get(self.path)
        lang = translation.get_language()
        response = HttpResponse()
        key = learn_cache_key(request, response)
        self.assertIn(
            lang,
            key,
            "Cache keys should include the language name when translation is active",
        )
        key2 = get_cache_key(request)
        self.assertEqual(key, key2)

    def check_accept_language_vary(self, accept_language, vary, reference_key):
        request = self.factory.get(self.path)
        request.META["HTTP_ACCEPT_LANGUAGE"] = accept_language
        request.META["HTTP_ACCEPT_ENCODING"] = "gzip;q=1.0, identity; q=0.5, *;q=0"
        response = HttpResponse()
        response.headers["Vary"] = vary
        key = learn_cache_key(request, response)
        key2 = get_cache_key(request)
        self.assertEqual(key, reference_key)
        self.assertEqual(key2, reference_key)

    @override_settings(USE_I18N=True, USE_TZ=False)
    def test_cache_key_i18n_translation_accept_language(self):
        lang = translation.get_language()
        self.assertEqual(lang, "en")
        request = self.factory.get(self.path)
        request.META["HTTP_ACCEPT_ENCODING"] = "gzip;q=1.0, identity; q=0.5, *;q=0"
        response = HttpResponse()
        response.headers["Vary"] = "accept-encoding"
        key = learn_cache_key(request, response)
        self.assertIn(
            lang,
            key,
            "Cache keys should include the language name when translation is active",
        )
        self.check_accept_language_vary(
            "en-us", "cookie, accept-language, accept-encoding", key
        )
        self.check_accept_language_vary(
            "en-US", "cookie, accept-encoding, accept-language", key
        )
        self.check_accept_language_vary(
            "en-US,en;q=0.8", "accept-encoding, accept-language, cookie", key
        )
        self.check_accept_language_vary(
            "en-US,en;q=0.8,ko;q=0.6", "accept-language, cookie, accept-encoding", key
        )
        self.check_accept_language_vary(
            "ko-kr,ko;q=0.8,en-us;q=0.5,en;q=0.3 ",
            "accept-encoding, cookie, accept-language",
            key,
        )
        self.check_accept_language_vary(
            "ko-KR,ko;q=0.8,en-US;q=0.6,en;q=0.4",
            "accept-language, accept-encoding, cookie",
            key,
        )
        self.check_accept_language_vary(
            "ko;q=1.0,en;q=0.5", "cookie, accept-language, accept-encoding", key
        )
        self.check_accept_language_vary(
            "ko, en", "cookie, accept-encoding, accept-language", key
        )
        self.check_accept_language_vary(
            "ko-KR, en-US", "accept-encoding, accept-language, cookie", key
        )

    @override_settings(USE_I18N=False, USE_TZ=True)
    def test_cache_key_i18n_timezone(self):
        request = self.factory.get(self.path)
        tz = timezone.get_current_timezone_name()
        response = HttpResponse()
        key = learn_cache_key(request, response)
        self.assertIn(
            tz,
            key,
            "Cache keys should include the time zone name when time zones are active",
        )
        key2 = get_cache_key(request)
        self.assertEqual(key, key2)

    @override_settings(USE_I18N=False)
    def test_cache_key_no_i18n(self):
        request = self.factory.get(self.path)
        lang = translation.get_language()
        tz = timezone.get_current_timezone_name()
        response = HttpResponse()
        key = learn_cache_key(request, response)
        self.assertNotIn(
            lang,
            key,
            "Cache keys shouldn't include the language name when i18n isn't active",
        )
        self.assertNotIn(
            tz,
            key,
            "Cache keys shouldn't include the time zone name when i18n isn't active",
        )

    @override_settings(
        CACHE_MIDDLEWARE_KEY_PREFIX="test",
        CACHE_MIDDLEWARE_SECONDS=60,
        USE_I18N=True,
    )
    def test_middleware(self):
        def set_cache(request, lang, msg):
            def get_response(req):
                return HttpResponse(msg)

            translation.activate(lang)
            return UpdateCacheMiddleware(get_response)(request)

        
        request = self.factory.get(self.path, {"foo": "bar", "other": "true"})
        request._cache_update_cache = True

        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        
        self.assertIsNone(get_cache_data)
        content = "Check for cache with QUERY_STRING"

        def get_response(req):
            return HttpResponse(content)

        UpdateCacheMiddleware(get_response)(request)
        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        
        self.assertIsNotNone(get_cache_data)
        self.assertEqual(get_cache_data.content, content.encode())
        
        request = self.factory.get(self.path, {"foo": "bar", "somethingelse": "true"})
        request._cache_update_cache = True
        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        self.assertIsNone(get_cache_data)

        
        en_message = "Hello world!"
        es_message = "Hola mundo!"

        request = self.factory.get(self.path)
        request._cache_update_cache = True
        set_cache(request, "en", en_message)
        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        
        self.assertIsNotNone(get_cache_data)
        self.assertEqual(get_cache_data.content, en_message.encode())
        
        request = self.factory.get(self.path)
        request._cache_update_cache = True
        set_cache(request, "es", es_message)
        
        translation.activate("en")
        
        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        self.assertEqual(get_cache_data.content, en_message.encode())
        
        translation.activate("es")
        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        self.assertEqual(get_cache_data.content, es_message.encode())
        
        translation.deactivate()

    @override_settings(
        CACHE_MIDDLEWARE_KEY_PREFIX="test",
        CACHE_MIDDLEWARE_SECONDS=60,
    )
    def test_middleware_doesnt_cache_streaming_response(self):
        request = self.factory.get(self.path)
        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        self.assertIsNone(get_cache_data)

        def get_stream_response(req):
            return StreamingHttpResponse(["Check for cache with streaming content."])

        UpdateCacheMiddleware(get_stream_response)(request)

        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        self.assertIsNone(get_cache_data)


@override_settings(
    CACHES={
        "default": {
            "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
            "KEY_PREFIX": "cacheprefix",
        },
    },
)
class PrefixedCacheI18nTest(CacheI18nTest):
    pass


def hello_world_view(request, value):
    return HttpResponse("Hello World %s" % value)


def csrf_view(request):
    return HttpResponse(csrf(request)["csrf_token"])


@override_settings(
    CACHE_MIDDLEWARE_ALIAS="other",
    CACHE_MIDDLEWARE_KEY_PREFIX="middlewareprefix",
    CACHE_MIDDLEWARE_SECONDS=30,
    CACHES={
        "default": {
            "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
        },
        "other": {
            "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
            "LOCATION": "other",
            "TIMEOUT": "1",
        },
    },
)
class CacheMiddlewareTest(SimpleTestCase):
    factory = RequestFactory()

    def setUp(self):
        self.default_cache = caches["default"]
        self.other_cache = caches["other"]

    def tearDown(self):
        self.default_cache.clear()
        self.other_cache.clear()
        super().tearDown()

    def test_constructor(self):
        
        
        
        middleware = CacheMiddleware(empty_response)

        
        self.assertEqual(middleware.cache_timeout, 30)
        self.assertEqual(middleware.key_prefix, "middlewareprefix")
        self.assertEqual(middleware.cache_alias, "other")
        self.assertEqual(middleware.cache, self.other_cache)

        
        
        as_view_decorator = CacheMiddleware(
            empty_response, cache_alias=None, key_prefix=None
        )

        self.assertEqual(
            as_view_decorator.cache_timeout, 30
        )  
        self.assertEqual(as_view_decorator.key_prefix, "")
        
        self.assertEqual(as_view_decorator.cache_alias, "default")
        self.assertEqual(as_view_decorator.cache, self.default_cache)

        
        as_view_decorator_with_custom = CacheMiddleware(
            hello_world_view, cache_timeout=60, cache_alias="other", key_prefix="foo"
        )

        self.assertEqual(as_view_decorator_with_custom.cache_timeout, 60)
        self.assertEqual(as_view_decorator_with_custom.key_prefix, "foo")
        self.assertEqual(as_view_decorator_with_custom.cache_alias, "other")
        self.assertEqual(as_view_decorator_with_custom.cache, self.other_cache)

    def test_update_cache_middleware_constructor(self):
        middleware = UpdateCacheMiddleware(empty_response)
        self.assertEqual(middleware.cache_timeout, 30)
        self.assertIsNone(middleware.page_timeout)
        self.assertEqual(middleware.key_prefix, "middlewareprefix")
        self.assertEqual(middleware.cache_alias, "other")
        self.assertEqual(middleware.cache, self.other_cache)

    def test_fetch_cache_middleware_constructor(self):
        middleware = FetchFromCacheMiddleware(empty_response)
        self.assertEqual(middleware.key_prefix, "middlewareprefix")
        self.assertEqual(middleware.cache_alias, "other")
        self.assertEqual(middleware.cache, self.other_cache)

    def test_middleware(self):
        middleware = CacheMiddleware(hello_world_view)
        prefix_middleware = CacheMiddleware(hello_world_view, key_prefix="prefix1")
        timeout_middleware = CacheMiddleware(hello_world_view, cache_timeout=1)

        request = self.factory.get("/view/")

        
        result = middleware.process_request(request)
        self.assertIsNone(result)

        response = hello_world_view(request, "1")

        
        response = middleware.process_response(request, response)

        
        result = middleware.process_request(request)
        self.assertIsNotNone(result)
        self.assertEqual(result.content, b"Hello World 1")

        
        result = prefix_middleware.process_request(request)
        self.assertIsNone(result)

        
        result = timeout_middleware.process_request(request)
        self.assertIsNotNone(result)
        self.assertEqual(result.content, b"Hello World 1")

    def test_view_decorator(self):
        
        default_view = cache_page(3)(hello_world_view)
        default_with_prefix_view = cache_page(3, key_prefix="prefix1")(hello_world_view)

        explicit_default_view = cache_page(3, cache="default")(hello_world_view)
        explicit_default_with_prefix_view = cache_page(
            3, cache="default", key_prefix="prefix1"
        )(hello_world_view)

        other_view = cache_page(1, cache="other")(hello_world_view)
        other_with_prefix_view = cache_page(1, cache="other", key_prefix="prefix2")(
            hello_world_view
        )

        request = self.factory.get("/view/")

        
        response = default_view(request, "1")
        self.assertEqual(response.content, b"Hello World 1")

        
        response = default_view(request, "2")
        self.assertEqual(response.content, b"Hello World 1")

        
        response = explicit_default_view(request, "3")
        self.assertEqual(response.content, b"Hello World 1")

        
        response = explicit_default_with_prefix_view(request, "4")
        self.assertEqual(response.content, b"Hello World 4")

        
        response = explicit_default_with_prefix_view(request, "5")
        self.assertEqual(response.content, b"Hello World 4")

        
        response = default_with_prefix_view(request, "6")
        self.assertEqual(response.content, b"Hello World 4")

        
        response = other_view(request, "7")
        self.assertEqual(response.content, b"Hello World 7")

        
        response = other_view(request, "8")
        self.assertEqual(response.content, b"Hello World 7")

        
        response = other_with_prefix_view(request, "9")
        self.assertEqual(response.content, b"Hello World 9")

        
        time.sleep(2)

        
        caches["default"]
        response = default_view(request, "11")
        self.assertEqual(response.content, b"Hello World 1")

        
        response = default_with_prefix_view(request, "12")
        self.assertEqual(response.content, b"Hello World 4")

        
        response = explicit_default_view(request, "13")
        self.assertEqual(response.content, b"Hello World 1")

        
        response = explicit_default_with_prefix_view(request, "14")
        self.assertEqual(response.content, b"Hello World 4")

        
        response = other_view(request, "15")
        self.assertEqual(response.content, b"Hello World 15")

        
        response = other_with_prefix_view(request, "16")
        self.assertEqual(response.content, b"Hello World 16")

    def test_cache_page_timeout(self):
        
        
        tests = [
            (1, 3),  
            (3, 1),  
        ]
        for max_age, page_timeout in tests:
            with self.subTest(max_age=max_age, page_timeout=page_timeout):
                view = cache_page(timeout=page_timeout)(
                    cache_control(max_age=max_age)(hello_world_view)
                )
                request = self.factory.get("/view/")
                response = view(request, "1")
                self.assertEqual(response.content, b"Hello World 1")
                time.sleep(1)
                response = view(request, "2")
                self.assertEqual(
                    response.content,
                    b"Hello World 1" if page_timeout > max_age else b"Hello World 2",
                )
            cache.clear()

    def test_cached_control_private_not_cached(self):
        
        view_with_private_cache = cache_page(3)(
            cache_control(private=True)(hello_world_view)
        )
        request = self.factory.get("/view/")
        response = view_with_private_cache(request, "1")
        self.assertEqual(response.content, b"Hello World 1")
        response = view_with_private_cache(request, "2")
        self.assertEqual(response.content, b"Hello World 2")

    def test_sensitive_cookie_not_cached(self):
        
        request = self.factory.get("/view/")
        csrf_middleware = CsrfViewMiddleware(csrf_view)
        csrf_middleware.process_view(request, csrf_view, (), {})
        cache_middleware = CacheMiddleware(csrf_middleware)

        self.assertIsNone(cache_middleware.process_request(request))
        cache_middleware(request)

        
        self.assertIsNone(cache_middleware.process_request(request))

    def test_304_response_has_http_caching_headers_but_not_cached(self):
        original_view = mock.Mock(return_value=HttpResponseNotModified())
        view = cache_page(2)(original_view)
        request = self.factory.get("/view/")
        
        view(request).close()
        response = view(request)
        response.close()
        self.assertEqual(original_view.call_count, 2)
        self.assertIsInstance(response, HttpResponseNotModified)
        self.assertIn("Cache-Control", response)
        self.assertIn("Expires", response)

    def test_per_thread(self):
        
        thread_caches = []
        middleware = CacheMiddleware(empty_response)

        def runner():
            thread_caches.append(middleware.cache)

        for _ in range(2):
            thread = threading.Thread(target=runner)
            thread.start()
            thread.join()

        self.assertIsNot(thread_caches[0], thread_caches[1])


@override_settings(
    CACHE_MIDDLEWARE_KEY_PREFIX="settingsprefix",
    CACHE_MIDDLEWARE_SECONDS=1,
    CACHES={
        "default": {
            "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
        },
    },
    USE_I18N=False,
)
class TestWithTemplateResponse(SimpleTestCase):
    

    path = "/cache/test/"
    factory = RequestFactory()

    def tearDown(self):
        cache.clear()

    def test_patch_vary_headers(self):
        headers = (
            
            (None, ("Accept-Encoding",), "Accept-Encoding"),
            ("Accept-Encoding", ("accept-encoding",), "Accept-Encoding"),
            ("Accept-Encoding", ("ACCEPT-ENCODING",), "Accept-Encoding"),
            ("Cookie", ("Accept-Encoding",), "Cookie, Accept-Encoding"),
            (
                "Cookie, Accept-Encoding",
                ("Accept-Encoding",),
                "Cookie, Accept-Encoding",
            ),
            (
                "Cookie, Accept-Encoding",
                ("Accept-Encoding", "cookie"),
                "Cookie, Accept-Encoding",
            ),
            (None, ("Accept-Encoding", "COOKIE"), "Accept-Encoding, COOKIE"),
            (
                "Cookie,     Accept-Encoding",
                ("Accept-Encoding", "cookie"),
                "Cookie, Accept-Encoding",
            ),
            (
                "Cookie    ,     Accept-Encoding",
                ("Accept-Encoding", "cookie"),
                "Cookie, Accept-Encoding",
            ),
        )
        for initial_vary, newheaders, resulting_vary in headers:
            with self.subTest(initial_vary=initial_vary, newheaders=newheaders):
                template = engines["django"].from_string("This is a test")
                response = TemplateResponse(HttpRequest(), template)
                if initial_vary is not None:
                    response.headers["Vary"] = initial_vary
                patch_vary_headers(response, newheaders)
                self.assertEqual(response.headers["Vary"], resulting_vary)

    def test_get_cache_key(self):
        request = self.factory.get(self.path)
        template = engines["django"].from_string("This is a test")
        response = TemplateResponse(HttpRequest(), template)
        key_prefix = "localprefix"
        
        self.assertIsNone(get_cache_key(request))
        
        learn_cache_key(request, response)

        self.assertEqual(
            get_cache_key(request),
            "views.decorators.cache.cache_page.settingsprefix.GET."
            "58a0a05c8a5620f813686ff969c26853.d41d8cd98f00b204e9800998ecf8427e",
        )
        
        learn_cache_key(request, response, key_prefix=key_prefix)
        self.assertEqual(
            get_cache_key(request, key_prefix=key_prefix),
            "views.decorators.cache.cache_page.localprefix.GET."
            "58a0a05c8a5620f813686ff969c26853.d41d8cd98f00b204e9800998ecf8427e",
        )

    def test_get_cache_key_with_query(self):
        request = self.factory.get(self.path, {"test": 1})
        template = engines["django"].from_string("This is a test")
        response = TemplateResponse(HttpRequest(), template)
        
        self.assertIsNone(get_cache_key(request))
        
        learn_cache_key(request, response)
        
        self.assertEqual(
            get_cache_key(request),
            "views.decorators.cache.cache_page.settingsprefix.GET."
            "0f1c2d56633c943073c4569d9a9502fe.d41d8cd98f00b204e9800998ecf8427e",
        )


class TestMakeTemplateFragmentKey(SimpleTestCase):
    def test_without_vary_on(self):
        key = make_template_fragment_key("a.fragment")
        self.assertEqual(
            key, "template.cache.a.fragment.d41d8cd98f00b204e9800998ecf8427e"
        )

    def test_with_one_vary_on(self):
        key = make_template_fragment_key("foo", ["abc"])
        self.assertEqual(key, "template.cache.foo.493e283d571a73056196f1a68efd0f66")

    def test_with_many_vary_on(self):
        key = make_template_fragment_key("bar", ["abc", "def"])
        self.assertEqual(key, "template.cache.bar.17c1a507a0cb58384f4c639067a93520")

    def test_proper_escaping(self):
        key = make_template_fragment_key("spam", ["abc:def%"])
        self.assertEqual(key, "template.cache.spam.06c8ae8e8c430b69fb0a6443504153dc")

    def test_with_ints_vary_on(self):
        key = make_template_fragment_key("foo", [1, 2, 3, 4, 5])
        self.assertEqual(key, "template.cache.foo.7ae8fd2e0d25d651c683bdeebdb29461")

    def test_with_unicode_vary_on(self):
        key = make_template_fragment_key("foo", ["42º", "😀"])
        self.assertEqual(key, "template.cache.foo.7ced1c94e543668590ba39b3c08b0237")

    def test_long_vary_on(self):
        key = make_template_fragment_key("foo", ["x" * 10000])
        self.assertEqual(key, "template.cache.foo.3670b349b5124aa56bdb50678b02b23a")


class CacheHandlerTest(SimpleTestCase):
    def test_same_instance(self):
        
        cache1 = caches["default"]
        cache2 = caches["default"]

        self.assertIs(cache1, cache2)

    def test_per_thread(self):
        
        c = []

        def runner():
            c.append(caches["default"])

        for x in range(2):
            t = threading.Thread(target=runner)
            t.start()
            t.join()

        self.assertIsNot(c[0], c[1])

    def test_nonexistent_alias(self):
        msg = "The connection 'nonexistent' doesn't exist."
        with self.assertRaisesMessage(InvalidCacheBackendError, msg):
            caches["nonexistent"]

    def test_nonexistent_backend(self):
        test_caches = CacheHandler(
            {
                "invalid_backend": {
                    "BACKEND": "django.nonexistent.NonexistentBackend",
                },
            }
        )
        msg = (
            "Could not find backend 'django.nonexistent.NonexistentBackend': "
            "No module named 'django.nonexistent'"
        )
        with self.assertRaisesMessage(InvalidCacheBackendError, msg):
            test_caches["invalid_backend"]

    def test_all(self):
        test_caches = CacheHandler(
            {
                "cache_1": {
                    "BACKEND": "django.core.cache.backends.dummy.DummyCache",
                },
                "cache_2": {
                    "BACKEND": "django.core.cache.backends.dummy.DummyCache",
                },
            }
        )
        self.assertEqual(test_caches.all(initialized_only=True), [])
        cache_1 = test_caches["cache_1"]
        self.assertEqual(test_caches.all(initialized_only=True), [cache_1])
        self.assertEqual(len(test_caches.all()), 2)
        
        self.assertEqual(len(test_caches.all(initialized_only=True)), 2)
        self.assertEqual(test_caches.all(), test_caches.all(initialized_only=True))