如何在设置Django测试环境时创建外部数据库?

4
我的Django应用需要连接一个非Django管理的外部Postgres数据库。为了测试我的项目,我需要使用原始SQL创建外部数据库和表,并在之后删除该数据库。从阅读Django文档得知,可以通过使用现有的DiscoverRunner类来创建自己的TestRunner类。可以重写setup_test_environment和teardown_test_environment方法,以执行SQL以创建外部数据库。
from psycopg2 import connect
from django.test.runner import DiscoverRunner

class CustomTestRunner(DiscoverRunner):
    """
    Test runner that helps to setup external db if doesn`t exist.
    """

    def setup_test_environment(self, *args, **kwargs):
        conn = connect(database='postgres', host='localhost', user='my_user', password='password123')
        try:
            with conn.cursor() as c:
                query = """CREATE DATABASE "test_db" """
                query += """WITH OWNER = my_user """
                query += """ENCODING = 'UTF8' TABLESPACE = pg_default """
                query += """LC_COLLATE = 'en_ZA.UTF-8' LC_CTYPE = 'en_ZA.UTF-8' """
                query += """CONNECTION LIMIT = -1;"""
                c.execute(query)
        except Exception as e:
            print e
            conn.close()

        super(CustomTestRunner, self).setup_test_environment(*args,
            **kwargs)

    def teardown_test_environment(self, *args, **kwargs):
        super(CustomTestRunner, self).teardown_test_environment(*args,
            **kwargs)

        # Delete external database after tests
        try:
            conn = connect(database='postgres', host='localhost', user='my_user', password='password123')
            with conn.cursor() as c:
                query = """DROP DATABASE test_db;"""
                c.execute(query)
        except Exception as e:
            print e
            conn.close()
        conn.close()

然而,当我运行测试时,会出现以下错误信息:

CREATE DATABASE不能在事务块中运行

是否可以在Django TestRunner中运行和提交SQL?如果不行,那么在哪里可以创建这样一个外部数据库呢?

1个回答

1

一些数据库命令,例如CREATE DATABASE无法在事务中运行。您需要在数据库连接上设置适当的隔离级别:

conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

请查看http://initd.org/psycopg/docs/extensions.html#isolation-level-constants

在测试运行程序中也可以运行和提交SQL。更适合这样做的地方是setup_databasesteardown_databases方法。以下是一个示例

def setup_databases(self, **kwargs):
    conn = connect(database='postgres', host='localhost', ...)
    try:        
        # set the proper isolation level so CREATE DATABASE works
        conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
        with conn.cursor() as cur:
            # execute SQL in here
            query = 'create database TEST_DB'
            cur.execute(query)
    finally:
        conn.close()

    return super(CustomTestRunner, self).setup_databases(**kwargs)

实现teardown_databases应该非常类似。

我已经测试了您的解决方案,非常好用。非常感谢你! - AtlasStrategic
作为一个脚注,我们必须记得返回 return super(CustomTestRunner, self).setup_databases(*args, **kwargs)。注意,super 中也有 *args - AtlasStrategic

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接