579 lines
19 KiB
Python
579 lines
19 KiB
Python
"""
|
|
Testing signals emitted on changing m2m relations.
|
|
"""
|
|
|
|
from django.db import models
|
|
from django.test import TestCase
|
|
|
|
from .models import Car, Part, Person, SportsCar
|
|
|
|
|
|
class ManyToManySignalsTest(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.vw = Car.objects.create(name="VW")
|
|
cls.bmw = Car.objects.create(name="BMW")
|
|
cls.toyota = Car.objects.create(name="Toyota")
|
|
|
|
cls.wheelset = Part.objects.create(name="Wheelset")
|
|
cls.doors = Part.objects.create(name="Doors")
|
|
cls.engine = Part.objects.create(name="Engine")
|
|
cls.airbag = Part.objects.create(name="Airbag")
|
|
cls.sunroof = Part.objects.create(name="Sunroof")
|
|
|
|
cls.alice = Person.objects.create(name="Alice")
|
|
cls.bob = Person.objects.create(name="Bob")
|
|
cls.chuck = Person.objects.create(name="Chuck")
|
|
cls.daisy = Person.objects.create(name="Daisy")
|
|
|
|
def setUp(self):
|
|
self.m2m_changed_messages = []
|
|
|
|
def m2m_changed_signal_receiver(self, signal, sender, **kwargs):
|
|
message = {
|
|
"instance": kwargs["instance"],
|
|
"action": kwargs["action"],
|
|
"reverse": kwargs["reverse"],
|
|
"model": kwargs["model"],
|
|
}
|
|
if kwargs["pk_set"]:
|
|
message["objects"] = list(
|
|
kwargs["model"].objects.filter(pk__in=kwargs["pk_set"])
|
|
)
|
|
self.m2m_changed_messages.append(message)
|
|
|
|
def tearDown(self):
|
|
# disconnect all signal handlers
|
|
models.signals.m2m_changed.disconnect(
|
|
self.m2m_changed_signal_receiver, Car.default_parts.through
|
|
)
|
|
models.signals.m2m_changed.disconnect(
|
|
self.m2m_changed_signal_receiver, Car.optional_parts.through
|
|
)
|
|
models.signals.m2m_changed.disconnect(
|
|
self.m2m_changed_signal_receiver, Person.fans.through
|
|
)
|
|
models.signals.m2m_changed.disconnect(
|
|
self.m2m_changed_signal_receiver, Person.friends.through
|
|
)
|
|
|
|
def _initialize_signal_car(self, add_default_parts_before_set_signal=False):
|
|
"""Install a listener on the two m2m relations."""
|
|
models.signals.m2m_changed.connect(
|
|
self.m2m_changed_signal_receiver, Car.optional_parts.through
|
|
)
|
|
if add_default_parts_before_set_signal:
|
|
# adding a default part to our car - no signal listener installed
|
|
self.vw.default_parts.add(self.sunroof)
|
|
models.signals.m2m_changed.connect(
|
|
self.m2m_changed_signal_receiver, Car.default_parts.through
|
|
)
|
|
|
|
def test_pk_set_on_repeated_add_remove(self):
|
|
"""
|
|
m2m_changed is always fired, even for repeated calls to the same
|
|
method, but the behavior of pk_sets differs by action.
|
|
|
|
- For signals related to `add()`, only PKs that will actually be
|
|
inserted are sent.
|
|
- For `remove()` all PKs are sent, even if they will not affect the DB.
|
|
"""
|
|
pk_sets_sent = []
|
|
|
|
def handler(signal, sender, **kwargs):
|
|
if kwargs["action"] in ["pre_add", "pre_remove"]:
|
|
pk_sets_sent.append(kwargs["pk_set"])
|
|
|
|
models.signals.m2m_changed.connect(handler, Car.default_parts.through)
|
|
|
|
self.vw.default_parts.add(self.wheelset)
|
|
self.vw.default_parts.add(self.wheelset)
|
|
|
|
self.vw.default_parts.remove(self.wheelset)
|
|
self.vw.default_parts.remove(self.wheelset)
|
|
|
|
expected_pk_sets = [
|
|
{self.wheelset.pk},
|
|
set(),
|
|
{self.wheelset.pk},
|
|
{self.wheelset.pk},
|
|
]
|
|
self.assertEqual(pk_sets_sent, expected_pk_sets)
|
|
|
|
models.signals.m2m_changed.disconnect(handler, Car.default_parts.through)
|
|
|
|
def test_m2m_relations_add_remove_clear(self):
|
|
expected_messages = []
|
|
|
|
self._initialize_signal_car(add_default_parts_before_set_signal=True)
|
|
|
|
self.vw.default_parts.add(self.wheelset, self.doors, self.engine)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "pre_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.doors, self.engine, self.wheelset],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "post_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.doors, self.engine, self.wheelset],
|
|
}
|
|
)
|
|
self.assertEqual(self.m2m_changed_messages, expected_messages)
|
|
|
|
# give the BMW and Toyota some doors as well
|
|
self.doors.car_set.add(self.bmw, self.toyota)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.doors,
|
|
"action": "pre_add",
|
|
"reverse": True,
|
|
"model": Car,
|
|
"objects": [self.bmw, self.toyota],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.doors,
|
|
"action": "post_add",
|
|
"reverse": True,
|
|
"model": Car,
|
|
"objects": [self.bmw, self.toyota],
|
|
}
|
|
)
|
|
self.assertEqual(self.m2m_changed_messages, expected_messages)
|
|
|
|
def test_m2m_relations_signals_remove_relation(self):
|
|
self._initialize_signal_car()
|
|
# remove the engine from the self.vw and the airbag (which is not set
|
|
# but is returned)
|
|
self.vw.default_parts.remove(self.engine, self.airbag)
|
|
self.assertEqual(
|
|
self.m2m_changed_messages,
|
|
[
|
|
{
|
|
"instance": self.vw,
|
|
"action": "pre_remove",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.airbag, self.engine],
|
|
},
|
|
{
|
|
"instance": self.vw,
|
|
"action": "post_remove",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.airbag, self.engine],
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_m2m_relations_signals_give_the_self_vw_some_optional_parts(self):
|
|
expected_messages = []
|
|
|
|
self._initialize_signal_car()
|
|
|
|
# give the self.vw some optional parts (second relation to same model)
|
|
self.vw.optional_parts.add(self.airbag, self.sunroof)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "pre_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.airbag, self.sunroof],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "post_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.airbag, self.sunroof],
|
|
}
|
|
)
|
|
self.assertEqual(self.m2m_changed_messages, expected_messages)
|
|
|
|
# add airbag to all the cars (even though the self.vw already has one)
|
|
self.airbag.cars_optional.add(self.vw, self.bmw, self.toyota)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.airbag,
|
|
"action": "pre_add",
|
|
"reverse": True,
|
|
"model": Car,
|
|
"objects": [self.bmw, self.toyota],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.airbag,
|
|
"action": "post_add",
|
|
"reverse": True,
|
|
"model": Car,
|
|
"objects": [self.bmw, self.toyota],
|
|
}
|
|
)
|
|
self.assertEqual(self.m2m_changed_messages, expected_messages)
|
|
|
|
def test_m2m_relations_signals_reverse_relation_with_custom_related_name(self):
|
|
self._initialize_signal_car()
|
|
# remove airbag from the self.vw (reverse relation with custom
|
|
# related_name)
|
|
self.airbag.cars_optional.remove(self.vw)
|
|
self.assertEqual(
|
|
self.m2m_changed_messages,
|
|
[
|
|
{
|
|
"instance": self.airbag,
|
|
"action": "pre_remove",
|
|
"reverse": True,
|
|
"model": Car,
|
|
"objects": [self.vw],
|
|
},
|
|
{
|
|
"instance": self.airbag,
|
|
"action": "post_remove",
|
|
"reverse": True,
|
|
"model": Car,
|
|
"objects": [self.vw],
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_m2m_relations_signals_clear_all_parts_of_the_self_vw(self):
|
|
self._initialize_signal_car()
|
|
# clear all parts of the self.vw
|
|
self.vw.default_parts.clear()
|
|
self.assertEqual(
|
|
self.m2m_changed_messages,
|
|
[
|
|
{
|
|
"instance": self.vw,
|
|
"action": "pre_clear",
|
|
"reverse": False,
|
|
"model": Part,
|
|
},
|
|
{
|
|
"instance": self.vw,
|
|
"action": "post_clear",
|
|
"reverse": False,
|
|
"model": Part,
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_m2m_relations_signals_all_the_doors_off_of_cars(self):
|
|
self._initialize_signal_car()
|
|
# take all the doors off of cars
|
|
self.doors.car_set.clear()
|
|
self.assertEqual(
|
|
self.m2m_changed_messages,
|
|
[
|
|
{
|
|
"instance": self.doors,
|
|
"action": "pre_clear",
|
|
"reverse": True,
|
|
"model": Car,
|
|
},
|
|
{
|
|
"instance": self.doors,
|
|
"action": "post_clear",
|
|
"reverse": True,
|
|
"model": Car,
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_m2m_relations_signals_reverse_relation(self):
|
|
self._initialize_signal_car()
|
|
# take all the airbags off of cars (clear reverse relation with custom
|
|
# related_name)
|
|
self.airbag.cars_optional.clear()
|
|
self.assertEqual(
|
|
self.m2m_changed_messages,
|
|
[
|
|
{
|
|
"instance": self.airbag,
|
|
"action": "pre_clear",
|
|
"reverse": True,
|
|
"model": Car,
|
|
},
|
|
{
|
|
"instance": self.airbag,
|
|
"action": "post_clear",
|
|
"reverse": True,
|
|
"model": Car,
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_m2m_relations_signals_alternative_ways(self):
|
|
expected_messages = []
|
|
|
|
self._initialize_signal_car()
|
|
|
|
# alternative ways of setting relation:
|
|
self.vw.default_parts.create(name="Windows")
|
|
p6 = Part.objects.get(name="Windows")
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "pre_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [p6],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "post_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [p6],
|
|
}
|
|
)
|
|
self.assertEqual(self.m2m_changed_messages, expected_messages)
|
|
|
|
# direct assignment clears the set first, then adds
|
|
self.vw.default_parts.set([self.wheelset, self.doors, self.engine])
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "pre_remove",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [p6],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "post_remove",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [p6],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "pre_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.doors, self.engine, self.wheelset],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "post_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.doors, self.engine, self.wheelset],
|
|
}
|
|
)
|
|
self.assertEqual(self.m2m_changed_messages, expected_messages)
|
|
|
|
def test_m2m_relations_signals_clearing_removing(self):
|
|
expected_messages = []
|
|
|
|
self._initialize_signal_car(add_default_parts_before_set_signal=True)
|
|
|
|
# set by clearing.
|
|
self.vw.default_parts.set([self.wheelset, self.doors, self.engine], clear=True)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "pre_clear",
|
|
"reverse": False,
|
|
"model": Part,
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "post_clear",
|
|
"reverse": False,
|
|
"model": Part,
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "pre_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.doors, self.engine, self.wheelset],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "post_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.doors, self.engine, self.wheelset],
|
|
}
|
|
)
|
|
self.assertEqual(self.m2m_changed_messages, expected_messages)
|
|
|
|
# set by only removing what's necessary.
|
|
self.vw.default_parts.set([self.wheelset, self.doors], clear=False)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "pre_remove",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.engine],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.vw,
|
|
"action": "post_remove",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.engine],
|
|
}
|
|
)
|
|
self.assertEqual(self.m2m_changed_messages, expected_messages)
|
|
|
|
def test_m2m_relations_signals_when_inheritance(self):
|
|
expected_messages = []
|
|
|
|
self._initialize_signal_car(add_default_parts_before_set_signal=True)
|
|
|
|
# Signals still work when model inheritance is involved
|
|
c4 = SportsCar.objects.create(name="Bugatti", price="1000000")
|
|
c4b = Car.objects.get(name="Bugatti")
|
|
c4.default_parts.set([self.doors])
|
|
expected_messages.append(
|
|
{
|
|
"instance": c4,
|
|
"action": "pre_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.doors],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": c4,
|
|
"action": "post_add",
|
|
"reverse": False,
|
|
"model": Part,
|
|
"objects": [self.doors],
|
|
}
|
|
)
|
|
self.assertEqual(self.m2m_changed_messages, expected_messages)
|
|
|
|
self.engine.car_set.add(c4)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.engine,
|
|
"action": "pre_add",
|
|
"reverse": True,
|
|
"model": Car,
|
|
"objects": [c4b],
|
|
}
|
|
)
|
|
expected_messages.append(
|
|
{
|
|
"instance": self.engine,
|
|
"action": "post_add",
|
|
"reverse": True,
|
|
"model": Car,
|
|
"objects": [c4b],
|
|
}
|
|
)
|
|
self.assertEqual(self.m2m_changed_messages, expected_messages)
|
|
|
|
def _initialize_signal_person(self):
|
|
# Install a listener on the two m2m relations.
|
|
models.signals.m2m_changed.connect(
|
|
self.m2m_changed_signal_receiver, Person.fans.through
|
|
)
|
|
models.signals.m2m_changed.connect(
|
|
self.m2m_changed_signal_receiver, Person.friends.through
|
|
)
|
|
|
|
def test_m2m_relations_with_self_add_friends(self):
|
|
self._initialize_signal_person()
|
|
self.alice.friends.set([self.bob, self.chuck])
|
|
self.assertEqual(
|
|
self.m2m_changed_messages,
|
|
[
|
|
{
|
|
"instance": self.alice,
|
|
"action": "pre_add",
|
|
"reverse": False,
|
|
"model": Person,
|
|
"objects": [self.bob, self.chuck],
|
|
},
|
|
{
|
|
"instance": self.alice,
|
|
"action": "post_add",
|
|
"reverse": False,
|
|
"model": Person,
|
|
"objects": [self.bob, self.chuck],
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_m2m_relations_with_self_add_fan(self):
|
|
self._initialize_signal_person()
|
|
self.alice.fans.set([self.daisy])
|
|
self.assertEqual(
|
|
self.m2m_changed_messages,
|
|
[
|
|
{
|
|
"instance": self.alice,
|
|
"action": "pre_add",
|
|
"reverse": False,
|
|
"model": Person,
|
|
"objects": [self.daisy],
|
|
},
|
|
{
|
|
"instance": self.alice,
|
|
"action": "post_add",
|
|
"reverse": False,
|
|
"model": Person,
|
|
"objects": [self.daisy],
|
|
},
|
|
],
|
|
)
|
|
|
|
def test_m2m_relations_with_self_add_idols(self):
|
|
self._initialize_signal_person()
|
|
self.chuck.idols.set([self.alice, self.bob])
|
|
self.assertEqual(
|
|
self.m2m_changed_messages,
|
|
[
|
|
{
|
|
"instance": self.chuck,
|
|
"action": "pre_add",
|
|
"reverse": True,
|
|
"model": Person,
|
|
"objects": [self.alice, self.bob],
|
|
},
|
|
{
|
|
"instance": self.chuck,
|
|
"action": "post_add",
|
|
"reverse": True,
|
|
"model": Person,
|
|
"objects": [self.alice, self.bob],
|
|
},
|
|
],
|
|
)
|