Ticket #22014: prefetch_recursion_protection_fail.diff

tests/prefetch_related/models.py

@@ -51 +51 @@
 class Book(models.Model):
     title = models.CharField(max_length=255)
     authors = models.ManyToManyField(Author, related_name='books')
+    publishers = models.ManyToManyField('Publisher', related_name='books')
 
     def __str__(self):
         return self.title
@@ -58 +59 @@
     class Meta:
         ordering = ['id']
 
+class Publisher(models.Model):
+    authors = models.ManyToManyField('Author', related_name='publishers')
+
 
 class BookWithYear(Book):
     book = models.OneToOneField(Book, parent_link=True)

tests/prefetch_related/tests.py

@@ -1 +1 @@
 from __future__ import unicode_literals
+from collections import deque
 
 from django.contrib.contenttypes.models import ContentType
 from django.db import connection
@@ -9 +10 @@
 from .models import (Author, Book, Reader, Qualification, Teacher, Department,
     TaggedItem, Bookmark, AuthorAddress, FavoriteAuthors, AuthorWithAge,
     BookWithYear, BookReview, Person, House, Room, Employee, Comment,
-    LessonEntry, WordEntry, Author2)
+    LessonEntry, WordEntry, Author2, Publisher)
 
 
 class PrefetchRelatedTests(TestCase):
@@ -996 +997 @@
 
     def test_bug(self):
         list(Author2.objects.prefetch_related('first_book', 'favorite_books'))
+
+
+class RecursionProtectionTests(TestCase):
+    def setUp(self):
+        self.book1 = Book.objects.create()
+        self.book2 = Book.objects.create()
+        self.author1 = Author.objects.create(first_book=self.book1, name='Author #1')
+        self.author2 = Author.objects.create(first_book=self.book1, name='Author #2')
+        self.publisher = Publisher.objects.create()
+        self.author1.books.add(self.book1)
+        self.author1.save()
+        self.author2.books.add(self.book2)
+        self.author2.save()
+        self.book1.publishers.add(self.publisher)
+        self.book1.save()
+        self.publisher.authors.add(self.author1, self.author2)
+        self.publisher.save()
+
+    def walk(self, qs):
+        get_publishers = lambda b: b.publishers.all()
+        get_authors = lambda p: p.authors.all()
+        get_books = lambda a: a.books.all()
+
+        fetched_books = []
+        for e in qs:
+            for b in get_books(e):
+                for p in get_publishers(b):
+                    for a in get_authors(p):
+                        for fetched_book in get_books(a):
+                            fetched_books.append(fetched_book)
+        return fetched_books
+
+    @override_settings(DEBUG=True)
+    def test_descriptors_storing_protection(self):
+        """
+        What happens here:
+        ------------------
+        There is the check in `prefetch_related_objects`:
+        ::
+
+            if not (lookup in auto_lookups and descriptor in followed_descriptors):
+                done_queries[prefetch_to] = obj_list
+                auto_lookups.extend(normalize_prefetch_lookups(additional_lookups, prefetch_to))
+
+        if lookup is not autolookup and descriptor for it not been seen
+        lookup being added to `done_queries`.
+
+        This tests creates situation where:
+        lookup is autolookup and descriptor for it was added while
+        traversing same field but for different lookup.
+        So it never would be added to `done_lookups`.
+        """
+        qs1 = Author.objects.filter(pk=self.author1.pk).prefetch_related(Prefetch('books', Book.objects.prefetch_related('publishers__authors__books')), 'books__publishers__authors__books')
+        qs2 = Author.objects.filter(pk=self.author1.pk).prefetch_related(Prefetch('books', Book.objects.prefetch_related('publishers__authors__books')), 'books__publishers__authors__books', 'books__publishers__authors__books')
+
+        def check_num_queries(qs):
+            old_num_queries = len(connection.queries)
+            self.assertEqual(len(self.walk(qs)), 2)
+            return connection.queries[old_num_queries:]
+
+        queries_qs1 = check_num_queries(qs1)
+        queries_qs2 = check_num_queries(qs2)
+        self.assertEqual(len(queries_qs1), len(queries_qs2))
+
+    @override_settings(DEBUG=True)
+    def test_only_unique_queries(self):
+        qs = Author.objects.filter(pk=self.author1.pk).prefetch_related(Prefetch('books', Book.objects.prefetch_related('publishers__authors__books')), 'books__publishers__authors__books')
+        connection.queries = []
+        self.walk(qs)
+        queries = map(lambda e: e['sql'], connection.queries)
+        self.assertEqual(len(queries), len(set(queries)))