python/lib/django_1_4/tests/regressiontests/multiple_database/tests.py

   1 from __future__ import absolute_import
   2
   3 import datetime
   4 import pickle
   5 from StringIO import StringIO
   6
   7 from django.conf import settings
   8 from django.contrib.auth.models import User
   9 from django.contrib.contenttypes.models import ContentType
  10 from django.core import management
  11 from django.db import connections, router, DEFAULT_DB_ALIAS
  12 from django.db.models import signals
  13 from django.test import TestCase
  14
  15 from .models import Book, Person, Pet, Review, UserProfile
  16
  17
  18 def copy_content_types_from_default_to_other():
  19     # On post_syncdb, content types are created in the 'default' database.
  20     # However, tests of generic foreign keys require them in 'other' too.
  21     # The problem is masked on backends that defer constraints checks: at the
  22     # end of each test, there's a rollback, and constraints are never checked.
  23     # It only appears on MySQL + InnoDB.
  24     for ct in ContentType.objects.using('default').all():
  25         ct.save(using='other')
  26
  27
  28 class QueryTestCase(TestCase):
  29     multi_db = True
  30
  31     def test_db_selection(self):
  32         "Check that querysets will use the default database by default"
  33         self.assertEqual(Book.objects.db, DEFAULT_DB_ALIAS)
  34         self.assertEqual(Book.objects.all().db, DEFAULT_DB_ALIAS)
  35
  36         self.assertEqual(Book.objects.using('other').db, 'other')
  37
  38         self.assertEqual(Book.objects.db_manager('other').db, 'other')
  39         self.assertEqual(Book.objects.db_manager('other').all().db, 'other')
  40
  41     def test_default_creation(self):
  42         "Objects created on the default database don't leak onto other databases"
  43         # Create a book on the default database using create()
  44         Book.objects.create(title="Pro Django",
  45                             published=datetime.date(2008, 12, 16))
  46
  47         # Create a book on the default database using a save
  48         dive = Book()
  49         dive.title="Dive into Python"
  50         dive.published = datetime.date(2009, 5, 4)
  51         dive.save()
  52
  53         # Check that book exists on the default database, but not on other database
  54         try:
  55             Book.objects.get(title="Pro Django")
  56             Book.objects.using('default').get(title="Pro Django")
  57         except Book.DoesNotExist:
  58             self.fail('"Dive Into Python" should exist on default database')
  59
  60         self.assertRaises(Book.DoesNotExist,
  61             Book.objects.using('other').get,
  62             title="Pro Django"
  63         )
  64
  65         try:
  66             Book.objects.get(title="Dive into Python")
  67             Book.objects.using('default').get(title="Dive into Python")
  68         except Book.DoesNotExist:
  69             self.fail('"Dive into Python" should exist on default database')
  70
  71         self.assertRaises(Book.DoesNotExist,
  72             Book.objects.using('other').get,
  73             title="Dive into Python"
  74         )
  75
  76
  77     def test_other_creation(self):
  78         "Objects created on another database don't leak onto the default database"
  79         # Create a book on the second database
  80         Book.objects.using('other').create(title="Pro Django",
  81                                            published=datetime.date(2008, 12, 16))
  82
  83         # Create a book on the default database using a save
  84         dive = Book()
  85         dive.title="Dive into Python"
  86         dive.published = datetime.date(2009, 5, 4)
  87         dive.save(using='other')
  88
  89         # Check that book exists on the default database, but not on other database
  90         try:
  91             Book.objects.using('other').get(title="Pro Django")
  92         except Book.DoesNotExist:
  93             self.fail('"Dive Into Python" should exist on other database')
  94
  95         self.assertRaises(Book.DoesNotExist,
  96             Book.objects.get,
  97             title="Pro Django"
  98         )
  99         self.assertRaises(Book.DoesNotExist,
 100             Book.objects.using('default').get,
 101             title="Pro Django"
 102         )
 103
 104         try:
 105             Book.objects.using('other').get(title="Dive into Python")
 106         except Book.DoesNotExist:
 107             self.fail('"Dive into Python" should exist on other database')
 108
 109         self.assertRaises(Book.DoesNotExist,
 110             Book.objects.get,
 111             title="Dive into Python"
 112         )
 113         self.assertRaises(Book.DoesNotExist,
 114             Book.objects.using('default').get,
 115             title="Dive into Python"
 116         )
 117
 118     def test_basic_queries(self):
 119         "Queries are constrained to a single database"
 120         dive = Book.objects.using('other').create(title="Dive into Python",
 121                                                   published=datetime.date(2009, 5, 4))
 122
 123         dive =  Book.objects.using('other').get(published=datetime.date(2009, 5, 4))
 124         self.assertEqual(dive.title, "Dive into Python")
 125         self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, published=datetime.date(2009, 5, 4))
 126
 127         dive = Book.objects.using('other').get(title__icontains="dive")
 128         self.assertEqual(dive.title, "Dive into Python")
 129         self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, title__icontains="dive")
 130
 131         dive = Book.objects.using('other').get(title__iexact="dive INTO python")
 132         self.assertEqual(dive.title, "Dive into Python")
 133         self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, title__iexact="dive INTO python")
 134
 135         dive =  Book.objects.using('other').get(published__year=2009)
 136         self.assertEqual(dive.title, "Dive into Python")
 137         self.assertEqual(dive.published, datetime.date(2009, 5, 4))
 138         self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, published__year=2009)
 139
 140         years = Book.objects.using('other').dates('published', 'year')
 141         self.assertEqual([o.year for o in years], [2009])
 142         years = Book.objects.using('default').dates('published', 'year')
 143         self.assertEqual([o.year for o in years], [])
 144
 145         months = Book.objects.using('other').dates('published', 'month')
 146         self.assertEqual([o.month for o in months], [5])
 147         months = Book.objects.using('default').dates('published', 'month')
 148         self.assertEqual([o.month for o in months], [])
 149
 150     def test_m2m_separation(self):
 151         "M2M fields are constrained to a single database"
 152         # Create a book and author on the default database
 153         pro = Book.objects.create(title="Pro Django",
 154                                   published=datetime.date(2008, 12, 16))
 155
 156         marty = Person.objects.create(name="Marty Alchin")
 157
 158         # Create a book and author on the other database
 159         dive = Book.objects.using('other').create(title="Dive into Python",
 160                                                   published=datetime.date(2009, 5, 4))
 161
 162         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 163
 164         # Save the author relations
 165         pro.authors = [marty]
 166         dive.authors = [mark]
 167
 168         # Inspect the m2m tables directly.
 169         # There should be 1 entry in each database
 170         self.assertEqual(Book.authors.through.objects.using('default').count(), 1)
 171         self.assertEqual(Book.authors.through.objects.using('other').count(), 1)
 172
 173         # Check that queries work across m2m joins
 174         self.assertEqual(list(Book.objects.using('default').filter(authors__name='Marty Alchin').values_list('title', flat=True)),
 175                           [u'Pro Django'])
 176         self.assertEqual(list(Book.objects.using('other').filter(authors__name='Marty Alchin').values_list('title', flat=True)),
 177                           [])
 178
 179         self.assertEqual(list(Book.objects.using('default').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
 180                           [])
 181         self.assertEqual(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
 182                           [u'Dive into Python'])
 183
 184         # Reget the objects to clear caches
 185         dive = Book.objects.using('other').get(title="Dive into Python")
 186         mark = Person.objects.using('other').get(name="Mark Pilgrim")
 187
 188         # Retrive related object by descriptor. Related objects should be database-baound
 189         self.assertEqual(list(dive.authors.all().values_list('name', flat=True)),
 190                           [u'Mark Pilgrim'])
 191
 192         self.assertEqual(list(mark.book_set.all().values_list('title', flat=True)),
 193                           [u'Dive into Python'])
 194
 195     def test_m2m_forward_operations(self):
 196         "M2M forward manipulations are all constrained to a single DB"
 197         # Create a book and author on the other database
 198         dive = Book.objects.using('other').create(title="Dive into Python",
 199                                                   published=datetime.date(2009, 5, 4))
 200
 201         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 202
 203         # Save the author relations
 204         dive.authors = [mark]
 205
 206         # Add a second author
 207         john = Person.objects.using('other').create(name="John Smith")
 208         self.assertEqual(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),
 209                           [])
 210
 211
 212         dive.authors.add(john)
 213         self.assertEqual(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
 214                           [u'Dive into Python'])
 215         self.assertEqual(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),
 216                           [u'Dive into Python'])
 217
 218         # Remove the second author
 219         dive.authors.remove(john)
 220         self.assertEqual(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
 221                           [u'Dive into Python'])
 222         self.assertEqual(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),
 223                           [])
 224
 225         # Clear all authors
 226         dive.authors.clear()
 227         self.assertEqual(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
 228                           [])
 229         self.assertEqual(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),
 230                           [])
 231
 232         # Create an author through the m2m interface
 233         dive.authors.create(name='Jane Brown')
 234         self.assertEqual(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
 235                           [])
 236         self.assertEqual(list(Book.objects.using('other').filter(authors__name='Jane Brown').values_list('title', flat=True)),
 237                           [u'Dive into Python'])
 238
 239     def test_m2m_reverse_operations(self):
 240         "M2M reverse manipulations are all constrained to a single DB"
 241         # Create a book and author on the other database
 242         dive = Book.objects.using('other').create(title="Dive into Python",
 243                                                   published=datetime.date(2009, 5, 4))
 244
 245         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 246
 247         # Save the author relations
 248         dive.authors = [mark]
 249
 250         # Create a second book on the other database
 251         grease = Book.objects.using('other').create(title="Greasemonkey Hacks",
 252                                                     published=datetime.date(2005, 11, 1))
 253
 254         # Add a books to the m2m
 255         mark.book_set.add(grease)
 256         self.assertEqual(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),
 257                           [u'Mark Pilgrim'])
 258         self.assertEqual(list(Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True)),
 259                           [u'Mark Pilgrim'])
 260
 261         # Remove a book from the m2m
 262         mark.book_set.remove(grease)
 263         self.assertEqual(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),
 264                           [u'Mark Pilgrim'])
 265         self.assertEqual(list(Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True)),
 266                           [])
 267
 268         # Clear the books associated with mark
 269         mark.book_set.clear()
 270         self.assertEqual(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),
 271                           [])
 272         self.assertEqual(list(Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True)),
 273                           [])
 274
 275         # Create a book through the m2m interface
 276         mark.book_set.create(title="Dive into HTML5", published=datetime.date(2020, 1, 1))
 277         self.assertEqual(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),
 278                           [])
 279         self.assertEqual(list(Person.objects.using('other').filter(book__title='Dive into HTML5').values_list('name', flat=True)),
 280                           [u'Mark Pilgrim'])
 281
 282     def test_m2m_cross_database_protection(self):
 283         "Operations that involve sharing M2M objects across databases raise an error"
 284         # Create a book and author on the default database
 285         pro = Book.objects.create(title="Pro Django",
 286                                   published=datetime.date(2008, 12, 16))
 287
 288         marty = Person.objects.create(name="Marty Alchin")
 289
 290         # Create a book and author on the other database
 291         dive = Book.objects.using('other').create(title="Dive into Python",
 292                                                   published=datetime.date(2009, 5, 4))
 293
 294         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 295         # Set a foreign key set with an object from a different database
 296         try:
 297             marty.book_set = [pro, dive]
 298             self.fail("Shouldn't be able to assign across databases")
 299         except ValueError:
 300             pass
 301
 302         # Add to an m2m with an object from a different database
 303         try:
 304             marty.book_set.add(dive)
 305             self.fail("Shouldn't be able to assign across databases")
 306         except ValueError:
 307             pass
 308
 309         # Set a m2m with an object from a different database
 310         try:
 311             marty.book_set = [pro, dive]
 312             self.fail("Shouldn't be able to assign across databases")
 313         except ValueError:
 314             pass
 315
 316         # Add to a reverse m2m with an object from a different database
 317         try:
 318             dive.authors.add(marty)
 319             self.fail("Shouldn't be able to assign across databases")
 320         except ValueError:
 321             pass
 322
 323         # Set a reverse m2m with an object from a different database
 324         try:
 325             dive.authors = [mark, marty]
 326             self.fail("Shouldn't be able to assign across databases")
 327         except ValueError:
 328             pass
 329
 330     def test_m2m_deletion(self):
 331         "Cascaded deletions of m2m relations issue queries on the right database"
 332         # Create a book and author on the other database
 333         dive = Book.objects.using('other').create(title="Dive into Python",
 334                                                   published=datetime.date(2009, 5, 4))
 335
 336         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 337         dive.authors = [mark]
 338
 339         # Check the initial state
 340         self.assertEqual(Person.objects.using('default').count(), 0)
 341         self.assertEqual(Book.objects.using('default').count(), 0)
 342         self.assertEqual(Book.authors.through.objects.using('default').count(), 0)
 343
 344         self.assertEqual(Person.objects.using('other').count(), 1)
 345         self.assertEqual(Book.objects.using('other').count(), 1)
 346         self.assertEqual(Book.authors.through.objects.using('other').count(), 1)
 347
 348         # Delete the object on the other database
 349         dive.delete(using='other')
 350
 351         self.assertEqual(Person.objects.using('default').count(), 0)
 352         self.assertEqual(Book.objects.using('default').count(), 0)
 353         self.assertEqual(Book.authors.through.objects.using('default').count(), 0)
 354
 355         # The person still exists ...
 356         self.assertEqual(Person.objects.using('other').count(), 1)
 357         # ... but the book has been deleted
 358         self.assertEqual(Book.objects.using('other').count(), 0)
 359         # ... and the relationship object has also been deleted.
 360         self.assertEqual(Book.authors.through.objects.using('other').count(), 0)
 361
 362         # Now try deletion in the reverse direction. Set up the relation again
 363         dive = Book.objects.using('other').create(title="Dive into Python",
 364                                                   published=datetime.date(2009, 5, 4))
 365         dive.authors = [mark]
 366
 367         # Check the initial state
 368         self.assertEqual(Person.objects.using('default').count(), 0)
 369         self.assertEqual(Book.objects.using('default').count(), 0)
 370         self.assertEqual(Book.authors.through.objects.using('default').count(), 0)
 371
 372         self.assertEqual(Person.objects.using('other').count(), 1)
 373         self.assertEqual(Book.objects.using('other').count(), 1)
 374         self.assertEqual(Book.authors.through.objects.using('other').count(), 1)
 375
 376         # Delete the object on the other database
 377         mark.delete(using='other')
 378
 379         self.assertEqual(Person.objects.using('default').count(), 0)
 380         self.assertEqual(Book.objects.using('default').count(), 0)
 381         self.assertEqual(Book.authors.through.objects.using('default').count(), 0)
 382
 383         # The person has been deleted ...
 384         self.assertEqual(Person.objects.using('other').count(), 0)
 385         # ... but the book still exists
 386         self.assertEqual(Book.objects.using('other').count(), 1)
 387         # ... and the relationship object has been deleted.
 388         self.assertEqual(Book.authors.through.objects.using('other').count(), 0)
 389
 390     def test_foreign_key_separation(self):
 391         "FK fields are constrained to a single database"
 392         # Create a book and author on the default database
 393         pro = Book.objects.create(title="Pro Django",
 394                                   published=datetime.date(2008, 12, 16))
 395
 396         marty = Person.objects.create(name="Marty Alchin")
 397         george = Person.objects.create(name="George Vilches")
 398
 399         # Create a book and author on the other database
 400         dive = Book.objects.using('other').create(title="Dive into Python",
 401                                                   published=datetime.date(2009, 5, 4))
 402
 403         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 404         chris = Person.objects.using('other').create(name="Chris Mills")
 405
 406         # Save the author's favourite books
 407         pro.editor = george
 408         pro.save()
 409
 410         dive.editor = chris
 411         dive.save()
 412
 413         pro = Book.objects.using('default').get(title="Pro Django")
 414         self.assertEqual(pro.editor.name, "George Vilches")
 415
 416         dive = Book.objects.using('other').get(title="Dive into Python")
 417         self.assertEqual(dive.editor.name, "Chris Mills")
 418
 419         # Check that queries work across foreign key joins
 420         self.assertEqual(list(Person.objects.using('default').filter(edited__title='Pro Django').values_list('name', flat=True)),
 421                           [u'George Vilches'])
 422         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Pro Django').values_list('name', flat=True)),
 423                           [])
 424
 425         self.assertEqual(list(Person.objects.using('default').filter(edited__title='Dive into Python').values_list('name', flat=True)),
 426                           [])
 427         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),
 428                           [u'Chris Mills'])
 429
 430         # Reget the objects to clear caches
 431         chris = Person.objects.using('other').get(name="Chris Mills")
 432         dive = Book.objects.using('other').get(title="Dive into Python")
 433
 434         # Retrive related object by descriptor. Related objects should be database-baound
 435         self.assertEqual(list(chris.edited.values_list('title', flat=True)),
 436                           [u'Dive into Python'])
 437
 438     def test_foreign_key_reverse_operations(self):
 439         "FK reverse manipulations are all constrained to a single DB"
 440         dive = Book.objects.using('other').create(title="Dive into Python",
 441                                                        published=datetime.date(2009, 5, 4))
 442
 443         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 444         chris = Person.objects.using('other').create(name="Chris Mills")
 445
 446         # Save the author relations
 447         dive.editor = chris
 448         dive.save()
 449
 450         # Add a second book edited by chris
 451         html5 = Book.objects.using('other').create(title="Dive into HTML5", published=datetime.date(2010, 3, 15))
 452         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),
 453                           [])
 454
 455         chris.edited.add(html5)
 456         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),
 457                           [u'Chris Mills'])
 458         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),
 459                           [u'Chris Mills'])
 460
 461         # Remove the second editor
 462         chris.edited.remove(html5)
 463         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),
 464                           [])
 465         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),
 466                           [u'Chris Mills'])
 467
 468         # Clear all edited books
 469         chris.edited.clear()
 470         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),
 471                           [])
 472         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),
 473                           [])
 474
 475         # Create an author through the m2m interface
 476         chris.edited.create(title='Dive into Water', published=datetime.date(2010, 3, 15))
 477         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),
 478                           [])
 479         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Water').values_list('name', flat=True)),
 480                           [u'Chris Mills'])
 481         self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),
 482                           [])
 483
 484     def test_foreign_key_cross_database_protection(self):
 485         "Operations that involve sharing FK objects across databases raise an error"
 486         # Create a book and author on the default database
 487         pro = Book.objects.create(title="Pro Django",
 488                                   published=datetime.date(2008, 12, 16))
 489
 490         marty = Person.objects.create(name="Marty Alchin")
 491
 492         # Create a book and author on the other database
 493         dive = Book.objects.using('other').create(title="Dive into Python",
 494                                                   published=datetime.date(2009, 5, 4))
 495
 496         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 497
 498         # Set a foreign key with an object from a different database
 499         try:
 500             dive.editor = marty
 501             self.fail("Shouldn't be able to assign across databases")
 502         except ValueError:
 503             pass
 504
 505         # Set a foreign key set with an object from a different database
 506         try:
 507             marty.edited = [pro, dive]
 508             self.fail("Shouldn't be able to assign across databases")
 509         except ValueError:
 510             pass
 511
 512         # Add to a foreign key set with an object from a different database
 513         try:
 514             marty.edited.add(dive)
 515             self.fail("Shouldn't be able to assign across databases")
 516         except ValueError:
 517             pass
 518
 519         # BUT! if you assign a FK object when the base object hasn't
 520         # been saved yet, you implicitly assign the database for the
 521         # base object.
 522         chris = Person(name="Chris Mills")
 523         html5 = Book(title="Dive into HTML5", published=datetime.date(2010, 3, 15))
 524         # initially, no db assigned
 525         self.assertEqual(chris._state.db, None)
 526         self.assertEqual(html5._state.db, None)
 527
 528         # old object comes from 'other', so the new object is set to use 'other'...
 529         dive.editor = chris
 530         html5.editor = mark
 531         self.assertEqual(chris._state.db, 'other')
 532         self.assertEqual(html5._state.db, 'other')
 533         # ... but it isn't saved yet
 534         self.assertEqual(list(Person.objects.using('other').values_list('name',flat=True)),
 535                           [u'Mark Pilgrim'])
 536         self.assertEqual(list(Book.objects.using('other').values_list('title',flat=True)),
 537                            [u'Dive into Python'])
 538
 539         # When saved (no using required), new objects goes to 'other'
 540         chris.save()
 541         html5.save()
 542         self.assertEqual(list(Person.objects.using('default').values_list('name',flat=True)),
 543                           [u'Marty Alchin'])
 544         self.assertEqual(list(Person.objects.using('other').values_list('name',flat=True)),
 545                           [u'Chris Mills', u'Mark Pilgrim'])
 546         self.assertEqual(list(Book.objects.using('default').values_list('title',flat=True)),
 547                           [u'Pro Django'])
 548         self.assertEqual(list(Book.objects.using('other').values_list('title',flat=True)),
 549                           [u'Dive into HTML5', u'Dive into Python'])
 550
 551         # This also works if you assign the FK in the constructor
 552         water = Book(title="Dive into Water", published=datetime.date(2001, 1, 1), editor=mark)
 553         self.assertEqual(water._state.db, 'other')
 554         # ... but it isn't saved yet
 555         self.assertEqual(list(Book.objects.using('default').values_list('title',flat=True)),
 556                           [u'Pro Django'])
 557         self.assertEqual(list(Book.objects.using('other').values_list('title',flat=True)),
 558                           [u'Dive into HTML5', u'Dive into Python'])
 559
 560         # When saved, the new book goes to 'other'
 561         water.save()
 562         self.assertEqual(list(Book.objects.using('default').values_list('title',flat=True)),
 563                           [u'Pro Django'])
 564         self.assertEqual(list(Book.objects.using('other').values_list('title',flat=True)),
 565                           [u'Dive into HTML5', u'Dive into Python', u'Dive into Water'])
 566
 567     def test_foreign_key_deletion(self):
 568         "Cascaded deletions of Foreign Key relations issue queries on the right database"
 569         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 570         fido = Pet.objects.using('other').create(name="Fido", owner=mark)
 571
 572         # Check the initial state
 573         self.assertEqual(Person.objects.using('default').count(), 0)
 574         self.assertEqual(Pet.objects.using('default').count(), 0)
 575
 576         self.assertEqual(Person.objects.using('other').count(), 1)
 577         self.assertEqual(Pet.objects.using('other').count(), 1)
 578
 579         # Delete the person object, which will cascade onto the pet
 580         mark.delete(using='other')
 581
 582         self.assertEqual(Person.objects.using('default').count(), 0)
 583         self.assertEqual(Pet.objects.using('default').count(), 0)
 584
 585         # Both the pet and the person have been deleted from the right database
 586         self.assertEqual(Person.objects.using('other').count(), 0)
 587         self.assertEqual(Pet.objects.using('other').count(), 0)
 588
 589     def test_foreign_key_validation(self):
 590         "ForeignKey.validate() uses the correct database"
 591         mickey = Person.objects.using('other').create(name="Mickey")
 592         pluto = Pet.objects.using('other').create(name="Pluto", owner=mickey)
 593         self.assertEqual(None, pluto.full_clean())
 594
 595     def test_o2o_separation(self):
 596         "OneToOne fields are constrained to a single database"
 597         # Create a user and profile on the default database
 598         alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com')
 599         alice_profile = UserProfile.objects.using('default').create(user=alice, flavor='chocolate')
 600
 601         # Create a user and profile on the other database
 602         bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com')
 603         bob_profile = UserProfile.objects.using('other').create(user=bob, flavor='crunchy frog')
 604
 605         # Retrieve related objects; queries should be database constrained
 606         alice = User.objects.using('default').get(username="alice")
 607         self.assertEqual(alice.userprofile.flavor, "chocolate")
 608
 609         bob = User.objects.using('other').get(username="bob")
 610         self.assertEqual(bob.userprofile.flavor, "crunchy frog")
 611
 612         # Check that queries work across joins
 613         self.assertEqual(list(User.objects.using('default').filter(userprofile__flavor='chocolate').values_list('username', flat=True)),
 614                           [u'alice'])
 615         self.assertEqual(list(User.objects.using('other').filter(userprofile__flavor='chocolate').values_list('username', flat=True)),
 616                           [])
 617
 618         self.assertEqual(list(User.objects.using('default').filter(userprofile__flavor='crunchy frog').values_list('username', flat=True)),
 619                           [])
 620         self.assertEqual(list(User.objects.using('other').filter(userprofile__flavor='crunchy frog').values_list('username', flat=True)),
 621                           [u'bob'])
 622
 623         # Reget the objects to clear caches
 624         alice_profile = UserProfile.objects.using('default').get(flavor='chocolate')
 625         bob_profile = UserProfile.objects.using('other').get(flavor='crunchy frog')
 626
 627         # Retrive related object by descriptor. Related objects should be database-baound
 628         self.assertEqual(alice_profile.user.username, 'alice')
 629         self.assertEqual(bob_profile.user.username, 'bob')
 630
 631     def test_o2o_cross_database_protection(self):
 632         "Operations that involve sharing FK objects across databases raise an error"
 633         # Create a user and profile on the default database
 634         alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com')
 635
 636         # Create a user and profile on the other database
 637         bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com')
 638
 639         # Set a one-to-one relation with an object from a different database
 640         alice_profile = UserProfile.objects.using('default').create(user=alice, flavor='chocolate')
 641         try:
 642             bob.userprofile = alice_profile
 643             self.fail("Shouldn't be able to assign across databases")
 644         except ValueError:
 645             pass
 646
 647         # BUT! if you assign a FK object when the base object hasn't
 648         # been saved yet, you implicitly assign the database for the
 649         # base object.
 650         bob_profile = UserProfile.objects.using('other').create(user=bob, flavor='crunchy frog')
 651
 652         new_bob_profile = UserProfile(flavor="spring surprise")
 653
 654         charlie = User(username='charlie',email='charlie@example.com')
 655         charlie.set_unusable_password()
 656
 657         # initially, no db assigned
 658         self.assertEqual(new_bob_profile._state.db, None)
 659         self.assertEqual(charlie._state.db, None)
 660
 661         # old object comes from 'other', so the new object is set to use 'other'...
 662         new_bob_profile.user = bob
 663         charlie.userprofile = bob_profile
 664         self.assertEqual(new_bob_profile._state.db, 'other')
 665         self.assertEqual(charlie._state.db, 'other')
 666
 667         # ... but it isn't saved yet
 668         self.assertEqual(list(User.objects.using('other').values_list('username',flat=True)),
 669                           [u'bob'])
 670         self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor',flat=True)),
 671                            [u'crunchy frog'])
 672
 673         # When saved (no using required), new objects goes to 'other'
 674         charlie.save()
 675         bob_profile.save()
 676         new_bob_profile.save()
 677         self.assertEqual(list(User.objects.using('default').values_list('username',flat=True)),
 678                           [u'alice'])
 679         self.assertEqual(list(User.objects.using('other').values_list('username',flat=True)),
 680                           [u'bob', u'charlie'])
 681         self.assertEqual(list(UserProfile.objects.using('default').values_list('flavor',flat=True)),
 682                            [u'chocolate'])
 683         self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor',flat=True)),
 684                            [u'crunchy frog', u'spring surprise'])
 685
 686         # This also works if you assign the O2O relation in the constructor
 687         denise = User.objects.db_manager('other').create_user('denise','denise@example.com')
 688         denise_profile = UserProfile(flavor="tofu", user=denise)
 689
 690         self.assertEqual(denise_profile._state.db, 'other')
 691         # ... but it isn't saved yet
 692         self.assertEqual(list(UserProfile.objects.using('default').values_list('flavor',flat=True)),
 693                            [u'chocolate'])
 694         self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor',flat=True)),
 695                            [u'crunchy frog', u'spring surprise'])
 696
 697         # When saved, the new profile goes to 'other'
 698         denise_profile.save()
 699         self.assertEqual(list(UserProfile.objects.using('default').values_list('flavor',flat=True)),
 700                            [u'chocolate'])
 701         self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor',flat=True)),
 702                            [u'crunchy frog', u'spring surprise', u'tofu'])
 703
 704     def test_generic_key_separation(self):
 705         "Generic fields are constrained to a single database"
 706         copy_content_types_from_default_to_other()
 707
 708         # Create a book and author on the default database
 709         pro = Book.objects.create(title="Pro Django",
 710                                   published=datetime.date(2008, 12, 16))
 711
 712         review1 = Review.objects.create(source="Python Monthly", content_object=pro)
 713
 714         # Create a book and author on the other database
 715         dive = Book.objects.using('other').create(title="Dive into Python",
 716                                                   published=datetime.date(2009, 5, 4))
 717
 718         review2 = Review.objects.using('other').create(source="Python Weekly", content_object=dive)
 719
 720         review1 = Review.objects.using('default').get(source="Python Monthly")
 721         self.assertEqual(review1.content_object.title, "Pro Django")
 722
 723         review2 = Review.objects.using('other').get(source="Python Weekly")
 724         self.assertEqual(review2.content_object.title, "Dive into Python")
 725
 726         # Reget the objects to clear caches
 727         dive = Book.objects.using('other').get(title="Dive into Python")
 728
 729         # Retrive related object by descriptor. Related objects should be database-bound
 730         self.assertEqual(list(dive.reviews.all().values_list('source', flat=True)),
 731                           [u'Python Weekly'])
 732
 733     def test_generic_key_reverse_operations(self):
 734         "Generic reverse manipulations are all constrained to a single DB"
 735         copy_content_types_from_default_to_other()
 736
 737         dive = Book.objects.using('other').create(title="Dive into Python",
 738                                                   published=datetime.date(2009, 5, 4))
 739
 740         temp = Book.objects.using('other').create(title="Temp",
 741                                                   published=datetime.date(2009, 5, 4))
 742
 743         review1 = Review.objects.using('other').create(source="Python Weekly", content_object=dive)
 744         review2 = Review.objects.using('other').create(source="Python Monthly", content_object=temp)
 745
 746         self.assertEqual(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),
 747                           [])
 748         self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),
 749                           [u'Python Weekly'])
 750
 751         # Add a second review
 752         dive.reviews.add(review2)
 753         self.assertEqual(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),
 754                           [])
 755         self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),
 756                           [u'Python Monthly', u'Python Weekly'])
 757
 758         # Remove the second author
 759         dive.reviews.remove(review1)
 760         self.assertEqual(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),
 761                           [])
 762         self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),
 763                           [u'Python Monthly'])
 764
 765         # Clear all reviews
 766         dive.reviews.clear()
 767         self.assertEqual(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),
 768                           [])
 769         self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),
 770                           [])
 771
 772         # Create an author through the generic interface
 773         dive.reviews.create(source='Python Daily')
 774         self.assertEqual(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),
 775                           [])
 776         self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),
 777                           [u'Python Daily'])
 778
 779     def test_generic_key_cross_database_protection(self):
 780         "Operations that involve sharing generic key objects across databases raise an error"
 781         copy_content_types_from_default_to_other()
 782
 783         # Create a book and author on the default database
 784         pro = Book.objects.create(title="Pro Django",
 785                                   published=datetime.date(2008, 12, 16))
 786
 787         review1 = Review.objects.create(source="Python Monthly", content_object=pro)
 788
 789         # Create a book and author on the other database
 790         dive = Book.objects.using('other').create(title="Dive into Python",
 791                                                   published=datetime.date(2009, 5, 4))
 792
 793         review2 = Review.objects.using('other').create(source="Python Weekly", content_object=dive)
 794
 795         # Set a foreign key with an object from a different database
 796         try:
 797             review1.content_object = dive
 798             self.fail("Shouldn't be able to assign across databases")
 799         except ValueError:
 800             pass
 801
 802         # Add to a foreign key set with an object from a different database
 803         try:
 804             dive.reviews.add(review1)
 805             self.fail("Shouldn't be able to assign across databases")
 806         except ValueError:
 807             pass
 808
 809         # BUT! if you assign a FK object when the base object hasn't
 810         # been saved yet, you implicitly assign the database for the
 811         # base object.
 812         review3 = Review(source="Python Daily")
 813         # initially, no db assigned
 814         self.assertEqual(review3._state.db, None)
 815
 816         # Dive comes from 'other', so review3 is set to use 'other'...
 817         review3.content_object = dive
 818         self.assertEqual(review3._state.db, 'other')
 819         # ... but it isn't saved yet
 820         self.assertEqual(list(Review.objects.using('default').filter(object_id=pro.pk).values_list('source', flat=True)),
 821                           [u'Python Monthly'])
 822         self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source',flat=True)),
 823                           [u'Python Weekly'])
 824
 825         # When saved, John goes to 'other'
 826         review3.save()
 827         self.assertEqual(list(Review.objects.using('default').filter(object_id=pro.pk).values_list('source', flat=True)),
 828                           [u'Python Monthly'])
 829         self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source',flat=True)),
 830                           [u'Python Daily', u'Python Weekly'])
 831
 832     def test_generic_key_deletion(self):
 833         "Cascaded deletions of Generic Key relations issue queries on the right database"
 834         copy_content_types_from_default_to_other()
 835
 836         dive = Book.objects.using('other').create(title="Dive into Python",
 837                                                   published=datetime.date(2009, 5, 4))
 838         review = Review.objects.using('other').create(source="Python Weekly", content_object=dive)
 839
 840         # Check the initial state
 841         self.assertEqual(Book.objects.using('default').count(), 0)
 842         self.assertEqual(Review.objects.using('default').count(), 0)
 843
 844         self.assertEqual(Book.objects.using('other').count(), 1)
 845         self.assertEqual(Review.objects.using('other').count(), 1)
 846
 847         # Delete the Book object, which will cascade onto the pet
 848         dive.delete(using='other')
 849
 850         self.assertEqual(Book.objects.using('default').count(), 0)
 851         self.assertEqual(Review.objects.using('default').count(), 0)
 852
 853         # Both the pet and the person have been deleted from the right database
 854         self.assertEqual(Book.objects.using('other').count(), 0)
 855         self.assertEqual(Review.objects.using('other').count(), 0)
 856
 857     def test_ordering(self):
 858         "get_next_by_XXX commands stick to a single database"
 859         pro = Book.objects.create(title="Pro Django",
 860                                   published=datetime.date(2008, 12, 16))
 861
 862         dive = Book.objects.using('other').create(title="Dive into Python",
 863                                                   published=datetime.date(2009, 5, 4))
 864
 865         learn = Book.objects.using('other').create(title="Learning Python",
 866                                                    published=datetime.date(2008, 7, 16))
 867
 868         self.assertEqual(learn.get_next_by_published().title, "Dive into Python")
 869         self.assertEqual(dive.get_previous_by_published().title, "Learning Python")
 870
 871     def test_raw(self):
 872         "test the raw() method across databases"
 873         dive = Book.objects.using('other').create(title="Dive into Python",
 874             published=datetime.date(2009, 5, 4))
 875         val = Book.objects.db_manager("other").raw('SELECT id FROM multiple_database_book')
 876         self.assertEqual(map(lambda o: o.pk, val), [dive.pk])
 877
 878         val = Book.objects.raw('SELECT id FROM multiple_database_book').using('other')
 879         self.assertEqual(map(lambda o: o.pk, val), [dive.pk])
 880
 881     def test_select_related(self):
 882         "Database assignment is retained if an object is retrieved with select_related()"
 883         # Create a book and author on the other database
 884         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 885         dive = Book.objects.using('other').create(title="Dive into Python",
 886                                                   published=datetime.date(2009, 5, 4),
 887                                                   editor=mark)
 888
 889         # Retrieve the Person using select_related()
 890         book = Book.objects.using('other').select_related('editor').get(title="Dive into Python")
 891
 892         # The editor instance should have a db state
 893         self.assertEqual(book.editor._state.db, 'other')
 894
 895     def test_subquery(self):
 896         """Make sure as_sql works with subqueries and master/slave."""
 897         sub = Person.objects.using('other').filter(name='fff')
 898         qs = Book.objects.filter(editor__in=sub)
 899
 900         # When you call __str__ on the query object, it doesn't know about using
 901         # so it falls back to the default. If the subquery explicitly uses a
 902         # different database, an error should be raised.
 903         self.assertRaises(ValueError, str, qs.query)
 904
 905         # Evaluating the query shouldn't work, either
 906         try:
 907             for obj in qs:
 908                 pass
 909             self.fail('Iterating over query should raise ValueError')
 910         except ValueError:
 911             pass
 912
 913     def test_related_manager(self):
 914         "Related managers return managers, not querysets"
 915         mark = Person.objects.using('other').create(name="Mark Pilgrim")
 916
 917         # extra_arg is removed by the BookManager's implementation of
 918         # create(); but the BookManager's implementation won't get called
 919         # unless edited returns a Manager, not a queryset
 920         mark.book_set.create(title="Dive into Python",
 921                              published=datetime.date(2009, 5, 4),
 922                              extra_arg=True)
 923
 924         mark.book_set.get_or_create(title="Dive into Python",
 925                                     published=datetime.date(2009, 5, 4),
 926                                     extra_arg=True)
 927
 928         mark.edited.create(title="Dive into Water",
 929                            published=datetime.date(2009, 5, 4),
 930                            extra_arg=True)
 931
 932         mark.edited.get_or_create(title="Dive into Water",
 933                                   published=datetime.date(2009, 5, 4),
 934                                   extra_arg=True)
 935
 936 class TestRouter(object):
 937     # A test router. The behavior is vaguely master/slave, but the
 938     # databases aren't assumed to propagate changes.
 939     def db_for_read(self, model, instance=None, **hints):
 940         if instance:
 941             return instance._state.db or 'other'
 942         return 'other'
 943
 944     def db_for_write(self, model, **hints):
 945         return DEFAULT_DB_ALIAS
 946
 947     def allow_relation(self, obj1, obj2, **hints):
 948         return obj1._state.db in ('default', 'other') and obj2._state.db in ('default', 'other')
 949
 950     def allow_syncdb(self, db, model):
 951         return True
 952
 953 class AuthRouter(object):
 954     """A router to control all database operations on models in
 955     the contrib.auth application"""
 956
 957     def db_for_read(self, model, **hints):
 958         "Point all read operations on auth models to 'default'"
 959         if model._meta.app_label == 'auth':
 960             # We use default here to ensure we can tell the difference
 961             # between a read request and a write request for Auth objects
 962             return 'default'
 963         return None
 964
 965     def db_for_write(self, model, **hints):
 966         "Point all operations on auth models to 'other'"
 967         if model._meta.app_label == 'auth':
 968             return 'other'
 969         return None
 970
 971     def allow_relation(self, obj1, obj2, **hints):
 972         "Allow any relation if a model in Auth is involved"
 973         if obj1._meta.app_label == 'auth' or obj2._meta.app_label == 'auth':
 974             return True
 975         return None
 976
 977     def allow_syncdb(self, db, model):
 978         "Make sure the auth app only appears on the 'other' db"
 979         if db == 'other':
 980             return model._meta.app_label == 'auth'
 981         elif model._meta.app_label == 'auth':
 982             return False
 983         return None
 984
 985 class WriteRouter(object):
 986     # A router that only expresses an opinion on writes
 987     def db_for_write(self, model, **hints):
 988         return 'writer'
 989
 990 class RouterTestCase(TestCase):
 991     multi_db = True
 992
 993     def setUp(self):
 994         # Make the 'other' database appear to be a slave of the 'default'
 995         self.old_routers = router.routers
 996         router.routers = [TestRouter()]
 997
 998     def tearDown(self):
 999         # Restore the 'other' database as an independent database
1000         router.routers = self.old_routers
1001
1002     def test_db_selection(self):
1003         "Check that querysets obey the router for db suggestions"
1004         self.assertEqual(Book.objects.db, 'other')
1005         self.assertEqual(Book.objects.all().db, 'other')
1006
1007         self.assertEqual(Book.objects.using('default').db, 'default')
1008
1009         self.assertEqual(Book.objects.db_manager('default').db, 'default')
1010         self.assertEqual(Book.objects.db_manager('default').all().db, 'default')
1011
1012     def test_syncdb_selection(self):
1013         "Synchronization behavior is predictable"
1014
1015         self.assertTrue(router.allow_syncdb('default', User))
1016         self.assertTrue(router.allow_syncdb('default', Book))
1017
1018         self.assertTrue(router.allow_syncdb('other', User))
1019         self.assertTrue(router.allow_syncdb('other', Book))
1020
1021         # Add the auth router to the chain.
1022         # TestRouter is a universal synchronizer, so it should have no effect.
1023         router.routers = [TestRouter(), AuthRouter()]
1024
1025         self.assertTrue(router.allow_syncdb('default', User))
1026         self.assertTrue(router.allow_syncdb('default', Book))
1027
1028         self.assertTrue(router.allow_syncdb('other', User))
1029         self.assertTrue(router.allow_syncdb('other', Book))
1030
1031         # Now check what happens if the router order is the other way around
1032         router.routers = [AuthRouter(), TestRouter()]
1033
1034         self.assertFalse(router.allow_syncdb('default', User))
1035         self.assertTrue(router.allow_syncdb('default', Book))
1036
1037         self.assertTrue(router.allow_syncdb('other', User))
1038         self.assertFalse(router.allow_syncdb('other', Book))
1039
1040     def test_partial_router(self):
1041         "A router can choose to implement a subset of methods"
1042         dive = Book.objects.using('other').create(title="Dive into Python",
1043                                                   published=datetime.date(2009, 5, 4))
1044
1045         # First check the baseline behavior.
1046
1047         self.assertEqual(router.db_for_read(User), 'other')
1048         self.assertEqual(router.db_for_read(Book), 'other')
1049
1050         self.assertEqual(router.db_for_write(User), 'default')
1051         self.assertEqual(router.db_for_write(Book), 'default')
1052
1053         self.assertTrue(router.allow_relation(dive, dive))
1054
1055         self.assertTrue(router.allow_syncdb('default', User))
1056         self.assertTrue(router.allow_syncdb('default', Book))
1057
1058         router.routers = [WriteRouter(), AuthRouter(), TestRouter()]
1059
1060         self.assertEqual(router.db_for_read(User), 'default')
1061         self.assertEqual(router.db_for_read(Book), 'other')
1062
1063         self.assertEqual(router.db_for_write(User), 'writer')
1064         self.assertEqual(router.db_for_write(Book), 'writer')
1065
1066         self.assertTrue(router.allow_relation(dive, dive))
1067
1068         self.assertFalse(router.allow_syncdb('default', User))
1069         self.assertTrue(router.allow_syncdb('default', Book))
1070
1071
1072     def test_database_routing(self):
1073         marty = Person.objects.using('default').create(name="Marty Alchin")
1074         pro = Book.objects.using('default').create(title="Pro Django",
1075                                                    published=datetime.date(2008, 12, 16),
1076                                                    editor=marty)
1077         pro.authors = [marty]
1078
1079         # Create a book and author on the other database
1080         dive = Book.objects.using('other').create(title="Dive into Python",
1081                                                   published=datetime.date(2009, 5, 4))
1082
1083         # An update query will be routed to the default database
1084         Book.objects.filter(title='Pro Django').update(pages=200)
1085
1086         try:
1087             # By default, the get query will be directed to 'other'
1088             Book.objects.get(title='Pro Django')
1089             self.fail("Shouldn't be able to find the book")
1090         except Book.DoesNotExist:
1091             pass
1092
1093         # But the same query issued explicitly at a database will work.
1094         pro = Book.objects.using('default').get(title='Pro Django')
1095
1096         # Check that the update worked.
1097         self.assertEqual(pro.pages, 200)
1098
1099         # An update query with an explicit using clause will be routed
1100         # to the requested database.
1101         Book.objects.using('other').filter(title='Dive into Python').update(pages=300)
1102         self.assertEqual(Book.objects.get(title='Dive into Python').pages, 300)
1103
1104         # Related object queries stick to the same database
1105         # as the original object, regardless of the router
1106         self.assertEqual(list(pro.authors.values_list('name', flat=True)), [u'Marty Alchin'])
1107         self.assertEqual(pro.editor.name, u'Marty Alchin')
1108
1109         # get_or_create is a special case. The get needs to be targeted at
1110         # the write database in order to avoid potential transaction
1111         # consistency problems
1112         book, created = Book.objects.get_or_create(title="Pro Django")
1113         self.assertFalse(created)
1114
1115         book, created = Book.objects.get_or_create(title="Dive Into Python",
1116                                                    defaults={'published':datetime.date(2009, 5, 4)})
1117         self.assertTrue(created)
1118
1119         # Check the head count of objects
1120         self.assertEqual(Book.objects.using('default').count(), 2)
1121         self.assertEqual(Book.objects.using('other').count(), 1)
1122         # If a database isn't specified, the read database is used
1123         self.assertEqual(Book.objects.count(), 1)
1124
1125         # A delete query will also be routed to the default database
1126         Book.objects.filter(pages__gt=150).delete()
1127
1128         # The default database has lost the book.
1129         self.assertEqual(Book.objects.using('default').count(), 1)
1130         self.assertEqual(Book.objects.using('other').count(), 1)
1131
1132     def test_foreign_key_cross_database_protection(self):
1133         "Foreign keys can cross databases if they two databases have a common source"
1134         # Create a book and author on the default database
1135         pro = Book.objects.using('default').create(title="Pro Django",
1136                                                    published=datetime.date(2008, 12, 16))
1137
1138         marty = Person.objects.using('default').create(name="Marty Alchin")
1139
1140         # Create a book and author on the other database
1141         dive = Book.objects.using('other').create(title="Dive into Python",
1142                                                   published=datetime.date(2009, 5, 4))
1143
1144         mark = Person.objects.using('other').create(name="Mark Pilgrim")
1145
1146         # Set a foreign key with an object from a different database
1147         try:
1148             dive.editor = marty
1149         except ValueError:
1150             self.fail("Assignment across master/slave databases with a common source should be ok")
1151
1152         # Database assignments of original objects haven't changed...
1153         self.assertEqual(marty._state.db, 'default')
1154         self.assertEqual(pro._state.db, 'default')
1155         self.assertEqual(dive._state.db, 'other')
1156         self.assertEqual(mark._state.db, 'other')
1157
1158         # ... but they will when the affected object is saved.
1159         dive.save()
1160         self.assertEqual(dive._state.db, 'default')
1161
1162         # ...and the source database now has a copy of any object saved
1163         try:
1164             Book.objects.using('default').get(title='Dive into Python').delete()
1165         except Book.DoesNotExist:
1166             self.fail('Source database should have a copy of saved object')
1167
1168         # This isn't a real master-slave database, so restore the original from other
1169         dive = Book.objects.using('other').get(title='Dive into Python')
1170         self.assertEqual(dive._state.db, 'other')
1171
1172         # Set a foreign key set with an object from a different database
1173         try:
1174             marty.edited = [pro, dive]
1175         except ValueError:
1176             self.fail("Assignment across master/slave databases with a common source should be ok")
1177
1178         # Assignment implies a save, so database assignments of original objects have changed...
1179         self.assertEqual(marty._state.db, 'default')
1180         self.assertEqual(pro._state.db, 'default')
1181         self.assertEqual(dive._state.db, 'default')
1182         self.assertEqual(mark._state.db, 'other')
1183
1184         # ...and the source database now has a copy of any object saved
1185         try:
1186             Book.objects.using('default').get(title='Dive into Python').delete()
1187         except Book.DoesNotExist:
1188             self.fail('Source database should have a copy of saved object')
1189
1190         # This isn't a real master-slave database, so restore the original from other
1191         dive = Book.objects.using('other').get(title='Dive into Python')
1192         self.assertEqual(dive._state.db, 'other')
1193
1194         # Add to a foreign key set with an object from a different database
1195         try:
1196             marty.edited.add(dive)
1197         except ValueError:
1198             self.fail("Assignment across master/slave databases with a common source should be ok")
1199
1200         # Add implies a save, so database assignments of original objects have changed...
1201         self.assertEqual(marty._state.db, 'default')
1202         self.assertEqual(pro._state.db, 'default')
1203         self.assertEqual(dive._state.db, 'default')
1204         self.assertEqual(mark._state.db, 'other')
1205
1206         # ...and the source database now has a copy of any object saved
1207         try:
1208             Book.objects.using('default').get(title='Dive into Python').delete()
1209         except Book.DoesNotExist:
1210             self.fail('Source database should have a copy of saved object')
1211
1212         # This isn't a real master-slave database, so restore the original from other
1213         dive = Book.objects.using('other').get(title='Dive into Python')
1214
1215         # If you assign a FK object when the base object hasn't
1216         # been saved yet, you implicitly assign the database for the
1217         # base object.
1218         chris = Person(name="Chris Mills")
1219         html5 = Book(title="Dive into HTML5", published=datetime.date(2010, 3, 15))
1220         # initially, no db assigned
1221         self.assertEqual(chris._state.db, None)
1222         self.assertEqual(html5._state.db, None)
1223
1224         # old object comes from 'other', so the new object is set to use the
1225         # source of 'other'...
1226         self.assertEqual(dive._state.db, 'other')
1227         dive.editor = chris
1228         html5.editor = mark
1229
1230         self.assertEqual(dive._state.db, 'other')
1231         self.assertEqual(mark._state.db, 'other')
1232         self.assertEqual(chris._state.db, 'default')
1233         self.assertEqual(html5._state.db, 'default')
1234
1235         # This also works if you assign the FK in the constructor
1236         water = Book(title="Dive into Water", published=datetime.date(2001, 1, 1), editor=mark)
1237         self.assertEqual(water._state.db, 'default')
1238
1239         # For the remainder of this test, create a copy of 'mark' in the
1240         # 'default' database to prevent integrity errors on backends that
1241         # don't defer constraints checks until the end of the transaction
1242         mark.save(using='default')
1243
1244         # This moved 'mark' in the 'default' database, move it back in 'other'
1245         mark.save(using='other')
1246         self.assertEqual(mark._state.db, 'other')
1247
1248         # If you create an object through a FK relation, it will be
1249         # written to the write database, even if the original object
1250         # was on the read database
1251         cheesecake = mark.edited.create(title='Dive into Cheesecake', published=datetime.date(2010, 3, 15))
1252         self.assertEqual(cheesecake._state.db, 'default')
1253
1254         # Same goes for get_or_create, regardless of whether getting or creating
1255         cheesecake, created = mark.edited.get_or_create(title='Dive into Cheesecake', published=datetime.date(2010, 3, 15))
1256         self.assertEqual(cheesecake._state.db, 'default')
1257
1258         puddles, created = mark.edited.get_or_create(title='Dive into Puddles', published=datetime.date(2010, 3, 15))
1259         self.assertEqual(puddles._state.db, 'default')
1260
1261     def test_m2m_cross_database_protection(self):
1262         "M2M relations can cross databases if the database share a source"
1263         # Create books and authors on the inverse to the usual database
1264         pro = Book.objects.using('other').create(pk=1, title="Pro Django",
1265                                                  published=datetime.date(2008, 12, 16))
1266
1267         marty = Person.objects.using('other').create(pk=1, name="Marty Alchin")
1268
1269         dive = Book.objects.using('default').create(pk=2, title="Dive into Python",
1270                                                     published=datetime.date(2009, 5, 4))
1271
1272         mark = Person.objects.using('default').create(pk=2, name="Mark Pilgrim")
1273
1274         # Now save back onto the usual database.
1275         # This simulates master/slave - the objects exist on both database,
1276         # but the _state.db is as it is for all other tests.
1277         pro.save(using='default')
1278         marty.save(using='default')
1279         dive.save(using='other')
1280         mark.save(using='other')
1281
1282         # Check that we have 2 of both types of object on both databases
1283         self.assertEqual(Book.objects.using('default').count(), 2)
1284         self.assertEqual(Book.objects.using('other').count(), 2)
1285         self.assertEqual(Person.objects.using('default').count(), 2)
1286         self.assertEqual(Person.objects.using('other').count(), 2)
1287
1288         # Set a m2m set with an object from a different database
1289         try:
1290             marty.book_set = [pro, dive]
1291         except ValueError:
1292             self.fail("Assignment across master/slave databases with a common source should be ok")
1293
1294         # Database assignments don't change
1295         self.assertEqual(marty._state.db, 'default')
1296         self.assertEqual(pro._state.db, 'default')
1297         self.assertEqual(dive._state.db, 'other')
1298         self.assertEqual(mark._state.db, 'other')
1299
1300         # All m2m relations should be saved on the default database
1301         self.assertEqual(Book.authors.through.objects.using('default').count(), 2)
1302         self.assertEqual(Book.authors.through.objects.using('other').count(), 0)
1303
1304         # Reset relations
1305         Book.authors.through.objects.using('default').delete()
1306
1307         # Add to an m2m with an object from a different database
1308         try:
1309             marty.book_set.add(dive)
1310         except ValueError:
1311             self.fail("Assignment across master/slave databases with a common source should be ok")
1312
1313         # Database assignments don't change
1314         self.assertEqual(marty._state.db, 'default')
1315         self.assertEqual(pro._state.db, 'default')
1316         self.assertEqual(dive._state.db, 'other')
1317         self.assertEqual(mark._state.db, 'other')
1318
1319         # All m2m relations should be saved on the default database
1320         self.assertEqual(Book.authors.through.objects.using('default').count(), 1)
1321         self.assertEqual(Book.authors.through.objects.using('other').count(), 0)
1322
1323         # Reset relations
1324         Book.authors.through.objects.using('default').delete()
1325
1326         # Set a reverse m2m with an object from a different database
1327         try:
1328             dive.authors = [mark, marty]
1329         except ValueError:
1330             self.fail("Assignment across master/slave databases with a common source should be ok")
1331
1332         # Database assignments don't change
1333         self.assertEqual(marty._state.db, 'default')
1334         self.assertEqual(pro._state.db, 'default')
1335         self.assertEqual(dive._state.db, 'other')
1336         self.assertEqual(mark._state.db, 'other')
1337
1338         # All m2m relations should be saved on the default database
1339         self.assertEqual(Book.authors.through.objects.using('default').count(), 2)
1340         self.assertEqual(Book.authors.through.objects.using('other').count(), 0)
1341
1342         # Reset relations
1343         Book.authors.through.objects.using('default').delete()
1344
1345         self.assertEqual(Book.authors.through.objects.using('default').count(), 0)
1346         self.assertEqual(Book.authors.through.objects.using('other').count(), 0)
1347
1348         # Add to a reverse m2m with an object from a different database
1349         try:
1350             dive.authors.add(marty)
1351         except ValueError:
1352             self.fail("Assignment across master/slave databases with a common source should be ok")
1353
1354         # Database assignments don't change
1355         self.assertEqual(marty._state.db, 'default')
1356         self.assertEqual(pro._state.db, 'default')
1357         self.assertEqual(dive._state.db, 'other')
1358         self.assertEqual(mark._state.db, 'other')
1359
1360         # All m2m relations should be saved on the default database
1361         self.assertEqual(Book.authors.through.objects.using('default').count(), 1)
1362         self.assertEqual(Book.authors.through.objects.using('other').count(), 0)
1363
1364         # If you create an object through a M2M relation, it will be
1365         # written to the write database, even if the original object
1366         # was on the read database
1367         alice = dive.authors.create(name='Alice')
1368         self.assertEqual(alice._state.db, 'default')
1369
1370         # Same goes for get_or_create, regardless of whether getting or creating
1371         alice, created = dive.authors.get_or_create(name='Alice')
1372         self.assertEqual(alice._state.db, 'default')
1373
1374         bob, created = dive.authors.get_or_create(name='Bob')
1375         self.assertEqual(bob._state.db, 'default')
1376
1377     def test_o2o_cross_database_protection(self):
1378         "Operations that involve sharing FK objects across databases raise an error"
1379         # Create a user and profile on the default database
1380         alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com')
1381
1382         # Create a user and profile on the other database
1383         bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com')
1384
1385         # Set a one-to-one relation with an object from a different database
1386         alice_profile = UserProfile.objects.create(user=alice, flavor='chocolate')
1387         try:
1388             bob.userprofile = alice_profile
1389         except ValueError:
1390             self.fail("Assignment across master/slave databases with a common source should be ok")
1391
1392         # Database assignments of original objects haven't changed...
1393         self.assertEqual(alice._state.db, 'default')
1394         self.assertEqual(alice_profile._state.db, 'default')
1395         self.assertEqual(bob._state.db, 'other')
1396
1397         # ... but they will when the affected object is saved.
1398         bob.save()
1399         self.assertEqual(bob._state.db, 'default')
1400
1401     def test_generic_key_cross_database_protection(self):
1402         "Generic Key operations can span databases if they share a source"
1403         copy_content_types_from_default_to_other()
1404
1405         # Create a book and author on the default database
1406         pro = Book.objects.using('default'
1407                 ).create(title="Pro Django", published=datetime.date(2008, 12, 16))
1408
1409         review1 = Review.objects.using('default'
1410                     ).create(source="Python Monthly", content_object=pro)
1411
1412         # Create a book and author on the other database
1413         dive = Book.objects.using('other'
1414                 ).create(title="Dive into Python", published=datetime.date(2009, 5, 4))
1415
1416         review2 = Review.objects.using('other'
1417                     ).create(source="Python Weekly", content_object=dive)
1418
1419         # Set a generic foreign key with an object from a different database
1420         try:
1421             review1.content_object = dive
1422         except ValueError:
1423             self.fail("Assignment across master/slave databases with a common source should be ok")
1424
1425         # Database assignments of original objects haven't changed...
1426         self.assertEqual(pro._state.db, 'default')
1427         self.assertEqual(review1._state.db, 'default')
1428         self.assertEqual(dive._state.db, 'other')
1429         self.assertEqual(review2._state.db, 'other')
1430
1431         # ... but they will when the affected object is saved.
1432         dive.save()
1433         self.assertEqual(review1._state.db, 'default')
1434         self.assertEqual(dive._state.db, 'default')
1435
1436         # ...and the source database now has a copy of any object saved
1437         try:
1438             Book.objects.using('default').get(title='Dive into Python').delete()
1439         except Book.DoesNotExist:
1440             self.fail('Source database should have a copy of saved object')
1441
1442         # This isn't a real master-slave database, so restore the original from other
1443         dive = Book.objects.using('other').get(title='Dive into Python')
1444         self.assertEqual(dive._state.db, 'other')
1445
1446         # Add to a generic foreign key set with an object from a different database
1447         try:
1448             dive.reviews.add(review1)
1449         except ValueError:
1450             self.fail("Assignment across master/slave databases with a common source should be ok")
1451
1452         # Database assignments of original objects haven't changed...
1453         self.assertEqual(pro._state.db, 'default')
1454         self.assertEqual(review1._state.db, 'default')
1455         self.assertEqual(dive._state.db, 'other')
1456         self.assertEqual(review2._state.db, 'other')
1457
1458         # ... but they will when the affected object is saved.
1459         dive.save()
1460         self.assertEqual(dive._state.db, 'default')
1461
1462         # ...and the source database now has a copy of any object saved
1463         try:
1464             Book.objects.using('default').get(title='Dive into Python').delete()
1465         except Book.DoesNotExist:
1466             self.fail('Source database should have a copy of saved object')
1467
1468         # BUT! if you assign a FK object when the base object hasn't
1469         # been saved yet, you implicitly assign the database for the
1470         # base object.
1471         review3 = Review(source="Python Daily")
1472         # initially, no db assigned
1473         self.assertEqual(review3._state.db, None)
1474
1475         # Dive comes from 'other', so review3 is set to use the source of 'other'...
1476         review3.content_object = dive
1477         self.assertEqual(review3._state.db, 'default')
1478
1479         # If you create an object through a M2M relation, it will be
1480         # written to the write database, even if the original object
1481         # was on the read database
1482         dive = Book.objects.using('other').get(title='Dive into Python')
1483         nyt = dive.reviews.create(source="New York Times", content_object=dive)
1484         self.assertEqual(nyt._state.db, 'default')
1485
1486     def test_m2m_managers(self):
1487         "M2M relations are represented by managers, and can be controlled like managers"
1488         pro = Book.objects.using('other').create(pk=1, title="Pro Django",
1489                                                  published=datetime.date(2008, 12, 16))
1490
1491         marty = Person.objects.using('other').create(pk=1, name="Marty Alchin")
1492         pro_authors = pro.authors.using('other')
1493         authors = [marty]
1494
1495         self.assertEqual(pro.authors.db, 'other')
1496         self.assertEqual(pro.authors.db_manager('default').db, 'default')
1497         self.assertEqual(pro.authors.db_manager('default').all().db, 'default')
1498
1499         self.assertEqual(marty.book_set.db, 'other')
1500         self.assertEqual(marty.book_set.db_manager('default').db, 'default')
1501         self.assertEqual(marty.book_set.db_manager('default').all().db, 'default')
1502
1503     def test_foreign_key_managers(self):
1504         "FK reverse relations are represented by managers, and can be controlled like managers"
1505         marty = Person.objects.using('other').create(pk=1, name="Marty Alchin")
1506         pro = Book.objects.using('other').create(pk=1, title="Pro Django",
1507                                                  published=datetime.date(2008, 12, 16),
1508                                                  editor=marty)
1509
1510         self.assertEqual(marty.edited.db, 'other')
1511         self.assertEqual(marty.edited.db_manager('default').db, 'default')
1512         self.assertEqual(marty.edited.db_manager('default').all().db, 'default')
1513
1514     def test_generic_key_managers(self):
1515         "Generic key relations are represented by managers, and can be controlled like managers"
1516         copy_content_types_from_default_to_other()
1517
1518         pro = Book.objects.using('other').create(title="Pro Django",
1519                                                  published=datetime.date(2008, 12, 16))
1520
1521         review1 = Review.objects.using('other').create(source="Python Monthly",
1522                                                        content_object=pro)
1523
1524         self.assertEqual(pro.reviews.db, 'other')
1525         self.assertEqual(pro.reviews.db_manager('default').db, 'default')
1526         self.assertEqual(pro.reviews.db_manager('default').all().db, 'default')
1527
1528     def test_subquery(self):
1529         """Make sure as_sql works with subqueries and master/slave."""
1530         # Create a book and author on the other database
1531
1532         mark = Person.objects.using('other').create(name="Mark Pilgrim")
1533         dive = Book.objects.using('other').create(title="Dive into Python",
1534                                                   published=datetime.date(2009, 5, 4),
1535                                                   editor=mark)
1536
1537         sub = Person.objects.filter(name='Mark Pilgrim')
1538         qs = Book.objects.filter(editor__in=sub)
1539
1540         # When you call __str__ on the query object, it doesn't know about using
1541         # so it falls back to the default. Don't let routing instructions
1542         # force the subquery to an incompatible database.
1543         str(qs.query)
1544
1545         # If you evaluate the query, it should work, running on 'other'
1546         self.assertEqual(list(qs.values_list('title', flat=True)), [u'Dive into Python'])
1547
1548 class AuthTestCase(TestCase):
1549     multi_db = True
1550
1551     def setUp(self):
1552         # Make the 'other' database appear to be a slave of the 'default'
1553         self.old_routers = router.routers
1554         router.routers = [AuthRouter()]
1555
1556     def tearDown(self):
1557         # Restore the 'other' database as an independent database
1558         router.routers = self.old_routers
1559
1560     def test_auth_manager(self):
1561         "The methods on the auth manager obey database hints"
1562         # Create one user using default allocation policy
1563         User.objects.create_user('alice', 'alice@example.com')
1564
1565         # Create another user, explicitly specifying the database
1566         User.objects.db_manager('default').create_user('bob', 'bob@example.com')
1567
1568         # The second user only exists on the other database
1569         alice = User.objects.using('other').get(username='alice')
1570
1571         self.assertEqual(alice.username, 'alice')
1572         self.assertEqual(alice._state.db, 'other')
1573
1574         self.assertRaises(User.DoesNotExist, User.objects.using('default').get, username='alice')
1575
1576         # The second user only exists on the default database
1577         bob = User.objects.using('default').get(username='bob')
1578
1579         self.assertEqual(bob.username, 'bob')
1580         self.assertEqual(bob._state.db, 'default')
1581
1582         self.assertRaises(User.DoesNotExist, User.objects.using('other').get, username='bob')
1583
1584         # That is... there is one user on each database
1585         self.assertEqual(User.objects.using('default').count(), 1)
1586         self.assertEqual(User.objects.using('other').count(), 1)
1587
1588     def test_dumpdata(self):
1589         "Check that dumpdata honors allow_syncdb restrictions on the router"
1590         User.objects.create_user('alice', 'alice@example.com')
1591         User.objects.db_manager('default').create_user('bob', 'bob@example.com')
1592
1593         # Check that dumping the default database doesn't try to include auth
1594         # because allow_syncdb prohibits auth on default
1595         new_io = StringIO()
1596         management.call_command('dumpdata', 'auth', format='json', database='default', stdout=new_io)
1597         command_output = new_io.getvalue().strip()
1598         self.assertEqual(command_output, '[]')
1599
1600         # Check that dumping the other database does include auth
1601         new_io = StringIO()
1602         management.call_command('dumpdata', 'auth', format='json', database='other', stdout=new_io)
1603         command_output = new_io.getvalue().strip()
1604         self.assertTrue('"email": "alice@example.com",' in command_output)
1605
1606 _missing = object()
1607 class UserProfileTestCase(TestCase):
1608     def setUp(self):
1609         self.old_auth_profile_module = getattr(settings, 'AUTH_PROFILE_MODULE', _missing)
1610         settings.AUTH_PROFILE_MODULE = 'multiple_database.UserProfile'
1611
1612     def tearDown(self):
1613         if self.old_auth_profile_module is _missing:
1614             del settings.AUTH_PROFILE_MODULE
1615         else:
1616             settings.AUTH_PROFILE_MODULE = self.old_auth_profile_module
1617
1618     def test_user_profiles(self):
1619
1620         alice = User.objects.create_user('alice', 'alice@example.com')
1621         bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com')
1622
1623         alice_profile = UserProfile(user=alice, flavor='chocolate')
1624         alice_profile.save()
1625
1626         bob_profile = UserProfile(user=bob, flavor='crunchy frog')
1627         bob_profile.save()
1628
1629         self.assertEqual(alice.get_profile().flavor, 'chocolate')
1630         self.assertEqual(bob.get_profile().flavor, 'crunchy frog')
1631
1632 class AntiPetRouter(object):
1633     # A router that only expresses an opinion on syncdb,
1634     # passing pets to the 'other' database
1635
1636     def allow_syncdb(self, db, model):
1637         "Make sure the auth app only appears on the 'other' db"
1638         if db == 'other':
1639             return model._meta.object_name == 'Pet'
1640         else:
1641             return model._meta.object_name != 'Pet'
1642
1643 class FixtureTestCase(TestCase):
1644     multi_db = True
1645     fixtures = ['multidb-common', 'multidb']
1646
1647     def setUp(self):
1648         # Install the anti-pet router
1649         self.old_routers = router.routers
1650         router.routers = [AntiPetRouter()]
1651
1652     def tearDown(self):
1653         # Restore the 'other' database as an independent database
1654         router.routers = self.old_routers
1655
1656     def test_fixture_loading(self):
1657         "Multi-db fixtures are loaded correctly"
1658         # Check that "Pro Django" exists on the default database, but not on other database
1659         try:
1660             Book.objects.get(title="Pro Django")
1661             Book.objects.using('default').get(title="Pro Django")
1662         except Book.DoesNotExist:
1663             self.fail('"Pro Django" should exist on default database')
1664
1665         self.assertRaises(Book.DoesNotExist,
1666             Book.objects.using('other').get,
1667             title="Pro Django"
1668         )
1669
1670         # Check that "Dive into Python" exists on the default database, but not on other database
1671         try:
1672             Book.objects.using('other').get(title="Dive into Python")
1673         except Book.DoesNotExist:
1674             self.fail('"Dive into Python" should exist on other database')
1675
1676         self.assertRaises(Book.DoesNotExist,
1677             Book.objects.get,
1678             title="Dive into Python"
1679         )
1680         self.assertRaises(Book.DoesNotExist,
1681             Book.objects.using('default').get,
1682             title="Dive into Python"
1683         )
1684
1685         # Check that "Definitive Guide" exists on the both databases
1686         try:
1687             Book.objects.get(title="The Definitive Guide to Django")
1688             Book.objects.using('default').get(title="The Definitive Guide to Django")
1689             Book.objects.using('other').get(title="The Definitive Guide to Django")
1690         except Book.DoesNotExist:
1691             self.fail('"The Definitive Guide to Django" should exist on both databases')
1692
1693     def test_pseudo_empty_fixtures(self):
1694         "A fixture can contain entries, but lead to nothing in the database; this shouldn't raise an error (ref #14068)"
1695         new_io = StringIO()
1696         management.call_command('loaddata', 'pets', stdout=new_io, stderr=new_io)
1697         command_output = new_io.getvalue().strip()
1698         # No objects will actually be loaded
1699         self.assertEqual(command_output, "Installed 0 object(s) (of 2) from 1 fixture(s)")
1700
1701 class PickleQuerySetTestCase(TestCase):
1702     multi_db = True
1703
1704     def test_pickling(self):
1705         for db in connections:
1706             Book.objects.using(db).create(title='Dive into Python', published=datetime.date(2009, 5, 4))
1707             qs = Book.objects.all()
1708             self.assertEqual(qs.db, pickle.loads(pickle.dumps(qs)).db)
1709
1710
1711 class DatabaseReceiver(object):
1712     """
1713     Used in the tests for the database argument in signals (#13552)
1714     """
1715     def __call__(self, signal, sender, **kwargs):
1716         self._database = kwargs['using']
1717
1718 class WriteToOtherRouter(object):
1719     """
1720     A router that sends all writes to the other database.
1721     """
1722     def db_for_write(self, model, **hints):
1723         return "other"
1724
1725 class SignalTests(TestCase):
1726     multi_db = True
1727
1728     def setUp(self):
1729         self.old_routers = router.routers
1730
1731     def tearDown(self):
1732         router.routers = self.old_routers
1733
1734     def _write_to_other(self):
1735         "Sends all writes to 'other'."
1736         router.routers = [WriteToOtherRouter()]
1737
1738     def _write_to_default(self):
1739         "Sends all writes to the default DB"
1740         router.routers = self.old_routers
1741
1742     def test_database_arg_save_and_delete(self):
1743         """
1744         Tests that the pre/post_save signal contains the correct database.
1745         (#13552)
1746         """
1747         # Make some signal receivers
1748         pre_save_receiver = DatabaseReceiver()
1749         post_save_receiver = DatabaseReceiver()
1750         pre_delete_receiver = DatabaseReceiver()
1751         post_delete_receiver = DatabaseReceiver()
1752         # Make model and connect receivers
1753         signals.pre_save.connect(sender=Person, receiver=pre_save_receiver)
1754         signals.post_save.connect(sender=Person, receiver=post_save_receiver)
1755         signals.pre_delete.connect(sender=Person, receiver=pre_delete_receiver)
1756         signals.post_delete.connect(sender=Person, receiver=post_delete_receiver)
1757         p = Person.objects.create(name='Darth Vader')
1758         # Save and test receivers got calls
1759         p.save()
1760         self.assertEqual(pre_save_receiver._database, DEFAULT_DB_ALIAS)
1761         self.assertEqual(post_save_receiver._database, DEFAULT_DB_ALIAS)
1762         # Delete, and test
1763         p.delete()
1764         self.assertEqual(pre_delete_receiver._database, DEFAULT_DB_ALIAS)
1765         self.assertEqual(post_delete_receiver._database, DEFAULT_DB_ALIAS)
1766         # Save again to a different database
1767         p.save(using="other")
1768         self.assertEqual(pre_save_receiver._database, "other")
1769         self.assertEqual(post_save_receiver._database, "other")
1770         # Delete, and test
1771         p.delete(using="other")
1772         self.assertEqual(pre_delete_receiver._database, "other")
1773         self.assertEqual(post_delete_receiver._database, "other")
1774
1775     def test_database_arg_m2m(self):
1776         """
1777         Test that the m2m_changed signal has a correct database arg (#13552)
1778         """
1779         # Make a receiver
1780         receiver = DatabaseReceiver()
1781         # Connect it
1782         signals.m2m_changed.connect(receiver=receiver)
1783
1784         # Create the models that will be used for the tests
1785         b = Book.objects.create(title="Pro Django",
1786                                 published=datetime.date(2008, 12, 16))
1787         p = Person.objects.create(name="Marty Alchin")
1788
1789         # Create a copy of the models on the 'other' database to prevent
1790         # integrity errors on backends that don't defer constraints checks
1791         Book.objects.using('other').create(pk=b.pk, title=b.title,
1792                                            published=b.published)
1793         Person.objects.using('other').create(pk=p.pk, name=p.name)
1794
1795         # Test addition
1796         b.authors.add(p)
1797         self.assertEqual(receiver._database, DEFAULT_DB_ALIAS)
1798         self._write_to_other()
1799         b.authors.add(p)
1800         self._write_to_default()
1801         self.assertEqual(receiver._database, "other")
1802
1803         # Test removal
1804         b.authors.remove(p)
1805         self.assertEqual(receiver._database, DEFAULT_DB_ALIAS)
1806         self._write_to_other()
1807         b.authors.remove(p)
1808         self._write_to_default()
1809         self.assertEqual(receiver._database, "other")
1810
1811         # Test addition in reverse
1812         p.book_set.add(b)
1813         self.assertEqual(receiver._database, DEFAULT_DB_ALIAS)
1814         self._write_to_other()
1815         p.book_set.add(b)
1816         self._write_to_default()
1817         self.assertEqual(receiver._database, "other")
1818
1819         # Test clearing
1820         b.authors.clear()
1821         self.assertEqual(receiver._database, DEFAULT_DB_ALIAS)
1822         self._write_to_other()
1823         b.authors.clear()
1824         self._write_to_default()
1825         self.assertEqual(receiver._database, "other")
1826
1827 class AttributeErrorRouter(object):
1828     "A router to test the exception handling of ConnectionRouter"
1829     def db_for_read(self, model, **hints):
1830         raise AttributeError
1831
1832     def db_for_write(self, model, **hints):
1833         raise AttributeError
1834
1835 class RouterAttributeErrorTestCase(TestCase):
1836     multi_db = True
1837
1838     def setUp(self):
1839         self.old_routers = router.routers
1840         router.routers = [AttributeErrorRouter()]
1841
1842     def tearDown(self):
1843         router.routers = self.old_routers
1844
1845     def test_attribute_error_read(self):
1846         "Check that the AttributeError from AttributeErrorRouter bubbles up"
1847         router.routers = [] # Reset routers so we can save a Book instance
1848         b = Book.objects.create(title="Pro Django",
1849                                 published=datetime.date(2008, 12, 16))
1850         router.routers = [AttributeErrorRouter()] # Install our router
1851         self.assertRaises(AttributeError, Book.objects.get, pk=b.pk)
1852
1853     def test_attribute_error_save(self):
1854         "Check that the AttributeError from AttributeErrorRouter bubbles up"
1855         dive = Book()
1856         dive.title="Dive into Python"
1857         dive.published = datetime.date(2009, 5, 4)
1858         self.assertRaises(AttributeError, dive.save)
1859
1860     def test_attribute_error_delete(self):
1861         "Check that the AttributeError from AttributeErrorRouter bubbles up"
1862         router.routers = [] # Reset routers so we can save our Book, Person instances
1863         b = Book.objects.create(title="Pro Django",
1864                                 published=datetime.date(2008, 12, 16))
1865         p = Person.objects.create(name="Marty Alchin")
1866         b.authors = [p]
1867         b.editor = p
1868         router.routers = [AttributeErrorRouter()] # Install our router
1869         self.assertRaises(AttributeError, b.delete)
1870
1871     def test_attribute_error_m2m(self):
1872         "Check that the AttributeError from AttributeErrorRouter bubbles up"
1873         router.routers = [] # Reset routers so we can save our Book, Person instances
1874         b = Book.objects.create(title="Pro Django",
1875                                 published=datetime.date(2008, 12, 16))
1876         p = Person.objects.create(name="Marty Alchin")
1877         router.routers = [AttributeErrorRouter()] # Install our router
1878         self.assertRaises(AttributeError, setattr, b, 'authors', [p])
1879
1880 class ModelMetaRouter(object):
1881     "A router to ensure model arguments are real model classes"
1882     def db_for_write(self, model, **hints):
1883         if not hasattr(model, '_meta'):
1884             raise ValueError
1885
1886 class RouterModelArgumentTestCase(TestCase):
1887     multi_db = True
1888
1889     def setUp(self):
1890         self.old_routers = router.routers
1891         router.routers = [ModelMetaRouter()]
1892
1893     def tearDown(self):
1894         router.routers = self.old_routers
1895
1896     def test_m2m_collection(self):
1897         b = Book.objects.create(title="Pro Django",
1898                                 published=datetime.date(2008, 12, 16))
1899
1900         p = Person.objects.create(name="Marty Alchin")
1901         # test add
1902         b.authors.add(p)
1903         # test remove
1904         b.authors.remove(p)
1905         # test clear
1906         b.authors.clear()
1907         # test setattr
1908         b.authors = [p]
1909         # test M2M collection
1910         b.delete()
1911
1912     def test_foreignkey_collection(self):
1913         person = Person.objects.create(name='Bob')
1914         pet = Pet.objects.create(owner=person, name='Wart')
1915         # test related FK collection
1916         person.delete()