Kiểm tra xem tôi đã kết nối với một tín hiệu cụ thể ở Django

Question

Tôi có một ứng dụng Django được gọi là "nhà xuất bản", nó kết nối với các tín hiệu khác nhau trong dự án django của tôi và khi nhận được nó sẽ gửi một tin nhắn đến hàng đợi thỏmq. Những gì tôi muốn làm là để có thể kiểm tra rằng mã thiết lập của tôi đang kết nối với các tín hiệu chính xác.

cấu trúc Ứng dụng của tôi trông giống như:

publisher 
    - __init__.py 
    - signals.py 
    - tests.py 

__init__.py của tôi trông giống như:

import signals 

và signals.py tôi:

def receiver_function(*args, **kwargs): 
    #Does rabbitmq stuff 

my_interesting_signal.connect(receiver_function) 

Tôi đã nghĩ đến việc vá các chức năng thu và kiểm tra xem mô hình có được gọi khi tôi gửi tín hiệu không:

tests.py:

class SignalsTeste(TestCase): 

    def_test_connection(self): 

     with patch('publisher.signals.receiver_function') as receiver_mock: 
      my_interesting_signal.application_created.send(None) 
      self.assertEquals(receiver_mock.call_count, 1) 

Tuy nhiên vì các module tín hiệu được nhập khẩu, và do đó các kết nối tín hiệu được thực hiện trước khi các cuộc thử nghiệm đang chạy, phương pháp này không làm việc như các kết nối được thực hiện trước khi chức năng là vá ......

Có ai có thể đề xuất chiến lược thay thế không?

Answer 1

Tôi đã gặp phải sự cố mô phỏng tương tự mà bạn mô tả. Giải pháp của tôi là để đạt được vào đăng ký tín hiệu của Django và khẳng định rằng chức năng của tôi đã được đăng ký với tín hiệu chính xác.

Dưới đây là thử nghiệm của tôi:

def test_signal_registry(self): 
    from foo.models import bar_func # The function I want to register. 
    from django.db.models import signals 
    registered_functions = [r[1]() for r in signals.pre_delete.receivers] 
    self.assertIn(bar_func, registered_functions) 

Một lời giải thích chút về điều đó danh sách hiểu:

"pre_delete" là thể hiện của django.dispatch.dispatcher.Signal mà tôi quan tâm đến trong trường hợp này. Bạn sẽ sử dụng "my_interesting_signal" của riêng bạn trong ví dụ của bạn. Các tín hiệu có một thuộc tính bên trong được gọi là "các bộ thu", đó là danh sách hai bộ dữ liệu, trong đó phần tử thứ hai là một hàm yếu cho hàm bạn đăng ký (do đó r [1]). Gọi một weakref trả về referent.

tôi phải chơi xung quanh với weakrefs để con số đó nhiều ra:

import weakref 
def foo(): 
    pass 
w = weakref.ref(foo) 
w() == foo 

Hope this helps.

Answer 2

Cách kiểm tra xem tín hiệu được kết nối có đang ngắt kết nối và kiểm tra kết quả của hành động này không. Gọi số <some_signal>.disconnect(...) trả về True nếu tín hiệu bị ngắt kết nối hoặc False nếu không.

Ví dụ: chúng tôi muốn kiểm tra tín hiệu post_save được kết nối với receiver_function của chúng tôi.

modules.py

def receiver_function(*args, **kwargs): 
    pass 

signals.post_save.connect(receiver_function) 

kiểm tra.py

class SignalsTest(TestCase): 
    def test_connection(self): 
     result = signals.post_save.disconnect(receiver_function) 

     self.assertTrue(result) 

Các cuộc gọi đến disconnect phải sử dụng những lập luận tương tự rằng connect gọi (sender, dispatch_uid)

Nó là cần thiết để kết nối tín hiệu một lần nữa sau khi kiểm tra, nếu không nó sẽ vẫn bị ngắt kết nối

Answer 3

Điều này khá phức tạp, vì như bạn nói, nếu bạn giả hoặc nhập bất kỳ thứ gì từ tệp bạn đưa vào bộ thu, bạn sẽ tự động kết nối nó. Đây là trên toàn bộ toàn bộ bộ thử nghiệm của bạn và không chỉ trong tệp thử nghiệm được đề cập. Dưới đây là một đoạn trích mà bạn có thể sử dụng, nhưng bạn cần phải được xử lý kỷ luật theo các ý kiến ​​về việc tránh nhập khẩu trong tệp người nhận.

from django.test import TestCase 

class ReceiverConnectionTestCase(TestCase): 
    """TestCase that allows asserting that a given receiver is connected 
    to a signal. 

    Important: this will work correctly providing you: 
     1. Do not import or patch anything in the module containing the receiver 
      in any django.test.TestCase. 
     2. Do not import (except in the context of a method) the module 
      containing the receiver in any test module. 

    This is because as soon as you import/patch, the receiver will be connected 
    by your test and will be connected for the entire test suite run. 

    If you want to test the behaviour of the receiver, you may do this 
    providing it is a unittest.TestCase, and there is no import from the 
    receiver module in that test module. 

    Usage: 

     # myapp/receivers.py 
     from django.dispatch import receiver 
     from apples.signals import apple_eaten 
     from apples.models import Apple 

     @receiver(apple_eaten, sender=Apple) 
     def my_receiver(sender, **kwargs): 
      pass 


     # tests/integration_tests.py 
     from apples.signals import apple_eaten 
     from apples.models import Apple 

     class TestMyReceiverConnection(ReceiverConnectionTestCase): 
      def test_connection(self): 
       self.assert_receiver_is_connected(
        'myapp.receivers.my_receiver', 
        signal=apple_eaten, sender=Apple) 

    """ 
    def assert_receiver_is_connected(self, receiver_string, signal, sender): 
     receivers = signal._live_receivers(sender) 
     receiver_strings = [ 
      "{}.{}".format(r.__module__, r.__name__) for r in receivers] 
     if receiver_string not in receiver_strings: 
      raise AssertionError(
       '{} is not connected to signal.'.format(receiver_string)) 

Điều này hoạt động vì Django chạy django.test.TestCase trước unittest.TestCase.