Django: using same test database in a separate thread

Question

I am running 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 the @pytest.mark.django_db, my test functions access a database called 'test_postgres' created for the 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 a 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 can see that in that thread the cursor is using database named 'postgres' which is the non-testing database. Output:

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

Is there a way to pass a database connection argument to my thread from the original test function and tell my thread routine 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 setting:

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

    },
}

Notice that we define TEST_NAME, and it's the same as NAME, so that running through test runner or not, we will be accessing same database.

Now you need to create this database, and run 'syncdb' and 'migrate' on it 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 need to specify --reuse-db, database re-creation will never work since the default database is the same as the test database. If there are changes to your database you will need to recreate the database manually with the commands above.

For the test itself, if you are adding records to the database that you need to be accessed by the spawned child process, remember 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 your function_to_run() declaration you're doing from django.db import connection. Are you sure that will be using the correct test db settings? I suspect the decorator you're using modifies the connection import to use the test_postgres rather than postgres but because you're importing outside of the decorators scope it's not using the right one. What happens if you put it inside the decorator-wrapped function like 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'm not familiar with pytest_django so I'm shooting in the dark at this point, I imagine that the marker function allows you to decorate a class as well, so have you tried putting all the tests that want to use this shared function and the db in one TestCase class? Like so:

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