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