Django: использование одной тестовой базы данных в отдельном потоке

Я запускаю 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') в качестве моей тестовой функции?

Я нашел обходной путь к моей проблеме.

Сначала вы создаете отдельный файл настроек 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, восстановление базы данных никогда не будет работать, поскольку база данных по умолчанию такая же, как и тестовая база данных. Если в вашей базе данных есть изменения, вам необходимо будет обновить базу данных вручную с помощью приведенных выше команд.

Для самого теста, если вы добавляете записи в базу данных, к которой нужно обратиться с помощью порожденного дочернего процесса, не забудьте добавить транзакцию = 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() 

В вашем объявлении function_to_run() вы делаете from django.db import connection . Вы уверены, что будете использовать правильные тестовые настройки db? Я подозреваю, что используемый вами декоратор изменяет импорт 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, поэтому сейчас снимаю в темноте, я полагаю, что функция маркера позволяет вам также украсить класс, поэтому попробуйте поставить все тесты, которые хотят использовать эту общую функцию, и db в одном классе 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()