mirror of
https://github.com/django/django.git
synced 2025-09-26 12:09:19 +00:00
Fixed #1142 -- Added multiple database support.
This monster of a patch is the result of Alex Gaynor's 2009 Google Summer of Code project. Congratulations to Alex for a job well done. Big thanks also go to: * Justin Bronn for keeping GIS in line with the changes, * Karen Tracey and Jani Tiainen for their help testing Oracle support * Brett Hoerner, Jon Loyens, and Craig Kimmerer for their feedback. * Malcolm Treddinick for his guidance during the GSoC submission process. * Simon Willison for driving the original design process * Cal Henderson for complaining about ponies he wanted. ... and everyone else too numerous to mention that helped to bring this feature into fruition. git-svn-id: http://code.djangoproject.com/svn/django/trunk@11952 bcc190cf-cafb-0310-a4f2-bffc1f526a37
This commit is contained in:
Russell Keith-Magee
parent
7ef212af14
commit
ff60c5f9de
231 changed files with 7860 additions and 5668 deletions
705
tests/regressiontests/multiple_database/tests.py
Normal file
705
tests/regressiontests/multiple_database/tests.py
Normal file
|
|
@ -0,0 +1,705 @@
|
|||
import datetime
|
||||
import pickle
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import User
|
||||
from django.db import connections
|
||||
from django.test import TestCase
|
||||
|
||||
from models import Book, Person, Review, UserProfile
|
||||
|
||||
try:
|
||||
# we only have these models if the user is using multi-db, it's safe the
|
||||
# run the tests without them though.
|
||||
from models import Article, article_using
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
class QueryTestCase(TestCase):
|
||||
multi_db = True
|
||||
|
||||
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.assertEquals(Book.authors.through.objects.using('default').count(), 1)
|
||||
self.assertEquals(Book.authors.through.objects.using('other').count(), 1)
|
||||
|
||||
# Check that queries work across m2m joins
|
||||
self.assertEquals(list(Book.objects.using('default').filter(authors__name='Marty Alchin').values_list('title', flat=True)),
|
||||
[u'Pro Django'])
|
||||
self.assertEquals(list(Book.objects.using('other').filter(authors__name='Marty Alchin').values_list('title', flat=True)),
|
||||
[])
|
||||
|
||||
self.assertEquals(list(Book.objects.using('default').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
|
||||
[])
|
||||
self.assertEquals(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
|
||||
[u'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.assertEquals(list(dive.authors.all().values_list('name', flat=True)),
|
||||
[u'Mark Pilgrim'])
|
||||
|
||||
self.assertEquals(list(mark.book_set.all().values_list('title', flat=True)),
|
||||
[u'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.assertEquals(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),
|
||||
[])
|
||||
|
||||
|
||||
dive.authors.add(john)
|
||||
self.assertEquals(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
|
||||
[u'Dive into Python'])
|
||||
self.assertEquals(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),
|
||||
[u'Dive into Python'])
|
||||
|
||||
# Remove the second author
|
||||
dive.authors.remove(john)
|
||||
self.assertEquals(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
|
||||
[u'Dive into Python'])
|
||||
self.assertEquals(list(Book.objects.using('other').filter(authors__name='John Smith').values_list('title', flat=True)),
|
||||
[])
|
||||
|
||||
# Clear all authors
|
||||
dive.authors.clear()
|
||||
self.assertEquals(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
|
||||
[])
|
||||
self.assertEquals(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.assertEquals(list(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True)),
|
||||
[])
|
||||
self.assertEquals(list(Book.objects.using('other').filter(authors__name='Jane Brown').values_list('title', flat=True)),
|
||||
[u'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.assertEquals(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),
|
||||
[u'Mark Pilgrim'])
|
||||
self.assertEquals(list(Person.objects.using('other').filter(book__title='Greasemonkey Hacks').values_list('name', flat=True)),
|
||||
[u'Mark Pilgrim'])
|
||||
|
||||
# Remove a book from the m2m
|
||||
mark.book_set.remove(grease)
|
||||
self.assertEquals(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),
|
||||
[u'Mark Pilgrim'])
|
||||
self.assertEquals(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.assertEquals(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),
|
||||
[])
|
||||
self.assertEquals(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.assertEquals(list(Person.objects.using('other').filter(book__title='Dive into Python').values_list('name', flat=True)),
|
||||
[])
|
||||
self.assertEquals(list(Person.objects.using('other').filter(book__title='Dive into HTML5').values_list('name', flat=True)),
|
||||
[u'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_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.assertEquals(pro.editor.name, "George Vilches")
|
||||
|
||||
dive = Book.objects.using('other').get(title="Dive into Python")
|
||||
self.assertEquals(dive.editor.name, "Chris Mills")
|
||||
|
||||
# Check that queries work across foreign key joins
|
||||
self.assertEquals(list(Person.objects.using('default').filter(edited__title='Pro Django').values_list('name', flat=True)),
|
||||
[u'George Vilches'])
|
||||
self.assertEquals(list(Person.objects.using('other').filter(edited__title='Pro Django').values_list('name', flat=True)),
|
||||
[])
|
||||
|
||||
self.assertEquals(list(Person.objects.using('default').filter(edited__title='Dive into Python').values_list('name', flat=True)),
|
||||
[])
|
||||
self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),
|
||||
[u'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.assertEquals(list(chris.edited.values_list('title', flat=True)),
|
||||
[u'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.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),
|
||||
[])
|
||||
|
||||
chris.edited.add(html5)
|
||||
self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),
|
||||
[u'Chris Mills'])
|
||||
self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),
|
||||
[u'Chris Mills'])
|
||||
|
||||
# Remove the second editor
|
||||
chris.edited.remove(html5)
|
||||
self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),
|
||||
[])
|
||||
self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into Python').values_list('name', flat=True)),
|
||||
[u'Chris Mills'])
|
||||
|
||||
# Clear all edited books
|
||||
chris.edited.clear()
|
||||
self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),
|
||||
[])
|
||||
self.assertEquals(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.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into HTML5').values_list('name', flat=True)),
|
||||
[])
|
||||
self.assertEquals(list(Person.objects.using('other').filter(edited__title='Dive into Water').values_list('name', flat=True)),
|
||||
[u'Chris Mills'])
|
||||
self.assertEquals(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.assertEquals(chris._state.db, None)
|
||||
# self.assertEquals(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.assertEquals(chris._state.db, 'other')
|
||||
# self.assertEquals(html5._state.db, 'other')
|
||||
# # ... but it isn't saved yet
|
||||
# self.assertEquals(list(Person.objects.using('other').values_list('name',flat=True)),
|
||||
# [u'Mark Pilgrim'])
|
||||
# self.assertEquals(list(Book.objects.using('other').values_list('title',flat=True)),
|
||||
# [u'Dive into Python'])
|
||||
|
||||
# # When saved (no using required), new objects goes to 'other'
|
||||
# chris.save()
|
||||
# html5.save()
|
||||
# self.assertEquals(list(Person.objects.using('default').values_list('name',flat=True)),
|
||||
# [u'Marty Alchin'])
|
||||
# self.assertEquals(list(Person.objects.using('other').values_list('name',flat=True)),
|
||||
# [u'Chris Mills', u'Mark Pilgrim'])
|
||||
# self.assertEquals(list(Book.objects.using('default').values_list('title',flat=True)),
|
||||
# [u'Pro Django'])
|
||||
# self.assertEquals(list(Book.objects.using('other').values_list('title',flat=True)),
|
||||
# [u'Dive into HTML5', u'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.assertEquals(water._state.db, 'other')
|
||||
# # ... but it isn't saved yet
|
||||
# self.assertEquals(list(Book.objects.using('default').values_list('title',flat=True)),
|
||||
# [u'Pro Django'])
|
||||
# self.assertEquals(list(Book.objects.using('other').values_list('title',flat=True)),
|
||||
# [u'Dive into HTML5', u'Dive into Python'])
|
||||
|
||||
# # When saved, the new book goes to 'other'
|
||||
# water.save()
|
||||
# self.assertEquals(list(Book.objects.using('default').values_list('title',flat=True)),
|
||||
# [u'Pro Django'])
|
||||
# self.assertEquals(list(Book.objects.using('other').values_list('title',flat=True)),
|
||||
# [u'Dive into HTML5', u'Dive into Python', u'Dive into Water'])
|
||||
|
||||
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.assertEquals(review1.content_object.title, "Pro Django")
|
||||
|
||||
review2 = Review.objects.using('other').get(source="Python Weekly")
|
||||
self.assertEquals(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.assertEquals(list(dive.reviews.all().values_list('source', flat=True)),
|
||||
[u'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.assertEquals(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),
|
||||
[])
|
||||
self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),
|
||||
[u'Python Weekly'])
|
||||
|
||||
# Add a second review
|
||||
dive.reviews.add(review2)
|
||||
self.assertEquals(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),
|
||||
[])
|
||||
self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),
|
||||
[u'Python Monthly', u'Python Weekly'])
|
||||
|
||||
# Remove the second author
|
||||
dive.reviews.remove(review1)
|
||||
self.assertEquals(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),
|
||||
[])
|
||||
self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),
|
||||
[u'Python Monthly'])
|
||||
|
||||
# Clear all reviews
|
||||
dive.reviews.clear()
|
||||
self.assertEquals(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),
|
||||
[])
|
||||
self.assertEquals(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.assertEquals(list(Review.objects.using('default').filter(object_id=dive.pk).values_list('source', flat=True)),
|
||||
[])
|
||||
self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source', flat=True)),
|
||||
[u'Python Daily'])
|
||||
|
||||
# def test_generic_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))
|
||||
|
||||
## 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.assertEquals(review3._state.db, None)
|
||||
|
||||
# # Dive comes from 'other', so review3 is set to use 'other'...
|
||||
# review3.content_object = dive
|
||||
# self.assertEquals(review3._state.db, 'other')
|
||||
# # ... but it isn't saved yet
|
||||
# self.assertEquals(list(Review.objects.using('default').filter(object_id=pro.pk).values_list('source', flat=True)),
|
||||
# [u'Python Monthly'])
|
||||
# self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source',flat=True)),
|
||||
# [u'Python Weekly'])
|
||||
|
||||
# # When saved, John goes to 'other'
|
||||
# review3.save()
|
||||
# self.assertEquals(list(Review.objects.using('default').filter(object_id=pro.pk).values_list('source', flat=True)),
|
||||
# [u'Python Monthly'])
|
||||
# self.assertEquals(list(Review.objects.using('other').filter(object_id=dive.pk).values_list('source',flat=True)),
|
||||
# [u'Python Daily', u'Python Weekly'])
|
||||
|
||||
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.assertEquals(learn.get_next_by_published().title, "Dive into Python")
|
||||
self.assertEquals(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.assertEqual(map(lambda o: o.pk, val), [dive.pk])
|
||||
|
||||
val = Book.objects.raw('SELECT id FROM "multiple_database_book"').using('other')
|
||||
self.assertEqual(map(lambda o: o.pk, val), [dive.pk])
|
||||
|
||||
|
||||
class UserProfileTestCase(TestCase):
|
||||
def setUp(self):
|
||||
self.old_auth_profile_module = getattr(settings, 'AUTH_PROFILE_MODULE', None)
|
||||
settings.AUTH_PROFILE_MODULE = 'multiple_database.UserProfile'
|
||||
|
||||
def tearDown(self):
|
||||
settings.AUTH_PROFILE_MODULE = self.old_auth_profile_module
|
||||
|
||||
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()
|
||||
|
||||
self.assertEquals(alice.get_profile().flavor, 'chocolate')
|
||||
self.assertEquals(bob.get_profile().flavor, 'crunchy frog')
|
||||
|
||||
class FixtureTestCase(TestCase):
|
||||
multi_db = True
|
||||
fixtures = ['multidb-common', 'multidb']
|
||||
|
||||
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')
|
||||
|
||||
|
||||
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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue