在测试中连接 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

我不明白为什么会发生这种情况，但无论如何我都会尝试做出有用的答案。

首先，如果你想测试缓存是否有效，你不应该依赖它自己的副作用来检查它，信号是它的主要功能的副作用——防止数据库调用。尝试测试：

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()，sender 为None，func 为修饰函数。

Answer 2

在这个特定的案例中，结果证明我的测试用例继承了 django rest 框架的 APITestCase，而 APITestCase 又继承了 django 的 SimpleTestCase。

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

很想知道为什么会这样，但问题现在已经解决了。