""" 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], }, ], )