<s> from __future__ import absolute_import , unicode_literals <EOL> import sys <EOL> from datetime import timedelta <EOL> from billiard . einfo import ExceptionInfo <EOL> import django <EOL> from django . core . cache . backends . base import InvalidCacheBackendError <EOL> from celery import result <EOL> from celery import states <EOL> from celery . utils import gen_unique_id <EOL> from djcelery . app import app <EOL> from djcelery . backends . cache import CacheBackend <EOL> from djcelery . tests . utils import unittest <EOL> class SomeClass ( object ) : <EOL> def __init__ ( self , data ) : <EOL> self . data = data <EOL> class test_CacheBackend ( unittest . TestCase ) : <EOL> def test_mark_as_done ( self ) : <EOL> cb = CacheBackend ( app = app ) <EOL> tid = gen_unique_id ( ) <EOL> self . assertEqual ( cb . get_status ( tid ) , states . PENDING ) <EOL> self . assertIsNone ( cb . get_result ( tid ) ) <EOL> cb . mark_as_done ( tid , <NUM_LIT> ) <EOL> self . assertEqual ( cb . get_status ( tid ) , states . SUCCESS ) <EOL> self . assertEqual ( cb . get_result ( tid ) , <NUM_LIT> ) <EOL> self . assertTrue ( cb . get_result ( tid ) , <NUM_LIT> ) <EOL> def test_forget ( self ) : <EOL> b = CacheBackend ( app = app ) <EOL> tid = gen_unique_id ( ) <EOL> b . mark_as_done ( tid , { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> self . assertEqual ( b . get_result ( tid ) . get ( '<STR_LIT:foo>' ) , '<STR_LIT:bar>' ) <EOL> b . forget ( tid ) <EOL> self . assertNotIn ( tid , b . _cache ) <EOL> self . assertIsNone ( b . get_result ( tid ) ) <EOL> def test_save_restore_delete_group ( self ) : <EOL> backend = CacheBackend ( app = app ) <EOL> group_id = gen_unique_id ( ) <EOL> subtask_ids = [ gen_unique_id ( ) for i in range ( <NUM_LIT:10> ) ] <EOL> subtasks = list ( map ( result . AsyncResult , subtask_ids ) ) <EOL> res = result . GroupResult ( group_id , subtasks ) <EOL> res . save ( backend = backend ) <EOL> saved = result . GroupResult . restore ( group_id , backend = backend ) <EOL> self . assertListEqual ( saved . subtasks , subtasks ) <EOL> self . assertEqual ( saved . id , group_id ) <EOL> saved . delete ( backend = backend ) <EOL> self . assertIsNone ( result . GroupResult . restore ( group_id , <EOL> backend = backend ) ) <EOL> def test_is_pickled ( self ) : <EOL> cb = CacheBackend ( app = app ) <EOL> tid2 = gen_unique_id ( ) <EOL> result = { '<STR_LIT:foo>' : '<STR_LIT>' , '<STR_LIT:bar>' : SomeClass ( <NUM_LIT> ) } <EOL> cb . mark_as_done ( tid2 , result ) <EOL> rindb = cb . get_result ( tid2 ) <EOL> self . assertEqual ( rindb . get ( '<STR_LIT:foo>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( rindb . get ( '<STR_LIT:bar>' ) . data , <NUM_LIT> ) <EOL> def test_mark_as_failure ( self ) : <EOL> cb = CacheBackend ( app = app ) <EOL> einfo = None <EOL> tid3 = gen_unique_id ( ) <EOL> try : <EOL> raise KeyError ( '<STR_LIT:foo>' ) <EOL> except KeyError as exception : <EOL> einfo = ExceptionInfo ( sys . exc_info ( ) ) <EOL> cb . mark_as_failure ( tid3 , exception , traceback = einfo . traceback ) <EOL> self . assertEqual ( cb . get_status ( tid3 ) , states . FAILURE ) <EOL> self . assertIsInstance ( cb . get_result ( tid3 ) , KeyError ) <EOL> self . assertEqual ( cb . get_traceback ( tid3 ) , einfo . traceback ) <EOL> def test_process_cleanup ( self ) : <EOL> cb = CacheBackend ( app = app ) <EOL> cb . process_cleanup ( ) <EOL> def test_set_expires ( self ) : <EOL> cb1 = CacheBackend ( app = app , expires = timedelta ( seconds = <NUM_LIT:16> ) ) <EOL> self . assertEqual ( cb1 . expires , <NUM_LIT:16> ) <EOL> cb2 = CacheBackend ( app = app , expires = <NUM_LIT:32> ) <EOL> self . assertEqual ( cb2 . expires , <NUM_LIT:32> ) <EOL> class test_custom_CacheBackend ( unittest . TestCase ) : <EOL> def test_custom_cache_backend ( self ) : <EOL> from celery import current_app <EOL> prev_backend = current_app . conf . CELERY_CACHE_BACKEND <EOL> prev_module = sys . modules [ '<STR_LIT>' ] <EOL> if django . VERSION >= ( <NUM_LIT:1> , <NUM_LIT:3> ) : <EOL> current_app . conf . CELERY_CACHE_BACKEND = '<STR_LIT>' <EOL> else : <EOL> current_app . conf . CELERY_CACHE_BACKEND = '<STR_LIT>' <EOL> sys . modules . pop ( '<STR_LIT>' ) <EOL> try : <EOL> from djcelery . backends . cache import cache <EOL> from django . core . cache import cache as django_cache <EOL> self . assertEqual ( cache . __class__ . __module__ , <EOL> '<STR_LIT>' ) <EOL> self . assertIsNot ( cache , django_cache ) <EOL> finally : <EOL> current_app . conf . CELERY_CACHE_BACKEND = prev_backend <EOL> sys . modules [ '<STR_LIT>' ] = prev_module <EOL> class test_MemcacheWrapper ( unittest . TestCase ) : <EOL> def test_memcache_wrapper ( self ) : <EOL> try : <EOL> from django . core . cache . backends import memcached <EOL> from django . core . cache . backends import locmem <EOL> except InvalidCacheBackendError : <EOL> sys . stderr . write ( <EOL> '<STR_LIT>' ) <EOL> return <EOL> try : <EOL> prev_cache_cls = memcached . CacheClass <EOL> memcached . CacheClass = locmem . CacheClass <EOL> except AttributeError : <EOL> return <EOL> prev_backend_module = sys . modules . pop ( '<STR_LIT>' ) <EOL> try : <EOL> from djcelery . backends . cache import cache <EOL> key = '<STR_LIT>' <EOL> val = '<STR_LIT>' <EOL> default = '<STR_LIT>' <EOL> self . assertEqual ( cache . get ( key , default = default ) , default ) <EOL> cache . set ( key , val ) <EOL> self . assertEqual ( cache . get ( key , default = default ) , val ) <EOL> finally : <EOL> memcached . CacheClass = prev_cache_cls <EOL> sys . modules [ '<STR_LIT>' ] = prev_backend_module </s>
