светильники pytest и синхронизация потоков - PullRequest
Новая
       28

светильники pytest и синхронизация потоков

0 голосов
/ 04 марта 2019

Я пытаюсь использовать pytest-xdist для того, чтобы мои тесты выполнялись параллельно. Проблема в том, что каждый поток обращается к устройству, которое используется для всех тестов, и выполняет его в соответствии с номером потока.

Это вызывает у меня проблему, потому что эта роль фикстора заключается в создании данных для моих тестов, и как только они уже созданы, я получаю и ошибку, так как она уже создана (через REST).

conftest.py: 

lock = threading.Lock()

@pytest.fixture(scope=session)
def init_data(request):

    lock.acquire()
    try:
        data = []
        if checking_data_not_created():
            data.append(some_rest_command_creating_data_1())
            data.append(some_rest_command_creating_data_2())
    finally:
        lock.release()

    yield data

    lock.acquire()
    try:
        remove_all_data()
    finally:
        lock.release()

tests_class.py: 

class classTests():

    def first_test(init_data):
        test body and validation....

     def second_test(init_data):
        test body and validation....

Я использую команду: pytest -v -n2

Предполагается, что 1-й поток должен запустить first_test (), а 2-й поток должен запустить second_test () один из них всегда будет терпеть неудачу, потому что первый уже создал данные в секции фикстур, а другой поток получит исключение, и все тесты, которые он должен выполнить, не пройдут.

Как видите, я пытался использовать блокировку вчтобы синхронизировать потоки, но это также не работает.

есть идеи, как мне решить эту проблему?

Спасибо.

Ответы [ 2 ]

0 голосов
/ 26 июля 2019

Обратите внимание, что pytest-xdist не учитывает 'session' приборы с областью видимости.Приспособление с областью действия 'session' в pytest-xdist на самом деле означает специфическое для процесса приспособление уровня сеанса, поэтому оно создается и уничтожается отдельно для каждого процесса, и состояние этого приспособления не разделяется между процессами.Существуют некоторые давние предложения добавить блокировку и совместное использование истинных фиксированных сессий в pytest-xdist, но все они сталкиваются со многими классическими причинами, чтобы попытаться любой ценой избежать многопоточности или многопроцессорной обработки с блокировками исинхронизацию, и в результате она сильно деприоритизируется разработчиками pytest-xdist (и это понятно).

0 голосов
/ 23 мая 2019

Этот подход не будет работать с pytest-xdist, так как он использует многопроцессорность, а не многопоточность, однако его можно использовать с pytest -rallel с использованием опции --tests-per-worker, он будет запускать тесты с использованием несколькихпотоки.

Данные будут настроены только один раз и очищены один раз при выполнении многопоточного pytest со следующим фиксатором:

conftest.py: 

lock = threading.Lock()
threaded_count = 0

@pytest.fixture(scope='session')
def init_data():
    global lock
    global threaded_count

    lock.acquire()
    threaded_count += 1
    try:
        if threaded_count == 1:
            # Setup Data Once
            data = setup_data()
    finally:
        lock.release()

    yield data

    lock.acquire()
    threaded_count -= 1
    try:
        if threaded_count == 0:
            # Cleanup Data Once
            data = cleaup_data()
    finally:
        lock.release()

Команда:

pytest -v - тесты на одного работника 2

...