diff options
Diffstat (limited to 'tests/multiple_database')
| -rw-r--r-- | tests/multiple_database/models.py | 18 | ||||
| -rw-r--r-- | tests/multiple_database/routers.py | 25 | ||||
| -rw-r--r-- | tests/multiple_database/tests.py | 1717 |
3 files changed, 1079 insertions, 681 deletions
diff --git a/tests/multiple_database/models.py b/tests/multiple_database/models.py index e4e764d6a5..7de784e149 100644 --- a/tests/multiple_database/models.py +++ b/tests/multiple_database/models.py @@ -1,7 +1,5 @@ from django.contrib.auth.models import User -from django.contrib.contenttypes.fields import ( - GenericForeignKey, GenericRelation, -) +from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation from django.contrib.contenttypes.models import ContentType from django.db import models @@ -13,7 +11,7 @@ class Review(models.Model): content_object = GenericForeignKey() class Meta: - ordering = ('source',) + ordering = ("source",) def __str__(self): return self.source @@ -30,7 +28,7 @@ class Person(models.Model): objects = PersonManager() class Meta: - ordering = ('name',) + ordering = ("name",) def __str__(self): return self.name @@ -52,14 +50,16 @@ class Book(models.Model): title = models.CharField(max_length=100) published = models.DateField() authors = models.ManyToManyField(Person) - editor = models.ForeignKey(Person, models.SET_NULL, null=True, related_name='edited') + editor = models.ForeignKey( + Person, models.SET_NULL, null=True, related_name="edited" + ) reviews = GenericRelation(Review) pages = models.IntegerField(default=100) objects = BookManager() class Meta: - ordering = ('title',) + ordering = ("title",) def __str__(self): return self.title @@ -70,7 +70,7 @@ class Pet(models.Model): owner = models.ForeignKey(Person, models.CASCADE) class Meta: - ordering = ('name',) + ordering = ("name",) class UserProfile(models.Model): @@ -78,4 +78,4 @@ class UserProfile(models.Model): flavor = models.CharField(max_length=100) class Meta: - ordering = ('flavor',) + ordering = ("flavor",) diff --git a/tests/multiple_database/routers.py b/tests/multiple_database/routers.py index 7891e22969..0cc7f1729c 100644 --- a/tests/multiple_database/routers.py +++ b/tests/multiple_database/routers.py @@ -9,14 +9,17 @@ class TestRouter: def db_for_read(self, model, instance=None, **hints): if instance: - return instance._state.db or 'other' - return 'other' + return instance._state.db or "other" + return "other" def db_for_write(self, model, **hints): return DEFAULT_DB_ALIAS def allow_relation(self, obj1, obj2, **hints): - return obj1._state.db in ('default', 'other') and obj2._state.db in ('default', 'other') + return obj1._state.db in ("default", "other") and obj2._state.db in ( + "default", + "other", + ) def allow_migrate(self, db, app_label, **hints): return True @@ -29,30 +32,30 @@ class AuthRouter: def db_for_read(self, model, **hints): "Point all read operations on auth models to 'default'" - if model._meta.app_label == 'auth': + if model._meta.app_label == "auth": # We use default here to ensure we can tell the difference # between a read request and a write request for Auth objects - return 'default' + return "default" return None def db_for_write(self, model, **hints): "Point all operations on auth models to 'other'" - if model._meta.app_label == 'auth': - return 'other' + if model._meta.app_label == "auth": + return "other" return None def allow_relation(self, obj1, obj2, **hints): "Allow any relation if a model in Auth is involved" - return obj1._meta.app_label == 'auth' or obj2._meta.app_label == 'auth' or None + return obj1._meta.app_label == "auth" or obj2._meta.app_label == "auth" or None def allow_migrate(self, db, app_label, **hints): "Make sure the auth app only appears on the 'other' db" - if app_label == 'auth': - return db == 'other' + if app_label == "auth": + return db == "other" return None class WriteRouter: # A router that only expresses an opinion on writes def db_for_write(self, model, **hints): - return 'writer' + return "writer" diff --git a/tests/multiple_database/tests.py b/tests/multiple_database/tests.py index 946137dc12..3f5548d761 100644 --- a/tests/multiple_database/tests.py +++ b/tests/multiple_database/tests.py @@ -17,17 +17,17 @@ from .routers import AuthRouter, TestRouter, WriteRouter class QueryTestCase(TestCase): - databases = {'default', 'other'} + databases = {"default", "other"} def test_db_selection(self): "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.using("other").db, "other") - self.assertEqual(Book.objects.db_manager('other').db, 'other') - self.assertEqual(Book.objects.db_manager('other').all().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" @@ -43,122 +43,130 @@ class QueryTestCase(TestCase): # 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") + Book.objects.using("default").get(title="Pro Django") except Book.DoesNotExist: self.fail('"Pro Django" should exist on default database') with self.assertRaises(Book.DoesNotExist): - Book.objects.using('other').get(title="Pro Django") + 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") + Book.objects.using("default").get(title="Dive into Python") except Book.DoesNotExist: self.fail('"Dive into Python" should exist on default database') with self.assertRaises(Book.DoesNotExist): - Book.objects.using('other').get(title="Dive into Python") + 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)) + 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') + dive.save(using="other") # Book exists on the default database, but not on other database try: - Book.objects.using('other').get(title="Pro Django") + Book.objects.using("other").get(title="Pro Django") except Book.DoesNotExist: self.fail('"Pro Django" should exist on other database') with self.assertRaises(Book.DoesNotExist): Book.objects.get(title="Pro Django") with self.assertRaises(Book.DoesNotExist): - Book.objects.using('default').get(title="Pro Django") + Book.objects.using("default").get(title="Pro Django") try: - Book.objects.using('other').get(title="Dive into Python") + Book.objects.using("other").get(title="Dive into Python") except Book.DoesNotExist: self.fail('"Dive into Python" should exist on other database') with self.assertRaises(Book.DoesNotExist): Book.objects.get(title="Dive into Python") with self.assertRaises(Book.DoesNotExist): - Book.objects.using('default').get(title="Dive into Python") + Book.objects.using("default").get(title="Dive into Python") def test_refresh(self): dive = Book(title="Dive into Python", published=datetime.date(2009, 5, 4)) - dive.save(using='other') - dive2 = Book.objects.using('other').get() + dive.save(using="other") + dive2 = Book.objects.using("other").get() dive2.title = "Dive into Python (on default)" - dive2.save(using='default') + dive2.save(using="default") dive.refresh_from_db() self.assertEqual(dive.title, "Dive into Python") - dive.refresh_from_db(using='default') + dive.refresh_from_db(using="default") self.assertEqual(dive.title, "Dive into Python (on default)") self.assertEqual(dive._state.db, "default") def test_refresh_router_instance_hint(self): router = Mock() router.db_for_read.return_value = None - book = Book.objects.create(title='Dive Into Python', published=datetime.date(1957, 10, 12)) + book = Book.objects.create( + title="Dive Into Python", published=datetime.date(1957, 10, 12) + ) with self.settings(DATABASE_ROUTERS=[router]): book.refresh_from_db() router.db_for_read.assert_called_once_with(Book, instance=book) 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").create( + title="Dive into Python", published=datetime.date(2009, 5, 4) + ) - dive = Book.objects.using('other').get(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") with self.assertRaises(Book.DoesNotExist): - Book.objects.using('default').get(published=datetime.date(2009, 5, 4)) + Book.objects.using("default").get(published=datetime.date(2009, 5, 4)) - dive = Book.objects.using('other').get(title__icontains="dive") + dive = Book.objects.using("other").get(title__icontains="dive") self.assertEqual(dive.title, "Dive into Python") with self.assertRaises(Book.DoesNotExist): - Book.objects.using('default').get(title__icontains="dive") + Book.objects.using("default").get(title__icontains="dive") - dive = Book.objects.using('other').get(title__iexact="dive INTO python") + dive = Book.objects.using("other").get(title__iexact="dive INTO python") self.assertEqual(dive.title, "Dive into Python") with self.assertRaises(Book.DoesNotExist): - Book.objects.using('default').get(title__iexact="dive INTO python") + Book.objects.using("default").get(title__iexact="dive INTO python") - dive = Book.objects.using('other').get(published__year=2009) + 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)) with self.assertRaises(Book.DoesNotExist): - Book.objects.using('default').get(published__year=2009) + Book.objects.using("default").get(published__year=2009) - years = Book.objects.using('other').dates('published', 'year') + 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') + years = Book.objects.using("default").dates("published", "year") self.assertEqual([o.year for o in years], []) - months = Book.objects.using('other').dates('published', 'month') + 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') + 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)) + 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)) + 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") + mark = Person.objects.using("other").create(name="Mark Pilgrim") # Save the author relations pro.authors.set([marty]) @@ -166,176 +174,271 @@ class QueryTestCase(TestCase): # 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) + self.assertEqual(Book.authors.through.objects.using("default").count(), 1) + self.assertEqual(Book.authors.through.objects.using("other").count(), 1) # Queries work across m2m joins self.assertEqual( - list(Book.objects.using('default').filter(authors__name='Marty Alchin').values_list('title', flat=True)), - ['Pro Django'] + 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)), - [] + 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)), - [] + 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'] + 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") + 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(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']) + 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") + 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") + 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)), - [] + 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'] + 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'] + 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'] + 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)), - [] + 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)), - [] + 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)), - [] + 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') + dive.authors.create(name="Jane Brown") self.assertEqual( - list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), - [] + 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'] + 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") + 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)) + 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'] + 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) + Person.objects.using("other") + .filter(book__title="Greasemonkey Hacks") + .values_list("name", flat=True) ), - ['Mark Pilgrim'] + ["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'] + 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) + 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)), - [] + 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) + 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)) + 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)), - [] + 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'] + 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)) + 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)) + 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") + mark = Person.objects.using("other").create(name="Mark Pilgrim") # Set a foreign key set with an object from a different database msg = ( 'Cannot assign "<Person: Marty Alchin>": the current database ' - 'router prevents this relation.' + "router prevents this relation." ) with self.assertRaisesMessage(ValueError, msg): - with transaction.atomic(using='default'): + with transaction.atomic(using="default"): marty.edited.set([pro, dive]) # Add to an m2m with an object from a different database @@ -344,12 +447,12 @@ class QueryTestCase(TestCase): 'database "default", value is on database "other"' ) with self.assertRaisesMessage(ValueError, msg): - with transaction.atomic(using='default'): + with transaction.atomic(using="default"): marty.book_set.add(dive) # Set a m2m with an object from a different database with self.assertRaisesMessage(ValueError, msg): - with transaction.atomic(using='default'): + with transaction.atomic(using="default"): marty.book_set.set([pro, dive]) # Add to a reverse m2m with an object from a different database @@ -358,81 +461,89 @@ class QueryTestCase(TestCase): 'database "other", value is on database "default"' ) with self.assertRaisesMessage(ValueError, msg): - with transaction.atomic(using='other'): + with transaction.atomic(using="other"): dive.authors.add(marty) # Set a reverse m2m with an object from a different database with self.assertRaisesMessage(ValueError, msg): - with transaction.atomic(using='other'): + 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 = 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("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) + 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') + 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) + 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) + self.assertEqual(Person.objects.using("other").count(), 1) # ... but the book has been deleted - self.assertEqual(Book.objects.using('other').count(), 0) + 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) + 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 = 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("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) + 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') + 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) + 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) + self.assertEqual(Person.objects.using("other").count(), 0) # ... but the book still exists - self.assertEqual(Book.objects.using('other').count(), 1) + 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) + 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)) + 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") + 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 @@ -441,166 +552,222 @@ class QueryTestCase(TestCase): dive.editor = chris dive.save() - pro = Book.objects.using('default').get(title="Pro Django") + 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") + dive = Book.objects.using("other").get(title="Dive into Python") self.assertEqual(dive.editor.name, "Chris Mills") # 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'] + 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)), - [] + 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) + 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) + Person.objects.using("other") + .filter(edited__title="Dive into Python") + .values_list("name", flat=True) ), - ['Chris Mills'] + ["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") + 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']) + 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") + 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)) + 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)), - [] + 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'] + 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) + Person.objects.using("other") + .filter(edited__title="Dive into Python") + .values_list("name", flat=True) ), - ['Chris Mills'] + ["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)), - [] + 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) + Person.objects.using("other") + .filter(edited__title="Dive into Python") + .values_list("name", flat=True) ), - ['Chris Mills'] + ["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)), - [] + 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) + 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)) + 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)), - [] + 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'] + 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) + 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)) + 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)) + 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 msg = ( 'Cannot assign "<Person: Marty Alchin>": the current database ' - 'router prevents this relation.' + "router prevents this relation." ) with self.assertRaisesMessage(ValueError, msg): dive.editor = marty # Set a foreign key set with an object from a different database with self.assertRaisesMessage(ValueError, msg): - with transaction.atomic(using='default'): + 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.assertRaisesMessage(ValueError, msg): - with transaction.atomic(using='default'): + 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) + 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("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) + 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') + mark.delete(using="other") - self.assertEqual(Person.objects.using('default').count(), 0) - self.assertEqual(Pet.objects.using('default').count(), 0) + 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) + 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) + mickey = Person.objects.using("other").create(name="Mickey") + pluto = Pet.objects.using("other").create(name="Pluto", owner=mickey) self.assertIsNone(pluto.full_clean()) # Any router that accesses `model` in db_for_read() works here. @@ -611,78 +778,92 @@ class QueryTestCase(TestCase): model_instance=None. """ mickey = Person.objects.create(name="Mickey") - owner_field = Pet._meta.get_field('owner') + owner_field = Pet._meta.get_field("owner") self.assertEqual(owner_field.clean(mickey.pk, None), mickey.pk) 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') + 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') + 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") + alice = User.objects.using("default").get(username="alice") self.assertEqual(alice.userprofile.flavor, "chocolate") - bob = User.objects.using('other').get(username="bob") + bob = User.objects.using("other").get(username="bob") self.assertEqual(bob.userprofile.flavor, "crunchy frog") # Queries work across joins self.assertEqual( list( - User.objects.using('default') - .filter(userprofile__flavor='chocolate').values_list('username', flat=True) + User.objects.using("default") + .filter(userprofile__flavor="chocolate") + .values_list("username", flat=True) ), - ['alice'] + ["alice"], ) self.assertEqual( list( - User.objects.using('other') - .filter(userprofile__flavor='chocolate').values_list('username', flat=True) + 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) + 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) + User.objects.using("other") + .filter(userprofile__flavor="crunchy frog") + .values_list("username", flat=True) ), - ['bob'] + ["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') + 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') + 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') + 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') + 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') + alice_profile = UserProfile.objects.using("default").create( + user=alice, flavor="chocolate" + ) msg = ( 'Cannot assign "%r": the current database router prevents this ' - 'relation.' % alice_profile + "relation." % alice_profile ) with self.assertRaisesMessage(ValueError, msg): bob.userprofile = alice_profile @@ -690,12 +871,14 @@ class QueryTestCase(TestCase): # 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') + 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 = User(pk=51, username="charlie", email="charlie@example.com") charlie.set_unusable_password() # initially, no db assigned @@ -705,143 +888,228 @@ class QueryTestCase(TestCase): # 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') + 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']) + 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'] + 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 = 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') + 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'] + 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'] + 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)) + 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)) + 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) + review2 = Review.objects.using("other").create( + source="Python Weekly", content_object=dive + ) - review1 = Review.objects.using('default').get(source="Python Monthly") + 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") + 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") + 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']) + 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) + 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)), - [] + 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'] + 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)), - [] + 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'] + 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)), - [] + 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'] + 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)), - [] + 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)), - [] + 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') + dive.reviews.create(source="Python Daily") self.assertEqual( - list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), - [] + 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'] + 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)) + 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)) + 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) + Review.objects.using("other").create( + source="Python Weekly", content_object=dive + ) # Set a foreign key with an object from a different database msg = ( 'Cannot assign "<ContentType: multiple_database | book>": the ' - 'current database router prevents this relation.' + "current database router prevents this relation." ) with self.assertRaisesMessage(ValueError, msg): review1.content_object = dive @@ -852,7 +1120,7 @@ class QueryTestCase(TestCase): "Use bulk=False or save the object first." ) with self.assertRaisesMessage(ValueError, msg): - with transaction.atomic(using='other'): + with transaction.atomic(using="other"): dive.reviews.add(review1) # BUT! if you assign a FK object when the base object hasn't @@ -864,87 +1132,119 @@ class QueryTestCase(TestCase): # Dive comes from 'other', so review3 is set to use 'other'... review3.content_object = dive - self.assertEqual(review3._state.db, 'other') + 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'] + 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'] + 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'] + 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'] + 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) + 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("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) + 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') + dive.delete(using="other") - self.assertEqual(Book.objects.using('default').count(), 0) - self.assertEqual(Review.objects.using('default').count(), 0) + 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) + 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)) + 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') + 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') + 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( + 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") + 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') + 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') + 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 @@ -964,142 +1264,175 @@ class QueryTestCase(TestCase): def test_related_manager(self): "Related managers return managers, not querysets" - mark = Person.objects.using('other').create(name="Mark Pilgrim") + 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) + 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']) + @override_settings( + DATABASE_ROUTERS=[ + "multiple_database.tests.TestRouter", + "multiple_database.tests.WriteRouter", + ] + ) def test_router_init_default(self): connection_router = ConnectionRouter() - self.assertEqual([r.__class__.__name__ for r in connection_router.routers], ['TestRouter', 'WriteRouter']) + self.assertEqual( + [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.assertEqual([r.__class__.__name__ for r in connection_router.routers], ['TestRouter', 'WriteRouter']) + connection_router = ConnectionRouter( + [ + "multiple_database.tests.TestRouter", + "multiple_database.tests.WriteRouter", + ] + ) + self.assertEqual( + [r.__class__.__name__ for r in connection_router.routers], + ["TestRouter", "WriteRouter"], + ) # Init with instances instead of strings connection_router = ConnectionRouter([TestRouter(), WriteRouter()]) - self.assertEqual([r.__class__.__name__ for r in connection_router.routers], ['TestRouter', 'WriteRouter']) + self.assertEqual( + [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): - databases = {'default', 'other'} + databases = {"default", "other"} def test_db_selection(self): "Querysets obey the router for db suggestions" - self.assertEqual(Book.objects.db, 'other') - self.assertEqual(Book.objects.all().db, 'other') + 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.using("default").db, "default") - self.assertEqual(Book.objects.db_manager('default').db, 'default') - self.assertEqual(Book.objects.db_manager('default').all().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("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)) + 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("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)) + 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.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)) + 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)) + 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_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.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)) + 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') + 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.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)) + 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', + 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)) + 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) + 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') + 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') + pro = Book.objects.using("default").get(title="Pro Django") # 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) + 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') + 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 @@ -1107,13 +1440,14 @@ class RouterTestCase(TestCase): 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)}) + 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) + 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) @@ -1121,12 +1455,12 @@ class RouterTestCase(TestCase): 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) + 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( + 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), ) @@ -1138,67 +1472,69 @@ class RouterTestCase(TestCase): 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)) + pro = Book.objects.using("default").create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) - marty = Person.objects.using('default').create(name="Marty Alchin") + 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)) + 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") + mark = Person.objects.using("other").create(name="Mark Pilgrim") # Set a foreign key with an object from a different database dive.editor = marty # 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') + 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') + self.assertEqual(dive._state.db, "default") # ...and the source database now has a copy of any object saved - Book.objects.using('default').get(title='Dive into Python').delete() + Book.objects.using("default").get(title="Dive into Python").delete() # 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') + 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 marty.edited.set([pro, dive], bulk=False) # 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') + 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 - Book.objects.using('default').get(title='Dive into Python').delete() + Book.objects.using("default").get(title="Dive into Python").delete() # 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') + 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 marty.edited.add(dive, bulk=False) # 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') + 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 - Book.objects.using('default').get(title='Dive into Python').delete() + Book.objects.using("default").get(title="Dive into Python").delete() # This isn't a real primary/replica database, so restore the original from other - dive = Book.objects.using('other').get(title='Dive into Python') + 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 @@ -1211,222 +1547,236 @@ class RouterTestCase(TestCase): # old object comes from 'other', so the new object is set to use the # source of 'other'... - self.assertEqual(dive._state.db, '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') + 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') + 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') + 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') + 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') + 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', + title="Dive into Cheesecake", published=datetime.date(2010, 3, 15), ) - self.assertEqual(cheesecake._state.db, 'default') + 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') + 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)) + 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") + 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)) + 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") + 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') + pro.save(using="default") + marty.save(using="default") + dive.save(using="other") + mark.save(using="other") # 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) + 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 marty.book_set.set([pro, dive]) # 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') + 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) + 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() + Book.authors.through.objects.using("default").delete() # Add to an m2m with an object from a different database marty.book_set.add(dive) # 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') + 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) + 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() + Book.authors.through.objects.using("default").delete() # Set a reverse m2m with an object from a different database dive.authors.set([mark, marty]) # 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') + 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) + 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() + 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) + 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 dive.authors.add(marty) # 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') + 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) + 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', pk=3) - self.assertEqual(alice._state.db, 'default') + alice = dive.authors.create(name="Alice", pk=3) + 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') + alice, created = dive.authors.get_or_create(name="Alice") + self.assertEqual(alice._state.db, "default") - bob, created = dive.authors.get_or_create(name='Bob', defaults={'pk': 4}) - self.assertEqual(bob._state.db, 'default') + bob, created = dive.authors.get_or_create(name="Bob", defaults={"pk": 4}) + 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') + 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') + 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') + alice_profile = UserProfile.objects.create(user=alice, flavor="chocolate") bob.userprofile = alice_profile # 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') + 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') + 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)) + 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) + 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)) + 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) + review2 = Review.objects.using("other").create( + source="Python Weekly", content_object=dive + ) # Set a generic foreign key with an object from a different database review1.content_object = dive # 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') + 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') + self.assertEqual(review1._state.db, "default") + self.assertEqual(dive._state.db, "default") # ...and the source database now has a copy of any object saved - Book.objects.using('default').get(title='Dive into Python').delete() + Book.objects.using("default").get(title="Dive into Python").delete() # 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') + 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 dive.reviews.add(review1) # 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') + 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') + self.assertEqual(dive._state.db, "default") # ...and the source database now has a copy of any object saved - Book.objects.using('default').get(title='Dive into Python').delete() + Book.objects.using("default").get(title="Dive into Python").delete() # BUT! if you assign a FK object when the base object hasn't # been saved yet, you implicitly assign the database for the @@ -1437,66 +1787,70 @@ class RouterTestCase(TestCase): # Dive comes from 'other', so review3 is set to use the source of 'other'... review3.content_object = dive - self.assertEqual(review3._state.db, 'default') + 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') + 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') + 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)) + 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") + 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(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') + 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( + marty = Person.objects.using("other").create(pk=1, name="Marty Alchin") + Book.objects.using("other").create( pk=1, - title='Pro Django', + 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') + 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)) + 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) + 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') + 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', + 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') + 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 @@ -1505,74 +1859,80 @@ class RouterTestCase(TestCase): 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']) + 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', + 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) + 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 = 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)) + 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): - databases = {'default', 'other'} + databases = {"default", "other"} 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') + 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') + 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') + alice = User.objects.using("other").get(username="alice") - self.assertEqual(alice.username, 'alice') - self.assertEqual(alice._state.db, 'other') + self.assertEqual(alice.username, "alice") + self.assertEqual(alice._state.db, "other") with self.assertRaises(User.DoesNotExist): - User.objects.using('default').get(username='alice') + User.objects.using("default").get(username="alice") # The second user only exists on the default database - bob = User.objects.using('default').get(username='bob') + bob = User.objects.using("default").get(username="bob") - self.assertEqual(bob.username, 'bob') - self.assertEqual(bob._state.db, 'default') + self.assertEqual(bob.username, "bob") + self.assertEqual(bob._state.db, "default") with self.assertRaises(User.DoesNotExist): - User.objects.using('other').get(username='bob') + 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) + self.assertEqual(User.objects.using("default").count(), 1) + self.assertEqual(User.objects.using("other").count(), 1) def test_dumpdata(self): "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') + User.objects.create_user("alice", "alice@example.com") + User.objects.db_manager("default").create_user("bob", "bob@example.com") # 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) + management.call_command( + "dumpdata", "auth", format="json", database="default", stdout=new_io + ) command_output = new_io.getvalue().strip() - self.assertEqual(command_output, '[]') + self.assertEqual(command_output, "[]") # dumping the other database does include auth new_io = StringIO() - management.call_command('dumpdata', 'auth', format='json', database='other', stdout=new_io) + 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) @@ -1582,38 +1942,38 @@ class AntiPetRouter: # passing pets to the 'other' database def allow_migrate(self, db, app_label, model_name=None, **hints): - if db == 'other': - return model_name == 'pet' + if db == "other": + return model_name == "pet" else: - return model_name != 'pet' + return model_name != "pet" class FixtureTestCase(TestCase): - databases = {'default', 'other'} - fixtures = ['multidb-common', 'multidb'] + databases = {"default", "other"} + fixtures = ["multidb-common", "multidb"] @override_settings(DATABASE_ROUTERS=[AntiPetRouter()]) def test_fixture_loading(self): "Multi-db fixtures are loaded correctly" # "Pro Django" exists on the default database, but not on other database Book.objects.get(title="Pro Django") - Book.objects.using('default').get(title="Pro Django") + Book.objects.using("default").get(title="Pro Django") with self.assertRaises(Book.DoesNotExist): - Book.objects.using('other').get(title="Pro Django") + Book.objects.using("other").get(title="Pro Django") # "Dive into Python" exists on the default database, but not on other database - Book.objects.using('other').get(title="Dive into Python") + Book.objects.using("other").get(title="Dive into Python") with self.assertRaises(Book.DoesNotExist): Book.objects.get(title="Dive into Python") with self.assertRaises(Book.DoesNotExist): - Book.objects.using('default').get(title="Dive into Python") + Book.objects.using("default").get(title="Dive into Python") # "Definitive Guide" exists on the both databases 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") + Book.objects.using("default").get(title="The Definitive Guide to Django") + Book.objects.using("other").get(title="The Definitive Guide to Django") @override_settings(DATABASE_ROUTERS=[AntiPetRouter()]) def test_pseudo_empty_fixtures(self): @@ -1622,18 +1982,22 @@ class FixtureTestCase(TestCase): this shouldn't raise an error (#14068). """ new_io = StringIO() - management.call_command('loaddata', 'pets', stdout=new_io, stderr=new_io) + 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)") + self.assertEqual( + command_output, "Installed 0 object(s) (of 2) from 1 fixture(s)" + ) class PickleQuerySetTestCase(TestCase): - databases = {'default', 'other'} + databases = {"default", "other"} def test_pickling(self): for db in self.databases: - Book.objects.using(db).create(title='Dive into Python', published=datetime.date(2009, 5, 4)) + 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) @@ -1642,20 +2006,22 @@ class DatabaseReceiver: """ Used in the tests for the database argument in signals (#13552) """ + def __call__(self, signal, sender, **kwargs): - self._database = kwargs['using'] + self._database = kwargs["using"] class WriteToOtherRouter: """ A router that sends all writes to the other database. """ + def db_for_write(self, model, **hints): return "other" class SignalTests(TestCase): - databases = {'default', 'other'} + databases = {"default", "other"} def override_router(self): return override_settings(DATABASE_ROUTERS=[WriteToOtherRouter()]) @@ -1674,7 +2040,7 @@ class SignalTests(TestCase): 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') + p = Person.objects.create(name="Darth Vader") # Save and test receivers got calls p.save() self.assertEqual(pre_save_receiver._database, DEFAULT_DB_ALIAS) @@ -1707,14 +2073,17 @@ class SignalTests(TestCase): 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)) + 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) + 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) @@ -1747,6 +2116,7 @@ class SignalTests(TestCase): class AttributeErrorRouter: "A router to test the exception handling of ConnectionRouter" + def db_for_read(self, model, **hints): raise AttributeError @@ -1755,15 +2125,16 @@ class AttributeErrorRouter: class RouterAttributeErrorTestCase(TestCase): - databases = {'default', 'other'} + databases = {"default", "other"} def override_router(self): return override_settings(DATABASE_ROUTERS=[AttributeErrorRouter()]) def test_attribute_error_read(self): "The AttributeError from AttributeErrorRouter bubbles up" - b = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) + b = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) with self.override_router(): with self.assertRaises(AttributeError): Book.objects.get(pk=b.pk) @@ -1779,8 +2150,9 @@ class RouterAttributeErrorTestCase(TestCase): def test_attribute_error_delete(self): "The AttributeError from AttributeErrorRouter bubbles up" - b = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) + 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 @@ -1790,8 +2162,9 @@ class RouterAttributeErrorTestCase(TestCase): def test_attribute_error_m2m(self): "The AttributeError from AttributeErrorRouter bubbles up" - b = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) + 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): @@ -1800,18 +2173,20 @@ class RouterAttributeErrorTestCase(TestCase): class ModelMetaRouter: "A router to ensure model arguments are real model classes" + def db_for_write(self, model, **hints): - if not hasattr(model, '_meta'): + if not hasattr(model, "_meta"): raise ValueError @override_settings(DATABASE_ROUTERS=[ModelMetaRouter()]) class RouterModelArgumentTestCase(TestCase): - databases = {'default', 'other'} + databases = {"default", "other"} def test_m2m_collection(self): - b = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) + b = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) p = Person.objects.create(name="Marty Alchin") # test add @@ -1826,8 +2201,8 @@ class RouterModelArgumentTestCase(TestCase): b.delete() def test_foreignkey_collection(self): - person = Person.objects.create(name='Bob') - Pet.objects.create(owner=person, name='Wart') + person = Person.objects.create(name="Bob") + Pet.objects.create(owner=person, name="Wart") # test related FK collection person.delete() @@ -1841,36 +2216,40 @@ class MigrateTestCase(TestCase): # Limit memory usage when calling 'migrate'. available_apps = [ - 'multiple_database', - 'django.contrib.auth', - 'django.contrib.contenttypes' + "multiple_database", + "django.contrib.auth", + "django.contrib.contenttypes", ] - databases = {'default', 'other'} + databases = {"default", "other"} 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') + 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') + 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 = 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') + management.call_command( + "migrate", verbosity=0, interactive=False, database="other" + ) self.assertEqual(cts.count(), 0) class RouterUsed(Exception): - WRITE = 'write' + WRITE = "write" def __init__(self, mode, model, hints): self.mode = mode @@ -1879,28 +2258,30 @@ class RouterUsed(Exception): class RouteForWriteTestCase(TestCase): - databases = {'default', 'other'} + databases = {"default", "other"} class WriteCheckRouter: 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()]) + 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) + owner = Person.objects.create(name="Someone") + pet = Pet.objects.create(name="fido", owner=owner) with self.assertRaises(RouterUsed) as cm: with self.override_router(): pet.owner.delete() e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Person) - self.assertEqual(e.hints, {'instance': owner}) + self.assertEqual(e.hints, {"instance": owner}) def test_reverse_fk_delete(self): - owner = Person.objects.create(name='Someone') + owner = Person.objects.create(name="Someone") to_del_qs = owner.pet_set.all() with self.assertRaises(RouterUsed) as cm: with self.override_router(): @@ -1908,45 +2289,47 @@ class RouteForWriteTestCase(TestCase): e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Pet) - self.assertEqual(e.hints, {'instance': owner}) + self.assertEqual(e.hints, {"instance": owner}) def test_reverse_fk_get_or_create(self): - owner = Person.objects.create(name='Someone') + owner = Person.objects.create(name="Someone") with self.assertRaises(RouterUsed) as cm: with self.override_router(): - owner.pet_set.get_or_create(name='fido') + owner.pet_set.get_or_create(name="fido") e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Pet) - self.assertEqual(e.hints, {'instance': owner}) + 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) + owner = Person.objects.create(name="Someone") + Pet.objects.create(name="fido", owner=owner) with self.assertRaises(RouterUsed) as cm: with self.override_router(): - owner.pet_set.update(name='max') + owner.pet_set.update(name="max") e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Pet) - self.assertEqual(e.hints, {'instance': owner}) + 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)) + auth = Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) with self.assertRaises(RouterUsed) as cm: with self.override_router(): book.authors.add(auth) e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) - self.assertEqual(e.hints, {'instance': book}) + 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)) + auth = Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) book.authors.add(auth) with self.assertRaises(RouterUsed) as cm: with self.override_router(): @@ -1954,12 +2337,13 @@ class RouteForWriteTestCase(TestCase): e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) - self.assertEqual(e.hints, {'instance': book}) + 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)) + auth = Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) book.authors.add(auth) with self.assertRaises(RouterUsed) as cm: with self.override_router(): @@ -1967,24 +2351,26 @@ class RouteForWriteTestCase(TestCase): e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Person) - self.assertEqual(e.hints, {'instance': book}) + 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)) + Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) with self.assertRaises(RouterUsed) as cm: with self.override_router(): - book.authors.get_or_create(name='Someone else') + book.authors.get_or_create(name="Someone else") e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book) - self.assertEqual(e.hints, {'instance': 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)) + auth = Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) book.authors.add(auth) with self.assertRaises(RouterUsed) as cm: with self.override_router(): @@ -1992,37 +2378,40 @@ class RouteForWriteTestCase(TestCase): e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) - self.assertEqual(e.hints, {'instance': book}) + 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)) + auth = Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) book.authors.add(auth) with self.assertRaises(RouterUsed) as cm: with self.override_router(): - book.authors.all().update(name='Different') + book.authors.all().update(name="Different") e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Person) - self.assertEqual(e.hints, {'instance': book}) + 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)) + auth = Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) with self.assertRaises(RouterUsed) as cm: with self.override_router(): auth.book_set.add(book) e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) - self.assertEqual(e.hints, {'instance': auth}) + 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)) + auth = Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) book.authors.add(auth) with self.assertRaises(RouterUsed) as cm: with self.override_router(): @@ -2030,12 +2419,13 @@ class RouteForWriteTestCase(TestCase): e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) - self.assertEqual(e.hints, {'instance': auth}) + 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)) + auth = Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) book.authors.add(auth) with self.assertRaises(RouterUsed) as cm: with self.override_router(): @@ -2043,24 +2433,26 @@ class RouteForWriteTestCase(TestCase): e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book) - self.assertEqual(e.hints, {'instance': auth}) + 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)) + auth = Person.objects.create(name="Someone") + Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) with self.assertRaises(RouterUsed) as cm: with self.override_router(): - auth.book_set.get_or_create(title="New Book", published=datetime.datetime.now()) + auth.book_set.get_or_create( + title="New Book", published=datetime.datetime.now() + ) e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Person) - self.assertEqual(e.hints, {'instance': auth}) + 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)) + auth = Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) book.authors.add(auth) with self.assertRaises(RouterUsed) as cm: with self.override_router(): @@ -2068,24 +2460,26 @@ class RouteForWriteTestCase(TestCase): e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book.authors.through) - self.assertEqual(e.hints, {'instance': auth}) + 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)) + auth = Person.objects.create(name="Someone") + book = Book.objects.create( + title="Pro Django", published=datetime.date(2008, 12, 16) + ) book.authors.add(auth) with self.assertRaises(RouterUsed) as cm: with self.override_router(): - auth.book_set.all().update(title='Different') + auth.book_set.all().update(title="Different") e = cm.exception self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.model, Book) - self.assertEqual(e.hints, {'instance': auth}) + self.assertEqual(e.hints, {"instance": auth}) class NoRelationRouter: """Disallow all relations.""" + def allow_relation(self, obj1, obj2, **hints): return False @@ -2093,17 +2487,18 @@ class NoRelationRouter: @override_settings(DATABASE_ROUTERS=[NoRelationRouter()]) class RelationAssignmentTests(SimpleTestCase): """allow_relation() is called with unsaved model instances.""" - databases = {'default', 'other'} - router_prevents_msg = 'the current database router prevents this relation' + + databases = {"default", "other"} + router_prevents_msg = "the current database router prevents this relation" def test_foreign_key_relation(self): - person = Person(name='Someone') + person = Person(name="Someone") pet = Pet() with self.assertRaisesMessage(ValueError, self.router_prevents_msg): pet.owner = person def test_reverse_one_to_one_relation(self): - user = User(username='Someone', password='fake_hash') + user = User(username="Someone", password="fake_hash") profile = UserProfile() with self.assertRaisesMessage(ValueError, self.router_prevents_msg): user.userprofile = profile |