tests.py 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147
  1. import datetime
  2. from django.conf import settings
  3. from django.db import backend, connection, transaction, DEFAULT_DB_ALIAS
  4. from django.test import TestCase, TransactionTestCase, skipUnlessDBFeature
  5. from models import (Book, Award, AwardNote, Person, Child, Toy, PlayedWith,
  6. PlayedWithNote, Contact, Email, Researcher, Food, Eaten,
  7. Policy, Version, Location, Item)
  8. # Can't run this test under SQLite, because you can't
  9. # get two connections to an in-memory database.
  10. class DeleteLockingTest(TransactionTestCase):
  11. def setUp(self):
  12. # Create a second connection to the default database
  13. conn_settings = settings.DATABASES[DEFAULT_DB_ALIAS]
  14. self.conn2 = backend.DatabaseWrapper({
  15. 'HOST': conn_settings['HOST'],
  16. 'NAME': conn_settings['NAME'],
  17. 'OPTIONS': conn_settings['OPTIONS'],
  18. 'PASSWORD': conn_settings['PASSWORD'],
  19. 'PORT': conn_settings['PORT'],
  20. 'USER': conn_settings['USER'],
  21. 'TIME_ZONE': settings.TIME_ZONE,
  22. })
  23. # Put both DB connections into managed transaction mode
  24. transaction.enter_transaction_management()
  25. transaction.managed(True)
  26. self.conn2._enter_transaction_management(True)
  27. def tearDown(self):
  28. # Close down the second connection.
  29. transaction.leave_transaction_management()
  30. self.conn2.close()
  31. @skipUnlessDBFeature('test_db_allows_multiple_connections')
  32. def test_concurrent_delete(self):
  33. "Deletes on concurrent transactions don't collide and lock the database. Regression for #9479"
  34. # Create some dummy data
  35. b1 = Book(id=1, pagecount=100)
  36. b2 = Book(id=2, pagecount=200)
  37. b3 = Book(id=3, pagecount=300)
  38. b1.save()
  39. b2.save()
  40. b3.save()
  41. transaction.commit()
  42. self.assertEqual(3, Book.objects.count())
  43. # Delete something using connection 2.
  44. cursor2 = self.conn2.cursor()
  45. cursor2.execute('DELETE from delete_regress_book WHERE id=1')
  46. self.conn2._commit();
  47. # Now perform a queryset delete that covers the object
  48. # deleted in connection 2. This causes an infinite loop
  49. # under MySQL InnoDB unless we keep track of already
  50. # deleted objects.
  51. Book.objects.filter(pagecount__lt=250).delete()
  52. transaction.commit()
  53. self.assertEqual(1, Book.objects.count())
  54. transaction.commit()
  55. class DeleteCascadeTests(TestCase):
  56. def test_generic_relation_cascade(self):
  57. """
  58. Django cascades deletes through generic-related objects to their
  59. reverse relations.
  60. """
  61. person = Person.objects.create(name='Nelson Mandela')
  62. award = Award.objects.create(name='Nobel', content_object=person)
  63. note = AwardNote.objects.create(note='a peace prize',
  64. award=award)
  65. self.assertEqual(AwardNote.objects.count(), 1)
  66. person.delete()
  67. self.assertEqual(Award.objects.count(), 0)
  68. # first two asserts are just sanity checks, this is the kicker:
  69. self.assertEqual(AwardNote.objects.count(), 0)
  70. def test_fk_to_m2m_through(self):
  71. """
  72. If an M2M relationship has an explicitly-specified through model, and
  73. some other model has an FK to that through model, deletion is cascaded
  74. from one of the participants in the M2M, to the through model, to its
  75. related model.
  76. """
  77. juan = Child.objects.create(name='Juan')
  78. paints = Toy.objects.create(name='Paints')
  79. played = PlayedWith.objects.create(child=juan, toy=paints,
  80. date=datetime.date.today())
  81. note = PlayedWithNote.objects.create(played=played,
  82. note='the next Jackson Pollock')
  83. self.assertEqual(PlayedWithNote.objects.count(), 1)
  84. paints.delete()
  85. self.assertEqual(PlayedWith.objects.count(), 0)
  86. # first two asserts just sanity checks, this is the kicker:
  87. self.assertEqual(PlayedWithNote.objects.count(), 0)
  88. def test_15776(self):
  89. policy = Policy.objects.create(pk=1, policy_number="1234")
  90. version = Version.objects.create(policy=policy)
  91. location = Location.objects.create(version=version)
  92. item = Item.objects.create(version=version, location=location)
  93. policy.delete()
  94. class DeleteCascadeTransactionTests(TransactionTestCase):
  95. def test_inheritance(self):
  96. """
  97. Auto-created many-to-many through tables referencing a parent model are
  98. correctly found by the delete cascade when a child of that parent is
  99. deleted.
  100. Refs #14896.
  101. """
  102. r = Researcher.objects.create()
  103. email = Email.objects.create(
  104. label="office-email", email_address="carl@science.edu"
  105. )
  106. r.contacts.add(email)
  107. email.delete()
  108. def test_to_field(self):
  109. """
  110. Cascade deletion works with ForeignKey.to_field set to non-PK.
  111. """
  112. apple = Food.objects.create(name="apple")
  113. eaten = Eaten.objects.create(food=apple, meal="lunch")
  114. apple.delete()
  115. class LargeDeleteTests(TestCase):
  116. def test_large_deletes(self):
  117. "Regression for #13309 -- if the number of objects > chunk size, deletion still occurs"
  118. for x in range(300):
  119. track = Book.objects.create(pagecount=x+100)
  120. Book.objects.all().delete()
  121. self.assertEqual(Book.objects.count(), 0)