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

Question

我正在使用具有以下数据库设置的测试数据库运行pytests.

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

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

首先，准备一个单独的 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，数据库重新创建将永远不会工作，因为默认数据库与测试数据库相同。如果您的数据库发生更改，您将需要使用上述命令手动重新创建数据库。

对于测试本身，如果您要将记录添加到需要由生成的子进程访问的数据库中，请记住将 transaction=True 添加到 pytest 装饰器中。

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()