Django: using one test database in a separate thread

Question

I run pytests using a test database with the following DB settings.

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

Using @ pytest.mark.django_db, my test functions access a database called "test_postgres" created for tests.

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

outputs:

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

but if I run the thread inside 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() 

I see that in this thread the cursor uses a database called "postgres", which is not a testable database. Output:

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

Is there a way to pass the argument of connecting the database to my stream from the original test function and tell my streaming program to use the same database name ('test_postgres') as my test function?

Answer 1

I found a workaround to my problem.

First, you prepare a separate Django settings file for testing (settings_pytest.py) with the following DATABASES settings:

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

Please note that we define TEST_NAME, and this is the same as NAME, therefore, passing through the test runner or not, we will access the same database.

Now you need to create this database and run "syncdb" and "migrate" first:

 sql> CREATE DATABASE test_database; manage.py syncdb --settings=settings_pytest manage.py migrate --settings=settings_pytest 

Finally, you can run your tests with:

 py.test --reuse-db 

You must specify --reuse-db, database recovery will never work, because the default database is the same as the test database. If there is a change in your database, you will need to manually update the database using the commands above.

For the test itself, if you add records to the database that you need to access using the child process, be sure to add transaction = True to the pytest decorator.

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

Answer 2

In the declaration of function_to_run() you do from django.db import connection . Are you sure you will use the correct db test settings? I suspect that the decorator you are using modifies the connection import to use test_postgres , not postgres , but since you are importing outside the decorators scope, it does not use the correct one. What happens if you put it inside a function wrapped by a decorator so ...

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

Edit:

I am not familiar with pytest_django, so at the moment I am shooting in the dark, I believe that the marker function allows you to also decorate the class, so you tried to put all the tests that want to use this common function and db in one TestCase class? For instance:

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