1
0
mirror of https://github.com/django/django.git synced 2025-04-01 12:06:43 +00:00

Completed django.dispatch test coverage.

This commit is contained in:
Adam Chainz 2016-11-20 18:02:10 +00:00 committed by Tim Graham
parent 66505b72c5
commit c70ffeeb8b
2 changed files with 83 additions and 19 deletions

View File

@ -0,0 +1,24 @@
import warnings
from django.dispatch import Signal
from django.test import SimpleTestCase
a_signal = Signal(providing_args=['val'])
def receiver_1_arg(val, **kwargs):
return val
class DispatcherTests(SimpleTestCase):
def test_disconnect_weak_deprecated(self):
a_signal.connect(receiver_1_arg)
with warnings.catch_warnings(record=True) as warns:
warnings.simplefilter('always')
a_signal.disconnect(receiver_1_arg, weak=True)
self.assertEqual(len(warns), 1)
self.assertEqual(
str(warns[0].message),
'Passing `weak` to disconnect has no effect.',
)

View File

@ -1,11 +1,12 @@
import gc import gc
import sys import sys
import time import time
import unittest
import weakref import weakref
from types import TracebackType from types import TracebackType
from django.dispatch import Signal, receiver from django.dispatch import Signal, receiver
from django.test import SimpleTestCase
from django.test.utils import override_settings
if sys.platform.startswith('java'): if sys.platform.startswith('java'):
def garbage_collect(): def garbage_collect():
@ -42,8 +43,7 @@ c_signal = Signal(providing_args=["val"])
d_signal = Signal(providing_args=["val"], use_caching=True) d_signal = Signal(providing_args=["val"], use_caching=True)
class DispatcherTests(unittest.TestCase): class DispatcherTests(SimpleTestCase):
"""Test suite for dispatcher (barely started)"""
def assertTestIsClean(self, signal): def assertTestIsClean(self, signal):
"""Assert that everything has been cleaned up automatically""" """Assert that everything has been cleaned up automatically"""
@ -53,30 +53,55 @@ class DispatcherTests(unittest.TestCase):
self.assertFalse(signal.has_listeners()) self.assertFalse(signal.has_listeners())
self.assertEqual(signal.receivers, []) self.assertEqual(signal.receivers, [])
def test_exact(self): @override_settings(DEBUG=True)
def test_cannot_connect_no_kwargs(self):
def receiver_no_kwargs(sender):
pass
msg = 'Signal receivers must accept keyword arguments (**kwargs).'
with self.assertRaisesMessage(ValueError, msg):
a_signal.connect(receiver_no_kwargs)
self.assertTestIsClean(a_signal)
@override_settings(DEBUG=True)
def test_cannot_connect_non_callable(self):
msg = 'Signal receivers must be callable.'
with self.assertRaisesMessage(AssertionError, msg):
a_signal.connect(object())
self.assertTestIsClean(a_signal)
def test_send(self):
a_signal.connect(receiver_1_arg, sender=self) a_signal.connect(receiver_1_arg, sender=self)
expected = [(receiver_1_arg, "test")] result = a_signal.send(sender=self, val='test')
result = a_signal.send(sender=self, val="test") self.assertEqual(result, [(receiver_1_arg, 'test')])
self.assertEqual(result, expected)
a_signal.disconnect(receiver_1_arg, sender=self) a_signal.disconnect(receiver_1_arg, sender=self)
self.assertTestIsClean(a_signal) self.assertTestIsClean(a_signal)
def test_ignored_sender(self): def test_send_no_receivers(self):
result = a_signal.send(sender=self, val='test')
self.assertEqual(result, [])
def test_send_connected_no_sender(self):
a_signal.connect(receiver_1_arg) a_signal.connect(receiver_1_arg)
expected = [(receiver_1_arg, "test")] result = a_signal.send(sender=self, val='test')
result = a_signal.send(sender=self, val="test") self.assertEqual(result, [(receiver_1_arg, 'test')])
self.assertEqual(result, expected)
a_signal.disconnect(receiver_1_arg) a_signal.disconnect(receiver_1_arg)
self.assertTestIsClean(a_signal) self.assertTestIsClean(a_signal)
def test_send_different_no_sender(self):
a_signal.connect(receiver_1_arg, sender=object)
result = a_signal.send(sender=self, val='test')
self.assertEqual(result, [])
a_signal.disconnect(receiver_1_arg, sender=object)
self.assertTestIsClean(a_signal)
def test_garbage_collected(self): def test_garbage_collected(self):
a = Callable() a = Callable()
a_signal.connect(a.a, sender=self) a_signal.connect(a.a, sender=self)
expected = []
del a del a
garbage_collect() garbage_collect()
result = a_signal.send(sender=self, val="test") result = a_signal.send(sender=self, val="test")
self.assertEqual(result, expected) self.assertEqual(result, [])
self.assertTestIsClean(a_signal) self.assertTestIsClean(a_signal)
def test_cached_garbaged_collected(self): def test_cached_garbaged_collected(self):
@ -126,8 +151,25 @@ class DispatcherTests(unittest.TestCase):
a_signal.disconnect(dispatch_uid="uid") a_signal.disconnect(dispatch_uid="uid")
self.assertTestIsClean(a_signal) self.assertTestIsClean(a_signal)
def test_robust(self): def test_send_robust_success(self):
"""Test the send_robust() function""" a_signal.connect(receiver_1_arg)
result = a_signal.send_robust(sender=self, val='test')
self.assertEqual(result, [(receiver_1_arg, 'test')])
a_signal.disconnect(receiver_1_arg)
self.assertTestIsClean(a_signal)
def test_send_robust_no_receivers(self):
result = a_signal.send_robust(sender=self, val='test')
self.assertEqual(result, [])
def test_send_robust_ignored_sender(self):
a_signal.connect(receiver_1_arg)
result = a_signal.send_robust(sender=self, val='test')
self.assertEqual(result, [(receiver_1_arg, 'test')])
a_signal.disconnect(receiver_1_arg)
self.assertTestIsClean(a_signal)
def test_send_robust_fail(self):
def fails(val, **kwargs): def fails(val, **kwargs):
raise ValueError('this') raise ValueError('this')
a_signal.connect(fails) a_signal.connect(fails)
@ -175,10 +217,8 @@ class DispatcherTests(unittest.TestCase):
self.assertFalse(a_signal.has_listeners(sender=object())) self.assertFalse(a_signal.has_listeners(sender=object()))
class ReceiverTestCase(unittest.TestCase): class ReceiverTestCase(SimpleTestCase):
"""
Test suite for receiver.
"""
def test_receiver_single_signal(self): def test_receiver_single_signal(self):
@receiver(a_signal) @receiver(a_signal)
def f(val, **kwargs): def f(val, **kwargs):