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

Question

使用django-cacheops,我想测试我的视图是否按照我的意图进行缓存.在我的测试用例中,我将cacheops cache_read信号连接到一个处理程序,该处理程序应该在缓存中为命中或未命中增加一个值.但是,信号永远不会被触发.有没有人知道在测试用例中连接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}])

Answer 1

在这个特定的例子中，事实证明我的测试用例是 django Rest 框架的 APITestCase 的子类，而 APITestCase 又是 django 的 SimpleTestCase 的子类。

查看cacheops源代码，我发现这些测试子类TransactionTestCase，并切换测试用例解决了这个问题。

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