Django: sử dụng cùng một cơ sở dữ liệu thử nghiệm trong một chủ đề riêng biệt

Question

Tôi đang chạy các pytests bằng cách sử dụng cơ sở dữ liệu thử nghiệm với các cài đặt DB sau.

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

    }, 
} 

Sử dụng @ pytest.mark.django_db, hàm thử nghiệm của tôi truy cập cơ sở dữ liệu được gọi là 'test_postgres' được tạo cho thử nghiệm.

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

kết quả đầu ra:

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

nhưng nếu tôi chạy một sợi bên test_example:

Tôi có thể thấy rằng trong chủ đề mà con trỏ đang sử dụng cơ sở dữ liệu có tên là 'postgres' là cơ sở dữ liệu không kiểm tra. Output:

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

Có cách nào để vượt qua một cuộc tranh luận kết nối cơ sở dữ liệu quan đến chủ đề của tôi từ chức năng kiểm tra ban đầu và nói với thói quen chủ đề của tôi sử dụng tên cơ sở dữ liệu tương tự ('test_postgres') như chức năng thử nghiệm của tôi?

Answer 1

Tôi đã tìm được giải pháp cho vấn đề của mình.

Trước tiên, bạn chuẩn bị một cài đặt Django tập tin riêng biệt để thử nghiệm (settings_pytest.py), với cơ sở dữ liệu thiết lập sau:

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

    }, 
} 

Chú ý rằng chúng ta định nghĩa TEST_NAME, và nó giống như TÊN, do đó chạy qua thử nghiệm á hậu hay không, chúng tôi sẽ được truy cập cùng một cơ sở dữ liệu.

Bây giờ bạn cần để tạo ra cơ sở dữ liệu này, và chạy 'syncdb' và 'di cư' vào nó đầu tiên:

sql> CREATE DATABASE test_database; 

manage.py syncdb --settings=settings_pytest 

manage.py migrate --settings=settings_pytest 

Cuối cùng bạn có thể chạy thử nghiệm của bạn với:

py.test --reuse-db 

Bạn cần phải xác định --reuse-db, tái tạo cơ sở dữ liệu sẽ không bao giờ làm việc vì cơ sở dữ liệu mặc định giống như cơ sở dữ liệu thử nghiệm. Nếu có thay đổi đối với cơ sở dữ liệu của bạn, bạn sẽ cần phải tạo lại cơ sở dữ liệu theo cách thủ công bằng các lệnh ở trên.

Đối với bản thân kiểm tra, nếu bạn đang thêm bản ghi vào cơ sở dữ liệu mà bạn cần truy cập bằng quy trình con sinh ra, hãy nhớ thêm giao dịch = Đúng với trình trang trí trang trí.

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

Trong tờ khai function_to_run() bạn đang thực hiện from django.db import connection. Bạn có chắc chắn sẽ sử dụng các cài đặt db thử nghiệm chính xác không? Tôi nghi ngờ trang trí bạn đang sử dụng sửa đổi nhập khẩu connection để sử dụng test_postgres thay vì postgres nhưng vì bạn đang nhập bên ngoài phạm vi trang trí mà không sử dụng đúng. Chuyện gì xảy ra nếu bạn đặt nó bên trong hàm trang trí bọc như vậy ...

@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:

Tôi không quen thuộc với pytest_django vì vậy tôi chụp trong bóng tối vào thời điểm này, tôi hãy tưởng tượng rằng chức năng đánh dấu cho phép bạn trang trí một lớp học, vì vậy bạn đã thử đặt tất cả các bài kiểm tra muốn sử dụng hàm chia sẻ này và db trong một lớp TestCase? Giống như vậy:

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