Merged the signals and signals_regress test packages.

This patch also made the tests less likely to pollute the global state
in case of failure.
This commit is contained in:
Loic Bistuer 2013-11-19 00:27:09 +07:00 committed by Simon Charette
parent a0f3eeccf3
commit 058e434064
5 changed files with 220 additions and 257 deletions

View File

@ -23,3 +23,20 @@ class Car(models.Model):
def __str__(self): def __str__(self):
return "%s %s" % (self.make, self.model) return "%s %s" % (self.make, self.model)
@python_2_unicode_compatible
class Author(models.Model):
name = models.CharField(max_length=20)
def __str__(self):
return self.name
@python_2_unicode_compatible
class Book(models.Model):
name = models.CharField(max_length=20)
authors = models.ManyToManyField(Author)
def __str__(self):
return self.name

View File

@ -5,9 +5,97 @@ from django.dispatch import receiver
from django.test import TestCase from django.test import TestCase
from django.utils import six from django.utils import six
from .models import Person, Car from .models import Author, Book, Car, Person
class SignalTests(TestCase):
def setUp(self):
# Save up the number of connected signals so that we can check at the
# end that all the signals we register get properly unregistered (#9989)
self.pre_signals = (
len(signals.pre_save.receivers),
len(signals.post_save.receivers),
len(signals.pre_delete.receivers),
len(signals.post_delete.receivers),
)
def tearDown(self):
# Check that all our signals got disconnected properly.
post_signals = (
len(signals.pre_save.receivers),
len(signals.post_save.receivers),
len(signals.pre_delete.receivers),
len(signals.post_delete.receivers),
)
self.assertEqual(self.pre_signals, post_signals)
def test_save_signals(self):
data = []
def pre_save_handler(signal, sender, instance, **kwargs):
data.append(
(instance, kwargs.get("raw", False))
)
def post_save_handler(signal, sender, instance, **kwargs):
data.append(
(instance, kwargs.get("created"), kwargs.get("raw", False))
)
signals.pre_save.connect(pre_save_handler)
signals.post_save.connect(post_save_handler)
try:
p1 = Person.objects.create(first_name="John", last_name="Smith")
self.assertEqual(data, [
(p1, False),
(p1, True, False),
])
data[:] = []
p1.first_name = "Tom"
p1.save()
self.assertEqual(data, [
(p1, False),
(p1, False, False),
])
data[:] = []
# Calling an internal method purely so that we can trigger a "raw" save.
p1.save_base(raw=True)
self.assertEqual(data, [
(p1, True),
(p1, False, True),
])
data[:] = []
p2 = Person(first_name="James", last_name="Jones")
p2.id = 99999
p2.save()
self.assertEqual(data, [
(p2, False),
(p2, True, False),
])
data[:] = []
p2.id = 99998
p2.save()
self.assertEqual(data, [
(p2, False),
(p2, True, False),
])
finally:
signals.post_delete.disconnect(pre_save_handler)
signals.pre_delete.disconnect(post_save_handler)
def test_delete_signals(self):
data = []
def pre_delete_handler(signal, sender, instance, **kwargs):
data.append(
(instance, instance.id is None)
)
# #8285: signals can be any callable # #8285: signals can be any callable
class PostDeleteHandler(object): class PostDeleteHandler(object):
def __init__(self, data): def __init__(self, data):
@ -17,100 +105,12 @@ class PostDeleteHandler(object):
self.data.append( self.data.append(
(instance, instance.id is None) (instance, instance.id is None)
) )
post_delete_handler = PostDeleteHandler(data)
signals.pre_delete.connect(pre_delete_handler)
class MyReceiver(object): signals.post_delete.connect(post_delete_handler)
def __init__(self, param): try:
self.param = param p1 = Person.objects.create(first_name="John", last_name="Smith")
self._run = False
def __call__(self, signal, sender, **kwargs):
self._run = True
signal.disconnect(receiver=self, sender=sender)
class SignalTests(TestCase):
def test_basic(self):
# Save up the number of connected signals so that we can check at the
# end that all the signals we register get properly unregistered (#9989)
pre_signals = (
len(signals.pre_save.receivers),
len(signals.post_save.receivers),
len(signals.pre_delete.receivers),
len(signals.post_delete.receivers),
)
data = []
def pre_save_test(signal, sender, instance, **kwargs):
data.append(
(instance, kwargs.get("raw", False))
)
signals.pre_save.connect(pre_save_test)
def post_save_test(signal, sender, instance, **kwargs):
data.append(
(instance, kwargs.get("created"), kwargs.get("raw", False))
)
signals.post_save.connect(post_save_test)
def pre_delete_test(signal, sender, instance, **kwargs):
data.append(
(instance, instance.id is None)
)
signals.pre_delete.connect(pre_delete_test)
post_delete_test = PostDeleteHandler(data)
signals.post_delete.connect(post_delete_test)
# throw a decorator syntax receiver into the mix
@receiver(signals.pre_save)
def pre_save_decorator_test(signal, sender, instance, **kwargs):
data.append(instance)
@receiver(signals.pre_save, sender=Car)
def pre_save_decorator_sender_test(signal, sender, instance, **kwargs):
data.append(instance)
p1 = Person(first_name="John", last_name="Smith")
self.assertEqual(data, [])
p1.save()
self.assertEqual(data, [
(p1, False),
p1,
(p1, True, False),
])
data[:] = []
p1.first_name = "Tom"
p1.save()
self.assertEqual(data, [
(p1, False),
p1,
(p1, False, False),
])
data[:] = []
# Car signal (sender defined)
c1 = Car(make="Volkswagon", model="Passat")
c1.save()
self.assertEqual(data, [
(c1, False),
c1,
c1,
(c1, True, False),
])
data[:] = []
# Calling an internal method purely so that we can trigger a "raw" save.
p1.save_base(raw=True)
self.assertEqual(data, [
(p1, True),
p1,
(p1, False, True),
])
data[:] = []
p1.delete() p1.delete()
self.assertEqual(data, [ self.assertEqual(data, [
(p1, False), (p1, False),
@ -121,27 +121,14 @@ class SignalTests(TestCase):
p2 = Person(first_name="James", last_name="Jones") p2 = Person(first_name="James", last_name="Jones")
p2.id = 99999 p2.id = 99999
p2.save() p2.save()
self.assertEqual(data, [
(p2, False),
p2,
(p2, True, False),
])
data[:] = []
p2.id = 99998 p2.id = 99998
p2.save() p2.save()
self.assertEqual(data, [
(p2, False),
p2,
(p2, True, False),
])
data[:] = []
p2.delete() p2.delete()
self.assertEqual(data, [ self.assertEqual(data, [
(p2, False), (p2, False),
(p2, False) (p2, False)
]) ])
data[:] = []
self.assertQuerysetEqual( self.assertQuerysetEqual(
Person.objects.all(), [ Person.objects.all(), [
@ -149,29 +136,102 @@ class SignalTests(TestCase):
], ],
six.text_type six.text_type
) )
finally:
signals.post_delete.disconnect(pre_delete_handler)
signals.pre_delete.disconnect(post_delete_handler)
signals.post_delete.disconnect(post_delete_test) def test_decorators(self):
signals.pre_delete.disconnect(pre_delete_test) data = []
signals.post_save.disconnect(post_save_test)
signals.pre_save.disconnect(pre_save_test)
signals.pre_save.disconnect(pre_save_decorator_test)
signals.pre_save.disconnect(pre_save_decorator_sender_test, sender=Car)
# Check that all our signals got disconnected properly. @receiver(signals.pre_save)
post_signals = ( def decorated_handler(signal, sender, instance, **kwargs):
len(signals.pre_save.receivers), data.append(instance)
len(signals.post_save.receivers),
len(signals.pre_delete.receivers), @receiver(signals.pre_save, sender=Car)
len(signals.post_delete.receivers), def decorated_handler_with_sender_arg(signal, sender, instance, **kwargs):
) data.append(instance)
self.assertEqual(pre_signals, post_signals)
try:
c1 = Car.objects.create(make="Volkswagon", model="Passat")
self.assertEqual(data, [c1, c1])
finally:
signals.post_delete.disconnect(decorated_handler)
signals.pre_delete.disconnect(decorated_handler_with_sender_arg, sender=Car)
def test_save_and_delete_signals_with_m2m(self):
data = []
def pre_save_handler(signal, sender, instance, **kwargs):
data.append('pre_save signal, %s' % instance)
if kwargs.get('raw'):
data.append('Is raw')
def post_save_handler(signal, sender, instance, **kwargs):
data.append('post_save signal, %s' % instance)
if 'created' in kwargs:
if kwargs['created']:
data.append('Is created')
else:
data.append('Is updated')
if kwargs.get('raw'):
data.append('Is raw')
def pre_delete_handler(signal, sender, instance, **kwargs):
data.append('pre_save signal, %s' % instance)
data.append('instance.id is not None: %s' % (instance.id is not None))
def post_delete_handler(signal, sender, instance, **kwargs):
data.append('post_delete signal, %s' % instance)
data.append('instance.id is not None: %s' % (instance.id is not None))
signals.pre_save.connect(pre_save_handler)
signals.post_save.connect(post_save_handler)
signals.pre_delete.connect(pre_delete_handler)
signals.post_delete.connect(post_delete_handler)
try:
a1 = Author.objects.create(name='Neal Stephenson')
self.assertEqual(data, [
"pre_save signal, Neal Stephenson",
"post_save signal, Neal Stephenson",
"Is created"
])
data[:] = []
b1 = Book.objects.create(name='Snow Crash')
self.assertEqual(data, [
"pre_save signal, Snow Crash",
"post_save signal, Snow Crash",
"Is created"
])
data[:] = []
# Assigning and removing to/from m2m shouldn't generate an m2m signal.
b1.authors = [a1]
self.assertEqual(data, [])
b1.authors = []
self.assertEqual(data, [])
finally:
signals.post_delete.disconnect(pre_save_handler)
signals.pre_delete.disconnect(post_save_handler)
signals.post_save.disconnect(pre_delete_handler)
signals.pre_save.disconnect(post_delete_handler)
def test_disconnect_in_dispatch(self): def test_disconnect_in_dispatch(self):
""" """
Test that signals that disconnect when being called don't mess future Test that signals that disconnect when being called don't mess future
dispatching. dispatching.
""" """
a, b = MyReceiver(1), MyReceiver(2)
class Handler(object):
def __init__(self, param):
self.param = param
self._run = False
def __call__(self, signal, sender, **kwargs):
self._run = True
signal.disconnect(receiver=self, sender=sender)
a, b = Handler(1), Handler(2)
signals.post_save.connect(sender=Person, receiver=a) signals.post_save.connect(sender=Person, receiver=a)
signals.post_save.connect(sender=Person, receiver=b) signals.post_save.connect(sender=Person, receiver=b)
Person.objects.create(first_name='John', last_name='Smith') Person.objects.create(first_name='John', last_name='Smith')

View File

@ -1,19 +0,0 @@
from django.db import models
from django.utils.encoding import python_2_unicode_compatible
@python_2_unicode_compatible
class Author(models.Model):
name = models.CharField(max_length=20)
def __str__(self):
return self.name
@python_2_unicode_compatible
class Book(models.Model):
name = models.CharField(max_length=20)
authors = models.ManyToManyField(Author)
def __str__(self):
return self.name

View File

@ -1,95 +0,0 @@
from __future__ import unicode_literals
from django.db import models
from django.test import TestCase
from .models import Author, Book
class SignalsRegressTests(TestCase):
"""
Testing signals before/after saving and deleting.
"""
def get_signal_output(self, fn, *args, **kwargs):
# Flush any existing signal output
self.signal_output = []
fn(*args, **kwargs)
return self.signal_output
def pre_save_test(self, signal, sender, instance, **kwargs):
self.signal_output.append('pre_save signal, %s' % instance)
if kwargs.get('raw'):
self.signal_output.append('Is raw')
def post_save_test(self, signal, sender, instance, **kwargs):
self.signal_output.append('post_save signal, %s' % instance)
if 'created' in kwargs:
if kwargs['created']:
self.signal_output.append('Is created')
else:
self.signal_output.append('Is updated')
if kwargs.get('raw'):
self.signal_output.append('Is raw')
def pre_delete_test(self, signal, sender, instance, **kwargs):
self.signal_output.append('pre_save signal, %s' % instance)
self.signal_output.append('instance.id is not None: %s' % (instance.id is not None))
def post_delete_test(self, signal, sender, instance, **kwargs):
self.signal_output.append('post_delete signal, %s' % instance)
self.signal_output.append('instance.id is not None: %s' % (instance.id is not None))
def setUp(self):
self.signal_output = []
# Save up the number of connected signals so that we can check at the end
# that all the signals we register get properly unregistered (#9989)
self.pre_signals = (len(models.signals.pre_save.receivers),
len(models.signals.post_save.receivers),
len(models.signals.pre_delete.receivers),
len(models.signals.post_delete.receivers))
models.signals.pre_save.connect(self.pre_save_test)
models.signals.post_save.connect(self.post_save_test)
models.signals.pre_delete.connect(self.pre_delete_test)
models.signals.post_delete.connect(self.post_delete_test)
def tearDown(self):
models.signals.post_delete.disconnect(self.post_delete_test)
models.signals.pre_delete.disconnect(self.pre_delete_test)
models.signals.post_save.disconnect(self.post_save_test)
models.signals.pre_save.disconnect(self.pre_save_test)
# Check that all our signals got disconnected properly.
post_signals = (len(models.signals.pre_save.receivers),
len(models.signals.post_save.receivers),
len(models.signals.pre_delete.receivers),
len(models.signals.post_delete.receivers))
self.assertEqual(self.pre_signals, post_signals)
def test_model_signals(self):
""" Model saves should throw some signals. """
a1 = Author(name='Neal Stephenson')
self.assertEqual(self.get_signal_output(a1.save), [
"pre_save signal, Neal Stephenson",
"post_save signal, Neal Stephenson",
"Is created"
])
b1 = Book(name='Snow Crash')
self.assertEqual(self.get_signal_output(b1.save), [
"pre_save signal, Snow Crash",
"post_save signal, Snow Crash",
"Is created"
])
def test_m2m_signals(self):
""" Assigning and removing to/from m2m shouldn't generate an m2m signal """
b1 = Book(name='Snow Crash')
self.get_signal_output(b1.save)
a1 = Author(name='Neal Stephenson')
self.get_signal_output(a1.save)
self.assertEqual(self.get_signal_output(setattr, b1, 'authors', [a1]), [])
self.assertEqual(self.get_signal_output(setattr, b1, 'authors', []), [])