Django：在单独线程中使用相同的测试数据库

Question

我正在使用一个测试数据库运行pytest，以下是我的数据库设置。

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql_psycopg2',
        'NAME': 'postgres',
        'USER': 'something',
        'PASSWORD': 'password',

    },
}

通过使用 @pytest.mark.django_db，我的测试函数可以访问一个名为'test_postgres'的数据库，这个数据库是专门为测试创建的。

@pytest.mark.django_db
def test_example():
    from django.db import connection
    cur_ = connection.cursor()
    print cur_.db.settings_dict

输出结果：

{'ENGINE': 'django.db.backends.postgresql_psycopg2', 'AUTOCOMMIT': True, 'ATOMIC_REQUESTS': False, 'NAME': 'test_postgres', 'TEST_MIRROR': None,...

但是如果我在 test_example 里面运行一个线程：

def function_to_run():
    from django.db import connection
    cur_ = connection.cursor
    logger.error(cur_.db.settings_dict)

@pytest.mark.django_db
def test_example():
    p = multiprocessing.Process(target=function_to_run)
    p.start()

我发现那个线程里的数据库连接使用的是名为'postgres'的数据库，这个是非测试用的数据库。输出结果：

{'ENGINE': 'django.db.backends.postgresql_psycopg2', 'AUTOCOMMIT': True, 'ATOMIC_REQUESTS': False, 'NAME': 'postgres', 'TEST_MIRROR': None,...

有没有办法从原始的测试函数把数据库连接的参数传递给我的线程，并告诉我的线程使用和测试函数一样的数据库名（'test_postgres'）呢？

Answer 1

在你的 function_to_run() 函数声明中，你写了 from django.db import connection。你确定这样会使用正确的测试数据库设置吗？我怀疑你使用的装饰器会把 connection 的导入修改为使用 test_postgres 而不是 postgres，但是因为你是在装饰器的作用域外导入的，所以它没有使用正确的连接。如果你把它放在被装饰的函数内部，会发生什么呢...

@pytest.mark.django_db
def test_example():

    def function_to_run():
        from django.db import connection
        cur_ = connection.cursor
        logger.error(cur_.db.settings_dict)

    p = multiprocessing.Process(target=function_to_run)
    p.start()

编辑：

我对 pytest_django 不是很熟悉，所以现在有点摸不着头脑。我想这个标记函数也允许你装饰一个类，那么你有没有尝试把所有想要使用这个共享函数和数据库的测试放在一个 TestCase 类里呢？像这样：

from django.test import TestCase

@pytest.mark.django_db
class ThreadDBTests(TestCase):

    # The function we want to share among tests
    def function_to_run():
        from django.db import connection
        cur_ = connection.cursor
        logger.error(cur_.db.settings_dict)

    # One of our tests taht needs the db
    def test_example1():
        p = multiprocessing.Process(target=function_to_run)
        p.start()

    # Another test that needs the DB
    def test_example2():
        p = multiprocessing.Process(target=function_to_run)
        p.start()

Answer 2

我找到了解决我问题的方法。

首先，你需要准备一个单独的Django设置文件用于测试（settings_pytest.py），里面有以下的DATABASES设置：

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql_psycopg2',
        'NAME': 'test_database',
        'TEST_NAME': 'test_database',
        'USER': 'something',
        'PASSWORD': 'password',

    },
}

注意，我们定义了TEST_NAME，并且它和NAME是一样的，这样无论是通过测试运行器还是其他方式，我们都能访问同一个数据库。

接下来，你需要先创建这个数据库，并运行'syncdb'和'migrate'命令：

sql> CREATE DATABASE test_database;

manage.py syncdb --settings=settings_pytest

manage.py migrate --settings=settings_pytest

最后，你可以用以下命令来运行你的测试：

py.test --reuse-db

你需要指定--reuse-db，因为重新创建数据库是行不通的，因为默认数据库和测试数据库是一样的。如果你的数据库有变动，你需要手动用上面的命令重新创建数据库。

对于测试本身，如果你要向数据库添加记录，并且这些记录需要被子进程访问，记得在pytest装饰器中加上transaction=True。

def function_to_run():

    Model.objects.count() == 1

@pytest.mark.django_db(transaction=True)
def test_example():
    obj_ = Model()
    obj_.save()
    p = multiprocessing.Process(target=function_to_run)
    p.start()