diff options
| author | Florian Apolloner <florian@apolloner.eu> | 2013-02-26 09:53:47 +0100 |
|---|---|---|
| committer | Florian Apolloner <florian@apolloner.eu> | 2013-02-26 14:36:57 +0100 |
| commit | 89f40e36246100df6a11316c31a76712ebc6c501 (patch) | |
| tree | 6e65639683ddaf2027908d1ecb1739e0e2ff853b /tests/regressiontests/multiple_database | |
| parent | b3d2ccb5bfbaf6e7fe1f98843baaa48c35a70950 (diff) | |
Merged regressiontests and modeltests into the test root.
Diffstat (limited to 'tests/regressiontests/multiple_database')
7 files changed, 0 insertions, 2109 deletions
diff --git a/tests/regressiontests/multiple_database/__init__.py b/tests/regressiontests/multiple_database/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 --- a/tests/regressiontests/multiple_database/__init__.py +++ /dev/null diff --git a/tests/regressiontests/multiple_database/fixtures/multidb-common.json b/tests/regressiontests/multiple_database/fixtures/multidb-common.json deleted file mode 100644 index 33134173b9..0000000000 --- a/tests/regressiontests/multiple_database/fixtures/multidb-common.json +++ /dev/null @@ -1,10 +0,0 @@ -[ - { - "pk": 1, - "model": "multiple_database.book", - "fields": { - "title": "The Definitive Guide to Django", - "published": "2009-7-8" - } - } -]
\ No newline at end of file diff --git a/tests/regressiontests/multiple_database/fixtures/multidb.default.json b/tests/regressiontests/multiple_database/fixtures/multidb.default.json deleted file mode 100644 index 379b18a803..0000000000 --- a/tests/regressiontests/multiple_database/fixtures/multidb.default.json +++ /dev/null @@ -1,26 +0,0 @@ -[ - { - "pk": 1, - "model": "multiple_database.person", - "fields": { - "name": "Marty Alchin" - } - }, - { - "pk": 2, - "model": "multiple_database.person", - "fields": { - "name": "George Vilches" - } - }, - { - "pk": 2, - "model": "multiple_database.book", - "fields": { - "title": "Pro Django", - "published": "2008-12-16", - "authors": [["Marty Alchin"]], - "editor": ["George Vilches"] - } - } -] diff --git a/tests/regressiontests/multiple_database/fixtures/multidb.other.json b/tests/regressiontests/multiple_database/fixtures/multidb.other.json deleted file mode 100644 index c64f490201..0000000000 --- a/tests/regressiontests/multiple_database/fixtures/multidb.other.json +++ /dev/null @@ -1,26 +0,0 @@ -[ - { - "pk": 1, - "model": "multiple_database.person", - "fields": { - "name": "Mark Pilgrim" - } - }, - { - "pk": 2, - "model": "multiple_database.person", - "fields": { - "name": "Chris Mills" - } - }, - { - "pk": 2, - "model": "multiple_database.book", - "fields": { - "title": "Dive into Python", - "published": "2009-5-4", - "authors": [["Mark Pilgrim"]], - "editor": ["Chris Mills"] - } - } -]
\ No newline at end of file diff --git a/tests/regressiontests/multiple_database/fixtures/pets.json b/tests/regressiontests/multiple_database/fixtures/pets.json deleted file mode 100644 index 89756a3e5b..0000000000 --- a/tests/regressiontests/multiple_database/fixtures/pets.json +++ /dev/null @@ -1,18 +0,0 @@ -[ - { - "pk": 1, - "model": "multiple_database.pet", - "fields": { - "name": "Mr Bigglesworth", - "owner": 1 - } - }, - { - "pk": 2, - "model": "multiple_database.pet", - "fields": { - "name": "Spot", - "owner": 2 - } - } -]
\ No newline at end of file diff --git a/tests/regressiontests/multiple_database/models.py b/tests/regressiontests/multiple_database/models.py deleted file mode 100644 index e46438649b..0000000000 --- a/tests/regressiontests/multiple_database/models.py +++ /dev/null @@ -1,83 +0,0 @@ -from __future__ import absolute_import - -from django.contrib.auth.models import User -from django.contrib.contenttypes.models import ContentType -from django.contrib.contenttypes import generic -from django.db import models -from django.utils.encoding import python_2_unicode_compatible - - -@python_2_unicode_compatible -class Review(models.Model): - source = models.CharField(max_length=100) - content_type = models.ForeignKey(ContentType) - object_id = models.PositiveIntegerField() - content_object = generic.GenericForeignKey() - - def __str__(self): - return self.source - - class Meta: - ordering = ('source',) - -class PersonManager(models.Manager): - def get_by_natural_key(self, name): - return self.get(name=name) - -@python_2_unicode_compatible -class Person(models.Model): - objects = PersonManager() - name = models.CharField(max_length=100) - - def __str__(self): - return self.name - - class Meta: - ordering = ('name',) - -# This book manager doesn't do anything interesting; it just -# exists to strip out the 'extra_arg' argument to certain -# calls. This argument is used to establish that the BookManager -# is actually getting used when it should be. -class BookManager(models.Manager): - def create(self, *args, **kwargs): - kwargs.pop('extra_arg', None) - return super(BookManager, self).create(*args, **kwargs) - - def get_or_create(self, *args, **kwargs): - kwargs.pop('extra_arg', None) - return super(BookManager, self).get_or_create(*args, **kwargs) - -@python_2_unicode_compatible -class Book(models.Model): - objects = BookManager() - title = models.CharField(max_length=100) - published = models.DateField() - authors = models.ManyToManyField(Person) - editor = models.ForeignKey(Person, null=True, related_name='edited') - reviews = generic.GenericRelation(Review) - pages = models.IntegerField(default=100) - - def __str__(self): - return self.title - - class Meta: - ordering = ('title',) - -@python_2_unicode_compatible -class Pet(models.Model): - name = models.CharField(max_length=100) - owner = models.ForeignKey(Person) - - def __str__(self): - return self.name - - class Meta: - ordering = ('name',) - -class UserProfile(models.Model): - user = models.OneToOneField(User, null=True) - flavor = models.CharField(max_length=100) - - class Meta: - ordering = ('flavor',) diff --git a/tests/regressiontests/multiple_database/tests.py b/tests/regressiontests/multiple_database/tests.py deleted file mode 100644 index 2bff7b3b66..0000000000 --- a/tests/regressiontests/multiple_database/tests.py +++ /dev/null @@ -1,1946 +0,0 @@ -from __future__ import absolute_import, unicode_literals - -import datetime -import pickle -from operator import attrgetter -import warnings - -from django.contrib.auth.models import User -from django.contrib.contenttypes.models import ContentType -from django.core import management -from django.db import connections, router, DEFAULT_DB_ALIAS -from django.db.models import signals -from django.test import TestCase -from django.test.utils import override_settings -from django.utils.six import StringIO - -from .models import Book, Person, Pet, Review, UserProfile - - -class QueryTestCase(TestCase): - multi_db = True - - def test_db_selection(self): - "Check that querysets will use the default database by default" - self.assertEqual(Book.objects.db, DEFAULT_DB_ALIAS) - self.assertEqual(Book.objects.all().db, DEFAULT_DB_ALIAS) - - self.assertEqual(Book.objects.using('other').db, 'other') - - self.assertEqual(Book.objects.db_manager('other').db, 'other') - self.assertEqual(Book.objects.db_manager('other').all().db, 'other') - - def test_default_creation(self): - "Objects created on the default database don't leak onto other databases" - # Create a book on the default database using create() - Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - # Create a book on the default database using a save - dive = Book() - dive.title="Dive into Python" - dive.published = datetime.date(2009, 5, 4) - dive.save() - - # Check that book exists on the default database, but not on other database - try: - Book.objects.get(title="Pro Django") - Book.objects.using('default').get(title="Pro Django") - except Book.DoesNotExist: - self.fail('"Dive Into Python" should exist on default database') - - self.assertRaises(Book.DoesNotExist, - Book.objects.using('other').get, - title="Pro Django" - ) - - try: - Book.objects.get(title="Dive into Python") - Book.objects.using('default').get(title="Dive into Python") - except Book.DoesNotExist: - self.fail('"Dive into Python" should exist on default database') - - self.assertRaises(Book.DoesNotExist, - Book.objects.using('other').get, - title="Dive into Python" - ) - - - def test_other_creation(self): - "Objects created on another database don't leak onto the default database" - # Create a book on the second database - Book.objects.using('other').create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - # Create a book on the default database using a save - dive = Book() - dive.title="Dive into Python" - dive.published = datetime.date(2009, 5, 4) - dive.save(using='other') - - # Check that book exists on the default database, but not on other database - try: - Book.objects.using('other').get(title="Pro Django") - except Book.DoesNotExist: - self.fail('"Dive Into Python" should exist on other database') - - self.assertRaises(Book.DoesNotExist, - Book.objects.get, - title="Pro Django" - ) - self.assertRaises(Book.DoesNotExist, - Book.objects.using('default').get, - title="Pro Django" - ) - - try: - Book.objects.using('other').get(title="Dive into Python") - except Book.DoesNotExist: - self.fail('"Dive into Python" should exist on other database') - - self.assertRaises(Book.DoesNotExist, - Book.objects.get, - title="Dive into Python" - ) - self.assertRaises(Book.DoesNotExist, - Book.objects.using('default').get, - title="Dive into Python" - ) - - def test_basic_queries(self): - "Queries are constrained to a single database" - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - dive = Book.objects.using('other').get(published=datetime.date(2009, 5, 4)) - self.assertEqual(dive.title, "Dive into Python") - self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, published=datetime.date(2009, 5, 4)) - - dive = Book.objects.using('other').get(title__icontains="dive") - self.assertEqual(dive.title, "Dive into Python") - self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, title__icontains="dive") - - dive = Book.objects.using('other').get(title__iexact="dive INTO python") - self.assertEqual(dive.title, "Dive into Python") - self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, title__iexact="dive INTO python") - - dive = Book.objects.using('other').get(published__year=2009) - self.assertEqual(dive.title, "Dive into Python") - self.assertEqual(dive.published, datetime.date(2009, 5, 4)) - self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, published__year=2009) - - years = Book.objects.using('other').dates('published', 'year') - self.assertEqual([o.year for o in years], [2009]) - years = Book.objects.using('default').dates('published', 'year') - self.assertEqual([o.year for o in years], []) - - months = Book.objects.using('other').dates('published', 'month') - self.assertEqual([o.month for o in months], [5]) - months = Book.objects.using('default').dates('published', 'month') - self.assertEqual([o.month for o in months], []) - - def test_m2m_separation(self): - "M2M fields are constrained to a single database" - # Create a book and author on the default database - pro = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - marty = Person.objects.create(name="Marty Alchin") - - # Create a book and author on the other database - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - mark = Person.objects.using('other').create(name="Mark Pilgrim") - - # Save the author relations - pro.authors = [marty] - dive.authors = [mark] - - # Inspect the m2m tables directly. - # There should be 1 entry in each database - self.assertEqual(Book.authors.through.objects.using('default').count(), 1) - self.assertEqual(Book.authors.through.objects.using('other').count(), 1) - - # Check that queries work across m2m joins - self.assertEqual(list(Book.objects.using('default').filter(authors__name='Marty Alchin').values_list('title', flat=True)), - ['Pro Django']) - self.assertEqual(list(Book.objects.using('other').filter(authors__name='Marty Alchin').values_list('title', flat=True)), - []) - - self.assertEqual(list(Book.objects.using('default').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), - []) - self.assertEqual(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), - ['Dive into Python']) - - # Reget the objects to clear caches - dive = Book.objects.using('other').get(title="Dive into Python") - mark = Person.objects.using('other').get(name="Mark Pilgrim") - - # Retrive related object by descriptor. Related objects should be database-baound - self.assertEqual(list(dive.authors.all().values_list('name', flat=True)), - ['Mark Pilgrim']) - - self.assertEqual(list(mark.book_set.all().values_list('title', flat=True)), - ['Dive into Python']) - - def test_m2m_forward_operations(self): - "M2M forward manipulations are all constrained to a single DB" - # Create a book and author on the other database - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - mark = Person.objects.using('other').create(name="Mark Pilgrim") - - # Save the author relations - dive.authors = [mark] - - # Add a second author - john = Person.objects.using('other').create(name="John Smith") - self.assertEqual(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)), - []) - - - dive.authors.add(john) - self.assertEqual(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), - ['Dive into Python']) - self.assertEqual(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)), - ['Dive into Python']) - - # Remove the second author - dive.authors.remove(john) - self.assertEqual(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), - ['Dive into Python']) - self.assertEqual(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)), - []) - - # Clear all authors - dive.authors.clear() - self.assertEqual(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), - []) - self.assertEqual(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)), - []) - - # Create an author through the m2m interface - dive.authors.create(name='Jane Brown') - self.assertEqual(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)), - []) - self.assertEqual(list(Book.objects.using('other').filter(authors__name='Jane Brown').values_list('title', flat=True)), - ['Dive into Python']) - - def test_m2m_reverse_operations(self): - "M2M reverse manipulations are all constrained to a single DB" - # Create a book and author on the other database - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - mark = Person.objects.using('other').create(name="Mark Pilgrim") - - # Save the author relations - dive.authors = [mark] - - # Create a second book on the other database - grease = Book.objects.using('other').create(title="Greasemonkey Hacks", - published=datetime.date(2005, 11, 1)) - - # Add a books to the m2m - mark.book_set.add(grease) - self.assertEqual(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)), - ['Mark Pilgrim']) - self.assertEqual(list(Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True)), - ['Mark Pilgrim']) - - # Remove a book from the m2m - mark.book_set.remove(grease) - self.assertEqual(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)), - ['Mark Pilgrim']) - self.assertEqual(list(Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True)), - []) - - # Clear the books associated with mark - mark.book_set.clear() - self.assertEqual(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)), - []) - self.assertEqual(list(Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True)), - []) - - # Create a book through the m2m interface - mark.book_set.create(title="Dive into HTML5", published=datetime.date(2020, 1, 1)) - self.assertEqual(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)), - []) - self.assertEqual(list(Person.objects.using('other').filter(book__title='Dive into HTML5').values_list('name', flat=True)), - ['Mark Pilgrim']) - - def test_m2m_cross_database_protection(self): - "Operations that involve sharing M2M objects across databases raise an error" - # Create a book and author on the default database - pro = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - marty = Person.objects.create(name="Marty Alchin") - - # Create a book and author on the other database - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - mark = Person.objects.using('other').create(name="Mark Pilgrim") - # Set a foreign key set with an object from a different database - try: - marty.book_set = [pro, dive] - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - # Add to an m2m with an object from a different database - try: - marty.book_set.add(dive) - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - # Set a m2m with an object from a different database - try: - marty.book_set = [pro, dive] - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - # Add to a reverse m2m with an object from a different database - try: - dive.authors.add(marty) - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - # Set a reverse m2m with an object from a different database - try: - dive.authors = [mark, marty] - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - def test_m2m_deletion(self): - "Cascaded deletions of m2m relations issue queries on the right database" - # Create a book and author on the other database - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - mark = Person.objects.using('other').create(name="Mark Pilgrim") - dive.authors = [mark] - - # Check the initial state - self.assertEqual(Person.objects.using('default').count(), 0) - self.assertEqual(Book.objects.using('default').count(), 0) - self.assertEqual(Book.authors.through.objects.using('default').count(), 0) - - self.assertEqual(Person.objects.using('other').count(), 1) - self.assertEqual(Book.objects.using('other').count(), 1) - self.assertEqual(Book.authors.through.objects.using('other').count(), 1) - - # Delete the object on the other database - dive.delete(using='other') - - self.assertEqual(Person.objects.using('default').count(), 0) - self.assertEqual(Book.objects.using('default').count(), 0) - self.assertEqual(Book.authors.through.objects.using('default').count(), 0) - - # The person still exists ... - self.assertEqual(Person.objects.using('other').count(), 1) - # ... but the book has been deleted - self.assertEqual(Book.objects.using('other').count(), 0) - # ... and the relationship object has also been deleted. - self.assertEqual(Book.authors.through.objects.using('other').count(), 0) - - # Now try deletion in the reverse direction. Set up the relation again - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - dive.authors = [mark] - - # Check the initial state - self.assertEqual(Person.objects.using('default').count(), 0) - self.assertEqual(Book.objects.using('default').count(), 0) - self.assertEqual(Book.authors.through.objects.using('default').count(), 0) - - self.assertEqual(Person.objects.using('other').count(), 1) - self.assertEqual(Book.objects.using('other').count(), 1) - self.assertEqual(Book.authors.through.objects.using('other').count(), 1) - - # Delete the object on the other database - mark.delete(using='other') - - self.assertEqual(Person.objects.using('default').count(), 0) - self.assertEqual(Book.objects.using('default').count(), 0) - self.assertEqual(Book.authors.through.objects.using('default').count(), 0) - - # The person has been deleted ... - self.assertEqual(Person.objects.using('other').count(), 0) - # ... but the book still exists - self.assertEqual(Book.objects.using('other').count(), 1) - # ... and the relationship object has been deleted. - self.assertEqual(Book.authors.through.objects.using('other').count(), 0) - - def test_foreign_key_separation(self): - "FK fields are constrained to a single database" - # Create a book and author on the default database - pro = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - marty = Person.objects.create(name="Marty Alchin") - 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)) - - mark = Person.objects.using('other').create(name="Mark Pilgrim") - chris = Person.objects.using('other').create(name="Chris Mills") - - # Save the author's favourite books - pro.editor = george - pro.save() - - dive.editor = chris - dive.save() - - pro = Book.objects.using('default').get(title="Pro Django") - self.assertEqual(pro.editor.name, "George Vilches") - - dive = Book.objects.using('other').get(title="Dive into Python") - self.assertEqual(dive.editor.name, "Chris Mills") - - # Check that queries work across foreign key joins - self.assertEqual(list(Person.objects.using('default').filter(edited__title='Pro Django').values_list('name', flat=True)), - ['George Vilches']) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Pro Django').values_list('name', flat=True)), - []) - - self.assertEqual(list(Person.objects.using('default').filter(edited__title='Dive into Python').values_list('name', flat=True)), - []) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)), - ['Chris Mills']) - - # Reget the objects to clear caches - chris = Person.objects.using('other').get(name="Chris Mills") - dive = Book.objects.using('other').get(title="Dive into Python") - - # Retrive related object by descriptor. Related objects should be database-baound - 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)) - - mark = Person.objects.using('other').create(name="Mark Pilgrim") - chris = Person.objects.using('other').create(name="Chris Mills") - - # Save the author relations - dive.editor = chris - dive.save() - - # Add a second book edited by chris - html5 = Book.objects.using('other').create(title="Dive into HTML5", published=datetime.date(2010, 3, 15)) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)), - []) - - chris.edited.add(html5) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)), - ['Chris Mills']) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)), - ['Chris Mills']) - - # Remove the second editor - chris.edited.remove(html5) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)), - []) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)), - ['Chris Mills']) - - # Clear all edited books - chris.edited.clear() - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)), - []) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)), - []) - - # Create an author through the m2m interface - chris.edited.create(title='Dive into Water', published=datetime.date(2010, 3, 15)) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)), - []) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Water').values_list('name', flat=True)), - ['Chris Mills']) - self.assertEqual(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)), - []) - - def test_foreign_key_cross_database_protection(self): - "Operations that involve sharing FK objects across databases raise an error" - # Create a book and author on the default database - pro = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - marty = Person.objects.create(name="Marty Alchin") - - # Create a book and author on the other database - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - mark = Person.objects.using('other').create(name="Mark Pilgrim") - - # Set a foreign key with an object from a different database - try: - dive.editor = marty - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - # Set a foreign key set with an object from a different database - try: - marty.edited = [pro, dive] - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - # Add to a foreign key set with an object from a different database - try: - marty.edited.add(dive) - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - # 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. - chris = Person(name="Chris Mills") - html5 = Book(title="Dive into HTML5", published=datetime.date(2010, 3, 15)) - # initially, no db assigned - self.assertEqual(chris._state.db, None) - self.assertEqual(html5._state.db, None) - - # old object comes from 'other', so the new object is set to use 'other'... - dive.editor = chris - html5.editor = mark - self.assertEqual(chris._state.db, 'other') - self.assertEqual(html5._state.db, 'other') - # ... but it isn't saved yet - self.assertEqual(list(Person.objects.using('other').values_list('name',flat=True)), - ['Mark Pilgrim']) - self.assertEqual(list(Book.objects.using('other').values_list('title',flat=True)), - ['Dive into Python']) - - # When saved (no using required), new objects goes to 'other' - chris.save() - html5.save() - self.assertEqual(list(Person.objects.using('default').values_list('name',flat=True)), - ['Marty Alchin']) - self.assertEqual(list(Person.objects.using('other').values_list('name',flat=True)), - ['Chris Mills', 'Mark Pilgrim']) - self.assertEqual(list(Book.objects.using('default').values_list('title',flat=True)), - ['Pro Django']) - self.assertEqual(list(Book.objects.using('other').values_list('title',flat=True)), - ['Dive into HTML5', 'Dive into Python']) - - # 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, 'other') - # ... but it isn't saved yet - self.assertEqual(list(Book.objects.using('default').values_list('title',flat=True)), - ['Pro Django']) - self.assertEqual(list(Book.objects.using('other').values_list('title',flat=True)), - ['Dive into HTML5', 'Dive into Python']) - - # When saved, the new book goes to 'other' - water.save() - self.assertEqual(list(Book.objects.using('default').values_list('title',flat=True)), - ['Pro Django']) - self.assertEqual(list(Book.objects.using('other').values_list('title',flat=True)), - ['Dive into HTML5', 'Dive into Python', 'Dive into Water']) - - 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") - fido = Pet.objects.using('other').create(name="Fido", owner=mark) - - # Check the initial state - self.assertEqual(Person.objects.using('default').count(), 0) - self.assertEqual(Pet.objects.using('default').count(), 0) - - self.assertEqual(Person.objects.using('other').count(), 1) - self.assertEqual(Pet.objects.using('other').count(), 1) - - # Delete the person object, which will cascade onto the pet - mark.delete(using='other') - - self.assertEqual(Person.objects.using('default').count(), 0) - self.assertEqual(Pet.objects.using('default').count(), 0) - - # Both the pet and the person have been deleted from the right database - self.assertEqual(Person.objects.using('other').count(), 0) - self.assertEqual(Pet.objects.using('other').count(), 0) - - def test_foreign_key_validation(self): - "ForeignKey.validate() uses the correct database" - mickey = Person.objects.using('other').create(name="Mickey") - pluto = Pet.objects.using('other').create(name="Pluto", owner=mickey) - self.assertEqual(None, pluto.full_clean()) - - def test_o2o_separation(self): - "OneToOne fields are constrained to a single database" - # Create a user and profile on the default database - alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com') - alice_profile = UserProfile.objects.using('default').create(user=alice, flavor='chocolate') - - # Create a user and profile on the other database - bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com') - bob_profile = UserProfile.objects.using('other').create(user=bob, flavor='crunchy frog') - - # Retrieve related objects; queries should be database constrained - alice = User.objects.using('default').get(username="alice") - self.assertEqual(alice.userprofile.flavor, "chocolate") - - bob = User.objects.using('other').get(username="bob") - self.assertEqual(bob.userprofile.flavor, "crunchy frog") - - # Check that queries work across joins - self.assertEqual(list(User.objects.using('default').filter(userprofile__flavor='chocolate').values_list('username', flat=True)), - ['alice']) - self.assertEqual(list(User.objects.using('other').filter(userprofile__flavor='chocolate').values_list('username', flat=True)), - []) - - self.assertEqual(list(User.objects.using('default').filter(userprofile__flavor='crunchy frog').values_list('username', flat=True)), - []) - self.assertEqual(list(User.objects.using('other').filter(userprofile__flavor='crunchy frog').values_list('username', flat=True)), - ['bob']) - - # Reget the objects to clear caches - alice_profile = UserProfile.objects.using('default').get(flavor='chocolate') - bob_profile = UserProfile.objects.using('other').get(flavor='crunchy frog') - - # Retrive related object by descriptor. Related objects should be database-baound - self.assertEqual(alice_profile.user.username, 'alice') - self.assertEqual(bob_profile.user.username, 'bob') - - def test_o2o_cross_database_protection(self): - "Operations that involve sharing FK objects across databases raise an error" - # Create a user and profile on the default database - alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com') - - # Create a user and profile on the other database - bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com') - - # Set a one-to-one relation with an object from a different database - alice_profile = UserProfile.objects.using('default').create(user=alice, flavor='chocolate') - try: - bob.userprofile = alice_profile - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - # BUT! if you assign a FK object when the base object hasn't - # been saved yet, you implicitly assign the database for the - # base object. - bob_profile = UserProfile.objects.using('other').create(user=bob, flavor='crunchy frog') - - new_bob_profile = UserProfile(flavor="spring surprise") - - # assigning a profile requires a explicit pk as the object isn't saved - charlie = User(pk=51, username='charlie', email='charlie@example.com') - charlie.set_unusable_password() - - # initially, no db assigned - self.assertEqual(new_bob_profile._state.db, None) - self.assertEqual(charlie._state.db, None) - - # old object comes from 'other', so the new object is set to use 'other'... - new_bob_profile.user = bob - charlie.userprofile = bob_profile - self.assertEqual(new_bob_profile._state.db, 'other') - self.assertEqual(charlie._state.db, 'other') - - # ... but it isn't saved yet - self.assertEqual(list(User.objects.using('other').values_list('username',flat=True)), - ['bob']) - self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor',flat=True)), - ['crunchy frog']) - - # When saved (no using required), new objects goes to 'other' - charlie.save() - bob_profile.save() - new_bob_profile.save() - self.assertEqual(list(User.objects.using('default').values_list('username',flat=True)), - ['alice']) - self.assertEqual(list(User.objects.using('other').values_list('username',flat=True)), - ['bob', 'charlie']) - self.assertEqual(list(UserProfile.objects.using('default').values_list('flavor',flat=True)), - ['chocolate']) - self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor',flat=True)), - ['crunchy frog', 'spring surprise']) - - # This also works if you assign the O2O relation in the constructor - denise = User.objects.db_manager('other').create_user('denise','denise@example.com') - denise_profile = UserProfile(flavor="tofu", user=denise) - - self.assertEqual(denise_profile._state.db, 'other') - # ... but it isn't saved yet - self.assertEqual(list(UserProfile.objects.using('default').values_list('flavor',flat=True)), - ['chocolate']) - self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor',flat=True)), - ['crunchy frog', 'spring surprise']) - - # When saved, the new profile goes to 'other' - denise_profile.save() - self.assertEqual(list(UserProfile.objects.using('default').values_list('flavor',flat=True)), - ['chocolate']) - self.assertEqual(list(UserProfile.objects.using('other').values_list('flavor',flat=True)), - ['crunchy frog', 'spring surprise', 'tofu']) - - def test_generic_key_separation(self): - "Generic fields are constrained to a single database" - # Create a book and author on the default database - pro = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - review1 = Review.objects.create(source="Python Monthly", content_object=pro) - - # Create a book and author on the other database - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - review2 = Review.objects.using('other').create(source="Python Weekly", content_object=dive) - - review1 = Review.objects.using('default').get(source="Python Monthly") - self.assertEqual(review1.content_object.title, "Pro Django") - - review2 = Review.objects.using('other').get(source="Python Weekly") - self.assertEqual(review2.content_object.title, "Dive into Python") - - # Reget the objects to clear caches - dive = Book.objects.using('other').get(title="Dive into Python") - - # Retrive related object by descriptor. Related objects should be database-bound - self.assertEqual(list(dive.reviews.all().values_list('source', flat=True)), - ['Python Weekly']) - - def test_generic_key_reverse_operations(self): - "Generic reverse manipulations are all constrained to a single DB" - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - temp = Book.objects.using('other').create(title="Temp", - published=datetime.date(2009, 5, 4)) - - review1 = Review.objects.using('other').create(source="Python Weekly", content_object=dive) - review2 = Review.objects.using('other').create(source="Python Monthly", content_object=temp) - - self.assertEqual(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), - []) - self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), - ['Python Weekly']) - - # Add a second review - dive.reviews.add(review2) - self.assertEqual(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), - []) - self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), - ['Python Monthly', 'Python Weekly']) - - # Remove the second author - dive.reviews.remove(review1) - self.assertEqual(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), - []) - self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), - ['Python Monthly']) - - # Clear all reviews - dive.reviews.clear() - self.assertEqual(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), - []) - self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), - []) - - # Create an author through the generic interface - dive.reviews.create(source='Python Daily') - self.assertEqual(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)), - []) - self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)), - ['Python Daily']) - - def test_generic_key_cross_database_protection(self): - "Operations that involve sharing generic key objects across databases raise an error" - # Create a book and author on the default database - pro = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - review1 = Review.objects.create(source="Python Monthly", content_object=pro) - - # Create a book and author on the other database - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - review2 = Review.objects.using('other').create(source="Python Weekly", content_object=dive) - - # Set a foreign key with an object from a different database - try: - review1.content_object = dive - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - # Add to a foreign key set with an object from a different database - try: - dive.reviews.add(review1) - self.fail("Shouldn't be able to assign across databases") - except ValueError: - pass - - # BUT! if you assign a FK object when the base object hasn't - # been saved yet, you implicitly assign the database for the - # base object. - review3 = Review(source="Python Daily") - # initially, no db assigned - self.assertEqual(review3._state.db, None) - - # Dive comes from 'other', so review3 is set to use 'other'... - review3.content_object = dive - self.assertEqual(review3._state.db, 'other') - # ... but it isn't saved yet - self.assertEqual(list(Review.objects.using('default').filter(object_id=pro.pk).values_list('source', flat=True)), - ['Python Monthly']) - self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source',flat=True)), - ['Python Weekly']) - - # When saved, John goes to 'other' - review3.save() - self.assertEqual(list(Review.objects.using('default').filter(object_id=pro.pk).values_list('source', flat=True)), - ['Python Monthly']) - self.assertEqual(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source',flat=True)), - ['Python Daily', 'Python Weekly']) - - def test_generic_key_deletion(self): - "Cascaded deletions of Generic Key relations issue queries on the right database" - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - review = Review.objects.using('other').create(source="Python Weekly", content_object=dive) - - # Check the initial state - self.assertEqual(Book.objects.using('default').count(), 0) - self.assertEqual(Review.objects.using('default').count(), 0) - - self.assertEqual(Book.objects.using('other').count(), 1) - self.assertEqual(Review.objects.using('other').count(), 1) - - # Delete the Book object, which will cascade onto the pet - dive.delete(using='other') - - self.assertEqual(Book.objects.using('default').count(), 0) - self.assertEqual(Review.objects.using('default').count(), 0) - - # Both the pet and the person have been deleted from the right database - self.assertEqual(Book.objects.using('other').count(), 0) - self.assertEqual(Review.objects.using('other').count(), 0) - - def test_ordering(self): - "get_next_by_XXX commands stick to a single database" - pro = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - learn = Book.objects.using('other').create(title="Learning Python", - published=datetime.date(2008, 7, 16)) - - self.assertEqual(learn.get_next_by_published().title, "Dive into Python") - self.assertEqual(dive.get_previous_by_published().title, "Learning Python") - - def test_raw(self): - "test the raw() method across databases" - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - val = Book.objects.db_manager("other").raw('SELECT id FROM multiple_database_book') - self.assertQuerysetEqual(val, [dive.pk], attrgetter("pk")) - - val = Book.objects.raw('SELECT id FROM multiple_database_book').using('other') - self.assertQuerysetEqual(val, [dive.pk], attrgetter("pk")) - - def test_select_related(self): - "Database assignment is retained if an object is retrieved with select_related()" - # Create a book and author on the other database - mark = Person.objects.using('other').create(name="Mark Pilgrim") - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4), - editor=mark) - - # Retrieve the Person using select_related() - book = Book.objects.using('other').select_related('editor').get(title="Dive into Python") - - # The editor instance should have a db state - self.assertEqual(book.editor._state.db, 'other') - - def test_subquery(self): - """Make sure as_sql works with subqueries and master/slave.""" - sub = Person.objects.using('other').filter(name='fff') - qs = Book.objects.filter(editor__in=sub) - - # When you call __str__ on the query object, it doesn't know about using - # so it falls back to the default. If the subquery explicitly uses a - # different database, an error should be raised. - self.assertRaises(ValueError, str, qs.query) - - # Evaluating the query shouldn't work, either - try: - for obj in qs: - pass - self.fail('Iterating over query should raise ValueError') - except ValueError: - pass - - def test_related_manager(self): - "Related managers return managers, not querysets" - mark = Person.objects.using('other').create(name="Mark Pilgrim") - - # extra_arg is removed by the BookManager's implementation of - # create(); but the BookManager's implementation won't get called - # unless edited returns a Manager, not a queryset - mark.book_set.create(title="Dive into Python", - published=datetime.date(2009, 5, 4), - extra_arg=True) - - mark.book_set.get_or_create(title="Dive into Python", - published=datetime.date(2009, 5, 4), - extra_arg=True) - - mark.edited.create(title="Dive into Water", - published=datetime.date(2009, 5, 4), - extra_arg=True) - - mark.edited.get_or_create(title="Dive into Water", - published=datetime.date(2009, 5, 4), - extra_arg=True) - -class TestRouter(object): - # A test router. The behavior is vaguely master/slave, but the - # databases aren't assumed to propagate changes. - def db_for_read(self, model, instance=None, **hints): - if instance: - 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') - - def allow_syncdb(self, db, model): - return True - -class AuthRouter(object): - """A router to control all database operations on models in - the contrib.auth application""" - - def db_for_read(self, model, **hints): - "Point all read operations on auth models to 'default'" - 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 None - - def db_for_write(self, model, **hints): - "Point all operations on auth models to '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" - if obj1._meta.app_label == 'auth' or obj2._meta.app_label == 'auth': - return True - return None - - def allow_syncdb(self, db, model): - "Make sure the auth app only appears on the 'other' db" - if db == 'other': - return model._meta.app_label == 'auth' - elif model._meta.app_label == 'auth': - return False - return None - -class WriteRouter(object): - # A router that only expresses an opinion on writes - def db_for_write(self, model, **hints): - return 'writer' - -class RouterTestCase(TestCase): - multi_db = True - - def setUp(self): - # Make the 'other' database appear to be a slave of the 'default' - self.old_routers = router.routers - router.routers = [TestRouter()] - - def tearDown(self): - # Restore the 'other' database as an independent database - router.routers = self.old_routers - - def test_db_selection(self): - "Check that querysets obey the router for db suggestions" - self.assertEqual(Book.objects.db, 'other') - self.assertEqual(Book.objects.all().db, 'other') - - self.assertEqual(Book.objects.using('default').db, 'default') - - self.assertEqual(Book.objects.db_manager('default').db, 'default') - self.assertEqual(Book.objects.db_manager('default').all().db, 'default') - - def test_syncdb_selection(self): - "Synchronization behavior is predictable" - - self.assertTrue(router.allow_syncdb('default', User)) - self.assertTrue(router.allow_syncdb('default', Book)) - - self.assertTrue(router.allow_syncdb('other', User)) - self.assertTrue(router.allow_syncdb('other', Book)) - - # Add the auth router to the chain. - # TestRouter is a universal synchronizer, so it should have no effect. - router.routers = [TestRouter(), AuthRouter()] - - self.assertTrue(router.allow_syncdb('default', User)) - self.assertTrue(router.allow_syncdb('default', Book)) - - self.assertTrue(router.allow_syncdb('other', User)) - self.assertTrue(router.allow_syncdb('other', Book)) - - # Now check what happens if the router order is the other way around - router.routers = [AuthRouter(), TestRouter()] - - self.assertFalse(router.allow_syncdb('default', User)) - self.assertTrue(router.allow_syncdb('default', Book)) - - self.assertTrue(router.allow_syncdb('other', User)) - self.assertFalse(router.allow_syncdb('other', Book)) - - def test_partial_router(self): - "A router can choose to implement a subset of methods" - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - # First check the baseline behavior. - - self.assertEqual(router.db_for_read(User), 'other') - self.assertEqual(router.db_for_read(Book), 'other') - - self.assertEqual(router.db_for_write(User), 'default') - self.assertEqual(router.db_for_write(Book), 'default') - - self.assertTrue(router.allow_relation(dive, dive)) - - self.assertTrue(router.allow_syncdb('default', User)) - self.assertTrue(router.allow_syncdb('default', Book)) - - router.routers = [WriteRouter(), AuthRouter(), TestRouter()] - - self.assertEqual(router.db_for_read(User), 'default') - self.assertEqual(router.db_for_read(Book), 'other') - - self.assertEqual(router.db_for_write(User), 'writer') - self.assertEqual(router.db_for_write(Book), 'writer') - - self.assertTrue(router.allow_relation(dive, dive)) - - self.assertFalse(router.allow_syncdb('default', User)) - self.assertTrue(router.allow_syncdb('default', Book)) - - - def test_database_routing(self): - marty = Person.objects.using('default').create(name="Marty Alchin") - pro = Book.objects.using('default').create(title="Pro Django", - published=datetime.date(2008, 12, 16), - editor=marty) - pro.authors = [marty] - - # 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)) - - # An update query will be routed to the default database - Book.objects.filter(title='Pro Django').update(pages=200) - - try: - # By default, the get query will be directed to 'other' - Book.objects.get(title='Pro Django') - self.fail("Shouldn't be able to find the book") - except Book.DoesNotExist: - pass - - # But the same query issued explicitly at a database will work. - pro = Book.objects.using('default').get(title='Pro Django') - - # Check that the update worked. - self.assertEqual(pro.pages, 200) - - # An update query with an explicit using clause will be routed - # to the requested database. - Book.objects.using('other').filter(title='Dive into Python').update(pages=300) - self.assertEqual(Book.objects.get(title='Dive into Python').pages, 300) - - # Related object queries stick to the same database - # as the original object, regardless of the router - self.assertEqual(list(pro.authors.values_list('name', flat=True)), ['Marty Alchin']) - self.assertEqual(pro.editor.name, 'Marty Alchin') - - # get_or_create is a special case. The get needs to be targeted at - # the write database in order to avoid potential transaction - # consistency problems - book, created = Book.objects.get_or_create(title="Pro Django") - self.assertFalse(created) - - book, created = Book.objects.get_or_create(title="Dive Into Python", - defaults={'published':datetime.date(2009, 5, 4)}) - self.assertTrue(created) - - # Check the head count of objects - self.assertEqual(Book.objects.using('default').count(), 2) - self.assertEqual(Book.objects.using('other').count(), 1) - # If a database isn't specified, the read database is used - self.assertEqual(Book.objects.count(), 1) - - # A delete query will also be routed to the default database - Book.objects.filter(pages__gt=150).delete() - - # The default database has lost the book. - self.assertEqual(Book.objects.using('default').count(), 1) - self.assertEqual(Book.objects.using('other').count(), 1) - - def test_foreign_key_cross_database_protection(self): - "Foreign keys can cross databases if they two databases have a common source" - # Create a book and author on the default database - pro = Book.objects.using('default').create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - marty = Person.objects.using('default').create(name="Marty Alchin") - - # Create a book and author on the other database - dive = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - mark = Person.objects.using('other').create(name="Mark Pilgrim") - - # Set a foreign key with an object from a different database - try: - dive.editor = marty - except ValueError: - self.fail("Assignment across master/slave databases with a common source should be ok") - - # Database assignments of original objects haven't changed... - self.assertEqual(marty._state.db, 'default') - self.assertEqual(pro._state.db, 'default') - self.assertEqual(dive._state.db, 'other') - self.assertEqual(mark._state.db, 'other') - - # ... but they will when the affected object is saved. - dive.save() - self.assertEqual(dive._state.db, 'default') - - # ...and the source database now has a copy of any object saved - try: - Book.objects.using('default').get(title='Dive into Python').delete() - except Book.DoesNotExist: - self.fail('Source database should have a copy of saved object') - - # This isn't a real master-slave database, so restore the original from other - dive = Book.objects.using('other').get(title='Dive into Python') - self.assertEqual(dive._state.db, 'other') - - # Set a foreign key set with an object from a different database - try: - marty.edited = [pro, dive] - except ValueError: - self.fail("Assignment across master/slave databases with a common source should be ok") - - # Assignment implies a save, so database assignments of original objects have changed... - self.assertEqual(marty._state.db, 'default') - self.assertEqual(pro._state.db, 'default') - self.assertEqual(dive._state.db, 'default') - self.assertEqual(mark._state.db, 'other') - - # ...and the source database now has a copy of any object saved - try: - Book.objects.using('default').get(title='Dive into Python').delete() - except Book.DoesNotExist: - self.fail('Source database should have a copy of saved object') - - # This isn't a real master-slave database, so restore the original from other - dive = Book.objects.using('other').get(title='Dive into Python') - self.assertEqual(dive._state.db, 'other') - - # Add to a foreign key set with an object from a different database - try: - marty.edited.add(dive) - except ValueError: - self.fail("Assignment across master/slave databases with a common source should be ok") - - # Add implies a save, so database assignments of original objects have changed... - self.assertEqual(marty._state.db, 'default') - self.assertEqual(pro._state.db, 'default') - self.assertEqual(dive._state.db, 'default') - self.assertEqual(mark._state.db, 'other') - - # ...and the source database now has a copy of any object saved - try: - Book.objects.using('default').get(title='Dive into Python').delete() - except Book.DoesNotExist: - self.fail('Source database should have a copy of saved object') - - # This isn't a real master-slave database, so restore the original from other - dive = Book.objects.using('other').get(title='Dive into Python') - - # If you assign a FK object when the base object hasn't - # been saved yet, you implicitly assign the database for the - # base object. - chris = Person(name="Chris Mills") - html5 = Book(title="Dive into HTML5", published=datetime.date(2010, 3, 15)) - # initially, no db assigned - self.assertEqual(chris._state.db, None) - self.assertEqual(html5._state.db, None) - - # old object comes from 'other', so the new object is set to use the - # source of 'other'... - self.assertEqual(dive._state.db, 'other') - dive.editor = chris - html5.editor = mark - - self.assertEqual(dive._state.db, 'other') - self.assertEqual(mark._state.db, 'other') - self.assertEqual(chris._state.db, 'default') - self.assertEqual(html5._state.db, 'default') - - # This also works if you assign the FK in the constructor - water = Book(title="Dive into Water", published=datetime.date(2001, 1, 1), editor=mark) - self.assertEqual(water._state.db, 'default') - - # For the remainder of this test, create a copy of 'mark' in the - # 'default' database to prevent integrity errors on backends that - # don't defer constraints checks until the end of the transaction - mark.save(using='default') - - # This moved 'mark' in the 'default' database, move it back in 'other' - mark.save(using='other') - self.assertEqual(mark._state.db, 'other') - - # If you create an object through a FK relation, it will be - # written to the write database, even if the original object - # was on the read database - cheesecake = mark.edited.create(title='Dive into Cheesecake', published=datetime.date(2010, 3, 15)) - self.assertEqual(cheesecake._state.db, 'default') - - # Same goes for get_or_create, regardless of whether getting or creating - cheesecake, created = mark.edited.get_or_create(title='Dive into Cheesecake', published=datetime.date(2010, 3, 15)) - self.assertEqual(cheesecake._state.db, 'default') - - puddles, created = mark.edited.get_or_create(title='Dive into Puddles', published=datetime.date(2010, 3, 15)) - self.assertEqual(puddles._state.db, 'default') - - def test_m2m_cross_database_protection(self): - "M2M relations can cross databases if the database share a source" - # Create books and authors on the inverse to the usual database - pro = Book.objects.using('other').create(pk=1, title="Pro Django", - published=datetime.date(2008, 12, 16)) - - marty = Person.objects.using('other').create(pk=1, name="Marty Alchin") - - dive = Book.objects.using('default').create(pk=2, title="Dive into Python", - published=datetime.date(2009, 5, 4)) - - mark = Person.objects.using('default').create(pk=2, name="Mark Pilgrim") - - # Now save back onto the usual database. - # This simulates master/slave - the objects exist on both database, - # but the _state.db is as it is for all other tests. - pro.save(using='default') - marty.save(using='default') - dive.save(using='other') - mark.save(using='other') - - # Check that we have 2 of both types of object on both databases - self.assertEqual(Book.objects.using('default').count(), 2) - self.assertEqual(Book.objects.using('other').count(), 2) - self.assertEqual(Person.objects.using('default').count(), 2) - self.assertEqual(Person.objects.using('other').count(), 2) - - # Set a m2m set with an object from a different database - try: - marty.book_set = [pro, dive] - except ValueError: - self.fail("Assignment across master/slave databases with a common source should be ok") - - # Database assignments don't change - self.assertEqual(marty._state.db, 'default') - self.assertEqual(pro._state.db, 'default') - self.assertEqual(dive._state.db, 'other') - self.assertEqual(mark._state.db, 'other') - - # All m2m relations should be saved on the default database - self.assertEqual(Book.authors.through.objects.using('default').count(), 2) - self.assertEqual(Book.authors.through.objects.using('other').count(), 0) - - # Reset relations - Book.authors.through.objects.using('default').delete() - - # Add to an m2m with an object from a different database - try: - marty.book_set.add(dive) - except ValueError: - self.fail("Assignment across master/slave databases with a common source should be ok") - - # Database assignments don't change - self.assertEqual(marty._state.db, 'default') - self.assertEqual(pro._state.db, 'default') - self.assertEqual(dive._state.db, 'other') - self.assertEqual(mark._state.db, 'other') - - # All m2m relations should be saved on the default database - self.assertEqual(Book.authors.through.objects.using('default').count(), 1) - self.assertEqual(Book.authors.through.objects.using('other').count(), 0) - - # Reset relations - Book.authors.through.objects.using('default').delete() - - # Set a reverse m2m with an object from a different database - try: - dive.authors = [mark, marty] - except ValueError: - self.fail("Assignment across master/slave databases with a common source should be ok") - - # Database assignments don't change - self.assertEqual(marty._state.db, 'default') - self.assertEqual(pro._state.db, 'default') - self.assertEqual(dive._state.db, 'other') - self.assertEqual(mark._state.db, 'other') - - # All m2m relations should be saved on the default database - self.assertEqual(Book.authors.through.objects.using('default').count(), 2) - self.assertEqual(Book.authors.through.objects.using('other').count(), 0) - - # Reset relations - Book.authors.through.objects.using('default').delete() - - self.assertEqual(Book.authors.through.objects.using('default').count(), 0) - self.assertEqual(Book.authors.through.objects.using('other').count(), 0) - - # Add to a reverse m2m with an object from a different database - try: - dive.authors.add(marty) - except ValueError: - self.fail("Assignment across master/slave databases with a common source should be ok") - - # Database assignments don't change - self.assertEqual(marty._state.db, 'default') - self.assertEqual(pro._state.db, 'default') - self.assertEqual(dive._state.db, 'other') - self.assertEqual(mark._state.db, 'other') - - # All m2m relations should be saved on the default database - self.assertEqual(Book.authors.through.objects.using('default').count(), 1) - self.assertEqual(Book.authors.through.objects.using('other').count(), 0) - - # If you create an object through a M2M relation, it will be - # written to the write database, even if the original object - # was on the read database - alice = dive.authors.create(name='Alice') - self.assertEqual(alice._state.db, 'default') - - # Same goes for get_or_create, regardless of whether getting or creating - alice, created = dive.authors.get_or_create(name='Alice') - self.assertEqual(alice._state.db, 'default') - - bob, created = dive.authors.get_or_create(name='Bob') - self.assertEqual(bob._state.db, 'default') - - def test_o2o_cross_database_protection(self): - "Operations that involve sharing FK objects across databases raise an error" - # Create a user and profile on the default database - alice = User.objects.db_manager('default').create_user('alice', 'alice@example.com') - - # Create a user and profile on the other database - bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com') - - # Set a one-to-one relation with an object from a different database - alice_profile = UserProfile.objects.create(user=alice, flavor='chocolate') - try: - bob.userprofile = alice_profile - except ValueError: - self.fail("Assignment across master/slave databases with a common source should be ok") - - # Database assignments of original objects haven't changed... - self.assertEqual(alice._state.db, 'default') - self.assertEqual(alice_profile._state.db, 'default') - self.assertEqual(bob._state.db, 'other') - - # ... but they will when the affected object is saved. - bob.save() - self.assertEqual(bob._state.db, 'default') - - def test_generic_key_cross_database_protection(self): - "Generic Key operations can span databases if they share a source" - # Create a book and author on the default database - pro = Book.objects.using('default' - ).create(title="Pro Django", published=datetime.date(2008, 12, 16)) - - review1 = Review.objects.using('default' - ).create(source="Python Monthly", content_object=pro) - - # Create a book and author on the other database - dive = Book.objects.using('other' - ).create(title="Dive into Python", published=datetime.date(2009, 5, 4)) - - review2 = Review.objects.using('other' - ).create(source="Python Weekly", content_object=dive) - - # Set a generic foreign key with an object from a different database - try: - review1.content_object = dive - except ValueError: - self.fail("Assignment across master/slave databases with a common source should be ok") - - # Database assignments of original objects haven't changed... - self.assertEqual(pro._state.db, 'default') - self.assertEqual(review1._state.db, 'default') - self.assertEqual(dive._state.db, 'other') - self.assertEqual(review2._state.db, 'other') - - # ... but they will when the affected object is saved. - dive.save() - self.assertEqual(review1._state.db, 'default') - self.assertEqual(dive._state.db, 'default') - - # ...and the source database now has a copy of any object saved - try: - Book.objects.using('default').get(title='Dive into Python').delete() - except Book.DoesNotExist: - self.fail('Source database should have a copy of saved object') - - # This isn't a real master-slave database, so restore the original from other - dive = Book.objects.using('other').get(title='Dive into Python') - self.assertEqual(dive._state.db, 'other') - - # Add to a generic foreign key set with an object from a different database - try: - dive.reviews.add(review1) - except ValueError: - self.fail("Assignment across master/slave databases with a common source should be ok") - - # Database assignments of original objects haven't changed... - self.assertEqual(pro._state.db, 'default') - self.assertEqual(review1._state.db, 'default') - self.assertEqual(dive._state.db, 'other') - self.assertEqual(review2._state.db, 'other') - - # ... but they will when the affected object is saved. - dive.save() - self.assertEqual(dive._state.db, 'default') - - # ...and the source database now has a copy of any object saved - try: - Book.objects.using('default').get(title='Dive into Python').delete() - except Book.DoesNotExist: - self.fail('Source database should have a copy of saved object') - - # BUT! if you assign a FK object when the base object hasn't - # been saved yet, you implicitly assign the database for the - # base object. - review3 = Review(source="Python Daily") - # initially, no db assigned - self.assertEqual(review3._state.db, None) - - # Dive comes from 'other', so review3 is set to use the source of 'other'... - review3.content_object = dive - self.assertEqual(review3._state.db, 'default') - - # If you create an object through a M2M relation, it will be - # written to the write database, even if the original object - # was on the read database - dive = Book.objects.using('other').get(title='Dive into Python') - nyt = dive.reviews.create(source="New York Times", content_object=dive) - self.assertEqual(nyt._state.db, 'default') - - def test_m2m_managers(self): - "M2M relations are represented by managers, and can be controlled like managers" - pro = Book.objects.using('other').create(pk=1, title="Pro Django", - published=datetime.date(2008, 12, 16)) - - marty = Person.objects.using('other').create(pk=1, name="Marty Alchin") - pro_authors = pro.authors.using('other') - authors = [marty] - - self.assertEqual(pro.authors.db, 'other') - self.assertEqual(pro.authors.db_manager('default').db, 'default') - self.assertEqual(pro.authors.db_manager('default').all().db, 'default') - - self.assertEqual(marty.book_set.db, 'other') - self.assertEqual(marty.book_set.db_manager('default').db, 'default') - self.assertEqual(marty.book_set.db_manager('default').all().db, 'default') - - def test_foreign_key_managers(self): - "FK reverse relations are represented by managers, and can be controlled like managers" - marty = Person.objects.using('other').create(pk=1, name="Marty Alchin") - pro = Book.objects.using('other').create(pk=1, title="Pro Django", - published=datetime.date(2008, 12, 16), - editor=marty) - - self.assertEqual(marty.edited.db, 'other') - self.assertEqual(marty.edited.db_manager('default').db, 'default') - self.assertEqual(marty.edited.db_manager('default').all().db, 'default') - - def test_generic_key_managers(self): - "Generic key relations are represented by managers, and can be controlled like managers" - pro = Book.objects.using('other').create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - review1 = Review.objects.using('other').create(source="Python Monthly", - content_object=pro) - - self.assertEqual(pro.reviews.db, 'other') - self.assertEqual(pro.reviews.db_manager('default').db, 'default') - self.assertEqual(pro.reviews.db_manager('default').all().db, 'default') - - def test_subquery(self): - """Make sure as_sql works with subqueries and master/slave.""" - # Create a book and author on the other database - - 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), - editor=mark) - - sub = Person.objects.filter(name='Mark Pilgrim') - qs = Book.objects.filter(editor__in=sub) - - # When you call __str__ on the query object, it doesn't know about using - # so it falls back to the default. Don't let routing instructions - # force the subquery to an incompatible database. - str(qs.query) - - # If you evaluate the query, it should work, running on 'other' - self.assertEqual(list(qs.values_list('title', flat=True)), ['Dive into Python']) - - def test_deferred_models(self): - mark_def = Person.objects.using('default').create(name="Mark Pilgrim") - mark_other = Person.objects.using('other').create(name="Mark Pilgrim") - orig_b = Book.objects.using('other').create(title="Dive into Python", - published=datetime.date(2009, 5, 4), - editor=mark_other) - b = Book.objects.using('other').only('title').get(pk=orig_b.pk) - self.assertEqual(b.published, datetime.date(2009, 5, 4)) - b = Book.objects.using('other').only('title').get(pk=orig_b.pk) - b.editor = mark_def - b.save(using='default') - self.assertEqual(Book.objects.using('default').get(pk=b.pk).published, - datetime.date(2009, 5, 4)) - - -class AuthTestCase(TestCase): - multi_db = True - - def setUp(self): - # Make the 'other' database appear to be a slave of the 'default' - self.old_routers = router.routers - router.routers = [AuthRouter()] - - def tearDown(self): - # Restore the 'other' database as an independent database - router.routers = self.old_routers - - def test_auth_manager(self): - "The methods on the auth manager obey database hints" - # Create one user using default allocation policy - User.objects.create_user('alice', 'alice@example.com') - - # Create another user, explicitly specifying the database - User.objects.db_manager('default').create_user('bob', 'bob@example.com') - - # The second user only exists on the other database - alice = User.objects.using('other').get(username='alice') - - self.assertEqual(alice.username, 'alice') - self.assertEqual(alice._state.db, 'other') - - self.assertRaises(User.DoesNotExist, User.objects.using('default').get, username='alice') - - # The second user only exists on the default database - bob = User.objects.using('default').get(username='bob') - - self.assertEqual(bob.username, 'bob') - self.assertEqual(bob._state.db, 'default') - - self.assertRaises(User.DoesNotExist, User.objects.using('other').get, username='bob') - - # That is... there is one user on each database - self.assertEqual(User.objects.using('default').count(), 1) - self.assertEqual(User.objects.using('other').count(), 1) - - def test_dumpdata(self): - "Check that dumpdata honors allow_syncdb restrictions on the router" - User.objects.create_user('alice', 'alice@example.com') - User.objects.db_manager('default').create_user('bob', 'bob@example.com') - - # Check that dumping the default database doesn't try to include auth - # because allow_syncdb prohibits auth on default - new_io = StringIO() - management.call_command('dumpdata', 'auth', format='json', database='default', stdout=new_io) - command_output = new_io.getvalue().strip() - self.assertEqual(command_output, '[]') - - # Check that dumping the other database does include auth - new_io = StringIO() - management.call_command('dumpdata', 'auth', format='json', database='other', stdout=new_io) - command_output = new_io.getvalue().strip() - self.assertTrue('"email": "alice@example.com"' in command_output) - - -@override_settings(AUTH_PROFILE_MODULE='multiple_database.UserProfile') -class UserProfileTestCase(TestCase): - - def test_user_profiles(self): - alice = User.objects.create_user('alice', 'alice@example.com') - bob = User.objects.db_manager('other').create_user('bob', 'bob@example.com') - - alice_profile = UserProfile(user=alice, flavor='chocolate') - alice_profile.save() - - bob_profile = UserProfile(user=bob, flavor='crunchy frog') - bob_profile.save() - - with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - self.assertEqual(alice.get_profile().flavor, 'chocolate') - self.assertEqual(bob.get_profile().flavor, 'crunchy frog') - -class AntiPetRouter(object): - # A router that only expresses an opinion on syncdb, - # passing pets to the 'other' database - - def allow_syncdb(self, db, model): - "Make sure the auth app only appears on the 'other' db" - if db == 'other': - return model._meta.object_name == 'Pet' - else: - return model._meta.object_name != 'Pet' - -class FixtureTestCase(TestCase): - multi_db = True - fixtures = ['multidb-common', 'multidb'] - - def setUp(self): - # Install the anti-pet router - self.old_routers = router.routers - router.routers = [AntiPetRouter()] - - def tearDown(self): - # Restore the 'other' database as an independent database - router.routers = self.old_routers - - def test_fixture_loading(self): - "Multi-db fixtures are loaded correctly" - # Check that "Pro Django" exists on the default database, but not on other database - try: - Book.objects.get(title="Pro Django") - Book.objects.using('default').get(title="Pro Django") - except Book.DoesNotExist: - self.fail('"Pro Django" should exist on default database') - - self.assertRaises(Book.DoesNotExist, - Book.objects.using('other').get, - title="Pro Django" - ) - - # Check that "Dive into Python" exists on the default database, but not on other database - try: - Book.objects.using('other').get(title="Dive into Python") - except Book.DoesNotExist: - self.fail('"Dive into Python" should exist on other database') - - self.assertRaises(Book.DoesNotExist, - Book.objects.get, - title="Dive into Python" - ) - self.assertRaises(Book.DoesNotExist, - Book.objects.using('default').get, - title="Dive into Python" - ) - - # Check that "Definitive Guide" exists on the both databases - try: - Book.objects.get(title="The Definitive Guide to Django") - Book.objects.using('default').get(title="The Definitive Guide to Django") - Book.objects.using('other').get(title="The Definitive Guide to Django") - except Book.DoesNotExist: - self.fail('"The Definitive Guide to Django" should exist on both databases') - - def test_pseudo_empty_fixtures(self): - "A fixture can contain entries, but lead to nothing in the database; this shouldn't raise an error (ref #14068)" - new_io = StringIO() - management.call_command('loaddata', 'pets', stdout=new_io, stderr=new_io) - command_output = new_io.getvalue().strip() - # No objects will actually be loaded - self.assertEqual(command_output, "Installed 0 object(s) (of 2) from 1 fixture(s)") - -class PickleQuerySetTestCase(TestCase): - multi_db = True - - def test_pickling(self): - for db in connections: - Book.objects.using(db).create(title='Dive into Python', published=datetime.date(2009, 5, 4)) - qs = Book.objects.all() - self.assertEqual(qs.db, pickle.loads(pickle.dumps(qs)).db) - - -class DatabaseReceiver(object): - """ - Used in the tests for the database argument in signals (#13552) - """ - def __call__(self, signal, sender, **kwargs): - self._database = kwargs['using'] - -class WriteToOtherRouter(object): - """ - A router that sends all writes to the other database. - """ - def db_for_write(self, model, **hints): - return "other" - -class SignalTests(TestCase): - multi_db = True - - def setUp(self): - self.old_routers = router.routers - - def tearDown(self): - router.routers = self.old_routers - - def _write_to_other(self): - "Sends all writes to 'other'." - router.routers = [WriteToOtherRouter()] - - def _write_to_default(self): - "Sends all writes to the default DB" - router.routers = self.old_routers - - def test_database_arg_save_and_delete(self): - """ - Tests that the pre/post_save signal contains the correct database. - (#13552) - """ - # Make some signal receivers - pre_save_receiver = DatabaseReceiver() - post_save_receiver = DatabaseReceiver() - pre_delete_receiver = DatabaseReceiver() - post_delete_receiver = DatabaseReceiver() - # Make model and connect receivers - signals.pre_save.connect(sender=Person, receiver=pre_save_receiver) - signals.post_save.connect(sender=Person, receiver=post_save_receiver) - signals.pre_delete.connect(sender=Person, receiver=pre_delete_receiver) - signals.post_delete.connect(sender=Person, receiver=post_delete_receiver) - p = Person.objects.create(name='Darth Vader') - # Save and test receivers got calls - p.save() - self.assertEqual(pre_save_receiver._database, DEFAULT_DB_ALIAS) - self.assertEqual(post_save_receiver._database, DEFAULT_DB_ALIAS) - # Delete, and test - p.delete() - self.assertEqual(pre_delete_receiver._database, DEFAULT_DB_ALIAS) - self.assertEqual(post_delete_receiver._database, DEFAULT_DB_ALIAS) - # Save again to a different database - p.save(using="other") - self.assertEqual(pre_save_receiver._database, "other") - self.assertEqual(post_save_receiver._database, "other") - # Delete, and test - p.delete(using="other") - self.assertEqual(pre_delete_receiver._database, "other") - self.assertEqual(post_delete_receiver._database, "other") - - signals.pre_save.disconnect(sender=Person, receiver=pre_save_receiver) - signals.post_save.disconnect(sender=Person, receiver=post_save_receiver) - signals.pre_delete.disconnect(sender=Person, receiver=pre_delete_receiver) - signals.post_delete.disconnect(sender=Person, receiver=post_delete_receiver) - - def test_database_arg_m2m(self): - """ - Test that the m2m_changed signal has a correct database arg (#13552) - """ - # Make a receiver - receiver = DatabaseReceiver() - # Connect it - signals.m2m_changed.connect(receiver=receiver) - - # Create the models that will be used for the tests - b = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - p = Person.objects.create(name="Marty Alchin") - - # Create a copy of the models on the 'other' database to prevent - # integrity errors on backends that don't defer constraints checks - Book.objects.using('other').create(pk=b.pk, title=b.title, - published=b.published) - Person.objects.using('other').create(pk=p.pk, name=p.name) - - # Test addition - b.authors.add(p) - self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) - self._write_to_other() - b.authors.add(p) - self._write_to_default() - self.assertEqual(receiver._database, "other") - - # Test removal - b.authors.remove(p) - self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) - self._write_to_other() - b.authors.remove(p) - self._write_to_default() - self.assertEqual(receiver._database, "other") - - # Test addition in reverse - p.book_set.add(b) - self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) - self._write_to_other() - p.book_set.add(b) - self._write_to_default() - self.assertEqual(receiver._database, "other") - - # Test clearing - b.authors.clear() - self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) - self._write_to_other() - b.authors.clear() - self._write_to_default() - self.assertEqual(receiver._database, "other") - -class AttributeErrorRouter(object): - "A router to test the exception handling of ConnectionRouter" - def db_for_read(self, model, **hints): - raise AttributeError - - def db_for_write(self, model, **hints): - raise AttributeError - -class RouterAttributeErrorTestCase(TestCase): - multi_db = True - - def setUp(self): - self.old_routers = router.routers - router.routers = [AttributeErrorRouter()] - - def tearDown(self): - router.routers = self.old_routers - - def test_attribute_error_read(self): - "Check that the AttributeError from AttributeErrorRouter bubbles up" - router.routers = [] # Reset routers so we can save a Book instance - b = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - router.routers = [AttributeErrorRouter()] # Install our router - self.assertRaises(AttributeError, Book.objects.get, pk=b.pk) - - def test_attribute_error_save(self): - "Check that the AttributeError from AttributeErrorRouter bubbles up" - dive = Book() - dive.title="Dive into Python" - dive.published = datetime.date(2009, 5, 4) - self.assertRaises(AttributeError, dive.save) - - def test_attribute_error_delete(self): - "Check that the AttributeError from AttributeErrorRouter bubbles up" - router.routers = [] # Reset routers so we can save our Book, Person instances - b = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - p = Person.objects.create(name="Marty Alchin") - b.authors = [p] - b.editor = p - router.routers = [AttributeErrorRouter()] # Install our router - self.assertRaises(AttributeError, b.delete) - - def test_attribute_error_m2m(self): - "Check that the AttributeError from AttributeErrorRouter bubbles up" - router.routers = [] # Reset routers so we can save our Book, Person instances - b = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - p = Person.objects.create(name="Marty Alchin") - router.routers = [AttributeErrorRouter()] # Install our router - self.assertRaises(AttributeError, setattr, b, 'authors', [p]) - -class ModelMetaRouter(object): - "A router to ensure model arguments are real model classes" - def db_for_write(self, model, **hints): - if not hasattr(model, '_meta'): - raise ValueError - -class RouterModelArgumentTestCase(TestCase): - multi_db = True - - def setUp(self): - self.old_routers = router.routers - router.routers = [ModelMetaRouter()] - - def tearDown(self): - router.routers = self.old_routers - - def test_m2m_collection(self): - b = Book.objects.create(title="Pro Django", - published=datetime.date(2008, 12, 16)) - - p = Person.objects.create(name="Marty Alchin") - # test add - b.authors.add(p) - # test remove - b.authors.remove(p) - # test clear - b.authors.clear() - # test setattr - b.authors = [p] - # test M2M collection - b.delete() - - def test_foreignkey_collection(self): - person = Person.objects.create(name='Bob') - pet = Pet.objects.create(owner=person, name='Wart') - # test related FK collection - person.delete() - - -class SyncOnlyDefaultDatabaseRouter(object): - def allow_syncdb(self, db, model): - return db == DEFAULT_DB_ALIAS - - -class SyncDBTestCase(TestCase): - multi_db = True - - def test_syncdb_to_other_database(self): - """Regression test for #16039: syncdb with --database option.""" - cts = ContentType.objects.using('other').filter(app_label='multiple_database') - - count = cts.count() - self.assertGreater(count, 0) - - cts.delete() - management.call_command('syncdb', verbosity=0, interactive=False, - load_initial_data=False, database='other') - self.assertEqual(cts.count(), count) - - def test_syncdb_to_other_database_with_router(self): - """Regression test for #16039: syncdb with --database option.""" - cts = ContentType.objects.using('other').filter(app_label='multiple_database') - - cts.delete() - try: - old_routers = router.routers - router.routers = [SyncOnlyDefaultDatabaseRouter()] - management.call_command('syncdb', verbosity=0, interactive=False, - load_initial_data=False, database='other') - finally: - router.routers = old_routers - - self.assertEqual(cts.count(), 0) |