[Django]-Choose test database?

14πŸ‘

βœ…

I had a similar issue. But I wanted Django to just bypass the creation of a test database for one of my instances (it is not a mirror tough). Following Mark’s suggestion, I created a custom test runner, as follows

from django.test.simple import DjangoTestSuiteRunner


class ByPassableDBDjangoTestSuiteRunner(DjangoTestSuiteRunner):

    def setup_databases(self, **kwargs):
        from django.db import connections
        old_names = []
        mirrors = []

        for alias in connections:
            connection = connections[alias]
            # If the database is a test mirror, redirect its connection
            # instead of creating a test database.
            if connection.settings_dict['TEST_MIRROR']:
                mirrors.append((alias, connection))
                mirror_alias = connection.settings_dict['TEST_MIRROR']
                connections._connections[alias] = connections[mirror_alias]
            elif connection.settings_dict.get('BYPASS_CREATION','no') == 'no':
                old_names.append((connection, connection.settings_dict['NAME']))
                connection.creation.create_test_db(self.verbosity, autoclobber=not self.interactive)
        return old_names, mirrors

Then I created an extra dict entry in one of my databases entries inside settings.py, 'BYPASS_CREATION':'yes',

Finally, I configured a new TestRunner with

TEST_RUNNER = 'auth.data.runner.ByPassableDBDjangoTestSuiteRunner'

10πŸ‘

I would suggest using sqlite3 for testing purposes while keeping on using mysql/postgres/etc for production.

This can be achieved by placing this in your settings file:

if 'test' in sys.argv:
    DATABASES['default'] = {'ENGINE': 'django.db.backends.sqlite3'}

see Running django tests with sqlite

a temporary sqlite database file will be created in your django project home which you will have write access to. The other advantage is that sqlite3 is much faster for testing. You may however run in to problems if you are using any mysql/postgres specific raw sql (which you should try to avoid anyway).

4πŸ‘

I think a better solution might be to define your own test runner.

3πŸ‘

I added this to the comments above but it got kind of lost – recent changes to webfaction make this MUCH easier. You can now create new private database instances.

Follow the instructions there, and when creating a new user make sure to give them the permission to ALTER USER new_username CREATEDB;.

You probably also should change the default cron settings so they don’t try to check if this database is up and runnings as frequently.

2πŸ‘

You could use django-nose as your TEST_RUNNER. Once installed, if you pass the following environment variable, it will not delete and re-create the database (create it manually yourself first).

REUSE_DB=1 ./manage.py test

You can also add the following to settings.py so you don’t have to write REUSE_DB=1 every time you want to run tests:

os.environ['REUSE_DB'] = "1"

Note: this will also leave all your tables in the databases which means test setup will be a little quicker, but you will have to manually update the tables (or delete and re-create the database yourself) when you change your models.

2πŸ‘

my variant to reusing database:

from django.test.simple import DjangoTestSuiteRunner
from django.core.management import call_command


class TestRunner(DjangoTestSuiteRunner):
    def setup_databases(self, **kwargs):
        from django.db import connections

        settings = connections['default'].settings_dict
        settings['NAME'] = settings['TEST_NAME']
        settings['USER'] = settings['TEST_USER']
        settings['PASSWORD'] = settings['TEST_PASSWD']

        call_command('syncdb', verbosity=1, interactive=False, load_initial_data=False)

    def teardown_databases(self, old_config, **kwargs):
        from django.db import connection

        cursor = connection.cursor()
        cursor.execute('show tables;')
        parts = ('DROP TABLE IF EXISTS %s;' % table for (table,) in cursor.fetchall())
        sql = 'SET FOREIGN_KEY_CHECKS = 0;\n' + '\n'.join(parts) + 'SET FOREIGN_KEY_CHECKS = 1;\n'
        connection.cursor().execute(sql)

1πŸ‘

The following is a django test suite runner to create database using Webfaction XML-RPC API. Note, setting up the database using the API may take up to a minute, and the script may appear to be stuck momentarily, just wait for a little while.

NOTE: there is a security risk of having control panel password in the webfaction server, because someone breaching into your web server SSH could take over your Webfaction account. If that is a concern, set USE_SESSKEY to True and use the fabric script below this script to pass a session id to the server. The session key expires in 1 hour from the last API call.

File test_runner.py: in the server, you need to configure ./manage.py test to use WebfactionTestRunner

"""
This test runner uses Webfaction XML-RPC API to create and destroy database
"""

# you can put your control panel username and password here.
# NOTE: there is a security risk of having control panel password in
# the webfaction server, because someone breaching into your web server
# SSH could take over your Webfaction account. If that is a concern,
# set USE_SESSKEY to True and use the fabric script below this script to
# generate a session.

USE_SESSKEY = True
# CP_USERNAME = 'webfactionusername' # required if and only if USE_SESSKEY is False
# CP_PASSWORD = 'webfactionpassword' # required if and only if USE_SESSKEY is False

import sys
import os
from django.test.simple import DjangoTestSuiteRunner
from django import db
from webfaction import Webfaction

def get_sesskey():
    f = os.path.expanduser("~/sesskey")
    sesskey = open(f).read().strip()
    os.remove(f)
    return sesskey

if USE_SESSKEY:
    wf = Webfaction(get_sesskey())
else:
    wf = Webfaction()
    wf.login(CP_USERNAME, CP_PASSWORD)


def get_db_user_and_type(connection):
    db_types = {
        'django.db.backends.postgresql_psycopg2': 'postgresql',
        'django.db.backends.mysql': 'mysql',
    }
    return (
        connection.settings_dict['USER'],
        db_types[connection.settings_dict['ENGINE']],
    )


def _create_test_db(self, verbosity, autoclobber):
    """
    Internal implementation - creates the test db tables.
    """

    test_database_name = self._get_test_db_name()

    db_user, db_type = get_db_user_and_type(self.connection)

    try:
        wf.create_db(db_user, test_database_name, db_type)
    except Exception as e:
        sys.stderr.write(
            "Got an error creating the test database: %s\n" % e)
        if not autoclobber:
            confirm = raw_input(
                "Type 'yes' if you would like to try deleting the test "
                "database '%s', or 'no' to cancel: " % test_database_name)
        if autoclobber or confirm == 'yes':
            try:
                if verbosity >= 1:
                    print("Destroying old test database '%s'..."
                        % self.connection.alias)
                wf.delete_db(test_database_name, db_type)
                wf.create_db(db_user, test_database_name, db_type)
            except Exception as e:
                sys.stderr.write(
                    "Got an error recreating the test database: %s\n" % e)
                sys.exit(2)
        else:
            print("Tests cancelled.")
            sys.exit(1)

    db.close_connection()
    return test_database_name


def _destroy_test_db(self, test_database_name, verbosity):
    """
    Internal implementation - remove the test db tables.
    """
    db_user, db_type = get_db_user_and_type(self.connection)
    wf.delete_db(test_database_name, db_type)
    self.connection.close()


class WebfactionTestRunner(DjangoTestSuiteRunner):
    def __init__(self, *args, **kwargs):
        # Monkey patch BaseDatabaseCreation with our own version
        from django.db.backends.creation import BaseDatabaseCreation
        BaseDatabaseCreation._create_test_db = _create_test_db
        BaseDatabaseCreation._destroy_test_db = _destroy_test_db

        return super(WebfactionTestRunner, self).__init__(*args, **kwargs)

File webfaction.py: this is a thin wrapper for Webfaction API, it need to be importable by both test_runner.py (in the remote server) and the fabfile.py (in the local machine)

import xmlrpclib

class Webfaction(object):
    def __init__(self, sesskey=None):
        self.connection = xmlrpclib.ServerProxy("https://api.webfaction.com/")
        self.sesskey = sesskey

    def login(self, username, password):
        self.sesskey, _ = self.connection.login(username, password)

    def create_db(self, db_user, db_name, db_type):
        """ Create a database owned by db_user """
        self.connection.create_db(self.sesskey, db_name, db_type, 'unused')

        # deletes the default user created by Webfaction API
        self.connection.make_user_owner_of_db(self.sesskey, db_user, db_name, db_type)
        self.connection.delete_db_user(self.sesskey, db_name, db_type)

    def delete_db(self, db_name, db_type):
        try:
            self.connection.delete_db_user(self.sesskey, db_name, db_type)
        except xmlrpclib.Fault as e:
            print 'ignored error:', e
        try:
            self.connection.delete_db(self.sesskey, db_name, db_type)
        except xmlrpclib.Fault as e:
            print 'ignored error:', e

File fabfile.py: A sample fabric script to generate session key, needed only if USE_SESSKEY=True

from fabric.api import *
from fabric.operations import run, put
from webfaction import Webfaction
import io

env.hosts = ["webfactionusername@webfactionusername.webfactional.com"]
env.password = "webfactionpassword"

def run_test():
    wf = Webfaction()
    wf.login(env.hosts[0].split('@')[0], env.password)
    sesskey_file = '~/sesskey'
    sesskey = wf.sesskey
    try:
        put(io.StringIO(unicode(sesskey)), sesskey_file, mode='0600')
        # put your test code here
        # e.g. run('DJANGO_SETTINGS_MODULE=settings /path/to/virtualenv/python /path/to/manage.py test --testrunner=test_runner.WebfactionTestRunner')
        raise Exception('write your test here')
    finally:
        run("rm -f %s" % sesskey_file)

1πŸ‘

The accepted answer didn’t work for me. It’s so outdated, that it didn’t run on my legacy codebase with djano 1.5.

I wrote a blogpost entirely describing how I solved this issue by creating an alternative test runner and changing django settings to provide all the required config and to use new test runner.

1πŸ‘

You need to specify a sqlite ENGINE when using unit tests. Open the settings.py and add the just after DATABASES section:

import sys
if 'test' in sys.argv or 'test_coverage' in sys.argv: #Covers regular testing and django-coverage
    DATABASES['default']['ENGINE'] = 'django.db.backends.sqlite3'
    DATABASES['default']['NAME'] = ':memory:'

0πŸ‘

Modify the following methods in django/db/backends/creation.py:

def _destroy_test_db(self, test_database_name, verbosity):
    "Internal implementation - remove the test db tables."

    # Remove the test database to clean up after
    # ourselves. Connect to the previous database (not the test database)
    # to do so, because it's not allowed to delete a database while being
    # connected to it.
    self._set_test_dict()
    cursor = self.connection.cursor()
    self.set_autocommit()
    time.sleep(1) # To avoid "database is being accessed by other users" errors.

    cursor.execute("""SELECT table_name FROM information_schema.tables WHERE table_schema='public'""")
    rows = cursor.fetchall()
    for row in rows:
        try:
            print "Dropping table '%s'" % row[0]
            cursor.execute('drop table %s cascade ' % row[0])
        except:
            print "Couldn't drop '%s'" % row[0] 

    #cursor.execute("DROP DATABASE %s" % self.connection.ops.quote_name(test_database_name))
    self.connection.close()

def _create_test_db(self, verbosity, autoclobber):
    "Internal implementation - creates the test db tables."

    suffix = self.sql_table_creation_suffix()

    if self.connection.settings_dict['TEST_NAME']:
        test_database_name = self.connection.settings_dict['TEST_NAME']
    else:
        test_database_name = TEST_DATABASE_PREFIX + self.connection.settings_dict['NAME']

    qn = self.connection.ops.quote_name

    # 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.
    self._set_test_dict()
    cursor = self.connection.cursor()
    self.set_autocommit()

    return test_database_name

def _set_test_dict(self):
    if "TEST_NAME" in self.connection.settings_dict:
        self.connection.settings_dict["NAME"] = self.connection.settings_dict["TEST_NAME"]
    if "TEST_USER" in self.connection.settings_dict:
        self.connection.settings_dict['USER'] = self.connection.settings_dict["TEST_USER"]
    if "TEST_PASSWORD" in self.connection.settings_dict:
        self.connection.settings_dict['PASSWORD'] = self.connection.settings_dict["TEST_PASSWORD"]

Seems to work… just add the extra settings to your settings.py if you need ’em.

0πŸ‘

Simple workaround: change TEST_DATABASE_PREFIX in django/db/backends/base/creation.py as you like.

Leave a comment