django - 选择测试数据库？

Question

我正在尝试运行

./manage.py test

但它告诉我

创建测试数据库时出错：创建数据库的权限被拒绝

显然它没有创建数据库的权限，但我在共享服务器上，所以我无能为力。我可以通过控制面板创建一个新数据库，但我认为没有任何方法可以让 Django 自动完成。

那么，我不能手动创建测试数据库，而是告诉 Django 每次都刷新它，而不是重新创建整个东西吗？

Answer 1

我有一个类似的问题。但我希望 Django 只是绕过为我的一个实例创建测试数据库（这不是镜像很难）。按照 Mark 的建议，我创建了一个自定义测试运行器，如下

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

然后我在 settings.py 中的一个数据库条目中创建了一个额外的 dict 条目，'BYPASS_CREATION':'yes',

最后，我配置了一个新的 TestRunner

TEST_RUNNER = 'auth.data.runner.ByPassableDBDjangoTestSuiteRunner'

Answer 2

我建议使用 sqlite3 进行测试，同时继续使用 mysql/postgres/etc 进行生产。

这可以通过将其放置在您的设置文件中来实现：

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

请参阅使用 sqlite 运行 django 测试

将在您的 django 项目主页中创建一个临时 sqlite 数据库文件，您将拥有该文件的写入权限。另一个优点是 sqlite3 的测试速度要快得多。但是，如果您使用任何 mysql/postgres 特定的原始 sql（无论如何您应该尽量避免），您可能会遇到问题。

Answer 3

我认为更好的解决方案可能是定义自己的 test runner。

Answer 4

我在上面的评论中添加了这个，但它有点迷失了——最近对 webfaction 的更改使这变得更容易了。您现在可以创建新的私有数据库实例。

按照那里的说明进行操作，并在创建新用户时确保授予他们ALTER USER new_username CREATEDB;.

您可能还应该更改默认的 cron 设置，这样他们就不会尝试检查该数据库是否已启动并频繁运行。

Answer 5

你可以使用django-nose作为你的 TEST_RUNNER。安装好后，如果传入如下环境变量，则不会删除并重新创建数据库（先自己手动创建）。

REUSE_DB=1 ./manage.py test

您还可以将以下内容添加到 settings.py 中，这样您就不必在每次要运行测试时都编写 REUSE_DB=1 ：

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

注意：这也会将所有表留在数据库中，这意味着测试设置会更快一些，但是当您更改模型时，您必须手动更新表（或自己删除并重新创建数据库）。

Answer 6

我重用数据库的变体：

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)

Answer 7

以下是使用Webfaction XML-RPC API创建数据库的 django 测试套件运行器。请注意，使用 API 设置数据库可能需要一分钟时间，并且脚本可能会出现暂时卡住的情况，请稍等片刻。

注意：在 webfaction 服务器中有控制面板密码存在安全风险，因为有人侵入您的 web 服务器 SSH 可能会接管您的 Webfaction 帐户。如果这是一个问题，请将 USE_SESSKEY 设置为 True 并使用此脚本下方的结构脚本将会话 ID 传递给服务器。会话密钥在最后一次 API 调用后 1 小时内过期。

文件test_runner.py：在服务器中，需要配置./manage.py test 才能使用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)

文件 webfaction.py：这是 Webfaction API 的精简包装，它需要由 test_runner.py（在远程服务器中）和 fabfile.py（在本地机器中）都可以导入

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

文件 fabfile.py：生成会话密钥的示例结构脚本，仅当 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)

Answer 8

接受的答案对我不起作用。它已经过时了，它没有在我的 djano 1.5 遗留代码库上运行。

我写了一篇博文，完整地描述了我如何通过创建替代测试运行程序并更改 django 设置以提供所有必需的配置并使用新的测试运行程序来解决这个问题。

Answer 9

修改以下方法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"]

似乎工作......settings.py如果你需要的话，只需将额外的设置添加到你的。

Answer 10

简单的解决方法：随意更改TEST_DATABASE_PREFIX。django/db/backends/base/creation.py

Answer 11

使用单元测试时需要指定 sqlite ENGINE。打开 settings.py 并添加紧随其后的 DATABASES 部分：

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:'