Fixed #13724: Corrected routing of write queries involving managers.

Previously, if a database request spanned a related object manager, the
first manager encountered would cause a request to the router, and this
would bind all subsequent queries to the same database returned by the
router. Unfortunately, the first router query would be performed using
a read request to the router, resulting in bad routing information being
used if the subsequent query was actually a write.

This change defers the call to the router until the final query is acutally
made.

It includes a small *BACKWARDS INCOMPATIBILITY* on an edge case - see the
release notes for details.

Thanks to Paul Collins (@paulcollinsiii) for the excellent debugging
work and patch.

1 files changed, 241 insertions, 0 deletions

diff --git a/tests/multiple_database/tests.py b/tests/multiple_database/tests.py
index 8a31f5f530..859330c40d 100644
--- a/tests/multiple_database/tests.py
+++ b/tests/multiple_database/tests.py
@@ -1902,3 +1902,244 @@ class MigrateTestCase(TestCase):
             router.routers = old_routers
 
         self.assertEqual(cts.count(), 0)
+
+
+
+
+class RouterUsed(Exception):
+    WRITE = 'write'
+
+    def __init__(self, mode, model, hints):
+        self.mode = mode
+        self.model = model
+        self.hints = hints
+
+
+class RouteForWriteTestCase(TestCase):
+    multi_db = True
+    RAISE_MSG = 'Db for write called'
+
+    class WriteCheckRouter(object):
+        def db_for_write(self, model, **hints):
+            raise RouterUsed(mode=RouterUsed.WRITE, model=model, hints=hints)
+
+    def setUp(self):
+        self._old_rtrs = router.routers
+
+    def tearDown(self):
+        router.routers = self._old_rtrs
+
+    def enable_router(self):
+        router.routers = [RouteForWriteTestCase.WriteCheckRouter()]
+
+    def test_fk_delete(self):
+        owner = Person.objects.create(name='Someone')
+        pet = Pet.objects.create(name='fido', owner=owner)
+        self.enable_router()
+        try:
+            pet.owner.delete()
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Person)
+            self.assertEqual(e.hints, {'instance': owner})
+
+    def test_reverse_fk_delete(self):
+        owner = Person.objects.create(name='Someone')
+        to_del_qs = owner.pet_set.all()
+        self.enable_router()
+        try:
+            to_del_qs.delete()
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Pet)
+            self.assertEqual(e.hints, {'instance': owner})
+
+    def test_reverse_fk_get_or_create(self):
+        owner = Person.objects.create(name='Someone')
+        self.enable_router()
+        try:
+            owner.pet_set.get_or_create(name='fido')
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Pet)
+            self.assertEqual(e.hints, {'instance': owner})
+
+    def test_reverse_fk_update(self):
+        owner = Person.objects.create(name='Someone')
+        pet = Pet.objects.create(name='fido', owner=owner)
+        self.enable_router()
+        try:
+            owner.pet_set.update(name='max')
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Pet)
+            self.assertEqual(e.hints, {'instance': owner})
+
+    def test_m2m_add(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        self.enable_router()
+        try:
+            book.authors.add(auth)
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Book.authors.through)
+            self.assertEqual(e.hints, {'instance': book})
+
+    def test_m2m_clear(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        book.authors.add(auth)
+        self.enable_router()
+        try:
+            book.authors.clear()
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Book.authors.through)
+            self.assertEqual(e.hints, {'instance': book})
+
+    def test_m2m_delete(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        book.authors.add(auth)
+        self.enable_router()
+        try:
+            book.authors.all().delete()
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Person)
+            self.assertEqual(e.hints, {'instance': book})
+
+    def test_m2m_get_or_create(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        self.enable_router()
+        try:
+            book.authors.get_or_create(name='Someone else')
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Book)
+            self.assertEqual(e.hints, {'instance': book})
+
+    def test_m2m_remove(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        book.authors.add(auth)
+        self.enable_router()
+        self.assertRaisesMessage(AttributeError, self.RAISE_MSG, )
+        try:
+            book.authors.remove(auth)
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Book.authors.through)
+            self.assertEqual(e.hints, {'instance': book})
+
+    def test_m2m_update(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        book.authors.add(auth)
+        self.enable_router()
+        try:
+            book.authors.all().update(name='Different')
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Person)
+            self.assertEqual(e.hints, {'instance': book})
+
+    def test_reverse_m2m_add(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        self.enable_router()
+        try:
+            auth.book_set.add(book)
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Book.authors.through)
+            self.assertEqual(e.hints, {'instance': auth})
+
+    def test_reverse_m2m_clear(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        book.authors.add(auth)
+        self.enable_router()
+        try:
+            auth.book_set.clear()
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Book.authors.through)
+            self.assertEqual(e.hints, {'instance': auth})
+
+    def test_reverse_m2m_delete(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        book.authors.add(auth)
+        self.enable_router()
+        try:
+            auth.book_set.all().delete()
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Book)
+            self.assertEqual(e.hints, {'instance': auth})
+
+    def test_reverse_m2m_get_or_create(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        self.enable_router()
+        try:
+            auth.book_set.get_or_create(title="New Book", published=datetime.datetime.now())
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Person)
+            self.assertEqual(e.hints, {'instance': auth})
+
+    def test_reverse_m2m_remove(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        book.authors.add(auth)
+        self.enable_router()
+        try:
+            auth.book_set.remove(book)
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Book.authors.through)
+            self.assertEqual(e.hints, {'instance': auth})
+
+    def test_reverse_m2m_update(self):
+        auth = Person.objects.create(name='Someone')
+        book = Book.objects.create(title="Pro Django",
+                                   published=datetime.date(2008, 12, 16))
+        book.authors.add(auth)
+        self.enable_router()
+        try:
+            auth.book_set.all().update(title='Different')
+            self.fail('db_for_write() not invoked on router')
+        except RouterUsed, e:
+            self.assertEqual(e.mode, RouterUsed.WRITE)
+            self.assertEqual(e.model, Book)
+            self.assertEqual(e.hints, {'instance': auth})