|
1 import datetime |
|
2 |
|
3 from django.conf import settings |
|
4 from django.db import backend, connection, transaction, DEFAULT_DB_ALIAS |
|
5 from django.test import TestCase, TransactionTestCase |
|
6 |
|
7 from models import Book, Award, AwardNote, Person, Child, Toy, PlayedWith, PlayedWithNote |
|
8 |
|
9 # Can't run this test under SQLite, because you can't |
|
10 # get two connections to an in-memory database. |
|
11 if settings.DATABASES[DEFAULT_DB_ALIAS]['ENGINE'] != 'django.db.backends.sqlite3': |
|
12 class DeleteLockingTest(TransactionTestCase): |
|
13 def setUp(self): |
|
14 # Create a second connection to the default database |
|
15 conn_settings = settings.DATABASES[DEFAULT_DB_ALIAS] |
|
16 self.conn2 = backend.DatabaseWrapper({ |
|
17 'HOST': conn_settings['HOST'], |
|
18 'NAME': conn_settings['NAME'], |
|
19 'OPTIONS': conn_settings['OPTIONS'], |
|
20 'PASSWORD': conn_settings['PASSWORD'], |
|
21 'PORT': conn_settings['PORT'], |
|
22 'USER': conn_settings['USER'], |
|
23 'TIME_ZONE': settings.TIME_ZONE, |
|
24 }) |
|
25 |
|
26 # Put both DB connections into managed transaction mode |
|
27 transaction.enter_transaction_management() |
|
28 transaction.managed(True) |
|
29 self.conn2._enter_transaction_management(True) |
|
30 |
|
31 def tearDown(self): |
|
32 # Close down the second connection. |
|
33 transaction.leave_transaction_management() |
|
34 self.conn2.close() |
|
35 |
|
36 def test_concurrent_delete(self): |
|
37 "Deletes on concurrent transactions don't collide and lock the database. Regression for #9479" |
|
38 |
|
39 # Create some dummy data |
|
40 b1 = Book(id=1, pagecount=100) |
|
41 b2 = Book(id=2, pagecount=200) |
|
42 b3 = Book(id=3, pagecount=300) |
|
43 b1.save() |
|
44 b2.save() |
|
45 b3.save() |
|
46 |
|
47 transaction.commit() |
|
48 |
|
49 self.assertEquals(3, Book.objects.count()) |
|
50 |
|
51 # Delete something using connection 2. |
|
52 cursor2 = self.conn2.cursor() |
|
53 cursor2.execute('DELETE from delete_regress_book WHERE id=1') |
|
54 self.conn2._commit(); |
|
55 |
|
56 # Now perform a queryset delete that covers the object |
|
57 # deleted in connection 2. This causes an infinite loop |
|
58 # under MySQL InnoDB unless we keep track of already |
|
59 # deleted objects. |
|
60 Book.objects.filter(pagecount__lt=250).delete() |
|
61 transaction.commit() |
|
62 self.assertEquals(1, Book.objects.count()) |
|
63 |
|
64 class DeleteCascadeTests(TestCase): |
|
65 def test_generic_relation_cascade(self): |
|
66 """ |
|
67 Test that Django cascades deletes through generic-related |
|
68 objects to their reverse relations. |
|
69 |
|
70 This might falsely succeed if the database cascades deletes |
|
71 itself immediately; the postgresql_psycopg2 backend does not |
|
72 give such a false success because ForeignKeys are created with |
|
73 DEFERRABLE INITIALLY DEFERRED, so its internal cascade is |
|
74 delayed until transaction commit. |
|
75 |
|
76 """ |
|
77 person = Person.objects.create(name='Nelson Mandela') |
|
78 award = Award.objects.create(name='Nobel', content_object=person) |
|
79 note = AwardNote.objects.create(note='a peace prize', |
|
80 award=award) |
|
81 self.assertEquals(AwardNote.objects.count(), 1) |
|
82 person.delete() |
|
83 self.assertEquals(Award.objects.count(), 0) |
|
84 # first two asserts are just sanity checks, this is the kicker: |
|
85 self.assertEquals(AwardNote.objects.count(), 0) |
|
86 |
|
87 def test_fk_to_m2m_through(self): |
|
88 """ |
|
89 Test that if a M2M relationship has an explicitly-specified |
|
90 through model, and some other model has an FK to that through |
|
91 model, deletion is cascaded from one of the participants in |
|
92 the M2M, to the through model, to its related model. |
|
93 |
|
94 Like the above test, this could in theory falsely succeed if |
|
95 the DB cascades deletes itself immediately. |
|
96 |
|
97 """ |
|
98 juan = Child.objects.create(name='Juan') |
|
99 paints = Toy.objects.create(name='Paints') |
|
100 played = PlayedWith.objects.create(child=juan, toy=paints, |
|
101 date=datetime.date.today()) |
|
102 note = PlayedWithNote.objects.create(played=played, |
|
103 note='the next Jackson Pollock') |
|
104 self.assertEquals(PlayedWithNote.objects.count(), 1) |
|
105 paints.delete() |
|
106 self.assertEquals(PlayedWith.objects.count(), 0) |
|
107 # first two asserts just sanity checks, this is the kicker: |
|
108 self.assertEquals(PlayedWithNote.objects.count(), 0) |
|
109 |
|
110 class LargeDeleteTests(TestCase): |
|
111 def test_large_deletes(self): |
|
112 "Regression for #13309 -- if the number of objects > chunk size, deletion still occurs" |
|
113 for x in range(300): |
|
114 track = Book.objects.create(pagecount=x+100) |
|
115 Book.objects.all().delete() |
|
116 self.assertEquals(Book.objects.count(), 0) |