mirror of
https://github.com/django/django.git
synced 2025-07-04 17:59:13 +00:00
[multi-db] Added initialization code to runtests.py that creates
two named test databases, 'django_test_db_a' and 'django_test_db_b', along with the default test database. Moved multiple databases test to modeltests and reorganized it to match other model tests. git-svn-id: http://code.djangoproject.com/svn/django/branches/multiple-db-support@3241 bcc190cf-cafb-0310-a4f2-bffc1f526a37
This commit is contained in:
Jason Pellerin
parent
763e087de1
commit
4190c9e16f
1
tests/modeltests/multiple_databases/__init__.py
Normal file
1
tests/modeltests/multiple_databases/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
pass
|
||||
96
tests/modeltests/multiple_databases/models.py
Normal file
96
tests/modeltests/multiple_databases/models.py
Normal file
@ -0,0 +1,96 @@
|
||||
"""
|
||||
XXX. Using multiple database connections
|
||||
|
||||
Django normally uses only a single database connection. However,
|
||||
support is available for using any number of different, named
|
||||
connections. Multiple database support is entirely optional and has
|
||||
no impact on your application if you don't use it.
|
||||
|
||||
Named connections are defined in your settings module. Create a
|
||||
`DATABASES` variable that is a dict, mapping connection names to their
|
||||
particulars. The particulars are defined in a dict with the same keys
|
||||
as the variable names as are used to define the default connection.
|
||||
|
||||
Access to named connections is through `django.db.connections`, which
|
||||
behaves like a dict: you access connections by name. Connections are
|
||||
established lazily, when accessed. `django.db.connections[database]`
|
||||
holds a `ConnectionInfo` instance, with the attributes:
|
||||
`DatabaseError`, `backend`, `get_introspection_module`,
|
||||
`get_creation_module`, and `runshell`.
|
||||
|
||||
Models can define which connection to use, by name. To use a named
|
||||
connection, set the `db_connection` property in the model's Meta class
|
||||
to the name of the connection. The name used must be a key in
|
||||
settings.DATABASES, of course.
|
||||
|
||||
To access a model's connection, use `model._meta.connection`. To find
|
||||
the backend or other connection metadata, use
|
||||
`model._meta.connection_info`.
|
||||
"""
|
||||
|
||||
from django.db import models
|
||||
|
||||
class Artist(models.Model):
|
||||
name = models.CharField(maxlength=100)
|
||||
alive = models.BooleanField(default=True)
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
||||
class Meta:
|
||||
db_connection = 'django_test_db_a'
|
||||
|
||||
class Widget(models.Model):
|
||||
code = models.CharField(maxlength=10, unique=True)
|
||||
weight = models.IntegerField()
|
||||
|
||||
def __str__(self):
|
||||
return self.code
|
||||
|
||||
class Meta:
|
||||
db_connection = 'django_test_db_b'
|
||||
|
||||
class Vehicle(models.Model):
|
||||
make = models.CharField(maxlength=20)
|
||||
model = models.CharField(maxlength=20)
|
||||
year = models.IntegerField()
|
||||
|
||||
def __str__(self):
|
||||
return "%d %s %s" % (self.year, self.make, self.model)
|
||||
|
||||
|
||||
API_TESTS = """
|
||||
|
||||
# See what connections are defined. django.db.connections acts like a dict.
|
||||
>>> from django.db import connection, connections
|
||||
>>> from django.conf import settings
|
||||
>>> connections.keys()
|
||||
['django_test_db_a', 'django_test_db_b']
|
||||
|
||||
# Each connection references its settings
|
||||
>>> connections['django_test_db_a'].settings.DATABASE_NAME == settings.DATABASES['django_test_db_a']['DATABASE_NAME']
|
||||
True
|
||||
>>> connections['django_test_db_b'].settings.DATABASE_NAME == settings.DATABASES['django_test_db_b']['DATABASE_NAME']
|
||||
True
|
||||
>>> connections['django_test_db_b'].settings.DATABASE_NAME == settings.DATABASES['django_test_db_a']['DATABASE_NAME']
|
||||
False
|
||||
|
||||
# Invalid connection names raise ImproperlyConfigured
|
||||
>>> connections['bad']
|
||||
Traceback (most recent call last):
|
||||
...
|
||||
ImproperlyConfigured: No database connection 'bad' has been configured
|
||||
|
||||
# Models can access their connections through their _meta properties
|
||||
>>> Artist._meta.connection.settings == connections['django_test_db_a'].settings
|
||||
True
|
||||
>>> Widget._meta.connection.settings == connections['django_test_db_b'].settings
|
||||
True
|
||||
>>> Vehicle._meta.connection.settings == connection.settings
|
||||
True
|
||||
>>> Artist._meta.connection.settings == Widget._meta.connection.settings
|
||||
False
|
||||
>>> Artist._meta.connection.settings == Vehicle._meta.connection.settings
|
||||
False
|
||||
|
||||
"""
|
||||
@ -1,175 +0,0 @@
|
||||
import copy
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from django.conf import settings
|
||||
from django import db
|
||||
from runtests import doctest, DjangoDoctestRunner, error_list
|
||||
|
||||
# database files
|
||||
fh, db_a = tempfile.mkstemp()
|
||||
os.close(fh)
|
||||
fh, db_b = tempfile.mkstemp()
|
||||
os.close(fh)
|
||||
|
||||
# patches
|
||||
tmp_settings = None
|
||||
tmp_connections = None
|
||||
|
||||
# tests
|
||||
test = r"""
|
||||
XXX. Using multiple database connections
|
||||
|
||||
Django normally uses only a single database connection. However, support is
|
||||
available for connecting to any number of different, named databases.
|
||||
|
||||
Named connections are defined in your settings module. Create a
|
||||
`DATABASES` variable that is a dict, mapping connection names to their
|
||||
particulars. The particulars are defined in a dict with the same keys
|
||||
as the variable names as are used to define the default connection.
|
||||
|
||||
.. note::
|
||||
|
||||
Please note that this uses the sqlite3 backend and writes temporary
|
||||
database files to disk. This test will fail unless sqlite3 is
|
||||
installed and your temp directory is writable.
|
||||
|
||||
>>> from django.conf import settings
|
||||
>>> settings.DATABASES = {
|
||||
... 'a': { 'DATABASE_ENGINE': 'sqlite3',
|
||||
... 'DATABASE_NAME': db_a
|
||||
... },
|
||||
... 'b': { 'DATABASE_ENGINE': 'sqlite3',
|
||||
... 'DATABASE_NAME': db_b
|
||||
... }}
|
||||
|
||||
Connections are established lazily, when requested by name. When
|
||||
accessed, `connections[database]` holds a `ConnectionInfo` instance,
|
||||
with the attributes: `DatabaseError`, `backend`,
|
||||
`get_introspection_module`, `get_creation_module`, and
|
||||
`runshell`. Access connections through the `connections` property of
|
||||
the `django.db` module:
|
||||
|
||||
>>> from django.db import connection, connections
|
||||
>>> connections['a'].settings.DATABASE_NAME == db_a
|
||||
True
|
||||
>>> connections['b'].settings.DATABASE_NAME == db_b
|
||||
True
|
||||
|
||||
Invalid connection names raise ImproperlyConfigured:
|
||||
|
||||
>>> connections['bad']
|
||||
Traceback (most recent call last):
|
||||
...
|
||||
ImproperlyConfigured: No database connection 'bad' has been configured
|
||||
|
||||
Models can define which connection to use, by name. To use a named
|
||||
connection, set the `db_connection` property in the model's Meta class
|
||||
to the name of the connection. The name used must be a key in
|
||||
settings.DATABASES, of course.
|
||||
|
||||
>>> from django.db import models
|
||||
>>> class Artist(models.Model):
|
||||
... name = models.CharField(maxlength=100)
|
||||
... alive = models.BooleanField(default=True)
|
||||
...
|
||||
... def __str__(self):
|
||||
... return self.name
|
||||
...
|
||||
... class Meta:
|
||||
... app_label = 'mdb'
|
||||
... db_connection = 'a'
|
||||
...
|
||||
>>> class Widget(models.Model):
|
||||
... code = models.CharField(maxlength=10, unique=True)
|
||||
... weight = models.IntegerField()
|
||||
...
|
||||
... def __str__(self):
|
||||
... return self.code
|
||||
...
|
||||
... class Meta:
|
||||
... app_label = 'mdb'
|
||||
... db_connection = 'b'
|
||||
|
||||
But they don't have to. Multiple database support is entirely optional
|
||||
and has no impact on your application if you don't use it.
|
||||
|
||||
>>> class Vehicle(models.Model):
|
||||
... make = models.CharField(maxlength=20)
|
||||
... model = models.CharField(maxlength=20)
|
||||
... year = models.IntegerField()
|
||||
...
|
||||
... def __str__(self):
|
||||
... return "%d %s %s" % (self.year, self.make, self.model)
|
||||
...
|
||||
... class Meta:
|
||||
... app_label = 'mdb'
|
||||
|
||||
>>> Artist._meta.connection.settings.DATABASE_NAME == \
|
||||
... connections['a'].connection.settings.DATABASE_NAME
|
||||
True
|
||||
>>> Widget._meta.connection.settings.DATABASE_NAME == \
|
||||
... connections['b'].connection.settings.DATABASE_NAME
|
||||
True
|
||||
>>> Vehicle._meta.connection.settings.DATABASE_NAME == \
|
||||
... connection.settings.DATABASE_NAME
|
||||
True
|
||||
>>> Artist._meta.connection.settings.DATABASE_NAME == \
|
||||
... Widget._meta.connection.settings.DATABASE_NAME
|
||||
False
|
||||
>>> Artist._meta.connection.settings.DATABASE_NAME == \
|
||||
... Vehicle._meta.connection.settings.DATABASE_NAME
|
||||
False
|
||||
|
||||
"""
|
||||
|
||||
def cleanup():
|
||||
if os.path.exists(db_a):
|
||||
os.unlink(db_a)
|
||||
if os.path.exists(db_b):
|
||||
os.unlink(db_b)
|
||||
|
||||
def setup():
|
||||
global tmp_settings, tmp_connections
|
||||
try:
|
||||
tmp_connections = db.connections
|
||||
db.connections = db.LazyConnectionManager()
|
||||
tmp_settings = copy.copy(settings.DATABASES)
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
def teardown():
|
||||
try:
|
||||
db.connections = tmp_connections
|
||||
settings.DATABASES = tmp_settings
|
||||
except AttributeError:
|
||||
pass
|
||||
cleanup()
|
||||
|
||||
def run_tests(verbosity_level):
|
||||
setup()
|
||||
try:
|
||||
main(verbosity_level)
|
||||
finally:
|
||||
teardown()
|
||||
|
||||
def main(verbosity_level):
|
||||
mod = sys.modules[__name__]
|
||||
p = doctest.DocTestParser()
|
||||
dtest = p.get_doctest(mod.test, mod.__dict__, __name__, None, None)
|
||||
runner = DjangoDoctestRunner(verbosity_level=verbosity_level,
|
||||
verbose=False)
|
||||
runner.run(dtest, clear_globs=True, out=sys.stdout.write)
|
||||
if error_list:
|
||||
out = []
|
||||
for d in error_list:
|
||||
out.extend([d['title'], "=" * len(d['title']),
|
||||
d['description']])
|
||||
|
||||
raise Exception, "%s multiple_databases test%s failed:\n\n %s" \
|
||||
% (len(error_list),
|
||||
len(error_list) != 1 and 's' or '',
|
||||
'\n'.join(out))
|
||||
|
||||
if __name__ == '__main__':
|
||||
run_tests(1)
|
||||
@ -85,39 +85,25 @@ class TestRunner:
|
||||
self.verbosity_level = verbosity_level
|
||||
self.which_tests = which_tests
|
||||
self.created_dbs = []
|
||||
self.sqlite_memory_db_used = False
|
||||
self.cleanup_files = []
|
||||
self.old_database_name = None
|
||||
self.old_databases = None
|
||||
self.sqlite_memory_db_used = False
|
||||
|
||||
def output(self, required_level, message):
|
||||
if self.verbosity_level > required_level - 1:
|
||||
print message
|
||||
|
||||
def create_test_db(self, connection):
|
||||
def create_test_db(self, db_name, connection):
|
||||
"""Create a test db, returning a ConnectionInfo object holding
|
||||
a connection to that db.
|
||||
"""
|
||||
from django.db import connect
|
||||
|
||||
# settings may be a dict or settings object
|
||||
settings = connection.settings
|
||||
|
||||
# If we're using SQLite, it's more convenient to test against an
|
||||
# in-memory database. But we can only do this for the first one; after
|
||||
# that we have to use temp files.
|
||||
if settings.DATABASE_ENGINE == "sqlite3":
|
||||
if self.sqlite_memory_db_used:
|
||||
import tempfile
|
||||
fd, filename = tempfile.mkstemp()
|
||||
os.close(fd)
|
||||
db_name = filename
|
||||
self.cleanup_files.append(filename)
|
||||
else:
|
||||
db_name = ":memory:"
|
||||
self.sqlite_memory_db_used = True
|
||||
else:
|
||||
db_name = TEST_DATABASE_NAME
|
||||
if self.created_dbs:
|
||||
db_name += "_%s" % (len(self.created_dbs))
|
||||
|
||||
|
||||
if settings.DATABASE_ENGINE != "sqlite3":
|
||||
# Create the test database and connect to it. We need to autocommit
|
||||
# if the database supports it because PostgreSQL doesn't allow
|
||||
# CREATE/DROP DATABASE statements within transactions.
|
||||
@ -136,6 +122,7 @@ class TestRunner:
|
||||
else:
|
||||
raise Exception("Tests cancelled.")
|
||||
|
||||
|
||||
settings.DATABASE_NAME = db_name
|
||||
connection.close()
|
||||
|
||||
@ -153,6 +140,7 @@ class TestRunner:
|
||||
self.teardown()
|
||||
|
||||
def setup(self):
|
||||
global TEST_DATABASE_NAME
|
||||
from django.conf import settings
|
||||
from django.db import connection, connections
|
||||
from django.core import management
|
||||
@ -170,17 +158,38 @@ class TestRunner:
|
||||
|
||||
self.output(0, "Running tests with database %r" % settings.DATABASE_ENGINE)
|
||||
|
||||
# Create test dbs for the default connection and any named connections
|
||||
# in settings. Remeber the original default db name so we can connect
|
||||
# there to drop the created test dbs.
|
||||
self._old_database_name = settings.DATABASE_NAME
|
||||
self.create_test_db(connection)
|
||||
# Create test dbs for the default connection and two named connections,
|
||||
# replacing any named connections defined in settings. All connections
|
||||
# will use the default DATABASE_ENGINE
|
||||
self.old_database_name = settings.DATABASE_NAME
|
||||
self.old_databases = settings.DATABASES
|
||||
|
||||
db_a = TEST_DATABASE_NAME + '_a'
|
||||
db_b = TEST_DATABASE_NAME + '_b'
|
||||
if settings.DATABASE_ENGINE == 'sqlite3':
|
||||
# If we're using SQLite, it's more convenient to test against an
|
||||
# in-memory database. But we can only do this for the default;
|
||||
# after that we have to use temp files.
|
||||
TEST_DATABASE_NAME = ':memory:'
|
||||
db_a_name = self._tempfile()
|
||||
db_b_name = self._tempfile()
|
||||
self.cleanup_files.append(db_a_name)
|
||||
self.cleanup_files.append(db_b_name)
|
||||
else:
|
||||
db_a_name = db_a
|
||||
db_b_name = db_b
|
||||
|
||||
if hasattr(settings, 'DATABASES'):
|
||||
for name, info in settings.DATABASES.items():
|
||||
cx = connections[name]
|
||||
test_connection = self.create_test_db(cx.connection)
|
||||
connections[name] = test_connection
|
||||
settings.DATABASES = {
|
||||
db_a: { 'DATABASE_NAME': db_a_name },
|
||||
db_b: { 'DATABASE_NAME': db_b_name }
|
||||
}
|
||||
|
||||
self.create_test_db(TEST_DATABASE_NAME, connection)
|
||||
for name, info in settings.DATABASES.items():
|
||||
cx = connections[name]
|
||||
test_connection = self.create_test_db(info['DATABASE_NAME'],
|
||||
cx.connection)
|
||||
connections[name] = test_connection
|
||||
|
||||
# Install the core always installed apps
|
||||
for app in ALWAYS_INSTALLED_APPS:
|
||||
@ -196,7 +205,8 @@ class TestRunner:
|
||||
from django.db import connection
|
||||
from django.conf import settings
|
||||
connection.close()
|
||||
settings.DATABASE_NAME = self._old_database_name
|
||||
settings.DATABASE_NAME = self.old_database_name
|
||||
settings.DATABASES = self.old_databases
|
||||
for db_name, cx in self.created_dbs:
|
||||
settings = cx.settings
|
||||
cx.close()
|
||||
@ -334,6 +344,12 @@ class TestRunner:
|
||||
elif hasattr(connection.connection, "set_isolation_level"):
|
||||
connection.connection.set_isolation_level(0)
|
||||
|
||||
def _tempfile(self):
|
||||
import tempfile
|
||||
fd, filename = tempfile.mkstemp()
|
||||
os.close(fd)
|
||||
return filename
|
||||
|
||||
if __name__ == "__main__":
|
||||
from optparse import OptionParser
|
||||
usage = "%prog [options] [model model model ...]"
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user