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)