mirror of
https://github.com/django/django.git
synced 2025-07-06 02:39:12 +00:00
[soc2009/multidb] Updated testing services to handle multiple databases better. Includes extra tests (some failing) for multiple database support. Patch from Russell Keith-Magee.
git-svn-id: http://code.djangoproject.com/svn/django/branches/soc2009/multidb@11764 bcc190cf-cafb-0310-a4f2-bffc1f526a37
This commit is contained in:
parent
0c167ae0ff
commit
f2604c331d
4
TODO
4
TODO
@ -13,10 +13,6 @@ Required for v1.2
|
|||||||
* Resolve the public facing UI issues around using multi-db
|
* Resolve the public facing UI issues around using multi-db
|
||||||
* Should we take the opportunity to modify DB backends to use fully qualified paths?
|
* Should we take the opportunity to modify DB backends to use fully qualified paths?
|
||||||
* Meta.using? Is is still required/desirable?
|
* Meta.using? Is is still required/desirable?
|
||||||
* Testing infrastructure
|
|
||||||
* Most tests don't need multidb. Some absolutely require it, but only to prove you
|
|
||||||
can write to a different db. Second DB could be a SQLite temp file. Need to have
|
|
||||||
test infrastructure to allow creation of the temp database.
|
|
||||||
* Cleanup of new API entry points
|
* Cleanup of new API entry points
|
||||||
* validate() on a field
|
* validate() on a field
|
||||||
* name/purpose clash with Honza?
|
* name/purpose clash with Honza?
|
||||||
|
@ -26,7 +26,7 @@ class BaseDatabaseWrapper(local):
|
|||||||
"""
|
"""
|
||||||
ops = None
|
ops = None
|
||||||
|
|
||||||
def __init__(self, settings_dict):
|
def __init__(self, settings_dict, alias='default'):
|
||||||
# `settings_dict` should be a dictionary containing keys such as
|
# `settings_dict` should be a dictionary containing keys such as
|
||||||
# DATABASE_NAME, DATABASE_USER, etc. It's called `settings_dict`
|
# DATABASE_NAME, DATABASE_USER, etc. It's called `settings_dict`
|
||||||
# instead of `settings` to disambiguate it from Django settings
|
# instead of `settings` to disambiguate it from Django settings
|
||||||
@ -34,6 +34,7 @@ class BaseDatabaseWrapper(local):
|
|||||||
self.connection = None
|
self.connection = None
|
||||||
self.queries = []
|
self.queries = []
|
||||||
self.settings_dict = settings_dict
|
self.settings_dict = settings_dict
|
||||||
|
self.alias = alias
|
||||||
|
|
||||||
def __eq__(self, other):
|
def __eq__(self, other):
|
||||||
return self.settings_dict == other.settings_dict
|
return self.settings_dict == other.settings_dict
|
||||||
@ -117,7 +118,7 @@ class BaseDatabaseOperations(object):
|
|||||||
row.
|
row.
|
||||||
"""
|
"""
|
||||||
compiler_module = "django.db.models.sql.compiler"
|
compiler_module = "django.db.models.sql.compiler"
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self._cache = {}
|
self._cache = {}
|
||||||
|
|
||||||
|
@ -316,13 +316,13 @@ class BaseDatabaseCreation(object):
|
|||||||
output.append(ds)
|
output.append(ds)
|
||||||
return output
|
return output
|
||||||
|
|
||||||
def create_test_db(self, verbosity=1, autoclobber=False, alias=None):
|
def create_test_db(self, verbosity=1, autoclobber=False):
|
||||||
"""
|
"""
|
||||||
Creates a test database, prompting the user for confirmation if the
|
Creates a test database, prompting the user for confirmation if the
|
||||||
database already exists. Returns the name of the test database created.
|
database already exists. Returns the name of the test database created.
|
||||||
"""
|
"""
|
||||||
if verbosity >= 1:
|
if verbosity >= 1:
|
||||||
print "Creating test database..."
|
print "Creating test database '%s'..." % self.connection.alias
|
||||||
|
|
||||||
test_database_name = self._create_test_db(verbosity, autoclobber)
|
test_database_name = self._create_test_db(verbosity, autoclobber)
|
||||||
|
|
||||||
@ -334,7 +334,7 @@ class BaseDatabaseCreation(object):
|
|||||||
# FIXME we end up loading the same fixture into the default DB for each
|
# FIXME we end up loading the same fixture into the default DB for each
|
||||||
# DB we have, this causes various test failures, but can't really be
|
# DB we have, this causes various test failures, but can't really be
|
||||||
# fixed until we have an API for saving to a specific DB
|
# fixed until we have an API for saving to a specific DB
|
||||||
call_command('syncdb', verbosity=verbosity, interactive=False, database=alias)
|
call_command('syncdb', verbosity=verbosity, interactive=False, database=self.connection.alias)
|
||||||
|
|
||||||
if settings.CACHE_BACKEND.startswith('db://'):
|
if settings.CACHE_BACKEND.startswith('db://'):
|
||||||
from django.core.cache import parse_backend_uri
|
from django.core.cache import parse_backend_uri
|
||||||
@ -404,7 +404,7 @@ class BaseDatabaseCreation(object):
|
|||||||
database already exists. Returns the name of the test database created.
|
database already exists. Returns the name of the test database created.
|
||||||
"""
|
"""
|
||||||
if verbosity >= 1:
|
if verbosity >= 1:
|
||||||
print "Destroying test database..."
|
print "Destroying test database '%s'..." % self.connection.alias
|
||||||
self.connection.close()
|
self.connection.close()
|
||||||
test_database_name = self.connection.settings_dict['DATABASE_NAME']
|
test_database_name = self.connection.settings_dict['DATABASE_NAME']
|
||||||
self.connection.settings_dict['DATABASE_NAME'] = old_database_name
|
self.connection.settings_dict['DATABASE_NAME'] = old_database_name
|
||||||
|
@ -67,7 +67,7 @@ class ConnectionHandler(object):
|
|||||||
self.ensure_defaults(alias)
|
self.ensure_defaults(alias)
|
||||||
db = self.databases[alias]
|
db = self.databases[alias]
|
||||||
backend = load_backend(db['DATABASE_ENGINE'])
|
backend = load_backend(db['DATABASE_ENGINE'])
|
||||||
conn = backend.DatabaseWrapper(db)
|
conn = backend.DatabaseWrapper(db, alias)
|
||||||
self._connections[alias] = conn
|
self._connections[alias] = conn
|
||||||
return conn
|
return conn
|
||||||
|
|
||||||
@ -76,19 +76,3 @@ class ConnectionHandler(object):
|
|||||||
|
|
||||||
def all(self):
|
def all(self):
|
||||||
return [self[alias] for alias in self]
|
return [self[alias] for alias in self]
|
||||||
|
|
||||||
def alias_for_connection(self, connection):
|
|
||||||
"""
|
|
||||||
Returns the alias for the given connection object.
|
|
||||||
"""
|
|
||||||
return self.alias_for_settings(connection.settings_dict)
|
|
||||||
|
|
||||||
def alias_for_settings(self, settings_dict):
|
|
||||||
"""
|
|
||||||
Returns the alias for the given settings dictionary.
|
|
||||||
"""
|
|
||||||
for alias in self:
|
|
||||||
conn_settings = self.databases[alias]
|
|
||||||
if conn_settings == settings_dict:
|
|
||||||
return alias
|
|
||||||
return None
|
|
||||||
|
@ -191,7 +191,7 @@ def run_tests(test_labels, verbosity=1, interactive=True, extra_tests=[]):
|
|||||||
for alias in connections:
|
for alias in connections:
|
||||||
connection = connections[alias]
|
connection = connections[alias]
|
||||||
old_names.append((connection, connection.settings_dict['DATABASE_NAME']))
|
old_names.append((connection, connection.settings_dict['DATABASE_NAME']))
|
||||||
connection.creation.create_test_db(verbosity, autoclobber=not interactive, alias=alias)
|
connection.creation.create_test_db(verbosity, autoclobber=not interactive)
|
||||||
result = unittest.TextTestRunner(verbosity=verbosity).run(suite)
|
result = unittest.TextTestRunner(verbosity=verbosity).run(suite)
|
||||||
for connection, old_name in old_names:
|
for connection, old_name in old_names:
|
||||||
connection.creation.destroy_test_db(old_name, verbosity)
|
connection.creation.destroy_test_db(old_name, verbosity)
|
||||||
|
@ -7,7 +7,7 @@ from django.conf import settings
|
|||||||
from django.core import mail
|
from django.core import mail
|
||||||
from django.core.management import call_command
|
from django.core.management import call_command
|
||||||
from django.core.urlresolvers import clear_url_caches
|
from django.core.urlresolvers import clear_url_caches
|
||||||
from django.db import transaction, connections
|
from django.db import transaction, connections, DEFAULT_DB_ALIAS
|
||||||
from django.http import QueryDict
|
from django.http import QueryDict
|
||||||
from django.test import _doctest as doctest
|
from django.test import _doctest as doctest
|
||||||
from django.test.client import Client
|
from django.test.client import Client
|
||||||
@ -225,11 +225,19 @@ class TransactionTestCase(unittest.TestCase):
|
|||||||
mail.outbox = []
|
mail.outbox = []
|
||||||
|
|
||||||
def _fixture_setup(self):
|
def _fixture_setup(self):
|
||||||
call_command('flush', verbosity=0, interactive=False)
|
# If the test case has a multi_db=True flag, flush all databases.
|
||||||
if hasattr(self, 'fixtures'):
|
# Otherwise, just flush default.
|
||||||
# We have to use this slightly awkward syntax due to the fact
|
if getattr(self, 'multi_db', False):
|
||||||
# that we're using *args and **kwargs together.
|
databases = connections
|
||||||
call_command('loaddata', *self.fixtures, **{'verbosity': 0})
|
else:
|
||||||
|
databases = [DEFAULT_DB_ALIAS]
|
||||||
|
for db in databases:
|
||||||
|
call_command('flush', verbosity=0, interactive=False, database=db)
|
||||||
|
|
||||||
|
if hasattr(self, 'fixtures'):
|
||||||
|
# We have to use this slightly awkward syntax due to the fact
|
||||||
|
# that we're using *args and **kwargs together.
|
||||||
|
call_command('loaddata', *self.fixtures, **{'verbosity': 0, 'database': db})
|
||||||
|
|
||||||
def _urlconf_setup(self):
|
def _urlconf_setup(self):
|
||||||
if hasattr(self, 'urls'):
|
if hasattr(self, 'urls'):
|
||||||
@ -453,27 +461,44 @@ class TestCase(TransactionTestCase):
|
|||||||
if not connections_support_transactions():
|
if not connections_support_transactions():
|
||||||
return super(TestCase, self)._fixture_setup()
|
return super(TestCase, self)._fixture_setup()
|
||||||
|
|
||||||
for conn in connections:
|
# If the test case has a multi_db=True flag, setup all databases.
|
||||||
transaction.enter_transaction_management(using=conn)
|
# Otherwise, just use default.
|
||||||
transaction.managed(True, using=conn)
|
if getattr(self, 'multi_db', False):
|
||||||
|
databases = connections
|
||||||
|
else:
|
||||||
|
databases = [DEFAULT_DB_ALIAS]
|
||||||
|
|
||||||
|
for db in databases:
|
||||||
|
transaction.enter_transaction_management(using=db)
|
||||||
|
transaction.managed(True, using=db)
|
||||||
disable_transaction_methods()
|
disable_transaction_methods()
|
||||||
|
|
||||||
from django.contrib.sites.models import Site
|
from django.contrib.sites.models import Site
|
||||||
Site.objects.clear_cache()
|
Site.objects.clear_cache()
|
||||||
|
|
||||||
if hasattr(self, 'fixtures'):
|
for db in databases:
|
||||||
call_command('loaddata', *self.fixtures, **{
|
if hasattr(self, 'fixtures'):
|
||||||
'verbosity': 0,
|
call_command('loaddata', *self.fixtures, **{
|
||||||
'commit': False
|
'verbosity': 0,
|
||||||
})
|
'commit': False,
|
||||||
|
'database': db
|
||||||
|
})
|
||||||
|
|
||||||
def _fixture_teardown(self):
|
def _fixture_teardown(self):
|
||||||
if not connections_support_transactions():
|
if not connections_support_transactions():
|
||||||
return super(TestCase, self)._fixture_teardown()
|
return super(TestCase, self)._fixture_teardown()
|
||||||
|
|
||||||
|
# If the test case has a multi_db=True flag, teardown all databases.
|
||||||
|
# Otherwise, just teardown default.
|
||||||
|
if getattr(self, 'multi_db', False):
|
||||||
|
databases = connections
|
||||||
|
else:
|
||||||
|
databases = [DEFAULT_DB_ALIAS]
|
||||||
|
|
||||||
restore_transaction_methods()
|
restore_transaction_methods()
|
||||||
for conn in connections:
|
for db in databases:
|
||||||
transaction.rollback(using=conn)
|
transaction.rollback(using=db)
|
||||||
transaction.leave_transaction_management(using=conn)
|
transaction.leave_transaction_management(using=db)
|
||||||
|
|
||||||
for connection in connections.all():
|
for connection in connections.all():
|
||||||
connection.close()
|
connection.close()
|
||||||
|
@ -752,20 +752,46 @@ To run the tests, ``cd`` to the ``tests/`` directory and type:
|
|||||||
./runtests.py --settings=path.to.django.settings
|
./runtests.py --settings=path.to.django.settings
|
||||||
|
|
||||||
Yes, the unit tests need a settings module, but only for database connection
|
Yes, the unit tests need a settings module, but only for database connection
|
||||||
info, with the ``DATABASES`` setting.
|
info. Your :setting:`DATABASES` setting needs to define two databases:
|
||||||
|
|
||||||
If you're using the ``sqlite3`` database backend, no further settings are
|
* A ``default`` database. This database should use the backend that
|
||||||
needed. A temporary database will be created in memory when running the tests.
|
you want to use for primary testing
|
||||||
|
|
||||||
If you're using another backend:
|
* A database with the alias ``other``. The ``other`` database is
|
||||||
|
used to establish that queries can be directed to different
|
||||||
|
databases. As a result, this database can use any backend you
|
||||||
|
want. It doesn't need to use the same backend as the ``default``
|
||||||
|
database (although it can use the same backend if you want to).
|
||||||
|
|
||||||
* Your the ``DATABASE_USER`` option for each of your databases needs to
|
If you're using the ``sqlite3`` database backend, you need to define
|
||||||
|
:setting:`DATABASE_ENGINE` for both databases, plus a
|
||||||
|
:setting:`TEST_DATABASE_NAME` for the ``other`` database. The
|
||||||
|
following is a minimal settings file that can be used to test SQLite::
|
||||||
|
|
||||||
|
DATABASES = {
|
||||||
|
'default': {
|
||||||
|
'DATABASE_ENGINE': 'sqlite3'
|
||||||
|
},
|
||||||
|
'other': {
|
||||||
|
'DATABASE_ENGINE': 'sqlite3',
|
||||||
|
'TEST_DATABASE_NAME: 'other_db'
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
If you're using another backend, you will need to provide other details for
|
||||||
|
each database:
|
||||||
|
|
||||||
|
* The :setting:`DATABASE_USER` option for each of your databases needs to
|
||||||
specify an existing user account for the database.
|
specify an existing user account for the database.
|
||||||
|
|
||||||
* The ``DATABASE_NAME`` option must be the name of an existing database to
|
* The :setting:`DATABASE_PASSWORD` option needs to provide the password for
|
||||||
|
the :setting:`DATABASE_USER` that has been specified.
|
||||||
|
|
||||||
|
* The :setting:`DATABASE_NAME` option must be the name of an existing database to
|
||||||
which the given user has permission to connect. The unit tests will not
|
which the given user has permission to connect. The unit tests will not
|
||||||
touch this database; the test runner creates a new database whose name is
|
touch this database; the test runner creates a new database whose name is
|
||||||
``DATABASE_NAME`` prefixed with ``test_``, and this test database is
|
:setting:`DATABASE_NAME` prefixed with ``test_``, and this test database is
|
||||||
deleted when the tests are finished. This means your user account needs
|
deleted when the tests are finished. This means your user account needs
|
||||||
permission to execute ``CREATE DATABASE``.
|
permission to execute ``CREATE DATABASE``.
|
||||||
|
|
||||||
|
@ -202,11 +202,11 @@ Django. It is a nested dictionary who's contents maps aliases to a dictionary
|
|||||||
containing the options for an individual database. The following inner options
|
containing the options for an individual database. The following inner options
|
||||||
are used:
|
are used:
|
||||||
|
|
||||||
.. admonition:: Note
|
.. deprecated: 1.2
|
||||||
|
|
||||||
In versions of Django prior to 1.2 each of the following were individual
|
In versions of Django prior to 1.2 each of the following were
|
||||||
settings, the usage of those has been deprecated but will be supported
|
individual settings. The usage of the standalone database settings
|
||||||
until Django 1.4.
|
has been deprecated, and will be removed in Django 1.4.
|
||||||
|
|
||||||
.. setting:: DATABASE_ENGINE
|
.. setting:: DATABASE_ENGINE
|
||||||
|
|
||||||
|
@ -1037,6 +1037,39 @@ URLconf for the duration of the test case.
|
|||||||
|
|
||||||
.. _emptying-test-outbox:
|
.. _emptying-test-outbox:
|
||||||
|
|
||||||
|
Multi-database support
|
||||||
|
~~~~~~~~~~~~~~~~~~~~~~
|
||||||
|
|
||||||
|
.. attribute:: TestCase.multi_db
|
||||||
|
|
||||||
|
.. versionadded:: 1.2
|
||||||
|
|
||||||
|
Django sets up a test database corresponding to every database that is
|
||||||
|
defined in the :setting:``DATABASES`` definition in your settings
|
||||||
|
file. However, a big part of the time taken to run a Django TestCase
|
||||||
|
is consumed by the call to ``flush`` that ensures that you have a
|
||||||
|
clean database at the start of each test run. If you have multiple
|
||||||
|
databases, multiple flushes are required (one for each database),
|
||||||
|
which can be a time consuming activity -- especially if your tests
|
||||||
|
don't need to test multi-database activity.
|
||||||
|
|
||||||
|
As an optimization, Django only flushes the ``default`` database at
|
||||||
|
the start of each test run. If your setup contains multiple databases,
|
||||||
|
and you have a test that requires every database to be clean, you can
|
||||||
|
use the ``multi_db`` attribute on the test suite to request a full
|
||||||
|
flush.
|
||||||
|
|
||||||
|
For example::
|
||||||
|
|
||||||
|
class TestMyViews(TestCase):
|
||||||
|
multi_db = True
|
||||||
|
|
||||||
|
def testIndexPageView(self):
|
||||||
|
call_some_test_code()
|
||||||
|
|
||||||
|
This test case will flush *all* the test databases before running
|
||||||
|
``testIndexPageView``.
|
||||||
|
|
||||||
Emptying the test outbox
|
Emptying the test outbox
|
||||||
~~~~~~~~~~~~~~~~~~~~~~~~
|
~~~~~~~~~~~~~~~~~~~~~~~~
|
||||||
|
|
||||||
|
@ -0,0 +1,10 @@
|
|||||||
|
[
|
||||||
|
{
|
||||||
|
"pk": 1,
|
||||||
|
"model": "multiple_database.book",
|
||||||
|
"fields": {
|
||||||
|
"title": "The Definitive Guide to Django",
|
||||||
|
"published": "2009-7-8"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
]
|
@ -0,0 +1,10 @@
|
|||||||
|
[
|
||||||
|
{
|
||||||
|
"pk": 2,
|
||||||
|
"model": "multiple_database.book",
|
||||||
|
"fields": {
|
||||||
|
"title": "Dive into Python",
|
||||||
|
"published": "2009-5-4"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
]
|
@ -0,0 +1,10 @@
|
|||||||
|
[
|
||||||
|
{
|
||||||
|
"pk": 2,
|
||||||
|
"model": "multiple_database.book",
|
||||||
|
"fields": {
|
||||||
|
"title": "Pro Django",
|
||||||
|
"published": "2008-12-16"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
]
|
@ -14,19 +14,7 @@ class Book(models.Model):
|
|||||||
|
|
||||||
class Author(models.Model):
|
class Author(models.Model):
|
||||||
name = models.CharField(max_length=100)
|
name = models.CharField(max_length=100)
|
||||||
|
favourite_book = models.ForeignKey(Book, null=True, related_name='favourite_of')
|
||||||
|
|
||||||
def __unicode__(self):
|
def __unicode__(self):
|
||||||
return self.name
|
return self.name
|
||||||
|
|
||||||
if len(settings.DATABASES) > 1:
|
|
||||||
article_using = filter(lambda o: o != DEFAULT_DB_ALIAS, settings.DATABASES.keys())[0]
|
|
||||||
class Article(models.Model):
|
|
||||||
title = models.CharField(max_length=100)
|
|
||||||
author = models.ForeignKey(Author)
|
|
||||||
|
|
||||||
def __unicode__(self):
|
|
||||||
return self.title
|
|
||||||
|
|
||||||
class Meta:
|
|
||||||
ordering = ('title',)
|
|
||||||
using = article_using
|
|
||||||
|
@ -14,82 +14,243 @@ try:
|
|||||||
except ImportError:
|
except ImportError:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
class ConnectionHandlerTestCase(TestCase):
|
|
||||||
def test_alias_for_connection(self):
|
|
||||||
for db in connections:
|
|
||||||
self.assertEqual(db, connections.alias_for_connection(connections[db]))
|
|
||||||
|
|
||||||
|
|
||||||
class QueryTestCase(TestCase):
|
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="Dive into Python",
|
||||||
|
published=datetime.date(2009, 5, 4))
|
||||||
|
|
||||||
|
# Create a book on the default database using a save
|
||||||
|
pro = Book()
|
||||||
|
pro.title="Pro Django"
|
||||||
|
pro.published = datetime.date(2008, 12, 16)
|
||||||
|
pro.save()
|
||||||
|
|
||||||
|
# Check that book exists on the default database, but not on other database
|
||||||
|
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"
|
||||||
|
)
|
||||||
|
|
||||||
|
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"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
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="Dive into Python",
|
||||||
|
published=datetime.date(2009, 5, 4))
|
||||||
|
|
||||||
|
# Create a book on the default database using a save
|
||||||
|
pro = Book()
|
||||||
|
pro.title="Pro Django"
|
||||||
|
pro.published = datetime.date(2008, 12, 16)
|
||||||
|
pro.save(using='other')
|
||||||
|
|
||||||
|
# Check that book 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"
|
||||||
|
)
|
||||||
|
|
||||||
|
try:
|
||||||
|
Book.objects.using('other').get(title="Pro Django")
|
||||||
|
except Book.DoesNotExist:
|
||||||
|
self.fail('"Pro Django" should exist on other database')
|
||||||
|
|
||||||
|
self.assertRaises(Book.DoesNotExist,
|
||||||
|
Book.objects.get,
|
||||||
|
title="Pro Django"
|
||||||
|
)
|
||||||
|
self.assertRaises(Book.DoesNotExist,
|
||||||
|
Book.objects.using('default').get,
|
||||||
|
title="Pro Django"
|
||||||
|
)
|
||||||
|
|
||||||
def test_basic_queries(self):
|
def test_basic_queries(self):
|
||||||
for db in connections:
|
"Queries are constrained to a single database"
|
||||||
self.assertRaises(Book.DoesNotExist,
|
dive = Book.objects.using('other').create(title="Dive into Python",
|
||||||
lambda: Book.objects.using(db).get(title="Dive into Python"))
|
published=datetime.date(2009, 5, 4))
|
||||||
Book.objects.using(db).create(title="Dive into Python",
|
|
||||||
published=datetime.date(2009, 5, 4))
|
|
||||||
|
|
||||||
for db in connections:
|
dive = Book.objects.using('other').get(published=datetime.date(2009, 5, 4))
|
||||||
books = Book.objects.all().using(db)
|
self.assertEqual(dive.title, "Dive into Python")
|
||||||
self.assertEqual(books.count(), 1)
|
self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, published=datetime.date(2009, 5, 4))
|
||||||
self.assertEqual(len(books), 1)
|
|
||||||
self.assertEqual(books[0].title, "Dive into Python")
|
|
||||||
self.assertEqual(books[0].published, datetime.date(2009, 5, 4))
|
|
||||||
|
|
||||||
for db in connections:
|
dive = Book.objects.using('other').get(title__icontains="dive")
|
||||||
self.assertRaises(Book.DoesNotExist,
|
self.assertEqual(dive.title, "Dive into Python")
|
||||||
lambda: Book.objects.using(db).get(title="Pro Django"))
|
self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, title__icontains="dive")
|
||||||
book = Book(title="Pro Django", published=datetime.date(2008, 12, 16))
|
|
||||||
book.save(using=db)
|
|
||||||
|
|
||||||
for db in connections:
|
dive = Book.objects.using('other').get(title__iexact="dive INTO python")
|
||||||
books = Book.objects.all().using(db)
|
self.assertEqual(dive.title, "Dive into Python")
|
||||||
self.assertEqual(books.count(), 2)
|
self.assertRaises(Book.DoesNotExist, Book.objects.using('default').get, title__iexact="dive INTO python")
|
||||||
self.assertEqual(len(books), 2)
|
|
||||||
self.assertEqual(books[0].title, "Dive into Python")
|
|
||||||
self.assertEqual(books[1].title, "Pro Django")
|
|
||||||
|
|
||||||
pro = Book.objects.using(db).get(published=datetime.date(2008, 12, 16))
|
dive = Book.objects.using('other').get(published__year=2009)
|
||||||
self.assertEqual(pro.title, "Pro Django")
|
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)
|
||||||
|
|
||||||
dive = Book.objects.using(db).get(title__icontains="dive")
|
years = Book.objects.using('other').dates('published', 'year')
|
||||||
self.assertEqual(dive.title, "Dive into Python")
|
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], [])
|
||||||
|
|
||||||
dive = Book.objects.using(db).get(title__iexact="dive INTO python")
|
months = Book.objects.using('other').dates('published', 'month')
|
||||||
self.assertEqual(dive.title, "Dive into Python")
|
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], [])
|
||||||
|
|
||||||
pro = Book.objects.using(db).get(published__year=2008)
|
def test_m2m(self):
|
||||||
self.assertEqual(pro.title, "Pro Django")
|
"M2M fields are constrained to a single database"
|
||||||
self.assertEqual(pro.published, datetime.date(2008, 12, 16))
|
# Create a book and author on the default database
|
||||||
|
dive = Book.objects.create(title="Dive into Python",
|
||||||
|
published=datetime.date(2009, 5, 4))
|
||||||
|
|
||||||
years = Book.objects.using(db).dates('published', 'year')
|
mark = Author.objects.create(name="Mark Pilgrim")
|
||||||
self.assertEqual([o.year for o in years], [2008, 2009])
|
|
||||||
|
# Create a book and author on the other database
|
||||||
|
pro = Book.objects.using('other').create(title="Pro Django",
|
||||||
|
published=datetime.date(2008, 12, 16))
|
||||||
|
|
||||||
|
marty = Author.objects.using('other').create(name="Marty Alchin")
|
||||||
|
|
||||||
|
# Save the author relations
|
||||||
|
dive.authors = [mark]
|
||||||
|
pro.authors = [marty]
|
||||||
|
|
||||||
|
# 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(Book.objects.using('default').filter(authors__name='Mark Pilgrim').values_list('title', flat=True),
|
||||||
|
['Dive into Python'])
|
||||||
|
self.assertEquals(Book.objects.using('other').filter(authors__name='Mark Pilgrim').values_list('title', flat=True),
|
||||||
|
[])
|
||||||
|
|
||||||
|
self.assertEquals(Book.objects.using('default').filter(authors__name='Marty Alchin').values_list('title', flat=True),
|
||||||
|
[])
|
||||||
|
self.assertEquals(Book.objects.using('other').filter(authors__name='Marty Alchin').values_list('title', flat=True),
|
||||||
|
['Pro Django'])
|
||||||
|
|
||||||
|
def test_foreign_key(self):
|
||||||
|
"FK fields are constrained to a single database"
|
||||||
|
# Create a book and author on the default database
|
||||||
|
dive = Book.objects.create(title="Dive into Python",
|
||||||
|
published=datetime.date(2009, 5, 4))
|
||||||
|
|
||||||
|
mark = Author.objects.create(name="Mark Pilgrim")
|
||||||
|
|
||||||
|
# Create a book and author on the other database
|
||||||
|
pro = Book.objects.using('other').create(title="Pro Django",
|
||||||
|
published=datetime.date(2008, 12, 16))
|
||||||
|
|
||||||
|
marty = Author.objects.using('other').create(name="Marty Alchin")
|
||||||
|
|
||||||
|
# Save the author's favourite books
|
||||||
|
mark.favourite_book = dive
|
||||||
|
mark.save()
|
||||||
|
|
||||||
|
marty.favourite_book = pro
|
||||||
|
marty.save() # FIXME Should this be save(using=alias)?
|
||||||
|
|
||||||
|
mark = Author.objects.using('default').get(name="Mark Pilgrim")
|
||||||
|
self.assertEquals(mark.favourite_book.title, "Dive into Python")
|
||||||
|
|
||||||
|
marty = Author.objects.using('other').get(name='Marty Alchin')
|
||||||
|
self.assertEquals(marty.favourite_book.title, "Dive into Python")
|
||||||
|
|
||||||
|
try:
|
||||||
|
mark.favourite_book = marty
|
||||||
|
self.fail("Shouldn't be able to assign across databases")
|
||||||
|
except Exception: # FIXME - this should be more explicit
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Check that queries work across foreign key joins
|
||||||
|
self.assertEquals(Book.objects.using('default').filter(favourite_of__name='Mark Pilgrim').values_list('title', flat=True),
|
||||||
|
['Dive into Python'])
|
||||||
|
self.assertEquals(Book.objects.using('other').filter(favourite_of__name='Mark Pilgrim').values_list('title', flat=True),
|
||||||
|
[])
|
||||||
|
|
||||||
|
self.assertEquals(Book.objects.using('default').filter(favourite_of__name='Marty Alchin').values_list('title', flat=True),
|
||||||
|
[])
|
||||||
|
self.assertEquals(Book.objects.using('other').filter(favourite_of__name='Marty Alchin').values_list('title', flat=True),
|
||||||
|
['Pro Django'])
|
||||||
|
|
||||||
|
class FixtureTestCase(TestCase):
|
||||||
|
multi_db = True
|
||||||
|
fixtures = ['multidb-common', 'multidb']
|
||||||
|
|
||||||
|
def test_fixture_loading(self):
|
||||||
|
"Multi-db fixtures are loaded correctly"
|
||||||
|
# Check that "Dive into Python" exists on the default database, but not on other database
|
||||||
|
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"
|
||||||
|
)
|
||||||
|
|
||||||
|
# Check that "Pro Django" exists on the default database, but not on other database
|
||||||
|
try:
|
||||||
|
Book.objects.using('other').get(title="Pro Django")
|
||||||
|
except Book.DoesNotExist:
|
||||||
|
self.fail('"Pro Django" should exist on other database')
|
||||||
|
|
||||||
|
self.assertRaises(Book.DoesNotExist,
|
||||||
|
Book.objects.get,
|
||||||
|
title="Pro Django"
|
||||||
|
)
|
||||||
|
self.assertRaises(Book.DoesNotExist,
|
||||||
|
Book.objects.using('default').get,
|
||||||
|
title="Pro Django"
|
||||||
|
)
|
||||||
|
|
||||||
|
# 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')
|
||||||
|
|
||||||
months = Book.objects.dates('published', 'month').using(db)
|
|
||||||
self.assertEqual(sorted(o.month for o in months), [5, 12])
|
|
||||||
|
|
||||||
class PickleQuerySetTestCase(TestCase):
|
class PickleQuerySetTestCase(TestCase):
|
||||||
|
multi_db = True
|
||||||
|
|
||||||
def test_pickling(self):
|
def test_pickling(self):
|
||||||
for db in connections:
|
for db in connections:
|
||||||
|
Book.objects.using(db).create(title='Pro Django', published=datetime.date(2008, 12, 16))
|
||||||
qs = Book.objects.all()
|
qs = Book.objects.all()
|
||||||
self.assertEqual(qs._using, pickle.loads(pickle.dumps(qs))._using)
|
self.assertEqual(qs._using, pickle.loads(pickle.dumps(qs))._using)
|
||||||
|
|
||||||
|
|
||||||
if len(settings.DATABASES) > 1:
|
|
||||||
class MetaUsingTestCase(TestCase):
|
|
||||||
def test_meta_using_queries(self):
|
|
||||||
auth = Author.objects.create(name="Zed Shaw")
|
|
||||||
a = Article.objects.create(title="Django Rules!", author=auth)
|
|
||||||
self.assertEqual(Article.objects.get(title="Django Rules!"), a)
|
|
||||||
for db in connections:
|
|
||||||
if db == article_using:
|
|
||||||
a1 = Article.objects.using(db).get(title="Django Rules!")
|
|
||||||
self.assertEqual(a1, a)
|
|
||||||
self.assertEqual(a1.author, auth)
|
|
||||||
else:
|
|
||||||
self.assertRaises(Article.DoesNotExist,
|
|
||||||
lambda: Article.objects.using(db).get(title="Django Rules!"))
|
|
||||||
a.delete()
|
|
||||||
self.assertRaises(Article.DoesNotExist,
|
|
||||||
lambda: Article.objects.get(title="Django Rules!"))
|
|
||||||
self.assertRaises(ValueError,
|
|
||||||
lambda: list(Article.objects.get(pk__in=Article.objects.using('default'))))
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user