在测试中连接django信号处理程序

使用django-cacheops，我想测试我的视图是否按照我的意图进行缓存。在我的测试案例中，我将cacheops cache_read信号连接到一个处理程序，该处理程序应该在缓存中为缓存或未命中增加一个值。但是，信号从未被触发。有没有人知道在测试用例中连接django信号处理程序的正确方法，纯粹是为了在测试用例中使用？在测试中连接django信号处理程序

这里是我迄今为止

from cacheops.signals import cache_read 

cache.set('test_cache_hits', 0) 
cache.set('test_cache_misses', 0) 

def cache_log(sender, func, hit, **kwargs): 
    # never called 
    if hit: 
     cache.incr('test_cache_hits') 
    else: 
     cache.incr('test_cache_misses') 


class BootstrapTests(TestCase): 

    @classmethod 
    def setUpClass(cls): 
     super(BootstrapTests, cls).setUpClass() 
     cache_read.connect(cache_log) 
     assert cache_read.has_listeners() 

    def test_something_that_should_fill_and_retrieve_cache(self): 
     .... 
     hits = cache.get('test_cache_hits') # always 0

我也试过在模块级连接信号处理程序，并在常规测试用例setUp方法，都具有相同的结果。

编辑：这是我的实际测试代码，加上我测试的对象。我正在使用装饰器缓存一个函数cached_as。该测试目前失败。

boostrap.py

class BootstrapData(object): 

    def __init__(self, app, person=None): 
     self.app = app 

    def get_homepage_dict(self, context={}): 

     url_name = self.app.url_name 

     @cached_as(App.objects.filter(url_name=url_name), extra=context) 
     def _get_homepage_dict(): 
      if self.app.homepage is None: 
       return None 

      concrete_module_class = MODULE_MAPPING[self.app.homepage.type] 
      serializer_class_name = f'{concrete_module_class.__name__}Serializer' 
      serializer_class = getattr(api.serializers, serializer_class_name) 
      concrete_module = concrete_module_class.objects.get(module=self.app.homepage) 
      serializer = serializer_class(context=context) 
      key = concrete_module_class.__name__ 
      return { 
       key: serializer.to_representation(instance=concrete_module) 
      } 
     return _get_homepage_dict()

test_bootstrap.py

class BootstrapDataTest(TestCase): 

    def setUp(self): 
     super(BootstrapDataTest, self).setUp() 

     def set_signal(signal=None, **kwargs): 
      self.signal_calls.append(kwargs) 
     self.signal_calls = [] 
     cache_read.connect(set_signal, dispatch_uid=1, weak=False) 
     self.app = self.setup_basic_app() # creates an 'App' model and saves it 

    def tearDown(self): 
     cache_read.disconnect(dispatch_uid=1) 

    def test_boostrap_data_is_cached(self): 

     obj = BootstrapData(self.app) 
     obj.get_homepage_dict() 

     # fails, self.signal_calls == [] 
     self.assertEqual(self.signal_calls, [{'sender': App, 'func': None, 'hit': False }]) 

     self.signal_calls = [] 

     obj.get_homepage_dict() 
     self.assertEqual(self.signal_calls, [{'sender': App, 'func': None, 'hit': True}])

来源

2017-06-29 bharling

你确定测试会触发'cache_read'信号吗？不只是任何缓存读取都会触发cacheops“cache_read”信号。 https://github.com/Suor/django-cacheops/blob/master/cacheops/query.py 同样在cachops包中，他们有一个测试例子，它们在'setUp'方法中连接一个信号。 https://github.com/Suor/django-cacheops/blob/daa907d6ec5dc98d5cc80a3d519469fb134bd0bb/tests/tests.py#L917 – ARJMP

我可能是错的，但django-cacheops被设计为缓存查询集而不是常规密钥。因此，'cache_read'信号将不会被调用，因为您试图获取的值不是查询集。 – mattjegan

你可能不会做缓存请求或永远不会点击。你应该提供你的测试代码。 – Suor

在这个特定的情况下，事实证明，我的测试用例将django rest框架的APITestCase继承了下来，这继承了django的SimpleTestCase的子类。

寻找cacheops的来源，我发现那些测试的子类TransactionTestCase，并将测试案例切换出来解决了这个问题。

有兴趣知道为什么会出现这种情况，但问题目前已解决。

来源

2017-09-27 14:25:26 bharling

这是因为'SimpleTestCase'将每个测试包装到一个事务中，并且一旦它处于肮脏状态，cacheops在事务中关闭。这是因为从事务内可见的db状态是不同的，不应该被缓存也不依赖于缓存。 – Suor

啊哈，谢谢你的澄清 – bharling

我不明白为什么会这样，但我会努力做一个有用的答案呢。

首先，如果你想测试缓存是否工作，你不应该依靠自己的副作用来检查它，并且信号是其主要功能的副作用 - 防止数据库调用。尝试为：

def test_it_works(self): 
    with self.assertNumQueries(1): 
     obj.get_homepage_dict() 

    with self.assertNumQueries(0): 
     obj.get_homepage_dict()

第二，如果你想知道发生了什么事你添加打印无处不在，包括cacheops代码可挖，看看它停止。或者，你可以做一个测试让我看到，指令在这里https://github.com/Suor/django-cacheops#writing-a-test。

最后，你的测试有点不对。对于@cached_as()发件人将是None和func将被装饰的功能。

来源

2017-06-30 16:21:07 Suor

感谢您的帮助 - 几个月后回来，我了解到，这是因为我的测试用例没有使用TransactionTestCase作为它的基础（就像在缓存中进行测试一样）。切换测试用例基类使其全部工作。 – bharling

在测试中连接django信号处理程序

回答

相关问题