1. import datetime
    
  2. 
    
  3. from django.db import connection, models, transaction
    
  4. from django.db.models import Exists, OuterRef
    
  5. from django.test import (
    
  6.     SimpleTestCase,
    
  7.     TestCase,
    
  8.     TransactionTestCase,
    
  9.     skipUnlessDBFeature,
    
  10. )
    
  11. 
    
  12. from .models import (
    
  13.     Award,
    
  14.     AwardNote,
    
  15.     Book,
    
  16.     Child,
    
  17.     Contact,
    
  18.     Eaten,
    
  19.     Email,
    
  20.     File,
    
  21.     Food,
    
  22.     FooFile,
    
  23.     FooFileProxy,
    
  24.     FooImage,
    
  25.     FooPhoto,
    
  26.     House,
    
  27.     Image,
    
  28.     Item,
    
  29.     Location,
    
  30.     Login,
    
  31.     OrderedPerson,
    
  32.     OrgUnit,
    
  33.     Person,
    
  34.     Photo,
    
  35.     PlayedWith,
    
  36.     PlayedWithNote,
    
  37.     Policy,
    
  38.     Researcher,
    
  39.     Toy,
    
  40.     Version,
    
  41. )
    
  42. 
    
  43. 
    
  44. # Can't run this test under SQLite, because you can't
    
  45. # get two connections to an in-memory database.
    
  46. @skipUnlessDBFeature("test_db_allows_multiple_connections")
    
  47. class DeleteLockingTest(TransactionTestCase):
    
  48.     available_apps = ["delete_regress"]
    
  49. 
    
  50.     def setUp(self):
    
  51.         # Create a second connection to the default database
    
  52.         self.conn2 = connection.copy()
    
  53.         self.conn2.set_autocommit(False)
    
  54. 
    
  55.     def tearDown(self):
    
  56.         # Close down the second connection.
    
  57.         self.conn2.rollback()
    
  58.         self.conn2.close()
    
  59. 
    
  60.     def test_concurrent_delete(self):
    
  61.         """Concurrent deletes don't collide and lock the database (#9479)."""
    
  62.         with transaction.atomic():
    
  63.             Book.objects.create(id=1, pagecount=100)
    
  64.             Book.objects.create(id=2, pagecount=200)
    
  65.             Book.objects.create(id=3, pagecount=300)
    
  66. 
    
  67.         with transaction.atomic():
    
  68.             # Start a transaction on the main connection.
    
  69.             self.assertEqual(3, Book.objects.count())
    
  70. 
    
  71.             # Delete something using another database connection.
    
  72.             with self.conn2.cursor() as cursor2:
    
  73.                 cursor2.execute("DELETE from delete_regress_book WHERE id = 1")
    
  74.             self.conn2.commit()
    
  75. 
    
  76.             # In the same transaction on the main connection, perform a
    
  77.             # queryset delete that covers the object deleted with the other
    
  78.             # connection. This causes an infinite loop under MySQL InnoDB
    
  79.             # unless we keep track of already deleted objects.
    
  80.             Book.objects.filter(pagecount__lt=250).delete()
    
  81. 
    
  82.         self.assertEqual(1, Book.objects.count())
    
  83. 
    
  84. 
    
  85. class DeleteCascadeTests(TestCase):
    
  86.     def test_generic_relation_cascade(self):
    
  87.         """
    
  88.         Django cascades deletes through generic-related objects to their
    
  89.         reverse relations.
    
  90.         """
    
  91.         person = Person.objects.create(name="Nelson Mandela")
    
  92.         award = Award.objects.create(name="Nobel", content_object=person)
    
  93.         AwardNote.objects.create(note="a peace prize", award=award)
    
  94.         self.assertEqual(AwardNote.objects.count(), 1)
    
  95.         person.delete()
    
  96.         self.assertEqual(Award.objects.count(), 0)
    
  97.         # first two asserts are just sanity checks, this is the kicker:
    
  98.         self.assertEqual(AwardNote.objects.count(), 0)
    
  99. 
    
  100.     def test_fk_to_m2m_through(self):
    
  101.         """
    
  102.         If an M2M relationship has an explicitly-specified through model, and
    
  103.         some other model has an FK to that through model, deletion is cascaded
    
  104.         from one of the participants in the M2M, to the through model, to its
    
  105.         related model.
    
  106.         """
    
  107.         juan = Child.objects.create(name="Juan")
    
  108.         paints = Toy.objects.create(name="Paints")
    
  109.         played = PlayedWith.objects.create(
    
  110.             child=juan, toy=paints, date=datetime.date.today()
    
  111.         )
    
  112.         PlayedWithNote.objects.create(played=played, note="the next Jackson Pollock")
    
  113.         self.assertEqual(PlayedWithNote.objects.count(), 1)
    
  114.         paints.delete()
    
  115.         self.assertEqual(PlayedWith.objects.count(), 0)
    
  116.         # first two asserts just sanity checks, this is the kicker:
    
  117.         self.assertEqual(PlayedWithNote.objects.count(), 0)
    
  118. 
    
  119.     def test_15776(self):
    
  120.         policy = Policy.objects.create(pk=1, policy_number="1234")
    
  121.         version = Version.objects.create(policy=policy)
    
  122.         location = Location.objects.create(version=version)
    
  123.         Item.objects.create(version=version, location=location)
    
  124.         policy.delete()
    
  125. 
    
  126. 
    
  127. class DeleteCascadeTransactionTests(TransactionTestCase):
    
  128.     available_apps = ["delete_regress"]
    
  129. 
    
  130.     def test_inheritance(self):
    
  131.         """
    
  132.         Auto-created many-to-many through tables referencing a parent model are
    
  133.         correctly found by the delete cascade when a child of that parent is
    
  134.         deleted.
    
  135. 
    
  136.         Refs #14896.
    
  137.         """
    
  138.         r = Researcher.objects.create()
    
  139.         email = Email.objects.create(
    
  140.             label="office-email", email_address="[email protected]"
    
  141.         )
    
  142.         r.contacts.add(email)
    
  143. 
    
  144.         email.delete()
    
  145. 
    
  146.     def test_to_field(self):
    
  147.         """
    
  148.         Cascade deletion works with ForeignKey.to_field set to non-PK.
    
  149.         """
    
  150.         apple = Food.objects.create(name="apple")
    
  151.         Eaten.objects.create(food=apple, meal="lunch")
    
  152. 
    
  153.         apple.delete()
    
  154.         self.assertFalse(Food.objects.exists())
    
  155.         self.assertFalse(Eaten.objects.exists())
    
  156. 
    
  157. 
    
  158. class LargeDeleteTests(TestCase):
    
  159.     def test_large_deletes(self):
    
  160.         """
    
  161.         If the number of objects > chunk size, deletion still occurs.
    
  162.         """
    
  163.         for x in range(300):
    
  164.             Book.objects.create(pagecount=x + 100)
    
  165.         # attach a signal to make sure we will not fast-delete
    
  166. 
    
  167.         def noop(*args, **kwargs):
    
  168.             pass
    
  169. 
    
  170.         models.signals.post_delete.connect(noop, sender=Book)
    
  171.         Book.objects.all().delete()
    
  172.         models.signals.post_delete.disconnect(noop, sender=Book)
    
  173.         self.assertEqual(Book.objects.count(), 0)
    
  174. 
    
  175. 
    
  176. class ProxyDeleteTest(TestCase):
    
  177.     """
    
  178.     Tests on_delete behavior for proxy models.
    
  179. 
    
  180.     See #16128.
    
  181.     """
    
  182. 
    
  183.     def create_image(self):
    
  184.         """Return an Image referenced by both a FooImage and a FooFile."""
    
  185.         # Create an Image
    
  186.         test_image = Image()
    
  187.         test_image.save()
    
  188.         foo_image = FooImage(my_image=test_image)
    
  189.         foo_image.save()
    
  190. 
    
  191.         # Get the Image instance as a File
    
  192.         test_file = File.objects.get(pk=test_image.pk)
    
  193.         foo_file = FooFile(my_file=test_file)
    
  194.         foo_file.save()
    
  195. 
    
  196.         return test_image
    
  197. 
    
  198.     def test_delete_proxy(self):
    
  199.         """
    
  200.         Deleting the *proxy* instance bubbles through to its non-proxy and
    
  201.         *all* referring objects are deleted.
    
  202.         """
    
  203.         self.create_image()
    
  204. 
    
  205.         Image.objects.all().delete()
    
  206. 
    
  207.         # An Image deletion == File deletion
    
  208.         self.assertEqual(len(Image.objects.all()), 0)
    
  209.         self.assertEqual(len(File.objects.all()), 0)
    
  210. 
    
  211.         # The Image deletion cascaded and *all* references to it are deleted.
    
  212.         self.assertEqual(len(FooImage.objects.all()), 0)
    
  213.         self.assertEqual(len(FooFile.objects.all()), 0)
    
  214. 
    
  215.     def test_delete_proxy_of_proxy(self):
    
  216.         """
    
  217.         Deleting a proxy-of-proxy instance should bubble through to its proxy
    
  218.         and non-proxy parents, deleting *all* referring objects.
    
  219.         """
    
  220.         test_image = self.create_image()
    
  221. 
    
  222.         # Get the Image as a Photo
    
  223.         test_photo = Photo.objects.get(pk=test_image.pk)
    
  224.         foo_photo = FooPhoto(my_photo=test_photo)
    
  225.         foo_photo.save()
    
  226. 
    
  227.         Photo.objects.all().delete()
    
  228. 
    
  229.         # A Photo deletion == Image deletion == File deletion
    
  230.         self.assertEqual(len(Photo.objects.all()), 0)
    
  231.         self.assertEqual(len(Image.objects.all()), 0)
    
  232.         self.assertEqual(len(File.objects.all()), 0)
    
  233. 
    
  234.         # The Photo deletion should have cascaded and deleted *all*
    
  235.         # references to it.
    
  236.         self.assertEqual(len(FooPhoto.objects.all()), 0)
    
  237.         self.assertEqual(len(FooFile.objects.all()), 0)
    
  238.         self.assertEqual(len(FooImage.objects.all()), 0)
    
  239. 
    
  240.     def test_delete_concrete_parent(self):
    
  241.         """
    
  242.         Deleting an instance of a concrete model should also delete objects
    
  243.         referencing its proxy subclass.
    
  244.         """
    
  245.         self.create_image()
    
  246. 
    
  247.         File.objects.all().delete()
    
  248. 
    
  249.         # A File deletion == Image deletion
    
  250.         self.assertEqual(len(File.objects.all()), 0)
    
  251.         self.assertEqual(len(Image.objects.all()), 0)
    
  252. 
    
  253.         # The File deletion should have cascaded and deleted *all* references
    
  254.         # to it.
    
  255.         self.assertEqual(len(FooFile.objects.all()), 0)
    
  256.         self.assertEqual(len(FooImage.objects.all()), 0)
    
  257. 
    
  258.     def test_delete_proxy_pair(self):
    
  259.         """
    
  260.         If a pair of proxy models are linked by an FK from one concrete parent
    
  261.         to the other, deleting one proxy model cascade-deletes the other, and
    
  262.         the deletion happens in the right order (not triggering an
    
  263.         IntegrityError on databases unable to defer integrity checks).
    
  264. 
    
  265.         Refs #17918.
    
  266.         """
    
  267.         # Create an Image (proxy of File) and FooFileProxy (proxy of FooFile,
    
  268.         # which has an FK to File)
    
  269.         image = Image.objects.create()
    
  270.         as_file = File.objects.get(pk=image.pk)
    
  271.         FooFileProxy.objects.create(my_file=as_file)
    
  272. 
    
  273.         Image.objects.all().delete()
    
  274. 
    
  275.         self.assertEqual(len(FooFileProxy.objects.all()), 0)
    
  276. 
    
  277.     def test_19187_values(self):
    
  278.         msg = "Cannot call delete() after .values() or .values_list()"
    
  279.         with self.assertRaisesMessage(TypeError, msg):
    
  280.             Image.objects.values().delete()
    
  281.         with self.assertRaisesMessage(TypeError, msg):
    
  282.             Image.objects.values_list().delete()
    
  283. 
    
  284. 
    
  285. class Ticket19102Tests(TestCase):
    
  286.     """
    
  287.     Test different queries which alter the SELECT clause of the query. We
    
  288.     also must be using a subquery for the deletion (that is, the original
    
  289.     query has a join in it). The deletion should be done as "fast-path"
    
  290.     deletion (that is, just one query for the .delete() call).
    
  291. 
    
  292.     Note that .values() is not tested here on purpose. .values().delete()
    
  293.     doesn't work for non fast-path deletes at all.
    
  294.     """
    
  295. 
    
  296.     @classmethod
    
  297.     def setUpTestData(cls):
    
  298.         cls.o1 = OrgUnit.objects.create(name="o1")
    
  299.         cls.o2 = OrgUnit.objects.create(name="o2")
    
  300.         cls.l1 = Login.objects.create(description="l1", orgunit=cls.o1)
    
  301.         cls.l2 = Login.objects.create(description="l2", orgunit=cls.o2)
    
  302. 
    
  303.     @skipUnlessDBFeature("update_can_self_select")
    
  304.     def test_ticket_19102_annotate(self):
    
  305.         with self.assertNumQueries(1):
    
  306.             Login.objects.order_by("description").filter(
    
  307.                 orgunit__name__isnull=False
    
  308.             ).annotate(n=models.Count("description")).filter(
    
  309.                 n=1, pk=self.l1.pk
    
  310.             ).delete()
    
  311.         self.assertFalse(Login.objects.filter(pk=self.l1.pk).exists())
    
  312.         self.assertTrue(Login.objects.filter(pk=self.l2.pk).exists())
    
  313. 
    
  314.     @skipUnlessDBFeature("update_can_self_select")
    
  315.     def test_ticket_19102_extra(self):
    
  316.         with self.assertNumQueries(1):
    
  317.             Login.objects.order_by("description").filter(
    
  318.                 orgunit__name__isnull=False
    
  319.             ).extra(select={"extraf": "1"}).filter(pk=self.l1.pk).delete()
    
  320.         self.assertFalse(Login.objects.filter(pk=self.l1.pk).exists())
    
  321.         self.assertTrue(Login.objects.filter(pk=self.l2.pk).exists())
    
  322. 
    
  323.     @skipUnlessDBFeature("update_can_self_select")
    
  324.     def test_ticket_19102_select_related(self):
    
  325.         with self.assertNumQueries(1):
    
  326.             Login.objects.filter(pk=self.l1.pk).filter(
    
  327.                 orgunit__name__isnull=False
    
  328.             ).order_by("description").select_related("orgunit").delete()
    
  329.         self.assertFalse(Login.objects.filter(pk=self.l1.pk).exists())
    
  330.         self.assertTrue(Login.objects.filter(pk=self.l2.pk).exists())
    
  331. 
    
  332.     @skipUnlessDBFeature("update_can_self_select")
    
  333.     def test_ticket_19102_defer(self):
    
  334.         with self.assertNumQueries(1):
    
  335.             Login.objects.filter(pk=self.l1.pk).filter(
    
  336.                 orgunit__name__isnull=False
    
  337.             ).order_by("description").only("id").delete()
    
  338.         self.assertFalse(Login.objects.filter(pk=self.l1.pk).exists())
    
  339.         self.assertTrue(Login.objects.filter(pk=self.l2.pk).exists())
    
  340. 
    
  341. 
    
  342. class DeleteTests(TestCase):
    
  343.     def test_meta_ordered_delete(self):
    
  344.         # When a subquery is performed by deletion code, the subquery must be
    
  345.         # cleared of all ordering. There was a but that caused _meta ordering
    
  346.         # to be used. Refs #19720.
    
  347.         h = House.objects.create(address="Foo")
    
  348.         OrderedPerson.objects.create(name="Jack", lives_in=h)
    
  349.         OrderedPerson.objects.create(name="Bob", lives_in=h)
    
  350.         OrderedPerson.objects.filter(lives_in__address="Foo").delete()
    
  351.         self.assertEqual(OrderedPerson.objects.count(), 0)
    
  352. 
    
  353.     def test_foreign_key_delete_nullifies_correct_columns(self):
    
  354.         """
    
  355.         With a model (Researcher) that has two foreign keys pointing to the
    
  356.         same model (Contact), deleting an instance of the target model
    
  357.         (contact1) nullifies the correct fields of Researcher.
    
  358.         """
    
  359.         contact1 = Contact.objects.create(label="Contact 1")
    
  360.         contact2 = Contact.objects.create(label="Contact 2")
    
  361.         researcher1 = Researcher.objects.create(
    
  362.             primary_contact=contact1,
    
  363.             secondary_contact=contact2,
    
  364.         )
    
  365.         researcher2 = Researcher.objects.create(
    
  366.             primary_contact=contact2,
    
  367.             secondary_contact=contact1,
    
  368.         )
    
  369.         contact1.delete()
    
  370.         researcher1.refresh_from_db()
    
  371.         researcher2.refresh_from_db()
    
  372.         self.assertIsNone(researcher1.primary_contact)
    
  373.         self.assertEqual(researcher1.secondary_contact, contact2)
    
  374.         self.assertEqual(researcher2.primary_contact, contact2)
    
  375.         self.assertIsNone(researcher2.secondary_contact)
    
  376. 
    
  377.     def test_self_reference_with_through_m2m_at_second_level(self):
    
  378.         toy = Toy.objects.create(name="Paints")
    
  379.         child = Child.objects.create(name="Juan")
    
  380.         Book.objects.create(pagecount=500, owner=child)
    
  381.         PlayedWith.objects.create(child=child, toy=toy, date=datetime.date.today())
    
  382.         Book.objects.filter(
    
  383.             Exists(
    
  384.                 Book.objects.filter(
    
  385.                     pk=OuterRef("pk"),
    
  386.                     owner__toys=toy.pk,
    
  387.                 ),
    
  388.             )
    
  389.         ).delete()
    
  390.         self.assertIs(Book.objects.exists(), False)
    
  391. 
    
  392. 
    
  393. class DeleteDistinct(SimpleTestCase):
    
  394.     def test_disallowed_delete_distinct(self):
    
  395.         msg = "Cannot call delete() after .distinct()."
    
  396.         with self.assertRaisesMessage(TypeError, msg):
    
  397.             Book.objects.distinct().delete()
    
  398.         with self.assertRaisesMessage(TypeError, msg):
    
  399.             Book.objects.distinct("id").delete()