from __future__ import unicode_literals import datetime import pickle from operator import attrgetter from django.contrib.auth.models import User from django.contrib.contenttypes.models import ContentType from django.core import management from django.db import DEFAULT_DB_ALIAS, connections, router, transaction from django.db.models import signals from django.db.utils import ConnectionRouter from django.test import SimpleTestCase, TestCase, override_settings from django.utils.six import StringIO from .models import Book, Person, Pet, Review, UserProfile from .routers import AuthRouter, TestRouter, WriteRouter class QueryTestCase(TestCase): multi_db = True def test_db_selection(self): "Check that querysets will use the default database by default" self.assertEqual(Book.objects.db, DEFAULT_DB_ALIAS) self.assertEqual(Book.objects.all().db, DEFAULT_DB_ALIAS) self.assertEqual(Book.objects.using('other').db, 'other') self.assertEqual(Book.objects.db_manager('other').db, 'other') self.assertEqual(Book.objects.db_manager('other').all().db, 'other') def test_default_creation(self): "Objects created on the default database don't leak onto other databases" # Create a book on the default database using create() Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) # Create a book on the default database using a save dive = Book() dive.title = "Dive into Python" dive.published = datetime.date(2009, 5, 4) dive.save() # Check that book exists on the default database, but not on other database try: Book.objects.get(title="Pro Django") Book.objects.using('default').get(title="Pro Django") except Book.DoesNotExist: self.fail('"Pro Django" should exist on default database') self.assertRaises( Book.DoesNotExist, Book.objects.using('other').get, title="Pro Django" ) try: Book.objects.get(title="Dive into Python") Book.objects.using('default').get(title="Dive into Python") except Book.DoesNotExist: self.fail('"Dive into Python" should exist on default database') self.assertRaises( Book.DoesNotExist, Book.objects.using('other').get, title="Dive into Python" ) def test_other_creation(self): "Objects created on another database don't leak onto the default database" # Create a book on the second database Book.objects.using('other').create(title="Pro Django", published=datetime.date(2008, 12, 16)) # Create a book on the default database using a save dive = Book() dive.title = "Dive into Python" dive.published = datetime.date(2009, 5, 4) dive.save(using='other') # Check that book exists on the default database, but not on other database try: Book.objects.using('other').get(title="Pro Django") except Book.DoesNotExist: self.fail('"Pro Django" should exist on other database') self.assertRaises( Book.DoesNotExist, Book.objects.get, title="Pro Django" ) self.assertRaises( Book.DoesNotExist, Book.objects.using('default').get, title="Pro Django" ) try: Book.objects.using('other').get(title="Dive into Python") except Book.DoesNotExist: self.fail('"Dive into Python" should exist on other database') self.assertRaises( Book.DoesNotExist, Book.objects.get, title="Dive into Python" ) self.assertRaises( Book.DoesNotExist, Book.objects.using('default').get, title="Dive into Python" ) def test_refresh(self): dive = Book() dive.title = "Dive into Python" dive = Book() dive.title = "Dive into Python" dive.published = datetime.date(2009, 5, 4) dive.save(using='other') dive.published = datetime.date(2009, 5, 4) dive.save(using='other') dive2 = Book.objects.using('other').get() dive2.title = "Dive into Python (on default)" dive2.save(using='default') dive.refresh_from_db() self.assertEqual(dive.title, "Dive into Python") dive.refresh_from_db(using='default') self.assertEqual(dive.title, "Dive into Python (on default)") self.assertEqual(dive._state.db, "default") def test_basic_queries(self): "Queries are constrained to a single database" dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) dive = Book.objects.using('other').get(published=datetime.date(2009, 5, 4)) self.assertEqual(dive.title, "Dive into Python") self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, published=datetime.date(2009, 5, 4)) dive = Book.objects.using('other').get(title__icontains="dive") self.assertEqual(dive.title, "Dive into Python") self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, title__icontains="dive") dive = Book.objects.using('other').get(title__iexact="dive INTO python") self.assertEqual(dive.title, "Dive into Python") self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, title__iexact="dive INTO python") dive = Book.objects.using('other').get(published__year=2009) self.assertEqual(dive.title, "Dive into Python") self.assertEqual(dive.published, datetime.date(2009, 5, 4)) self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, published__year=2009) years = Book.objects.using('other').dates('published', 'year') self.assertEqual([o.year for o in years], [2009]) years = Book.objects.using('default').dates('published', 'year') self.assertEqual([o.year for o in years], []) months = Book.objects.using('other').dates('published', 'month') self.assertEqual([o.month for o in months], [5]) months = Book.objects.using('default').dates('published', 'month') self.assertEqual([o.month for o in months], []) def test_m2m_separation(self): "M2M fields are constrained to a single database" # Create a book and author on the default database pro = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) marty = Person.objects.create(name="Marty Alchin") # Create a book and author on the other database dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) mark = Person.objects.using('other').create(name="Mark Pilgrim") # Save the author relations pro.authors.set([marty]) dive.authors.set([mark]) # Inspect the m2m tables directly. # There should be 1 entry in each database self.assertEqual(Book.authors.through.objects.using('default').count(), 1) self.assertEqual(Book.authors.through.objects.using('other').count(), 1) # Check that queries work across m2m joins self.assertEqual( list(Book.objects.using('default').filter(authors__name='Marty Alchin').values_list('title', flat=True)), ['Pro Django'] ) self.assertEqual( list(Book.objects.using('other').filter(authors__name='Marty Alchin').values_list('title', flat=True)), [] ) self.assertEqual( list(Book.objects.using('default').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), [] ) self.assertEqual( list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), ['Dive into Python'] ) # Reget the objects to clear caches dive = Book.objects.using('other').get(title="Dive into Python") mark = Person.objects.using('other').get(name="Mark Pilgrim") # Retrieve related object by descriptor. Related objects should be database-bound self.assertEqual(list(dive.authors.all().values_list('name', flat=True)), ['Mark Pilgrim']) self.assertEqual(list(mark.book_set.all().values_list('title', flat=True)), ['Dive into Python']) def test_m2m_forward_operations(self): "M2M forward manipulations are all constrained to a single DB" # Create a book and author on the other database dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) mark = Person.objects.using('other').create(name="Mark Pilgrim") # Save the author relations dive.authors.set([mark]) # Add a second author john = Person.objects.using('other').create(name="John Smith") self.assertEqual( list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)), [] ) dive.authors.add(john) self.assertEqual( list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), ['Dive into Python'] ) self.assertEqual( list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)), ['Dive into Python'] ) # Remove the second author dive.authors.remove(john) self.assertEqual( list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), ['Dive into Python'] ) self.assertEqual( list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)), [] ) # Clear all authors dive.authors.clear() self.assertEqual( list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), [] ) self.assertEqual( list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)), [] ) # Create an author through the m2m interface dive.authors.create(name='Jane Brown') self.assertEqual( list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), [] ) self.assertEqual( list(Book.objects.using('other').filter(authors__name='Jane Brown').values_list('title', flat=True)), ['Dive into Python'] ) def test_m2m_reverse_operations(self): "M2M reverse manipulations are all constrained to a single DB" # Create a book and author on the other database dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) mark = Person.objects.using('other').create(name="Mark Pilgrim") # Save the author relations dive.authors.set([mark]) # Create a second book on the other database grease = Book.objects.using('other').create(title="Greasemonkey Hacks", published=datetime.date(2005, 11, 1)) # Add a books to the m2m mark.book_set.add(grease) self.assertEqual( list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)), ['Mark Pilgrim'] ) self.assertEqual( list( Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True) ), ['Mark Pilgrim'] ) # Remove a book from the m2m mark.book_set.remove(grease) self.assertEqual( list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)), ['Mark Pilgrim'] ) self.assertEqual( list( Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True) ), [] ) # Clear the books associated with mark mark.book_set.clear() self.assertEqual( list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)), [] ) self.assertEqual( list( Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True) ), [] ) # Create a book through the m2m interface mark.book_set.create(title="Dive into HTML5", published=datetime.date(2020, 1, 1)) self.assertEqual( list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)), [] ) self.assertEqual( list(Person.objects.using('other').filter(book__title='Dive into HTML5').values_list('name', flat=True)), ['Mark Pilgrim'] ) def test_m2m_cross_database_protection(self): "Operations that involve sharing M2M objects across databases raise an error" # Create a book and author on the default database pro = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) marty = Person.objects.create(name="Marty Alchin") # Create a book and author on the other database dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) mark = Person.objects.using('other').create(name="Mark Pilgrim") # Set a foreign key set with an object from a different database with self.assertRaises(ValueError): with transaction.atomic(using='default'): marty.edited.set([pro, dive]) # Add to an m2m with an object from a different database with self.assertRaises(ValueError): with transaction.atomic(using='default'): marty.book_set.add(dive) # Set a m2m with an object from a different database with self.assertRaises(ValueError): with transaction.atomic(using='default'): marty.book_set.set([pro, dive]) # Add to a reverse m2m with an object from a different database with self.assertRaises(ValueError): with transaction.atomic(using='other'): dive.authors.add(marty) # Set a reverse m2m with an object from a different database with self.assertRaises(ValueError): with transaction.atomic(using='other'): dive.authors.set([mark, marty]) def test_m2m_deletion(self): "Cascaded deletions of m2m relations issue queries on the right database" # Create a book and author on the other database dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) mark = Person.objects.using('other').create(name="Mark Pilgrim") dive.authors.set([mark]) # Check the initial state self.assertEqual(Person.objects.using('default').count(), 0) self.assertEqual(Book.objects.using('default').count(), 0) self.assertEqual(Book.authors.through.objects.using('default').count(), 0) self.assertEqual(Person.objects.using('other').count(), 1) self.assertEqual(Book.objects.using('other').count(), 1) self.assertEqual(Book.authors.through.objects.using('other').count(), 1) # Delete the object on the other database dive.delete(using='other') self.assertEqual(Person.objects.using('default').count(), 0) self.assertEqual(Book.objects.using('default').count(), 0) self.assertEqual(Book.authors.through.objects.using('default').count(), 0) # The person still exists ... self.assertEqual(Person.objects.using('other').count(), 1) # ... but the book has been deleted self.assertEqual(Book.objects.using('other').count(), 0) # ... and the relationship object has also been deleted. self.assertEqual(Book.authors.through.objects.using('other').count(), 0) # Now try deletion in the reverse direction. Set up the relation again dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) dive.authors.set([mark]) # Check the initial state self.assertEqual(Person.objects.using('default').count(), 0) self.assertEqual(Book.objects.using('default').count(), 0) self.assertEqual(Book.authors.through.objects.using('default').count(), 0) self.assertEqual(Person.objects.using('other').count(), 1) self.assertEqual(Book.objects.using('other').count(), 1) self.assertEqual(Book.authors.through.objects.using('other').count(), 1) # Delete the object on the other database mark.delete(using='other') self.assertEqual(Person.objects.using('default').count(), 0) self.assertEqual(Book.objects.using('default').count(), 0) self.assertEqual(Book.authors.through.objects.using('default').count(), 0) # The person has been deleted ... self.assertEqual(Person.objects.using('other').count(), 0) # ... but the book still exists self.assertEqual(Book.objects.using('other').count(), 1) # ... and the relationship object has been deleted. self.assertEqual(Book.authors.through.objects.using('other').count(), 0) def test_foreign_key_separation(self): "FK fields are constrained to a single database" # Create a book and author on the default database pro = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) george = Person.objects.create(name="George Vilches") # Create a book and author on the other database dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) chris = Person.objects.using('other').create(name="Chris Mills") # Save the author's favorite books pro.editor = george pro.save() dive.editor = chris dive.save() pro = Book.objects.using('default').get(title="Pro Django") self.assertEqual(pro.editor.name, "George Vilches") dive = Book.objects.using('other').get(title="Dive into Python") self.assertEqual(dive.editor.name, "Chris Mills") # Check that queries work across foreign key joins self.assertEqual( list(Person.objects.using('default').filter(edited__title='Pro Django').values_list('name', flat=True)), ['George Vilches'] ) self.assertEqual( list(Person.objects.using('other').filter(edited__title='Pro Django').values_list('name', flat=True)), [] ) self.assertEqual( list( Person.objects.using('default').filter(edited__title='Dive into Python').values_list('name', flat=True) ), [] ) self.assertEqual( list( Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True) ), ['Chris Mills'] ) # Reget the objects to clear caches chris = Person.objects.using('other').get(name="Chris Mills") dive = Book.objects.using('other').get(title="Dive into Python") # Retrieve related object by descriptor. Related objects should be database-bound self.assertEqual(list(chris.edited.values_list('title', flat=True)), ['Dive into Python']) def test_foreign_key_reverse_operations(self): "FK reverse manipulations are all constrained to a single DB" dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) chris = Person.objects.using('other').create(name="Chris Mills") # Save the author relations dive.editor = chris dive.save() # Add a second book edited by chris html5 = Book.objects.using('other').create(title="Dive into HTML5", published=datetime.date(2010, 3, 15)) self.assertEqual( list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)), [] ) chris.edited.add(html5) self.assertEqual( list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)), ['Chris Mills'] ) self.assertEqual( list( Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True) ), ['Chris Mills'] ) # Remove the second editor chris.edited.remove(html5) self.assertEqual( list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)), [] ) self.assertEqual( list( Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True) ), ['Chris Mills'] ) # Clear all edited books chris.edited.clear() self.assertEqual( list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)), [] ) self.assertEqual( list( Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True) ), [] ) # Create an author through the m2m interface chris.edited.create(title='Dive into Water', published=datetime.date(2010, 3, 15)) self.assertEqual( list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)), [] ) self.assertEqual( list(Person.objects.using('other').filter(edited__title='Dive into Water').values_list('name', flat=True)), ['Chris Mills'] ) self.assertEqual( list( Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True) ), [] ) def test_foreign_key_cross_database_protection(self): "Operations that involve sharing FK objects across databases raise an error" # Create a book and author on the default database pro = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) marty = Person.objects.create(name="Marty Alchin") # Create a book and author on the other database dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) # Set a foreign key with an object from a different database with self.assertRaises(ValueError): dive.editor = marty # Set a foreign key set with an object from a different database with self.assertRaises(ValueError): with transaction.atomic(using='default'): marty.edited.set([pro, dive]) # Add to a foreign key set with an object from a different database with self.assertRaises(ValueError): with transaction.atomic(using='default'): marty.edited.add(dive) def test_foreign_key_deletion(self): "Cascaded deletions of Foreign Key relations issue queries on the right database" mark = Person.objects.using('other').create(name="Mark Pilgrim") Pet.objects.using('other').create(name="Fido", owner=mark) # Check the initial state self.assertEqual(Person.objects.using('default').count(), 0) self.assertEqual(Pet.objects.using('default').count(), 0) self.assertEqual(Person.objects.using('other').count(), 1) self.assertEqual(Pet.objects.using('other').count(), 1) # Delete the person object, which will cascade onto the pet mark.delete(using='other') self.assertEqual(Person.objects.using('default').count(), 0) self.assertEqual(Pet.objects.using('default').count(), 0) # Both the pet and the person have been deleted from the right database self.assertEqual(Person.objects.using('other').count(), 0) self.assertEqual(Pet.objects.using('other').count(), 0) def test_foreign_key_validation(self): "ForeignKey.validate() uses the correct database" mickey = Person.objects.using('other').create(name="Mickey") pluto = Pet.objects.using('other').create(name="Pluto", owner=mickey) self.assertIsNone(pluto.full_clean()) def test_o2o_separation(self): "OneToOne fields are constrained to a single database" # Create a user and profile on the default database alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com') alice_profile = UserProfile.objects.using('default').create(user=alice, flavor='chocolate') # Create a user and profile on the other database bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com') bob_profile = UserProfile.objects.using('other').create(user=bob, flavor='crunchy frog') # Retrieve related objects; queries should be database constrained alice = User.objects.using('default').get(username="alice") self.assertEqual(alice.userprofile.flavor, "chocolate") bob = User.objects.using('other').get(username="bob") self.assertEqual(bob.userprofile.flavor, "crunchy frog") # Check that queries work across joins self.assertEqual( list( User.objects.using('default') .filter(userprofile__flavor='chocolate').values_list('username', flat=True) ), ['alice'] ) self.assertEqual( list( User.objects.using('other') .filter(userprofile__flavor='chocolate').values_list('username', flat=True) ), [] ) self.assertEqual( list( User.objects.using('default') .filter(userprofile__flavor='crunchy frog').values_list('username', flat=True) ), [] ) self.assertEqual( list( User.objects.using('other') .filter(userprofile__flavor='crunchy frog').values_list('username', flat=True) ), ['bob'] ) # Reget the objects to clear caches alice_profile = UserProfile.objects.using('default').get(flavor='chocolate') bob_profile = UserProfile.objects.using('other').get(flavor='crunchy frog') # Retrieve related object by descriptor. Related objects should be database-bound self.assertEqual(alice_profile.user.username, 'alice') self.assertEqual(bob_profile.user.username, 'bob') def test_o2o_cross_database_protection(self): "Operations that involve sharing FK objects across databases raise an error" # Create a user and profile on the default database alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com') # Create a user and profile on the other database bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com') # Set a one-to-one relation with an object from a different database alice_profile = UserProfile.objects.using('default').create(user=alice, flavor='chocolate') with self.assertRaises(ValueError): bob.userprofile = alice_profile # BUT! if you assign a FK object when the base object hasn't # been saved yet, you implicitly assign the database for the # base object. bob_profile = UserProfile.objects.using('other').create(user=bob, flavor='crunchy frog') new_bob_profile = UserProfile(flavor="spring surprise") # assigning a profile requires an explicit pk as the object isn't saved charlie = User(pk=51, username='charlie', email='charlie@example.com') charlie.set_unusable_password() # initially, no db assigned self.assertEqual(new_bob_profile._state.db, None) self.assertEqual(charlie._state.db, None) # old object comes from 'other', so the new object is set to use 'other'... new_bob_profile.user = bob charlie.userprofile = bob_profile self.assertEqual(new_bob_profile._state.db, 'other') self.assertEqual(charlie._state.db, 'other') # ... but it isn't saved yet self.assertEqual(list(User.objects.using('other').values_list('username', flat=True)), ['bob']) self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor', flat=True)), ['crunchy frog']) # When saved (no using required), new objects goes to 'other' charlie.save() bob_profile.save() new_bob_profile.save() self.assertEqual(list(User.objects.using('default').values_list('username', flat=True)), ['alice']) self.assertEqual(list(User.objects.using('other').values_list('username', flat=True)), ['bob', 'charlie']) self.assertEqual(list(UserProfile.objects.using('default').values_list('flavor', flat=True)), ['chocolate']) self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor', flat=True)), ['crunchy frog', 'spring surprise']) # This also works if you assign the O2O relation in the constructor denise = User.objects.db_manager('other').create_user('denise', 'denise@example.com') denise_profile = UserProfile(flavor="tofu", user=denise) self.assertEqual(denise_profile._state.db, 'other') # ... but it isn't saved yet self.assertEqual(list(UserProfile.objects.using('default').values_list('flavor', flat=True)), ['chocolate']) self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor', flat=True)), ['crunchy frog', 'spring surprise']) # When saved, the new profile goes to 'other' denise_profile.save() self.assertEqual(list(UserProfile.objects.using('default').values_list('flavor', flat=True)), ['chocolate']) self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor', flat=True)), ['crunchy frog', 'spring surprise', 'tofu']) def test_generic_key_separation(self): "Generic fields are constrained to a single database" # Create a book and author on the default database pro = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) review1 = Review.objects.create(source="Python Monthly", content_object=pro) # Create a book and author on the other database dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) review2 = Review.objects.using('other').create(source="Python Weekly", content_object=dive) review1 = Review.objects.using('default').get(source="Python Monthly") self.assertEqual(review1.content_object.title, "Pro Django") review2 = Review.objects.using('other').get(source="Python Weekly") self.assertEqual(review2.content_object.title, "Dive into Python") # Reget the objects to clear caches dive = Book.objects.using('other').get(title="Dive into Python") # Retrieve related object by descriptor. Related objects should be database-bound self.assertEqual(list(dive.reviews.all().values_list('source', flat=True)), ['Python Weekly']) def test_generic_key_reverse_operations(self): "Generic reverse manipulations are all constrained to a single DB" dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) temp = Book.objects.using('other').create(title="Temp", published=datetime.date(2009, 5, 4)) review1 = Review.objects.using('other').create(source="Python Weekly", content_object=dive) review2 = Review.objects.using('other').create(source="Python Monthly", content_object=temp) self.assertEqual( list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), [] ) self.assertEqual( list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), ['Python Weekly'] ) # Add a second review dive.reviews.add(review2) self.assertEqual( list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), [] ) self.assertEqual( list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), ['Python Monthly', 'Python Weekly'] ) # Remove the second author dive.reviews.remove(review1) self.assertEqual( list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), [] ) self.assertEqual( list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), ['Python Monthly'] ) # Clear all reviews dive.reviews.clear() self.assertEqual( list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), [] ) self.assertEqual( list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), [] ) # Create an author through the generic interface dive.reviews.create(source='Python Daily') self.assertEqual( list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), [] ) self.assertEqual( list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), ['Python Daily'] ) def test_generic_key_cross_database_protection(self): "Operations that involve sharing generic key objects across databases raise an error" # Create a book and author on the default database pro = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) review1 = Review.objects.create(source="Python Monthly", content_object=pro) # Create a book and author on the other database dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) Review.objects.using('other').create(source="Python Weekly", content_object=dive) # Set a foreign key with an object from a different database with self.assertRaises(ValueError): review1.content_object = dive # Add to a foreign key set with an object from a different database with self.assertRaises(ValueError): with transaction.atomic(using='other'): dive.reviews.add(review1) # BUT! if you assign a FK object when the base object hasn't # been saved yet, you implicitly assign the database for the # base object. review3 = Review(source="Python Daily") # initially, no db assigned self.assertEqual(review3._state.db, None) # Dive comes from 'other', so review3 is set to use 'other'... review3.content_object = dive self.assertEqual(review3._state.db, 'other') # ... but it isn't saved yet self.assertEqual( list(Review.objects.using('default').filter(object_id=pro.pk).values_list('source', flat=True)), ['Python Monthly'] ) self.assertEqual( list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), ['Python Weekly'] ) # When saved, John goes to 'other' review3.save() self.assertEqual( list(Review.objects.using('default').filter(object_id=pro.pk).values_list('source', flat=True)), ['Python Monthly'] ) self.assertEqual( list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), ['Python Daily', 'Python Weekly'] ) def test_generic_key_deletion(self): "Cascaded deletions of Generic Key relations issue queries on the right database" dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) Review.objects.using('other').create(source="Python Weekly", content_object=dive) # Check the initial state self.assertEqual(Book.objects.using('default').count(), 0) self.assertEqual(Review.objects.using('default').count(), 0) self.assertEqual(Book.objects.using('other').count(), 1) self.assertEqual(Review.objects.using('other').count(), 1) # Delete the Book object, which will cascade onto the pet dive.delete(using='other') self.assertEqual(Book.objects.using('default').count(), 0) self.assertEqual(Review.objects.using('default').count(), 0) # Both the pet and the person have been deleted from the right database self.assertEqual(Book.objects.using('other').count(), 0) self.assertEqual(Review.objects.using('other').count(), 0) def test_ordering(self): "get_next_by_XXX commands stick to a single database" Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) learn = Book.objects.using('other').create(title="Learning Python", published=datetime.date(2008, 7, 16)) self.assertEqual(learn.get_next_by_published().title, "Dive into Python") self.assertEqual(dive.get_previous_by_published().title, "Learning Python") def test_raw(self): "test the raw() method across databases" dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) val = Book.objects.db_manager("other").raw('SELECT id FROM multiple_database_book') self.assertQuerysetEqual(val, [dive.pk], attrgetter("pk")) val = Book.objects.raw('SELECT id FROM multiple_database_book').using('other') self.assertQuerysetEqual(val, [dive.pk], attrgetter("pk")) def test_select_related(self): "Database assignment is retained if an object is retrieved with select_related()" # Create a book and author on the other database mark = Person.objects.using('other').create(name="Mark Pilgrim") Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4), editor=mark) # Retrieve the Person using select_related() book = Book.objects.using('other').select_related('editor').get(title="Dive into Python") # The editor instance should have a db state self.assertEqual(book.editor._state.db, 'other') def test_subquery(self): """Make sure as_sql works with subqueries and primary/replica.""" sub = Person.objects.using('other').filter(name='fff') qs = Book.objects.filter(editor__in=sub) # When you call __str__ on the query object, it doesn't know about using # so it falls back to the default. If the subquery explicitly uses a # different database, an error should be raised. self.assertRaises(ValueError, str, qs.query) # Evaluating the query shouldn't work, either with self.assertRaises(ValueError): for obj in qs: pass def test_related_manager(self): "Related managers return managers, not querysets" mark = Person.objects.using('other').create(name="Mark Pilgrim") # extra_arg is removed by the BookManager's implementation of # create(); but the BookManager's implementation won't get called # unless edited returns a Manager, not a queryset mark.book_set.create(title="Dive into Python", published=datetime.date(2009, 5, 4), extra_arg=True) mark.book_set.get_or_create(title="Dive into Python", published=datetime.date(2009, 5, 4), extra_arg=True) mark.edited.create(title="Dive into Water", published=datetime.date(2009, 5, 4), extra_arg=True) mark.edited.get_or_create(title="Dive into Water", published=datetime.date(2009, 5, 4), extra_arg=True) class ConnectionRouterTestCase(SimpleTestCase): @override_settings(DATABASE_ROUTERS=[ 'multiple_database.tests.TestRouter', 'multiple_database.tests.WriteRouter']) def test_router_init_default(self): connection_router = ConnectionRouter() self.assertListEqual([r.__class__.__name__ for r in connection_router.routers], ['TestRouter', 'WriteRouter']) def test_router_init_arg(self): connection_router = ConnectionRouter([ 'multiple_database.tests.TestRouter', 'multiple_database.tests.WriteRouter' ]) self.assertListEqual([r.__class__.__name__ for r in connection_router.routers], ['TestRouter', 'WriteRouter']) # Init with instances instead of strings connection_router = ConnectionRouter([TestRouter(), WriteRouter()]) self.assertListEqual([r.__class__.__name__ for r in connection_router.routers], ['TestRouter', 'WriteRouter']) # Make the 'other' database appear to be a replica of the 'default' @override_settings(DATABASE_ROUTERS=[TestRouter()]) class RouterTestCase(TestCase): multi_db = True def test_db_selection(self): "Check that querysets obey the router for db suggestions" self.assertEqual(Book.objects.db, 'other') self.assertEqual(Book.objects.all().db, 'other') self.assertEqual(Book.objects.using('default').db, 'default') self.assertEqual(Book.objects.db_manager('default').db, 'default') self.assertEqual(Book.objects.db_manager('default').all().db, 'default') def test_migrate_selection(self): "Synchronization behavior is predictable" self.assertTrue(router.allow_migrate_model('default', User)) self.assertTrue(router.allow_migrate_model('default', Book)) self.assertTrue(router.allow_migrate_model('other', User)) self.assertTrue(router.allow_migrate_model('other', Book)) with override_settings(DATABASE_ROUTERS=[TestRouter(), AuthRouter()]): # Add the auth router to the chain. TestRouter is a universal # synchronizer, so it should have no effect. self.assertTrue(router.allow_migrate_model('default', User)) self.assertTrue(router.allow_migrate_model('default', Book)) self.assertTrue(router.allow_migrate_model('other', User)) self.assertTrue(router.allow_migrate_model('other', Book)) with override_settings(DATABASE_ROUTERS=[AuthRouter(), TestRouter()]): # Now check what happens if the router order is reversed. self.assertFalse(router.allow_migrate_model('default', User)) self.assertTrue(router.allow_migrate_model('default', Book)) self.assertTrue(router.allow_migrate_model('other', User)) self.assertTrue(router.allow_migrate_model('other', Book)) def test_partial_router(self): "A router can choose to implement a subset of methods" dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) # First check the baseline behavior. self.assertEqual(router.db_for_read(User), 'other') self.assertEqual(router.db_for_read(Book), 'other') self.assertEqual(router.db_for_write(User), 'default') self.assertEqual(router.db_for_write(Book), 'default') self.assertTrue(router.allow_relation(dive, dive)) self.assertTrue(router.allow_migrate_model('default', User)) self.assertTrue(router.allow_migrate_model('default', Book)) with override_settings(DATABASE_ROUTERS=[WriteRouter(), AuthRouter(), TestRouter()]): self.assertEqual(router.db_for_read(User), 'default') self.assertEqual(router.db_for_read(Book), 'other') self.assertEqual(router.db_for_write(User), 'writer') self.assertEqual(router.db_for_write(Book), 'writer') self.assertTrue(router.allow_relation(dive, dive)) self.assertFalse(router.allow_migrate_model('default', User)) self.assertTrue(router.allow_migrate_model('default', Book)) def test_database_routing(self): marty = Person.objects.using('default').create(name="Marty Alchin") pro = Book.objects.using('default').create(title="Pro Django", published=datetime.date(2008, 12, 16), editor=marty) pro.authors.set([marty]) # Create a book and author on the other database Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) # An update query will be routed to the default database Book.objects.filter(title='Pro Django').update(pages=200) with self.assertRaises(Book.DoesNotExist): # By default, the get query will be directed to 'other' Book.objects.get(title='Pro Django') # But the same query issued explicitly at a database will work. pro = Book.objects.using('default').get(title='Pro Django') # Check that the update worked. self.assertEqual(pro.pages, 200) # An update query with an explicit using clause will be routed # to the requested database. Book.objects.using('other').filter(title='Dive into Python').update(pages=300) self.assertEqual(Book.objects.get(title='Dive into Python').pages, 300) # Related object queries stick to the same database # as the original object, regardless of the router self.assertEqual(list(pro.authors.values_list('name', flat=True)), ['Marty Alchin']) self.assertEqual(pro.editor.name, 'Marty Alchin') # get_or_create is a special case. The get needs to be targeted at # the write database in order to avoid potential transaction # consistency problems book, created = Book.objects.get_or_create(title="Pro Django") self.assertFalse(created) book, created = Book.objects.get_or_create(title="Dive Into Python", defaults={'published': datetime.date(2009, 5, 4)}) self.assertTrue(created) # Check the head count of objects self.assertEqual(Book.objects.using('default').count(), 2) self.assertEqual(Book.objects.using('other').count(), 1) # If a database isn't specified, the read database is used self.assertEqual(Book.objects.count(), 1) # A delete query will also be routed to the default database Book.objects.filter(pages__gt=150).delete() # The default database has lost the book. self.assertEqual(Book.objects.using('default').count(), 1) self.assertEqual(Book.objects.using('other').count(), 1) def test_invalid_set_foreign_key_assignment(self): marty = Person.objects.using('default').create(name="Marty Alchin") dive = Book.objects.using('other').create( title="Dive into Python", published=datetime.date(2009, 5, 4), ) # Set a foreign key set with an object from a different database msg = " instance isn't saved. Use bulk=False or save the object first." with self.assertRaisesMessage(ValueError, msg): marty.edited.set([dive]) def test_foreign_key_cross_database_protection(self): "Foreign keys can cross databases if they two databases have a common source" # Create a book and author on the default database pro = Book.objects.using('default').create(title="Pro Django", published=datetime.date(2008, 12, 16)) marty = Person.objects.using('default').create(name="Marty Alchin") # Create a book and author on the other database dive = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) mark = Person.objects.using('other').create(name="Mark Pilgrim") # Set a foreign key with an object from a different database try: dive.editor = marty except ValueError: self.fail("Assignment across primary/replica databases with a common source should be ok") # Database assignments of original objects haven't changed... self.assertEqual(marty._state.db, 'default') self.assertEqual(pro._state.db, 'default') self.assertEqual(dive._state.db, 'other') self.assertEqual(mark._state.db, 'other') # ... but they will when the affected object is saved. dive.save() self.assertEqual(dive._state.db, 'default') # ...and the source database now has a copy of any object saved try: Book.objects.using('default').get(title='Dive into Python').delete() except Book.DoesNotExist: self.fail('Source database should have a copy of saved object') # This isn't a real primary/replica database, so restore the original from other dive = Book.objects.using('other').get(title='Dive into Python') self.assertEqual(dive._state.db, 'other') # Set a foreign key set with an object from a different database try: marty.edited.set([pro, dive], bulk=False) except ValueError: self.fail("Assignment across primary/replica databases with a common source should be ok") # Assignment implies a save, so database assignments of original objects have changed... self.assertEqual(marty._state.db, 'default') self.assertEqual(pro._state.db, 'default') self.assertEqual(dive._state.db, 'default') self.assertEqual(mark._state.db, 'other') # ...and the source database now has a copy of any object saved try: Book.objects.using('default').get(title='Dive into Python').delete() except Book.DoesNotExist: self.fail('Source database should have a copy of saved object') # This isn't a real primary/replica database, so restore the original from other dive = Book.objects.using('other').get(title='Dive into Python') self.assertEqual(dive._state.db, 'other') # Add to a foreign key set with an object from a different database try: marty.edited.add(dive, bulk=False) except ValueError: self.fail("Assignment across primary/replica databases with a common source should be ok") # Add implies a save, so database assignments of original objects have changed... self.assertEqual(marty._state.db, 'default') self.assertEqual(pro._state.db, 'default') self.assertEqual(dive._state.db, 'default') self.assertEqual(mark._state.db, 'other') # ...and the source database now has a copy of any object saved try: Book.objects.using('default').get(title='Dive into Python').delete() except Book.DoesNotExist: self.fail('Source database should have a copy of saved object') # This isn't a real primary/replica database, so restore the original from other dive = Book.objects.using('other').get(title='Dive into Python') # If you assign a FK object when the base object hasn't # been saved yet, you implicitly assign the database for the # base object. chris = Person(name="Chris Mills") html5 = Book(title="Dive into HTML5", published=datetime.date(2010, 3, 15)) # initially, no db assigned self.assertEqual(chris._state.db, None) self.assertEqual(html5._state.db, None) # old object comes from 'other', so the new object is set to use the # source of 'other'... self.assertEqual(dive._state.db, 'other') chris.save() dive.editor = chris html5.editor = mark self.assertEqual(dive._state.db, 'other') self.assertEqual(mark._state.db, 'other') self.assertEqual(chris._state.db, 'default') self.assertEqual(html5._state.db, 'default') # This also works if you assign the FK in the constructor water = Book(title="Dive into Water", published=datetime.date(2001, 1, 1), editor=mark) self.assertEqual(water._state.db, 'default') # For the remainder of this test, create a copy of 'mark' in the # 'default' database to prevent integrity errors on backends that # don't defer constraints checks until the end of the transaction mark.save(using='default') # This moved 'mark' in the 'default' database, move it back in 'other' mark.save(using='other') self.assertEqual(mark._state.db, 'other') # If you create an object through a FK relation, it will be # written to the write database, even if the original object # was on the read database cheesecake = mark.edited.create(title='Dive into Cheesecake', published=datetime.date(2010, 3, 15)) self.assertEqual(cheesecake._state.db, 'default') # Same goes for get_or_create, regardless of whether getting or creating cheesecake, created = mark.edited.get_or_create( title='Dive into Cheesecake', published=datetime.date(2010, 3, 15), ) self.assertEqual(cheesecake._state.db, 'default') puddles, created = mark.edited.get_or_create(title='Dive into Puddles', published=datetime.date(2010, 3, 15)) self.assertEqual(puddles._state.db, 'default') def test_m2m_cross_database_protection(self): "M2M relations can cross databases if the database share a source" # Create books and authors on the inverse to the usual database pro = Book.objects.using('other').create(pk=1, title="Pro Django", published=datetime.date(2008, 12, 16)) marty = Person.objects.using('other').create(pk=1, name="Marty Alchin") dive = Book.objects.using('default').create(pk=2, title="Dive into Python", published=datetime.date(2009, 5, 4)) mark = Person.objects.using('default').create(pk=2, name="Mark Pilgrim") # Now save back onto the usual database. # This simulates primary/replica - the objects exist on both database, # but the _state.db is as it is for all other tests. pro.save(using='default') marty.save(using='default') dive.save(using='other') mark.save(using='other') # Check that we have 2 of both types of object on both databases self.assertEqual(Book.objects.using('default').count(), 2) self.assertEqual(Book.objects.using('other').count(), 2) self.assertEqual(Person.objects.using('default').count(), 2) self.assertEqual(Person.objects.using('other').count(), 2) # Set a m2m set with an object from a different database try: marty.book_set.set([pro, dive]) except ValueError: self.fail("Assignment across primary/replica databases with a common source should be ok") # Database assignments don't change self.assertEqual(marty._state.db, 'default') self.assertEqual(pro._state.db, 'default') self.assertEqual(dive._state.db, 'other') self.assertEqual(mark._state.db, 'other') # All m2m relations should be saved on the default database self.assertEqual(Book.authors.through.objects.using('default').count(), 2) self.assertEqual(Book.authors.through.objects.using('other').count(), 0) # Reset relations Book.authors.through.objects.using('default').delete() # Add to an m2m with an object from a different database try: marty.book_set.add(dive) except ValueError: self.fail("Assignment across primary/replica databases with a common source should be ok") # Database assignments don't change self.assertEqual(marty._state.db, 'default') self.assertEqual(pro._state.db, 'default') self.assertEqual(dive._state.db, 'other') self.assertEqual(mark._state.db, 'other') # All m2m relations should be saved on the default database self.assertEqual(Book.authors.through.objects.using('default').count(), 1) self.assertEqual(Book.authors.through.objects.using('other').count(), 0) # Reset relations Book.authors.through.objects.using('default').delete() # Set a reverse m2m with an object from a different database try: dive.authors.set([mark, marty]) except ValueError: self.fail("Assignment across primary/replica databases with a common source should be ok") # Database assignments don't change self.assertEqual(marty._state.db, 'default') self.assertEqual(pro._state.db, 'default') self.assertEqual(dive._state.db, 'other') self.assertEqual(mark._state.db, 'other') # All m2m relations should be saved on the default database self.assertEqual(Book.authors.through.objects.using('default').count(), 2) self.assertEqual(Book.authors.through.objects.using('other').count(), 0) # Reset relations Book.authors.through.objects.using('default').delete() self.assertEqual(Book.authors.through.objects.using('default').count(), 0) self.assertEqual(Book.authors.through.objects.using('other').count(), 0) # Add to a reverse m2m with an object from a different database try: dive.authors.add(marty) except ValueError: self.fail("Assignment across primary/replica databases with a common source should be ok") # Database assignments don't change self.assertEqual(marty._state.db, 'default') self.assertEqual(pro._state.db, 'default') self.assertEqual(dive._state.db, 'other') self.assertEqual(mark._state.db, 'other') # All m2m relations should be saved on the default database self.assertEqual(Book.authors.through.objects.using('default').count(), 1) self.assertEqual(Book.authors.through.objects.using('other').count(), 0) # If you create an object through a M2M relation, it will be # written to the write database, even if the original object # was on the read database alice = dive.authors.create(name='Alice') self.assertEqual(alice._state.db, 'default') # Same goes for get_or_create, regardless of whether getting or creating alice, created = dive.authors.get_or_create(name='Alice') self.assertEqual(alice._state.db, 'default') bob, created = dive.authors.get_or_create(name='Bob') self.assertEqual(bob._state.db, 'default') def test_o2o_cross_database_protection(self): "Operations that involve sharing FK objects across databases raise an error" # Create a user and profile on the default database alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com') # Create a user and profile on the other database bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com') # Set a one-to-one relation with an object from a different database alice_profile = UserProfile.objects.create(user=alice, flavor='chocolate') try: bob.userprofile = alice_profile except ValueError: self.fail("Assignment across primary/replica databases with a common source should be ok") # Database assignments of original objects haven't changed... self.assertEqual(alice._state.db, 'default') self.assertEqual(alice_profile._state.db, 'default') self.assertEqual(bob._state.db, 'other') # ... but they will when the affected object is saved. bob.save() self.assertEqual(bob._state.db, 'default') def test_generic_key_cross_database_protection(self): "Generic Key operations can span databases if they share a source" # Create a book and author on the default database pro = Book.objects.using( 'default').create(title="Pro Django", published=datetime.date(2008, 12, 16)) review1 = Review.objects.using( 'default').create(source="Python Monthly", content_object=pro) # Create a book and author on the other database dive = Book.objects.using( 'other').create(title="Dive into Python", published=datetime.date(2009, 5, 4)) review2 = Review.objects.using( 'other').create(source="Python Weekly", content_object=dive) # Set a generic foreign key with an object from a different database try: review1.content_object = dive except ValueError: self.fail("Assignment across primary/replica databases with a common source should be ok") # Database assignments of original objects haven't changed... self.assertEqual(pro._state.db, 'default') self.assertEqual(review1._state.db, 'default') self.assertEqual(dive._state.db, 'other') self.assertEqual(review2._state.db, 'other') # ... but they will when the affected object is saved. dive.save() self.assertEqual(review1._state.db, 'default') self.assertEqual(dive._state.db, 'default') # ...and the source database now has a copy of any object saved try: Book.objects.using('default').get(title='Dive into Python').delete() except Book.DoesNotExist: self.fail('Source database should have a copy of saved object') # This isn't a real primary/replica database, so restore the original from other dive = Book.objects.using('other').get(title='Dive into Python') self.assertEqual(dive._state.db, 'other') # Add to a generic foreign key set with an object from a different database try: dive.reviews.add(review1) except ValueError: self.fail("Assignment across primary/replica databases with a common source should be ok") # Database assignments of original objects haven't changed... self.assertEqual(pro._state.db, 'default') self.assertEqual(review1._state.db, 'default') self.assertEqual(dive._state.db, 'other') self.assertEqual(review2._state.db, 'other') # ... but they will when the affected object is saved. dive.save() self.assertEqual(dive._state.db, 'default') # ...and the source database now has a copy of any object saved try: Book.objects.using('default').get(title='Dive into Python').delete() except Book.DoesNotExist: self.fail('Source database should have a copy of saved object') # BUT! if you assign a FK object when the base object hasn't # been saved yet, you implicitly assign the database for the # base object. review3 = Review(source="Python Daily") # initially, no db assigned self.assertEqual(review3._state.db, None) # Dive comes from 'other', so review3 is set to use the source of 'other'... review3.content_object = dive self.assertEqual(review3._state.db, 'default') # If you create an object through a M2M relation, it will be # written to the write database, even if the original object # was on the read database dive = Book.objects.using('other').get(title='Dive into Python') nyt = dive.reviews.create(source="New York Times", content_object=dive) self.assertEqual(nyt._state.db, 'default') def test_m2m_managers(self): "M2M relations are represented by managers, and can be controlled like managers" pro = Book.objects.using('other').create(pk=1, title="Pro Django", published=datetime.date(2008, 12, 16)) marty = Person.objects.using('other').create(pk=1, name="Marty Alchin") self.assertEqual(pro.authors.db, 'other') self.assertEqual(pro.authors.db_manager('default').db, 'default') self.assertEqual(pro.authors.db_manager('default').all().db, 'default') self.assertEqual(marty.book_set.db, 'other') self.assertEqual(marty.book_set.db_manager('default').db, 'default') self.assertEqual(marty.book_set.db_manager('default').all().db, 'default') def test_foreign_key_managers(self): "FK reverse relations are represented by managers, and can be controlled like managers" marty = Person.objects.using('other').create(pk=1, name="Marty Alchin") Book.objects.using('other').create(pk=1, title="Pro Django", published=datetime.date(2008, 12, 16), editor=marty) self.assertEqual(marty.edited.db, 'other') self.assertEqual(marty.edited.db_manager('default').db, 'default') self.assertEqual(marty.edited.db_manager('default').all().db, 'default') def test_generic_key_managers(self): "Generic key relations are represented by managers, and can be controlled like managers" pro = Book.objects.using('other').create(title="Pro Django", published=datetime.date(2008, 12, 16)) Review.objects.using('other').create(source="Python Monthly", content_object=pro) self.assertEqual(pro.reviews.db, 'other') self.assertEqual(pro.reviews.db_manager('default').db, 'default') self.assertEqual(pro.reviews.db_manager('default').all().db, 'default') def test_subquery(self): """Make sure as_sql works with subqueries and primary/replica.""" # Create a book and author on the other database mark = Person.objects.using('other').create(name="Mark Pilgrim") Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4), editor=mark) sub = Person.objects.filter(name='Mark Pilgrim') qs = Book.objects.filter(editor__in=sub) # When you call __str__ on the query object, it doesn't know about using # so it falls back to the default. Don't let routing instructions # force the subquery to an incompatible database. str(qs.query) # If you evaluate the query, it should work, running on 'other' self.assertEqual(list(qs.values_list('title', flat=True)), ['Dive into Python']) def test_deferred_models(self): mark_def = Person.objects.using('default').create(name="Mark Pilgrim") mark_other = Person.objects.using('other').create(name="Mark Pilgrim") orig_b = Book.objects.using('other').create(title="Dive into Python", published=datetime.date(2009, 5, 4), editor=mark_other) b = Book.objects.using('other').only('title').get(pk=orig_b.pk) self.assertEqual(b.published, datetime.date(2009, 5, 4)) b = Book.objects.using('other').only('title').get(pk=orig_b.pk) b.editor = mark_def b.save(using='default') self.assertEqual(Book.objects.using('default').get(pk=b.pk).published, datetime.date(2009, 5, 4)) @override_settings(DATABASE_ROUTERS=[AuthRouter()]) class AuthTestCase(TestCase): multi_db = True def test_auth_manager(self): "The methods on the auth manager obey database hints" # Create one user using default allocation policy User.objects.create_user('alice', 'alice@example.com') # Create another user, explicitly specifying the database User.objects.db_manager('default').create_user('bob', 'bob@example.com') # The second user only exists on the other database alice = User.objects.using('other').get(username='alice') self.assertEqual(alice.username, 'alice') self.assertEqual(alice._state.db, 'other') self.assertRaises(User.DoesNotExist, User.objects.using('default').get, username='alice') # The second user only exists on the default database bob = User.objects.using('default').get(username='bob') self.assertEqual(bob.username, 'bob') self.assertEqual(bob._state.db, 'default') self.assertRaises(User.DoesNotExist, User.objects.using('other').get, username='bob') # That is... there is one user on each database self.assertEqual(User.objects.using('default').count(), 1) self.assertEqual(User.objects.using('other').count(), 1) def test_dumpdata(self): "Check that dumpdata honors allow_migrate restrictions on the router" User.objects.create_user('alice', 'alice@example.com') User.objects.db_manager('default').create_user('bob', 'bob@example.com') # Check that dumping the default database doesn't try to include auth # because allow_migrate prohibits auth on default new_io = StringIO() management.call_command('dumpdata', 'auth', format='json', database='default', stdout=new_io) command_output = new_io.getvalue().strip() self.assertEqual(command_output, '[]') # Check that dumping the other database does include auth new_io = StringIO() management.call_command('dumpdata', 'auth', format='json', database='other', stdout=new_io) command_output = new_io.getvalue().strip() self.assertIn('"email": "alice@example.com"', command_output) class AntiPetRouter(object): # A router that only expresses an opinion on migrate, # passing pets to the 'other' database def allow_migrate(self, db, app_label, model_name=None, **hints): if db == 'other': return model_name == 'pet' else: return model_name != 'pet' class FixtureTestCase(TestCase): multi_db = True fixtures = ['multidb-common', 'multidb'] @override_settings(DATABASE_ROUTERS=[AntiPetRouter()]) def test_fixture_loading(self): "Multi-db fixtures are loaded correctly" # Check that "Pro Django" exists on the default database, but not on other database try: Book.objects.get(title="Pro Django") Book.objects.using('default').get(title="Pro Django") except Book.DoesNotExist: self.fail('"Pro Django" should exist on default database') self.assertRaises( Book.DoesNotExist, Book.objects.using('other').get, title="Pro Django" ) # Check that "Dive into Python" exists on the default database, but not on other database try: Book.objects.using('other').get(title="Dive into Python") except Book.DoesNotExist: self.fail('"Dive into Python" should exist on other database') self.assertRaises( Book.DoesNotExist, Book.objects.get, title="Dive into Python" ) self.assertRaises( Book.DoesNotExist, Book.objects.using('default').get, title="Dive into Python" ) # Check that "Definitive Guide" exists on the both databases try: Book.objects.get(title="The Definitive Guide to Django") Book.objects.using('default').get(title="The Definitive Guide to Django") Book.objects.using('other').get(title="The Definitive Guide to Django") except Book.DoesNotExist: self.fail('"The Definitive Guide to Django" should exist on both databases') @override_settings(DATABASE_ROUTERS=[AntiPetRouter()]) def test_pseudo_empty_fixtures(self): """ A fixture can contain entries, but lead to nothing in the database; this shouldn't raise an error (#14068). """ new_io = StringIO() management.call_command('loaddata', 'pets', stdout=new_io, stderr=new_io) command_output = new_io.getvalue().strip() # No objects will actually be loaded self.assertEqual(command_output, "Installed 0 object(s) (of 2) from 1 fixture(s)") class PickleQuerySetTestCase(TestCase): multi_db = True def test_pickling(self): for db in connections: Book.objects.using(db).create(title='Dive into Python', published=datetime.date(2009, 5, 4)) qs = Book.objects.all() self.assertEqual(qs.db, pickle.loads(pickle.dumps(qs)).db) class DatabaseReceiver(object): """ Used in the tests for the database argument in signals (#13552) """ def __call__(self, signal, sender, **kwargs): self._database = kwargs['using'] class WriteToOtherRouter(object): """ A router that sends all writes to the other database. """ def db_for_write(self, model, **hints): return "other" class SignalTests(TestCase): multi_db = True def override_router(self): return override_settings(DATABASE_ROUTERS=[WriteToOtherRouter()]) def test_database_arg_save_and_delete(self): """ Tests that the pre/post_save signal contains the correct database. (#13552) """ # Make some signal receivers pre_save_receiver = DatabaseReceiver() post_save_receiver = DatabaseReceiver() pre_delete_receiver = DatabaseReceiver() post_delete_receiver = DatabaseReceiver() # Make model and connect receivers signals.pre_save.connect(sender=Person, receiver=pre_save_receiver) signals.post_save.connect(sender=Person, receiver=post_save_receiver) signals.pre_delete.connect(sender=Person, receiver=pre_delete_receiver) signals.post_delete.connect(sender=Person, receiver=post_delete_receiver) p = Person.objects.create(name='Darth Vader') # Save and test receivers got calls p.save() self.assertEqual(pre_save_receiver._database, DEFAULT_DB_ALIAS) self.assertEqual(post_save_receiver._database, DEFAULT_DB_ALIAS) # Delete, and test p.delete() self.assertEqual(pre_delete_receiver._database, DEFAULT_DB_ALIAS) self.assertEqual(post_delete_receiver._database, DEFAULT_DB_ALIAS) # Save again to a different database p.save(using="other") self.assertEqual(pre_save_receiver._database, "other") self.assertEqual(post_save_receiver._database, "other") # Delete, and test p.delete(using="other") self.assertEqual(pre_delete_receiver._database, "other") self.assertEqual(post_delete_receiver._database, "other") signals.pre_save.disconnect(sender=Person, receiver=pre_save_receiver) signals.post_save.disconnect(sender=Person, receiver=post_save_receiver) signals.pre_delete.disconnect(sender=Person, receiver=pre_delete_receiver) signals.post_delete.disconnect(sender=Person, receiver=post_delete_receiver) def test_database_arg_m2m(self): """ Test that the m2m_changed signal has a correct database arg (#13552) """ # Make a receiver receiver = DatabaseReceiver() # Connect it signals.m2m_changed.connect(receiver=receiver) # Create the models that will be used for the tests b = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) p = Person.objects.create(name="Marty Alchin") # Create a copy of the models on the 'other' database to prevent # integrity errors on backends that don't defer constraints checks Book.objects.using('other').create(pk=b.pk, title=b.title, published=b.published) Person.objects.using('other').create(pk=p.pk, name=p.name) # Test addition b.authors.add(p) self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) with self.override_router(): b.authors.add(p) self.assertEqual(receiver._database, "other") # Test removal b.authors.remove(p) self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) with self.override_router(): b.authors.remove(p) self.assertEqual(receiver._database, "other") # Test addition in reverse p.book_set.add(b) self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) with self.override_router(): p.book_set.add(b) self.assertEqual(receiver._database, "other") # Test clearing b.authors.clear() self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) with self.override_router(): b.authors.clear() self.assertEqual(receiver._database, "other") class AttributeErrorRouter(object): "A router to test the exception handling of ConnectionRouter" def db_for_read(self, model, **hints): raise AttributeError def db_for_write(self, model, **hints): raise AttributeError class RouterAttributeErrorTestCase(TestCase): multi_db = True def override_router(self): return override_settings(DATABASE_ROUTERS=[AttributeErrorRouter()]) def test_attribute_error_read(self): "Check that the AttributeError from AttributeErrorRouter bubbles up" b = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) with self.override_router(): self.assertRaises(AttributeError, Book.objects.get, pk=b.pk) def test_attribute_error_save(self): "Check that the AttributeError from AttributeErrorRouter bubbles up" dive = Book() dive.title = "Dive into Python" dive.published = datetime.date(2009, 5, 4) with self.override_router(): self.assertRaises(AttributeError, dive.save) def test_attribute_error_delete(self): "Check that the AttributeError from AttributeErrorRouter bubbles up" b = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) p = Person.objects.create(name="Marty Alchin") b.authors.set([p]) b.editor = p with self.override_router(): self.assertRaises(AttributeError, b.delete) def test_attribute_error_m2m(self): "Check that the AttributeError from AttributeErrorRouter bubbles up" b = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) p = Person.objects.create(name="Marty Alchin") with self.override_router(): with self.assertRaises(AttributeError): b.authors.set([p]) class ModelMetaRouter(object): "A router to ensure model arguments are real model classes" def db_for_write(self, model, **hints): if not hasattr(model, '_meta'): raise ValueError @override_settings(DATABASE_ROUTERS=[ModelMetaRouter()]) class RouterModelArgumentTestCase(TestCase): multi_db = True def test_m2m_collection(self): b = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) p = Person.objects.create(name="Marty Alchin") # test add b.authors.add(p) # test remove b.authors.remove(p) # test clear b.authors.clear() # test setattr b.authors.set([p]) # test M2M collection b.delete() def test_foreignkey_collection(self): person = Person.objects.create(name='Bob') Pet.objects.create(owner=person, name='Wart') # test related FK collection person.delete() class SyncOnlyDefaultDatabaseRouter(object): def allow_migrate(self, db, app_label, **hints): return db == DEFAULT_DB_ALIAS class MigrateTestCase(TestCase): available_apps = [ 'multiple_database', 'django.contrib.auth', 'django.contrib.contenttypes' ] multi_db = True def test_migrate_to_other_database(self): """Regression test for #16039: migrate with --database option.""" cts = ContentType.objects.using('other').filter(app_label='multiple_database') count = cts.count() self.assertGreater(count, 0) cts.delete() management.call_command('migrate', verbosity=0, interactive=False, database='other') self.assertEqual(cts.count(), count) def test_migrate_to_other_database_with_router(self): """Regression test for #16039: migrate with --database option.""" cts = ContentType.objects.using('other').filter(app_label='multiple_database') cts.delete() with override_settings(DATABASE_ROUTERS=[SyncOnlyDefaultDatabaseRouter()]): management.call_command('migrate', verbosity=0, interactive=False, database='other') self.assertEqual(cts.count(), 0) class RouterUsed(Exception): WRITE = 'write' def __init__(self, mode, model, hints): self.mode = mode self.model = model self.hints = hints class RouteForWriteTestCase(TestCase): multi_db = True class WriteCheckRouter(object): def db_for_write(self, model, **hints): raise RouterUsed(mode=RouterUsed.WRITE, model=model, hints=hints) def override_router(self): return override_settings(DATABASE_ROUTERS=[RouteForWriteTestCase.WriteCheckRouter()]) def test_fk_delete(self): owner = Person.objects.create(name='Someone') pet = Pet.objects.create(name='fido', owner=owner) try: with self.override_router(): pet.owner.delete() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Person) self.assertEqual(e.hints, {'instance': owner}) def test_reverse_fk_delete(self): owner = Person.objects.create(name='Someone') to_del_qs = owner.pet_set.all() try: with self.override_router(): to_del_qs.delete() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Pet) self.assertEqual(e.hints, {'instance': owner}) def test_reverse_fk_get_or_create(self): owner = Person.objects.create(name='Someone') try: with self.override_router(): owner.pet_set.get_or_create(name='fido') self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Pet) self.assertEqual(e.hints, {'instance': owner}) def test_reverse_fk_update(self): owner = Person.objects.create(name='Someone') Pet.objects.create(name='fido', owner=owner) try: with self.override_router(): owner.pet_set.update(name='max') self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Pet) self.assertEqual(e.hints, {'instance': owner}) def test_m2m_add(self): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) try: with self.override_router(): book.authors.add(auth) self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) self.assertEqual(e.hints, {'instance': book}) def test_m2m_clear(self): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) try: with self.override_router(): book.authors.clear() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) self.assertEqual(e.hints, {'instance': book}) def test_m2m_delete(self): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) try: with self.override_router(): book.authors.all().delete() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Person) self.assertEqual(e.hints, {'instance': book}) def test_m2m_get_or_create(self): Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) try: with self.override_router(): book.authors.get_or_create(name='Someone else') self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book) self.assertEqual(e.hints, {'instance': book}) def test_m2m_remove(self): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) try: with self.override_router(): book.authors.remove(auth) self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) self.assertEqual(e.hints, {'instance': book}) def test_m2m_update(self): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) try: with self.override_router(): book.authors.all().update(name='Different') self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Person) self.assertEqual(e.hints, {'instance': book}) def test_reverse_m2m_add(self): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) try: with self.override_router(): auth.book_set.add(book) self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) self.assertEqual(e.hints, {'instance': auth}) def test_reverse_m2m_clear(self): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) try: with self.override_router(): auth.book_set.clear() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) self.assertEqual(e.hints, {'instance': auth}) def test_reverse_m2m_delete(self): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) try: with self.override_router(): auth.book_set.all().delete() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book) self.assertEqual(e.hints, {'instance': auth}) def test_reverse_m2m_get_or_create(self): auth = Person.objects.create(name='Someone') Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) try: with self.override_router(): auth.book_set.get_or_create(title="New Book", published=datetime.datetime.now()) self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Person) self.assertEqual(e.hints, {'instance': auth}) def test_reverse_m2m_remove(self): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) try: with self.override_router(): auth.book_set.remove(book) self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) self.assertEqual(e.hints, {'instance': auth}) def test_reverse_m2m_update(self): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) try: with self.override_router(): auth.book_set.all().update(title='Different') self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book) self.assertEqual(e.hints, {'instance': auth})