testing signals emitted on changing m2m relations disconnect all signal handlers install a listener on the two m2m relations models signals m2mchanged connect self m2mchangedsignalreceiver car optionalparts through if adddefaultpartsbeforesetsignal adding a default part to our car no signal listener installed self vw defaultparts addself sunroof models signals m2mchanged connect self m2mchangedsignalreceiver car defaultparts through def testpksetonrepeatedaddremoveself pksetssent def handlersignal sender kwargs if kwargsaction in preadd preremove pksetssent appendkwargspkset models signals m2mchanged connecthandler car defaultparts through self vw defaultparts addself wheelset self vw defaultparts addself wheelset self vw defaultparts removeself wheelset self vw defaultparts removeself wheelset expectedpksets self wheelset pk set self wheelset pk self wheelset pk self assertequalpksetssent expectedpksets models signals m2mchanged disconnecthandler car defaultparts through def testm2mrelationsaddremoveclearself expectedmessages self initializesignalcaradddefaultpartsbeforesetsignaltrue self vw defaultparts addself wheelset self doors self engine expectedmessages append instance self vw action preadd reverse false model part objects self doors self engine self wheelset expectedmessages append instance self vw action postadd reverse false model part objects self doors self engine self wheelset self assertequalself m2mchangedmessages expectedmessages give the bmw and toyota some doors as well self doors carset addself bmw self toyota expectedmessages append instance self doors action preadd reverse true model car objects self bmw self toyota expectedmessages append instance self doors action postadd reverse true model car objects self bmw self toyota self assertequalself m2mchangedmessages expectedmessages def testm2mrelationssignalsremoverelationself self initializesignalcar remove the engine from the self vw and the airbag which is not set but is returned self vw defaultparts removeself engine self airbag self assertequal self m2mchangedmessages instance self vw action preremove reverse false model part objects self airbag self engine instance self vw action postremove reverse false model part objects self airbag self engine def testm2mrelationssignalsgivetheselfvwsomeoptionalpartsself expectedmessages self initializesignalcar give the self vw some optional parts second relation to same model self vw optionalparts addself airbag self sunroof expectedmessages append instance self vw action preadd reverse false model part objects self airbag self sunroof expectedmessages append instance self vw action postadd reverse false model part objects self airbag self sunroof self assertequalself m2mchangedmessages expectedmessages add airbag to all the cars even though the self vw already has one self airbag carsoptional addself vw self bmw self toyota expectedmessages append instance self airbag action preadd reverse true model car objects self bmw self toyota expectedmessages append instance self airbag action postadd reverse true model car objects self bmw self toyota self assertequalself m2mchangedmessages expectedmessages def testm2mrelationssignalsreverserelationwithcustomrelatednameself self initializesignalcar remove airbag from the self vw reverse relation with custom relatedname self airbag carsoptional removeself vw self assertequal self m2mchangedmessages instance self airbag action preremove reverse true model car objects self vw instance self airbag action postremove reverse true model car objects self vw def testm2mrelationssignalsclearallpartsoftheselfvwself self initializesignalcar clear all parts of the self vw self vw defaultparts clear self assertequal self m2mchangedmessages instance self vw action preclear reverse false model part instance self vw action postclear reverse false model part def testm2mrelationssignalsallthedoorsoffofcarsself self initializesignalcar take all the doors off of cars self doors carset clear self assertequal self m2mchangedmessages instance self doors action preclear reverse true model car instance self doors action postclear reverse true model car def testm2mrelationssignalsreverserelationself self initializesignalcar take all the airbags off of cars clear reverse relation with custom relatedname self airbag carsoptional clear self assertequal self m2mchangedmessages instance self airbag action preclear reverse true model car instance self airbag action postclear reverse true model car def testm2mrelationssignalsalternativewaysself expectedmessages self initializesignalcar alternative ways of setting relation self vw defaultparts createnamewindows p6 part objects getnamewindows expectedmessages append instance self vw action preadd reverse false model part objects p6 expectedmessages append instance self vw action postadd reverse false model part objects p6 self assertequalself m2mchangedmessages expectedmessages direct assignment clears the set first then adds self vw defaultparts setself wheelset self doors self engine expectedmessages append instance self vw action preremove reverse false model part objects p6 expectedmessages append instance self vw action postremove reverse false model part objects p6 expectedmessages append instance self vw action preadd reverse false model part objects self doors self engine self wheelset expectedmessages append instance self vw action postadd reverse false model part objects self doors self engine self wheelset self assertequalself m2mchangedmessages expectedmessages def testm2mrelationssignalsclearingremovingself expectedmessages self initializesignalcaradddefaultpartsbeforesetsignaltrue set by clearing self vw defaultparts setself wheelset self doors self engine cleartrue expectedmessages append instance self vw action preclear reverse false model part expectedmessages append instance self vw action postclear reverse false model part expectedmessages append instance self vw action preadd reverse false model part objects self doors self engine self wheelset expectedmessages append instance self vw action postadd reverse false model part objects self doors self engine self wheelset self assertequalself m2mchangedmessages expectedmessages set by only removing what s necessary self vw defaultparts setself wheelset self doors clearfalse expectedmessages append instance self vw action preremove reverse false model part objects self engine expectedmessages append instance self vw action postremove reverse false model part objects self engine self assertequalself m2mchangedmessages expectedmessages def testm2mrelationssignalswheninheritanceself expectedmessages self initializesignalcaradddefaultpartsbeforesetsignaltrue signals still work when model inheritance is involved c4 sportscar objects createnamebugatti price1000000 c4b car objects getnamebugatti c4 defaultparts setself doors expectedmessages append instance c4 action preadd reverse false model part objects self doors expectedmessages append instance c4 action postadd reverse false model part objects self doors self assertequalself m2mchangedmessages expectedmessages self engine carset addc4 expectedmessages append instance self engine action preadd reverse true model car objects c4b expectedmessages append instance self engine action postadd reverse true model car objects c4b self assertequalself m2mchangedmessages expectedmessages def initializesignalpersonself install a listener on the two m2m relations models signals m2mchanged connect self m2mchangedsignalreceiver person fans through models signals m2mchanged connect self m2mchangedsignalreceiver person friends through def testm2mrelationswithselfaddfriendsself self initializesignalperson self alice friends setself bob self chuck self assertequal self m2mchangedmessages instance self alice action preadd reverse false model person objects self bob self chuck instance self alice action postadd reverse false model person objects self bob self chuck def testm2mrelationswithselfaddfanself self initializesignalperson self alice fans setself daisy self assertequal self m2mchangedmessages instance self alice action preadd reverse false model person objects self daisy instance self alice action postadd reverse false model person objects self daisy def testm2mrelationswithselfaddidolsself self initializesignalperson self chuck idols setself alice self bob self assertequal self m2mchangedmessages instance self chuck action preadd reverse true model person objects self alice self bob instance self chuck action postadd reverse true model person objects self alice self bob disconnect all signal handlers install a listener on the two m2m relations adding a default part to our car no signal listener installed 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 give the bmw and toyota some doors as well remove the engine from the self vw and the airbag which is not set but is returned give the self vw some optional parts second relation to same model add airbag to all the cars even though the self vw already has one remove airbag from the self vw reverse relation with custom related_name clear all parts of the self vw take all the doors off of cars take all the airbags off of cars clear reverse relation with custom related_name alternative ways of setting relation direct assignment clears the set first then adds set by clearing set by only removing what s necessary signals still work when model inheritance is involved install a listener on the two 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):
        
        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):
        
        models.signals.m2m_changed.connect(
            self.m2m_changed_signal_receiver, Car.optional_parts.through
        )
        if add_default_parts_before_set_signal:
            
            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):
        
        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)

        
        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()
        
        
        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()

        
        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)

        
        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()
        
        
        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()
        
        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()
        
        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()
        
        
        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()

        
        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)

        
        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)

        
        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)

        
        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)

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