我的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?如果不行,那么在哪里可以创建这样一个外部数据库呢?
return super(CustomTestRunner, self).setup_databases(*args, **kwargs)
。注意,super
中也有*args
。 - AtlasStrategic