Django: Verwenden derselben Testdatenbank in einem separaten Thread

Question

Ich führe Pytests mit einer Testdatenbank mit den folgenden DB-Einstellungen.

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

    }, 
} 

Mit dem @ pytest.mark.django_db, greifen meine Testfunktionen eine Datenbank 'test_postgres' erstellt für die Tests genannt.

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

Ausgänge:

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

aber wenn ich ein Gewinde im Inneren test_example auszuführen:

I, die in diesem Thread sehen kann der Cursor unter Verwendung der Datenbank 'postgres' genannt wird, das ist die nicht testende Datenbank. Ausgang:

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

Gibt es eine Möglichkeit, eine Datenbankverbindung Argument zu meinem Thread von der ursprünglichen Testfunktion zu übergeben und mein Thread Routine sagen denselben Datenbanknamen (‚test_postgres‘) als meine Testfunktion zu benutzen?

Answer 1

Ich fand eine Abhilfe zu meinem Problem.

Zuerst bereiten Sie eine separate Datei Django-Einstellungen für den Test (settings_pytest.py), mit folgenden Datenbanken Einstellung:

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

    }, 
} 

Beachten Sie, dass wir Testnamen definieren, und es ist die gleiche wie NAME, so dass die Ausführung durch Test Runner oder nicht, werden wir auf die gleiche Datenbank zugreifen.

Jetzt müssen Sie diese Datenbank erstellen, und run 'syncdb' und 'Migrate' auf sie zuerst:

sql> CREATE DATABASE test_database; 

manage.py syncdb --settings=settings_pytest 

manage.py migrate --settings=settings_pytest 

Schließlich können Sie Ihre Tests mit ausführen:

py.test --reuse-db 

Sie müssen Geben Sie --reuse-db an, das erneute Erstellen von Datenbanken funktioniert nie, da die Standarddatenbank mit der Testdatenbank identisch ist. Wenn Änderungen an Ihrer Datenbank vorgenommen werden, müssen Sie die Datenbank manuell mit den obigen Befehlen neu erstellen.

Für den Test selbst, wenn Sie Datensätze zur Datenbank hinzufügen, auf die Sie vom generierten Child-Prozess zugreifen müssen, denken Sie daran, transaction = True zum Pytest Decorator hinzuzufügen.

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 Ihrer function_to_run() Deklaration tun Sie from django.db import connection. Sind Sie sicher, dass Sie die richtigen Test-Db-Einstellungen verwenden? Ich vermute, der Dekorateur, den Sie verwenden, ändert den connection Import, um den test_postgres statt postgres zu verwenden, aber da Sie außerhalb des Decorators-Bereichs importieren, verwendet es nicht den richtigen. Was passiert, wenn man es in der Dekorateur-wrapped Funktion setzen wie 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:

Ich bin nicht vertraut mit pytest_django so dass ich in der Dunkelheit an diesem Punkt Schießen, ich Stellen Sie sich vor, dass Sie mit der Markierungsfunktion auch eine Klasse dekorieren können. Haben Sie also versucht, alle Tests, die diese gemeinsame Funktion und die db in einer TestCase-Klasse verwenden möchten, zu setzen? Wie 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()