1 import datetime

     2 import pickle

     3 import sys

     4 from StringIO import StringIO

     5 

     6 from django.conf import settings

     7 from django.contrib.auth.models import User

     8 from django.core import management

     9 from django.db import connections, router, DEFAULT_DB_ALIAS

    10 from django.db.utils import ConnectionRouter

    11 from django.test import TestCase

    12 

    13 from models import Book, Person, Pet, Review, UserProfile

    14 

    15 try:

    16     # we only have these models if the user is using multi-db, it's safe the

    17     # run the tests without them though.

    18     from models import Article, article_using

    19 except ImportError:

    20     pass

    21 

    22 class QueryTestCase(TestCase):

    23     multi_db = True

    24 

    25     def test_db_selection(self):

    26         "Check that querysets will use the default databse by default"

    27         self.assertEquals(Book.objects.db, DEFAULT_DB_ALIAS)

    28         self.assertEquals(Book.objects.all().db, DEFAULT_DB_ALIAS)

    29 

    30         self.assertEquals(Book.objects.using('other').db, 'other')

    31 

    32         self.assertEquals(Book.objects.db_manager('other').db, 'other')

    33         self.assertEquals(Book.objects.db_manager('other').all().db, 'other')

    34 

    35     def test_default_creation(self):

    36         "Objects created on the default database don't leak onto other databases"

    37         # Create a book on the default database using create()

    38         Book.objects.create(title="Pro Django",

    39                             published=datetime.date(2008, 12, 16))

    40 

    41         # Create a book on the default database using a save

    42         dive = Book()

    43         dive.title="Dive into Python"

    44         dive.published = datetime.date(2009, 5, 4)

    45         dive.save()

    46 

    47         # Check that book exists on the default database, but not on other database

    48         try:

    49             Book.objects.get(title="Pro Django")

    50             Book.objects.using('default').get(title="Pro Django")

    51         except Book.DoesNotExist:

    52             self.fail('"Dive Into Python" should exist on default database')

    53 

    54         self.assertRaises(Book.DoesNotExist,

    55             Book.objects.using('other').get,

    56             title="Pro Django"

    57         )

    58 

    59         try:

    60             Book.objects.get(title="Dive into Python")

    61             Book.objects.using('default').get(title="Dive into Python")

    62         except Book.DoesNotExist:

    63             self.fail('"Dive into Python" should exist on default database')

    64 

    65         self.assertRaises(Book.DoesNotExist,

    66             Book.objects.using('other').get,

    67             title="Dive into Python"

    68         )

    69 

    70 

    71     def test_other_creation(self):

    72         "Objects created on another database don't leak onto the default database"

    73         # Create a book on the second database

    74         Book.objects.using('other').create(title="Pro Django",

    75                                            published=datetime.date(2008, 12, 16))

    76 

    77         # Create a book on the default database using a save

    78         dive = Book()

    79         dive.title="Dive into Python"

    80         dive.published = datetime.date(2009, 5, 4)

    81         dive.save(using='other')

    82 

    83         # Check that book exists on the default database, but not on other database

    84         try:

    85             Book.objects.using('other').get(title="Pro Django")

    86         except Book.DoesNotExist:

    87             self.fail('"Dive Into Python" should exist on other database')

    88 

    89         self.assertRaises(Book.DoesNotExist,

    90             Book.objects.get,

    91             title="Pro Django"

    92         )

    93         self.assertRaises(Book.DoesNotExist,

    94             Book.objects.using('default').get,

    95             title="Pro Django"

    96         )

    97 

    98         try:

    99             Book.objects.using('other').get(title="Dive into Python")

   100         except Book.DoesNotExist:

   101             self.fail('"Dive into Python" should exist on other database')

   102 

   103         self.assertRaises(Book.DoesNotExist,

   104             Book.objects.get,

   105             title="Dive into Python"

   106         )

   107         self.assertRaises(Book.DoesNotExist,

   108             Book.objects.using('default').get,

   109             title="Dive into Python"

   110         )

   111 

   112     def test_basic_queries(self):

   113         "Queries are constrained to a single database"

   114         dive = Book.objects.using('other').create(title="Dive into Python",

   115                                                   published=datetime.date(2009, 5, 4))

   116 

   117         dive =  Book.objects.using('other').get(published=datetime.date(2009, 5, 4))

   118         self.assertEqual(dive.title, "Dive into Python")

   119         self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, published=datetime.date(2009, 5, 4))

   120 

   121         dive = Book.objects.using('other').get(title__icontains="dive")

   122         self.assertEqual(dive.title, "Dive into Python")

   123         self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, title__icontains="dive")

   124 

   125         dive = Book.objects.using('other').get(title__iexact="dive INTO python")

   126         self.assertEqual(dive.title, "Dive into Python")

   127         self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, title__iexact="dive INTO python")

   128 

   129         dive =  Book.objects.using('other').get(published__year=2009)

   130         self.assertEqual(dive.title, "Dive into Python")

   131         self.assertEqual(dive.published, datetime.date(2009, 5, 4))

   132         self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, published__year=2009)

   133 

   134         years = Book.objects.using('other').dates('published', 'year')

   135         self.assertEqual([o.year for o in years], [2009])

   136         years = Book.objects.using('default').dates('published', 'year')

   137         self.assertEqual([o.year for o in years], [])

   138 

   139         months = Book.objects.using('other').dates('published', 'month')

   140         self.assertEqual([o.month for o in months], [5])

   141         months = Book.objects.using('default').dates('published', 'month')

   142         self.assertEqual([o.month for o in months], [])

   143 

   144     def test_m2m_separation(self):

   145         "M2M fields are constrained to a single database"

   146         # Create a book and author on the default database

   147         pro = Book.objects.create(title="Pro Django",

   148                                   published=datetime.date(2008, 12, 16))

   149 

   150         marty = Person.objects.create(name="Marty Alchin")

   151 

   152         # Create a book and author on the other database

   153         dive = Book.objects.using('other').create(title="Dive into Python",

   154                                                   published=datetime.date(2009, 5, 4))

   155 

   156         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   157 

   158         # Save the author relations

   159         pro.authors = [marty]

   160         dive.authors = [mark]

   161 

   162         # Inspect the m2m tables directly.

   163         # There should be 1 entry in each database

   164         self.assertEquals(Book.authors.through.objects.using('default').count(), 1)

   165         self.assertEquals(Book.authors.through.objects.using('other').count(), 1)

   166 

   167         # Check that queries work across m2m joins

   168         self.assertEquals(list(Book.objects.using('default').filter(authors__name='Marty Alchin').values_list('title', flat=True)),

   169                           [u'Pro Django'])

   170         self.assertEquals(list(Book.objects.using('other').filter(authors__name='Marty Alchin').values_list('title', flat=True)),

   171                           [])

   172 

   173         self.assertEquals(list(Book.objects.using('default').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),

   174                           [])

   175         self.assertEquals(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),

   176                           [u'Dive into Python'])

   177 

   178         # Reget the objects to clear caches

   179         dive = Book.objects.using('other').get(title="Dive into Python")

   180         mark = Person.objects.using('other').get(name="Mark Pilgrim")

   181 

   182         # Retrive related object by descriptor. Related objects should be database-baound

   183         self.assertEquals(list(dive.authors.all().values_list('name', flat=True)),

   184                           [u'Mark Pilgrim'])

   185 

   186         self.assertEquals(list(mark.book_set.all().values_list('title', flat=True)),

   187                           [u'Dive into Python'])

   188 

   189     def test_m2m_forward_operations(self):

   190         "M2M forward manipulations are all constrained to a single DB"

   191         # Create a book and author on the other database

   192         dive = Book.objects.using('other').create(title="Dive into Python",

   193                                                   published=datetime.date(2009, 5, 4))

   194 

   195         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   196 

   197         # Save the author relations

   198         dive.authors = [mark]

   199 

   200         # Add a second author

   201         john = Person.objects.using('other').create(name="John Smith")

   202         self.assertEquals(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),

   203                           [])

   204 

   205 

   206         dive.authors.add(john)

   207         self.assertEquals(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),

   208                           [u'Dive into Python'])

   209         self.assertEquals(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),

   210                           [u'Dive into Python'])

   211 

   212         # Remove the second author

   213         dive.authors.remove(john)

   214         self.assertEquals(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),

   215                           [u'Dive into Python'])

   216         self.assertEquals(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),

   217                           [])

   218 

   219         # Clear all authors

   220         dive.authors.clear()

   221         self.assertEquals(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),

   222                           [])

   223         self.assertEquals(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),

   224                           [])

   225 

   226         # Create an author through the m2m interface

   227         dive.authors.create(name='Jane Brown')

   228         self.assertEquals(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),

   229                           [])

   230         self.assertEquals(list(Book.objects.using('other').filter(authors__name='Jane Brown').values_list('title', flat=True)),

   231                           [u'Dive into Python'])

   232 

   233     def test_m2m_reverse_operations(self):

   234         "M2M reverse manipulations are all constrained to a single DB"

   235         # Create a book and author on the other database

   236         dive = Book.objects.using('other').create(title="Dive into Python",

   237                                                   published=datetime.date(2009, 5, 4))

   238 

   239         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   240 

   241         # Save the author relations

   242         dive.authors = [mark]

   243 

   244         # Create a second book on the other database

   245         grease = Book.objects.using('other').create(title="Greasemonkey Hacks",

   246                                                     published=datetime.date(2005, 11, 1))

   247 

   248         # Add a books to the m2m

   249         mark.book_set.add(grease)

   250         self.assertEquals(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),

   251                           [u'Mark Pilgrim'])

   252         self.assertEquals(list(Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True)),

   253                           [u'Mark Pilgrim'])

   254 

   255         # Remove a book from the m2m

   256         mark.book_set.remove(grease)

   257         self.assertEquals(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),

   258                           [u'Mark Pilgrim'])

   259         self.assertEquals(list(Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True)),

   260                           [])

   261 

   262         # Clear the books associated with mark

   263         mark.book_set.clear()

   264         self.assertEquals(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),

   265                           [])

   266         self.assertEquals(list(Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True)),

   267                           [])

   268 

   269         # Create a book through the m2m interface

   270         mark.book_set.create(title="Dive into HTML5", published=datetime.date(2020, 1, 1))

   271         self.assertEquals(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),

   272                           [])

   273         self.assertEquals(list(Person.objects.using('other').filter(book__title='Dive into HTML5').values_list('name', flat=True)),

   274                           [u'Mark Pilgrim'])

   275 

   276     def test_m2m_cross_database_protection(self):

   277         "Operations that involve sharing M2M objects across databases raise an error"

   278         # Create a book and author on the default database

   279         pro = Book.objects.create(title="Pro Django",

   280                                   published=datetime.date(2008, 12, 16))

   281 

   282         marty = Person.objects.create(name="Marty Alchin")

   283 

   284         # Create a book and author on the other database

   285         dive = Book.objects.using('other').create(title="Dive into Python",

   286                                                   published=datetime.date(2009, 5, 4))

   287 

   288         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   289         # Set a foreign key set with an object from a different database

   290         try:

   291             marty.book_set = [pro, dive]

   292             self.fail("Shouldn't be able to assign across databases")

   293         except ValueError:

   294             pass

   295 

   296         # Add to an m2m with an object from a different database

   297         try:

   298             marty.book_set.add(dive)

   299             self.fail("Shouldn't be able to assign across databases")

   300         except ValueError:

   301             pass

   302 

   303         # Set a m2m with an object from a different database

   304         try:

   305             marty.book_set = [pro, dive]

   306             self.fail("Shouldn't be able to assign across databases")

   307         except ValueError:

   308             pass

   309 

   310         # Add to a reverse m2m with an object from a different database

   311         try:

   312             dive.authors.add(marty)

   313             self.fail("Shouldn't be able to assign across databases")

   314         except ValueError:

   315             pass

   316 

   317         # Set a reverse m2m with an object from a different database

   318         try:

   319             dive.authors = [mark, marty]

   320             self.fail("Shouldn't be able to assign across databases")

   321         except ValueError:

   322             pass

   323 

   324     def test_m2m_deletion(self):

   325         "Cascaded deletions of m2m relations issue queries on the right database"

   326         # Create a book and author on the other database

   327         dive = Book.objects.using('other').create(title="Dive into Python",

   328                                                   published=datetime.date(2009, 5, 4))

   329 

   330         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   331         dive.authors = [mark]

   332 

   333         # Check the initial state

   334         self.assertEquals(Person.objects.using('default').count(), 0)

   335         self.assertEquals(Book.objects.using('default').count(), 0)

   336         self.assertEquals(Book.authors.through.objects.using('default').count(), 0)

   337 

   338         self.assertEquals(Person.objects.using('other').count(), 1)

   339         self.assertEquals(Book.objects.using('other').count(), 1)

   340         self.assertEquals(Book.authors.through.objects.using('other').count(), 1)

   341 

   342         # Delete the object on the other database

   343         dive.delete(using='other')

   344 

   345         self.assertEquals(Person.objects.using('default').count(), 0)

   346         self.assertEquals(Book.objects.using('default').count(), 0)

   347         self.assertEquals(Book.authors.through.objects.using('default').count(), 0)

   348 

   349         # The person still exists ...

   350         self.assertEquals(Person.objects.using('other').count(), 1)

   351         # ... but the book has been deleted

   352         self.assertEquals(Book.objects.using('other').count(), 0)

   353         # ... and the relationship object has also been deleted.

   354         self.assertEquals(Book.authors.through.objects.using('other').count(), 0)

   355 

   356         # Now try deletion in the reverse direction. Set up the relation again

   357         dive = Book.objects.using('other').create(title="Dive into Python",

   358                                                   published=datetime.date(2009, 5, 4))

   359         dive.authors = [mark]

   360 

   361         # Check the initial state

   362         self.assertEquals(Person.objects.using('default').count(), 0)

   363         self.assertEquals(Book.objects.using('default').count(), 0)

   364         self.assertEquals(Book.authors.through.objects.using('default').count(), 0)

   365 

   366         self.assertEquals(Person.objects.using('other').count(), 1)

   367         self.assertEquals(Book.objects.using('other').count(), 1)

   368         self.assertEquals(Book.authors.through.objects.using('other').count(), 1)

   369 

   370         # Delete the object on the other database

   371         mark.delete(using='other')

   372 

   373         self.assertEquals(Person.objects.using('default').count(), 0)

   374         self.assertEquals(Book.objects.using('default').count(), 0)

   375         self.assertEquals(Book.authors.through.objects.using('default').count(), 0)

   376 

   377         # The person has been deleted ...

   378         self.assertEquals(Person.objects.using('other').count(), 0)

   379         # ... but the book still exists

   380         self.assertEquals(Book.objects.using('other').count(), 1)

   381         # ... and the relationship object has been deleted.

   382         self.assertEquals(Book.authors.through.objects.using('other').count(), 0)

   383 

   384     def test_foreign_key_separation(self):

   385         "FK fields are constrained to a single database"

   386         # Create a book and author on the default database

   387         pro = Book.objects.create(title="Pro Django",

   388                                   published=datetime.date(2008, 12, 16))

   389 

   390         marty = Person.objects.create(name="Marty Alchin")

   391         george = Person.objects.create(name="George Vilches")

   392 

   393         # Create a book and author on the other database

   394         dive = Book.objects.using('other').create(title="Dive into Python",

   395                                                   published=datetime.date(2009, 5, 4))

   396 

   397         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   398         chris = Person.objects.using('other').create(name="Chris Mills")

   399 

   400         # Save the author's favourite books

   401         pro.editor = george

   402         pro.save()

   403 

   404         dive.editor = chris

   405         dive.save()

   406 

   407         pro = Book.objects.using('default').get(title="Pro Django")

   408         self.assertEquals(pro.editor.name, "George Vilches")

   409 

   410         dive = Book.objects.using('other').get(title="Dive into Python")

   411         self.assertEquals(dive.editor.name, "Chris Mills")

   412 

   413         # Check that queries work across foreign key joins

   414         self.assertEquals(list(Person.objects.using('default').filter(edited__title='Pro Django').values_list('name', flat=True)),

   415                           [u'George Vilches'])

   416         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Pro Django').values_list('name', flat=True)),

   417                           [])

   418 

   419         self.assertEquals(list(Person.objects.using('default').filter(edited__title='Dive into Python').values_list('name', flat=True)),

   420                           [])

   421         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),

   422                           [u'Chris Mills'])

   423 

   424         # Reget the objects to clear caches

   425         chris = Person.objects.using('other').get(name="Chris Mills")

   426         dive = Book.objects.using('other').get(title="Dive into Python")

   427 

   428         # Retrive related object by descriptor. Related objects should be database-baound

   429         self.assertEquals(list(chris.edited.values_list('title', flat=True)),

   430                           [u'Dive into Python'])

   431 

   432     def test_foreign_key_reverse_operations(self):

   433         "FK reverse manipulations are all constrained to a single DB"

   434         dive = Book.objects.using('other').create(title="Dive into Python",

   435                                                        published=datetime.date(2009, 5, 4))

   436 

   437         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   438         chris = Person.objects.using('other').create(name="Chris Mills")

   439 

   440         # Save the author relations

   441         dive.editor = chris

   442         dive.save()

   443 

   444         # Add a second book edited by chris

   445         html5 = Book.objects.using('other').create(title="Dive into HTML5", published=datetime.date(2010, 3, 15))

   446         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),

   447                           [])

   448 

   449         chris.edited.add(html5)

   450         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),

   451                           [u'Chris Mills'])

   452         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),

   453                           [u'Chris Mills'])

   454 

   455         # Remove the second editor

   456         chris.edited.remove(html5)

   457         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),

   458                           [])

   459         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),

   460                           [u'Chris Mills'])

   461 

   462         # Clear all edited books

   463         chris.edited.clear()

   464         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),

   465                           [])

   466         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),

   467                           [])

   468 

   469         # Create an author through the m2m interface

   470         chris.edited.create(title='Dive into Water', published=datetime.date(2010, 3, 15))

   471         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),

   472                           [])

   473         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into Water').values_list('name', flat=True)),

   474                           [u'Chris Mills'])

   475         self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),

   476                           [])

   477 

   478     def test_foreign_key_cross_database_protection(self):

   479         "Operations that involve sharing FK objects across databases raise an error"

   480         # Create a book and author on the default database

   481         pro = Book.objects.create(title="Pro Django",

   482                                   published=datetime.date(2008, 12, 16))

   483 

   484         marty = Person.objects.create(name="Marty Alchin")

   485 

   486         # Create a book and author on the other database

   487         dive = Book.objects.using('other').create(title="Dive into Python",

   488                                                   published=datetime.date(2009, 5, 4))

   489 

   490         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   491 

   492         # Set a foreign key with an object from a different database

   493         try:

   494             dive.editor = marty

   495             self.fail("Shouldn't be able to assign across databases")

   496         except ValueError:

   497             pass

   498 

   499         # Set a foreign key set with an object from a different database

   500         try:

   501             marty.edited = [pro, dive]

   502             self.fail("Shouldn't be able to assign across databases")

   503         except ValueError:

   504             pass

   505 

   506         # Add to a foreign key set with an object from a different database

   507         try:

   508             marty.edited.add(dive)

   509             self.fail("Shouldn't be able to assign across databases")

   510         except ValueError:

   511             pass

   512 

   513         # BUT! if you assign a FK object when the base object hasn't

   514         # been saved yet, you implicitly assign the database for the

   515         # base object.

   516         chris = Person(name="Chris Mills")

   517         html5 = Book(title="Dive into HTML5", published=datetime.date(2010, 3, 15))

   518         # initially, no db assigned

   519         self.assertEquals(chris._state.db, None)

   520         self.assertEquals(html5._state.db, None)

   521 

   522         # old object comes from 'other', so the new object is set to use 'other'...

   523         dive.editor = chris

   524         html5.editor = mark

   525         self.assertEquals(chris._state.db, 'other')

   526         self.assertEquals(html5._state.db, 'other')

   527         # ... but it isn't saved yet

   528         self.assertEquals(list(Person.objects.using('other').values_list('name',flat=True)),

   529                           [u'Mark Pilgrim'])

   530         self.assertEquals(list(Book.objects.using('other').values_list('title',flat=True)),

   531                            [u'Dive into Python'])

   532 

   533         # When saved (no using required), new objects goes to 'other'

   534         chris.save()

   535         html5.save()

   536         self.assertEquals(list(Person.objects.using('default').values_list('name',flat=True)),

   537                           [u'Marty Alchin'])

   538         self.assertEquals(list(Person.objects.using('other').values_list('name',flat=True)),

   539                           [u'Chris Mills', u'Mark Pilgrim'])

   540         self.assertEquals(list(Book.objects.using('default').values_list('title',flat=True)),

   541                           [u'Pro Django'])

   542         self.assertEquals(list(Book.objects.using('other').values_list('title',flat=True)),

   543                           [u'Dive into HTML5', u'Dive into Python'])

   544 

   545         # This also works if you assign the FK in the constructor

   546         water = Book(title="Dive into Water", published=datetime.date(2001, 1, 1), editor=mark)

   547         self.assertEquals(water._state.db, 'other')

   548         # ... but it isn't saved yet

   549         self.assertEquals(list(Book.objects.using('default').values_list('title',flat=True)),

   550                           [u'Pro Django'])

   551         self.assertEquals(list(Book.objects.using('other').values_list('title',flat=True)),

   552                           [u'Dive into HTML5', u'Dive into Python'])

   553 

   554         # When saved, the new book goes to 'other'

   555         water.save()

   556         self.assertEquals(list(Book.objects.using('default').values_list('title',flat=True)),

   557                           [u'Pro Django'])

   558         self.assertEquals(list(Book.objects.using('other').values_list('title',flat=True)),

   559                           [u'Dive into HTML5', u'Dive into Python', u'Dive into Water'])

   560 

   561     def test_foreign_key_deletion(self):

   562         "Cascaded deletions of Foreign Key relations issue queries on the right database"

   563         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   564         fido = Pet.objects.using('other').create(name="Fido", owner=mark)

   565 

   566         # Check the initial state

   567         self.assertEquals(Person.objects.using('default').count(), 0)

   568         self.assertEquals(Pet.objects.using('default').count(), 0)

   569 

   570         self.assertEquals(Person.objects.using('other').count(), 1)

   571         self.assertEquals(Pet.objects.using('other').count(), 1)

   572 

   573         # Delete the person object, which will cascade onto the pet

   574         mark.delete(using='other')

   575 

   576         self.assertEquals(Person.objects.using('default').count(), 0)

   577         self.assertEquals(Pet.objects.using('default').count(), 0)

   578 

   579         # Both the pet and the person have been deleted from the right database

   580         self.assertEquals(Person.objects.using('other').count(), 0)

   581         self.assertEquals(Pet.objects.using('other').count(), 0)

   582 

   583     def test_foreign_key_validation(self):

   584         "ForeignKey.validate() uses the correct database"

   585         mickey = Person.objects.using('other').create(name="Mickey")

   586         pluto = Pet.objects.using('other').create(name="Pluto", owner=mickey)

   587         self.assertEquals(None, pluto.full_clean())

   588 

   589     def test_o2o_separation(self):

   590         "OneToOne fields are constrained to a single database"

   591         # Create a user and profile on the default database

   592         alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com')

   593         alice_profile = UserProfile.objects.using('default').create(user=alice, flavor='chocolate')

   594 

   595         # Create a user and profile on the other database

   596         bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com')

   597         bob_profile = UserProfile.objects.using('other').create(user=bob, flavor='crunchy frog')

   598 

   599         # Retrieve related objects; queries should be database constrained

   600         alice = User.objects.using('default').get(username="alice")

   601         self.assertEquals(alice.userprofile.flavor, "chocolate")

   602 

   603         bob = User.objects.using('other').get(username="bob")

   604         self.assertEquals(bob.userprofile.flavor, "crunchy frog")

   605 

   606         # Check that queries work across joins

   607         self.assertEquals(list(User.objects.using('default').filter(userprofile__flavor='chocolate').values_list('username', flat=True)),

   608                           [u'alice'])

   609         self.assertEquals(list(User.objects.using('other').filter(userprofile__flavor='chocolate').values_list('username', flat=True)),

   610                           [])

   611 

   612         self.assertEquals(list(User.objects.using('default').filter(userprofile__flavor='crunchy frog').values_list('username', flat=True)),

   613                           [])

   614         self.assertEquals(list(User.objects.using('other').filter(userprofile__flavor='crunchy frog').values_list('username', flat=True)),

   615                           [u'bob'])

   616 

   617         # Reget the objects to clear caches

   618         alice_profile = UserProfile.objects.using('default').get(flavor='chocolate')

   619         bob_profile = UserProfile.objects.using('other').get(flavor='crunchy frog')

   620 

   621         # Retrive related object by descriptor. Related objects should be database-baound

   622         self.assertEquals(alice_profile.user.username, 'alice')

   623         self.assertEquals(bob_profile.user.username, 'bob')

   624 

   625     def test_o2o_cross_database_protection(self):

   626         "Operations that involve sharing FK objects across databases raise an error"

   627         # Create a user and profile on the default database

   628         alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com')

   629 

   630         # Create a user and profile on the other database

   631         bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com')

   632 

   633         # Set a one-to-one relation with an object from a different database

   634         alice_profile = UserProfile.objects.using('default').create(user=alice, flavor='chocolate')

   635         try:

   636             bob.userprofile = alice_profile

   637             self.fail("Shouldn't be able to assign across databases")

   638         except ValueError:

   639             pass

   640 

   641         # BUT! if you assign a FK object when the base object hasn't

   642         # been saved yet, you implicitly assign the database for the

   643         # base object.

   644         bob_profile = UserProfile.objects.using('other').create(user=bob, flavor='crunchy frog')

   645 

   646         new_bob_profile = UserProfile(flavor="spring surprise")

   647 

   648         charlie = User(username='charlie',email='charlie@example.com')

   649         charlie.set_unusable_password()

   650 

   651         # initially, no db assigned

   652         self.assertEquals(new_bob_profile._state.db, None)

   653         self.assertEquals(charlie._state.db, None)

   654 

   655         # old object comes from 'other', so the new object is set to use 'other'...

   656         new_bob_profile.user = bob

   657         charlie.userprofile = bob_profile

   658         self.assertEquals(new_bob_profile._state.db, 'other')

   659         self.assertEquals(charlie._state.db, 'other')

   660 

   661         # ... but it isn't saved yet

   662         self.assertEquals(list(User.objects.using('other').values_list('username',flat=True)),

   663                           [u'bob'])

   664         self.assertEquals(list(UserProfile.objects.using('other').values_list('flavor',flat=True)),

   665                            [u'crunchy frog'])

   666 

   667         # When saved (no using required), new objects goes to 'other'

   668         charlie.save()

   669         bob_profile.save()

   670         new_bob_profile.save()

   671         self.assertEquals(list(User.objects.using('default').values_list('username',flat=True)),

   672                           [u'alice'])

   673         self.assertEquals(list(User.objects.using('other').values_list('username',flat=True)),

   674                           [u'bob', u'charlie'])

   675         self.assertEquals(list(UserProfile.objects.using('default').values_list('flavor',flat=True)),

   676                            [u'chocolate'])

   677         self.assertEquals(list(UserProfile.objects.using('other').values_list('flavor',flat=True)),

   678                            [u'crunchy frog', u'spring surprise'])

   679 

   680         # This also works if you assign the O2O relation in the constructor

   681         denise = User.objects.db_manager('other').create_user('denise','denise@example.com')

   682         denise_profile = UserProfile(flavor="tofu", user=denise)

   683 

   684         self.assertEquals(denise_profile._state.db, 'other')

   685         # ... but it isn't saved yet

   686         self.assertEquals(list(UserProfile.objects.using('default').values_list('flavor',flat=True)),

   687                            [u'chocolate'])

   688         self.assertEquals(list(UserProfile.objects.using('other').values_list('flavor',flat=True)),

   689                            [u'crunchy frog', u'spring surprise'])

   690 

   691         # When saved, the new profile goes to 'other'

   692         denise_profile.save()

   693         self.assertEquals(list(UserProfile.objects.using('default').values_list('flavor',flat=True)),

   694                            [u'chocolate'])

   695         self.assertEquals(list(UserProfile.objects.using('other').values_list('flavor',flat=True)),

   696                            [u'crunchy frog', u'spring surprise', u'tofu'])

   697 

   698     def test_generic_key_separation(self):

   699         "Generic fields are constrained to a single database"

   700         # Create a book and author on the default database

   701         pro = Book.objects.create(title="Pro Django",

   702                                   published=datetime.date(2008, 12, 16))

   703 

   704         review1 = Review.objects.create(source="Python Monthly", content_object=pro)

   705 

   706         # Create a book and author on the other database

   707         dive = Book.objects.using('other').create(title="Dive into Python",

   708                                                   published=datetime.date(2009, 5, 4))

   709 

   710         review2 = Review.objects.using('other').create(source="Python Weekly", content_object=dive)

   711 

   712         review1 = Review.objects.using('default').get(source="Python Monthly")

   713         self.assertEquals(review1.content_object.title, "Pro Django")

   714 

   715         review2 = Review.objects.using('other').get(source="Python Weekly")

   716         self.assertEquals(review2.content_object.title, "Dive into Python")

   717 

   718         # Reget the objects to clear caches

   719         dive = Book.objects.using('other').get(title="Dive into Python")

   720 

   721         # Retrive related object by descriptor. Related objects should be database-bound

   722         self.assertEquals(list(dive.reviews.all().values_list('source', flat=True)),

   723                           [u'Python Weekly'])

   724 

   725     def test_generic_key_reverse_operations(self):

   726         "Generic reverse manipulations are all constrained to a single DB"

   727         dive = Book.objects.using('other').create(title="Dive into Python",

   728                                                   published=datetime.date(2009, 5, 4))

   729 

   730         temp = Book.objects.using('other').create(title="Temp",

   731                                                   published=datetime.date(2009, 5, 4))

   732 

   733         review1 = Review.objects.using('other').create(source="Python Weekly", content_object=dive)

   734         review2 = Review.objects.using('other').create(source="Python Monthly", content_object=temp)

   735 

   736         self.assertEquals(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),

   737                           [])

   738         self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),

   739                           [u'Python Weekly'])

   740 

   741         # Add a second review

   742         dive.reviews.add(review2)

   743         self.assertEquals(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),

   744                           [])

   745         self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),

   746                           [u'Python Monthly', u'Python Weekly'])

   747 

   748         # Remove the second author

   749         dive.reviews.remove(review1)

   750         self.assertEquals(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),

   751                           [])

   752         self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),

   753                           [u'Python Monthly'])

   754 

   755         # Clear all reviews

   756         dive.reviews.clear()

   757         self.assertEquals(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),

   758                           [])

   759         self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),

   760                           [])

   761 

   762         # Create an author through the generic interface

   763         dive.reviews.create(source='Python Daily')

   764         self.assertEquals(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),

   765                           [])

   766         self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),

   767                           [u'Python Daily'])

   768 

   769     def test_generic_key_cross_database_protection(self):

   770         "Operations that involve sharing generic key objects across databases raise an error"

   771         # Create a book and author on the default database

   772         pro = Book.objects.create(title="Pro Django",

   773                                   published=datetime.date(2008, 12, 16))

   774 

   775         review1 = Review.objects.create(source="Python Monthly", content_object=pro)

   776 

   777         # Create a book and author on the other database

   778         dive = Book.objects.using('other').create(title="Dive into Python",

   779                                                   published=datetime.date(2009, 5, 4))

   780 

   781         review2 = Review.objects.using('other').create(source="Python Weekly", content_object=dive)

   782 

   783         # Set a foreign key with an object from a different database

   784         try:

   785             review1.content_object = dive

   786             self.fail("Shouldn't be able to assign across databases")

   787         except ValueError:

   788             pass

   789 

   790         # Add to a foreign key set with an object from a different database

   791         try:

   792             dive.reviews.add(review1)

   793             self.fail("Shouldn't be able to assign across databases")

   794         except ValueError:

   795             pass

   796 

   797         # BUT! if you assign a FK object when the base object hasn't

   798         # been saved yet, you implicitly assign the database for the

   799         # base object.

   800         review3 = Review(source="Python Daily")

   801         # initially, no db assigned

   802         self.assertEquals(review3._state.db, None)

   803 

   804         # Dive comes from 'other', so review3 is set to use 'other'...

   805         review3.content_object = dive

   806         self.assertEquals(review3._state.db, 'other')

   807         # ... but it isn't saved yet

   808         self.assertEquals(list(Review.objects.using('default').filter(object_id=pro.pk).values_list('source', flat=True)),

   809                           [u'Python Monthly'])

   810         self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source',flat=True)),

   811                           [u'Python Weekly'])

   812 

   813         # When saved, John goes to 'other'

   814         review3.save()

   815         self.assertEquals(list(Review.objects.using('default').filter(object_id=pro.pk).values_list('source', flat=True)),

   816                           [u'Python Monthly'])

   817         self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source',flat=True)),

   818                           [u'Python Daily', u'Python Weekly'])

   819 

   820     def test_generic_key_deletion(self):

   821         "Cascaded deletions of Generic Key relations issue queries on the right database"

   822         dive = Book.objects.using('other').create(title="Dive into Python",

   823                                                   published=datetime.date(2009, 5, 4))

   824         review = Review.objects.using('other').create(source="Python Weekly", content_object=dive)

   825 

   826         # Check the initial state

   827         self.assertEquals(Book.objects.using('default').count(), 0)

   828         self.assertEquals(Review.objects.using('default').count(), 0)

   829 

   830         self.assertEquals(Book.objects.using('other').count(), 1)

   831         self.assertEquals(Review.objects.using('other').count(), 1)

   832 

   833         # Delete the Book object, which will cascade onto the pet

   834         dive.delete(using='other')

   835 

   836         self.assertEquals(Book.objects.using('default').count(), 0)

   837         self.assertEquals(Review.objects.using('default').count(), 0)

   838 

   839         # Both the pet and the person have been deleted from the right database

   840         self.assertEquals(Book.objects.using('other').count(), 0)

   841         self.assertEquals(Review.objects.using('other').count(), 0)

   842 

   843     def test_ordering(self):

   844         "get_next_by_XXX commands stick to a single database"

   845         pro = Book.objects.create(title="Pro Django",

   846                                   published=datetime.date(2008, 12, 16))

   847 

   848         dive = Book.objects.using('other').create(title="Dive into Python",

   849                                                   published=datetime.date(2009, 5, 4))

   850 

   851         learn = Book.objects.using('other').create(title="Learning Python",

   852                                                    published=datetime.date(2008, 7, 16))

   853 

   854         self.assertEquals(learn.get_next_by_published().title, "Dive into Python")

   855         self.assertEquals(dive.get_previous_by_published().title, "Learning Python")

   856 

   857     def test_raw(self):

   858         "test the raw() method across databases"

   859         dive = Book.objects.using('other').create(title="Dive into Python",

   860             published=datetime.date(2009, 5, 4))

   861         val = Book.objects.db_manager("other").raw('SELECT id FROM multiple_database_book')

   862         self.assertEqual(map(lambda o: o.pk, val), [dive.pk])

   863 

   864         val = Book.objects.raw('SELECT id FROM multiple_database_book').using('other')

   865         self.assertEqual(map(lambda o: o.pk, val), [dive.pk])

   866 

   867     def test_select_related(self):

   868         "Database assignment is retained if an object is retrieved with select_related()"

   869         # Create a book and author on the other database

   870         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   871         dive = Book.objects.using('other').create(title="Dive into Python",

   872                                                   published=datetime.date(2009, 5, 4),

   873                                                   editor=mark)

   874 

   875         # Retrieve the Person using select_related()

   876         book = Book.objects.using('other').select_related('editor').get(title="Dive into Python")

   877 

   878         # The editor instance should have a db state

   879         self.assertEqual(book.editor._state.db, 'other')

   880 

   881     def test_subquery(self):

   882         """Make sure as_sql works with subqueries and master/slave."""

   883         sub = Person.objects.using('other').filter(name='fff')

   884         qs = Book.objects.filter(editor__in=sub)

   885 

   886         # When you call __str__ on the query object, it doesn't know about using

   887         # so it falls back to the default. If the subquery explicitly uses a

   888         # different database, an error should be raised.

   889         self.assertRaises(ValueError, str, qs.query)

   890 

   891         # Evaluating the query shouldn't work, either

   892         try:

   893             for obj in qs:

   894                 pass

   895             self.fail('Iterating over query should raise ValueError')

   896         except ValueError:

   897             pass

   898 

   899     def test_related_manager(self):

   900         "Related managers return managers, not querysets"

   901         mark = Person.objects.using('other').create(name="Mark Pilgrim")

   902 

   903         # extra_arg is removed by the BookManager's implementation of

   904         # create(); but the BookManager's implementation won't get called

   905         # unless edited returns a Manager, not a queryset

   906         mark.book_set.create(title="Dive into Python",

   907                              published=datetime.date(2009, 5, 4),

   908                              extra_arg=True)

   909 

   910         mark.book_set.get_or_create(title="Dive into Python",

   911                                     published=datetime.date(2009, 5, 4),

   912                                     extra_arg=True)

   913 

   914         mark.edited.create(title="Dive into Water",

   915                            published=datetime.date(2009, 5, 4),

   916                            extra_arg=True)

   917 

   918         mark.edited.get_or_create(title="Dive into Water",

   919                                   published=datetime.date(2009, 5, 4),

   920                                   extra_arg=True)

   921 

   922 class TestRouter(object):

   923     # A test router. The behaviour is vaguely master/slave, but the

   924     # databases aren't assumed to propagate changes.

   925     def db_for_read(self, model, instance=None, **hints):

   926         if instance:

   927             return instance._state.db or 'other'

   928         return 'other'

   929 

   930     def db_for_write(self, model, **hints):

   931         return DEFAULT_DB_ALIAS

   932 

   933     def allow_relation(self, obj1, obj2, **hints):

   934         return obj1._state.db in ('default', 'other') and obj2._state.db in ('default', 'other')

   935 

   936     def allow_syncdb(self, db, model):

   937         return True

   938 

   939 class AuthRouter(object):

   940     """A router to control all database operations on models in

   941     the contrib.auth application"""

   942 

   943     def db_for_read(self, model, **hints):

   944         "Point all read operations on auth models to 'default'"

   945         if model._meta.app_label == 'auth':

   946             # We use default here to ensure we can tell the difference

   947             # between a read request and a write request for Auth objects

   948             return 'default'

   949         return None

   950 

   951     def db_for_write(self, model, **hints):

   952         "Point all operations on auth models to 'other'"

   953         if model._meta.app_label == 'auth':

   954             return 'other'

   955         return None

   956 

   957     def allow_relation(self, obj1, obj2, **hints):

   958         "Allow any relation if a model in Auth is involved"

   959         if obj1._meta.app_label == 'auth' or obj2._meta.app_label == 'auth':

   960             return True

   961         return None

   962 

   963     def allow_syncdb(self, db, model):

   964         "Make sure the auth app only appears on the 'other' db"

   965         if db == 'other':

   966             return model._meta.app_label == 'auth'

   967         elif model._meta.app_label == 'auth':

   968             return False

   969         return None

   970 

   971 class WriteRouter(object):

   972     # A router that only expresses an opinion on writes

   973     def db_for_write(self, model, **hints):

   974         return 'writer'

   975 

   976 class RouterTestCase(TestCase):

   977     multi_db = True

   978 

   979     def setUp(self):

   980         # Make the 'other' database appear to be a slave of the 'default'

   981         self.old_routers = router.routers

   982         router.routers = [TestRouter()]

   983 

   984     def tearDown(self):

   985         # Restore the 'other' database as an independent database

   986         router.routers = self.old_routers

   987 

   988     def test_db_selection(self):

   989         "Check that querysets obey the router for db suggestions"

   990         self.assertEquals(Book.objects.db, 'other')

   991         self.assertEquals(Book.objects.all().db, 'other')

   992 

   993         self.assertEquals(Book.objects.using('default').db, 'default')

   994 

   995         self.assertEquals(Book.objects.db_manager('default').db, 'default')

   996         self.assertEquals(Book.objects.db_manager('default').all().db, 'default')

   997 

   998     def test_syncdb_selection(self):

   999         "Synchronization behaviour is predicatable"

  1000 

  1001         self.assertTrue(router.allow_syncdb('default', User))

  1002         self.assertTrue(router.allow_syncdb('default', Book))

  1003 

  1004         self.assertTrue(router.allow_syncdb('other', User))

  1005         self.assertTrue(router.allow_syncdb('other', Book))

  1006 

  1007         # Add the auth router to the chain.

  1008         # TestRouter is a universal synchronizer, so it should have no effect.

  1009         router.routers = [TestRouter(), AuthRouter()]

  1010 

  1011         self.assertTrue(router.allow_syncdb('default', User))

  1012         self.assertTrue(router.allow_syncdb('default', Book))

  1013 

  1014         self.assertTrue(router.allow_syncdb('other', User))

  1015         self.assertTrue(router.allow_syncdb('other', Book))

  1016 

  1017         # Now check what happens if the router order is the other way around

  1018         router.routers = [AuthRouter(), TestRouter()]

  1019 

  1020         self.assertFalse(router.allow_syncdb('default', User))

  1021         self.assertTrue(router.allow_syncdb('default', Book))

  1022 

  1023         self.assertTrue(router.allow_syncdb('other', User))

  1024         self.assertFalse(router.allow_syncdb('other', Book))

  1025 

  1026     def test_partial_router(self):

  1027         "A router can choose to implement a subset of methods"

  1028         dive = Book.objects.using('other').create(title="Dive into Python",

  1029                                                   published=datetime.date(2009, 5, 4))

  1030 

  1031         # First check the baseline behaviour

  1032 

  1033         self.assertEquals(router.db_for_read(User), 'other')

  1034         self.assertEquals(router.db_for_read(Book), 'other')

  1035 

  1036         self.assertEquals(router.db_for_write(User), 'default')

  1037         self.assertEquals(router.db_for_write(Book), 'default')

  1038 

  1039         self.assertTrue(router.allow_relation(dive, dive))

  1040 

  1041         self.assertTrue(router.allow_syncdb('default', User))

  1042         self.assertTrue(router.allow_syncdb('default', Book))

  1043 

  1044         router.routers = [WriteRouter(), AuthRouter(), TestRouter()]

  1045 

  1046         self.assertEquals(router.db_for_read(User), 'default')

  1047         self.assertEquals(router.db_for_read(Book), 'other')

  1048 

  1049         self.assertEquals(router.db_for_write(User), 'writer')

  1050         self.assertEquals(router.db_for_write(Book), 'writer')

  1051 

  1052         self.assertTrue(router.allow_relation(dive, dive))

  1053 

  1054         self.assertFalse(router.allow_syncdb('default', User))

  1055         self.assertTrue(router.allow_syncdb('default', Book))

  1056 

  1057 

  1058     def test_database_routing(self):

  1059         marty = Person.objects.using('default').create(name="Marty Alchin")

  1060         pro = Book.objects.using('default').create(title="Pro Django",

  1061                                                    published=datetime.date(2008, 12, 16),

  1062                                                    editor=marty)

  1063         pro.authors = [marty]

  1064 

  1065         # Create a book and author on the other database

  1066         dive = Book.objects.using('other').create(title="Dive into Python",

  1067                                                   published=datetime.date(2009, 5, 4))

  1068 

  1069         # An update query will be routed to the default database

  1070         Book.objects.filter(title='Pro Django').update(pages=200)

  1071 

  1072         try:

  1073             # By default, the get query will be directed to 'other'

  1074             Book.objects.get(title='Pro Django')

  1075             self.fail("Shouldn't be able to find the book")

  1076         except Book.DoesNotExist:

  1077             pass

  1078 

  1079         # But the same query issued explicitly at a database will work.

  1080         pro = Book.objects.using('default').get(title='Pro Django')

  1081 

  1082         # Check that the update worked.

  1083         self.assertEquals(pro.pages, 200)

  1084 

  1085         # An update query with an explicit using clause will be routed

  1086         # to the requested database.

  1087         Book.objects.using('other').filter(title='Dive into Python').update(pages=300)

  1088         self.assertEquals(Book.objects.get(title='Dive into Python').pages, 300)

  1089 

  1090         # Related object queries stick to the same database

  1091         # as the original object, regardless of the router

  1092         self.assertEquals(list(pro.authors.values_list('name', flat=True)), [u'Marty Alchin'])

  1093         self.assertEquals(pro.editor.name, u'Marty Alchin')

  1094 

  1095         # get_or_create is a special case. The get needs to be targetted at

  1096         # the write database in order to avoid potential transaction

  1097         # consistency problems

  1098         book, created = Book.objects.get_or_create(title="Pro Django")

  1099         self.assertFalse(created)

  1100 

  1101         book, created = Book.objects.get_or_create(title="Dive Into Python",

  1102                                                    defaults={'published':datetime.date(2009, 5, 4)})

  1103         self.assertTrue(created)

  1104 

  1105         # Check the head count of objects

  1106         self.assertEquals(Book.objects.using('default').count(), 2)

  1107         self.assertEquals(Book.objects.using('other').count(), 1)

  1108         # If a database isn't specified, the read database is used

  1109         self.assertEquals(Book.objects.count(), 1)

  1110 

  1111         # A delete query will also be routed to the default database

  1112         Book.objects.filter(pages__gt=150).delete()

  1113 

  1114         # The default database has lost the book.

  1115         self.assertEquals(Book.objects.using('default').count(), 1)

  1116         self.assertEquals(Book.objects.using('other').count(), 1)

  1117 

  1118     def test_foreign_key_cross_database_protection(self):

  1119         "Foreign keys can cross databases if they two databases have a common source"

  1120         # Create a book and author on the default database

  1121         pro = Book.objects.using('default').create(title="Pro Django",

  1122                                                    published=datetime.date(2008, 12, 16))

  1123 

  1124         marty = Person.objects.using('default').create(name="Marty Alchin")

  1125 

  1126         # Create a book and author on the other database

  1127         dive = Book.objects.using('other').create(title="Dive into Python",

  1128                                                   published=datetime.date(2009, 5, 4))

  1129 

  1130         mark = Person.objects.using('other').create(name="Mark Pilgrim")

  1131 

  1132         # Set a foreign key with an object from a different database

  1133         try:

  1134             dive.editor = marty

  1135         except ValueError:

  1136             self.fail("Assignment across master/slave databases with a common source should be ok")