当测试还包含多处理时，使用 xdist 与工人（例如 -n 1）进行 pytest答案

【问题标题】：pytest with workers ( -n 1 for example ) using xdist when test contains also multiprocessing当测试还包含多处理时，使用 xdist 与工人（例如 -n 1）进行 pytest
【发布时间】：2020-10-27 18:19:37
【问题描述】：

我正在使用

Python 3.6
Pytest 6.1.1
Pytest xdist 2.1.0
Pycharm 2018.3.7

使用 HTML 插件进行报告以及自定义报告

我正在通过 pycharm >pytest 运行自动化测试，参数如下： --html=../../reports/report.html --self-contained-html -n 1 -k test_instalment_mtn_ng_unified_eco

为了在硒期间捕获屏幕，我正在使用多重处理，开始捕获，并继续测试

如果我没有在参数中使用 -n 1（表示没有工人）-测试成功通过（包括屏幕截图）

如果我使用 -n 1（表示 1 个工人）pytest crush 并出现以下错误：

mtn_ng_tests/mtn_ng_instalment_tests/test_instalment_mtn_ng_unified_eco.py::test_instalment_mtn_ng_unified_eco[customer buy 3 months happy flow while in light-instalment_mtn_ng_unified_eco_1-3-eco-48-40-0-10-ACTIVATED-ACTIVATED-owner_phone_unified_happy_flow_mtn_ng_eco-unified_mtn_instalment_subscribe_charge_event-activation_flow_mtn_unified_instalmant_and_unsub-no-unified_mtn_instalment_unsubscribe_charge_event-sms0-false-None-false-mtn_l20_unified--free_sms0-ACTIVATION-true-Assigned to customer-Ready For Install-false--components_end_time_delta0] 
INTERNALERROR> Traceback (most recent call last):
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/_pytest/main.py", line 257, in wrap_session
INTERNALERROR>     session.exitstatus = doit(config, session) or 0
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/_pytest/main.py", line 313, in _main
INTERNALERROR>     config.hook.pytest_runtestloop(session=session)
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/pluggy/hooks.py", line 286, in __call__
INTERNALERROR>     return self._hookexec(self, self.get_hookimpls(), kwargs)
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/pluggy/manager.py", line 93, in _hookexec
INTERNALERROR>     return self._inner_hookexec(hook, methods, kwargs)
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/pluggy/manager.py", line 87, in <lambda>
INTERNALERROR>     firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/pluggy/callers.py", line 208, in _multicall
INTERNALERROR>     return outcome.get_result()
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/pluggy/callers.py", line 80, in get_result
INTERNALERROR>     raise ex[1].with_traceback(ex[2])
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/pluggy/callers.py", line 187, in _multicall
INTERNALERROR>     res = hook_impl.function(*args)
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/xdist/dsession.py", line 112, in pytest_runtestloop
INTERNALERROR>     self.loop_once()
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/xdist/dsession.py", line 135, in loop_once
INTERNALERROR>     call(**kwargs)
INTERNALERROR>   File "/home/ohad/.local/lib/python2.7/e2e/lib/python3.6/site-packages/xdist/dsession.py", line 174, in worker_workerfinished
INTERNALERROR>     **assert not crashitem, (crashitem, node)**
INTERNALERROR> AssertionError: ('E2eAutomation/tests/mtn_ng_tests/mtn_ng_instalment_tests/test_instalment_mtn_ng_unified_eco.py::test_instalment_mtn_...ms0-ACTIVATION-true-Assigned to customer-Ready For Install-false--components_end_time_delta0]', <WorkerController gw0>)
INTERNALERROR> assert not 'E2eAutomation/tests/mtn_ng_tests/mtn_ng_instalment_tests/test_instalment_mtn_ng_unified_eco.py::test_instalment_mtn_n...e-mtn_l20_unified--free_sms0-ACTIVATION-true-Assigned to customer-Ready For Install-false--components_end_time_delta0]'

google了2天，没找到原因附上部分测试代码提前致谢

# @pytest.mark.skip(reason='relax for now :)')
@pytest.mark.sanity
@pytest.mark.regression
@pytest.mark.instalment_mtn_ng
@pytest.mark.parametrize("scenario_description,unit,month,unit_type,customer_deal,time_unit_purchased,"
                         "set_desired,light_dark,unit_set_state,unit_expected_state,"
                         "owner_phone,charge_event_flow,activation_flow_job_scenario,"
                         "real_unit,charge_event_flow_unsub,sms,ipp_flow,set_product_id,free_triggered,instalment_type,"
                         "free_job_flow,free_sms,unit_status_during_process,valid_product,"
                         "set_crm_unit_status,get_crm_unit_status,set_last_day_of_month,notes,components_end_time_delta",
                         load_data_driven_for_test(profile='instalment_mtn_ng_eco'))
@pytest.mark.skipif(os.environ.get('DRY_RUN') == "True", reason="Jenkins Dry Run")
def test_instalment_mtn_ng_unified_eco(scenario_description, unit, month, unit_type, customer_deal, time_unit_purchased,
                                       set_desired, light_dark,
                                       unit_set_state, unit_expected_state, owner_phone, charge_event_flow,
                                       activation_flow_job_scenario, real_unit, charge_event_flow_unsub, sms, ipp_flow,
                                       set_product_id, free_triggered, instalment_type, free_job_flow, free_sms,
                                       unit_status_during_process, valid_product, set_crm_unit_status,
                                       get_crm_unit_status,
                                       set_last_day_of_month, notes, components_end_time_delta,
                                       setup, record_property):
    # ==================================================================================== Test yield from fixture

    lms_cursor, db_conn, lms_user, locale_sms, lms_db_connection, crm_db_connection, crm_cursor, soap_port = \
        setup[0], setup[1], setup[2], setup[3], setup[4], setup[5], setup[6], setup[7]

    # ====================================== Log creation ================================================
    os.environ['TEST_NUMBER'] = str(int(os.environ.get('TEST_NUMBER')) + 1)
    now = datetime.datetime.now()
    time_stamp = f"{now.year}{now.month}{now.day}{now.hour}{now.minute}{now.second}"
    os.environ['TEST_NAME'] = f"{os.path.basename(__file__)[:-3]}_#_{os.environ.get('TEST_NUMBER')}_" \
                                  f"{scenario_description.replace(' ', '_')}_" + time_stamp
    logging = setting_log.log_config.setup_logger()

    # ==================================================================================== Video recording
    if   os.environ.get('PYTEST_XDIST_WORKER_COUNT'):  # no workers so run with video capture
        video_record = multiprocessing.Process(target=ScreenVideo().execute_video_capturing)
        video_record.start()

        # ==================================================================================== Test Tear Down
        logging.info("test tear down")
        record_property("test_returns", {"screen_recorder": video_record})
        # record_property("test_returns", {"screen_recorder": video_record, "selenium_driver": automation_utils})
    else:
        logging.info("test tear down")
        record_property("test_returns", {})

    # ========================================================Initiate logs that not created in main fixture
    assert 1==2 (just to falied the test quicker)

【问题讨论】：

-n1 并不意味着“没有工人”。使用-n0 重新运行测试并检查是否引发了断言错误。
我的意思是当我从参数中删除 -n 1 时 >>> 这意味着我没有使用工人，并且一切正常。示例：html=../../reports/report.html --self-contained-html -k test_instalment_mtn_ng_unified_eco。当我使用 -n 1 时出现问题 - 目标当然是使用多个工作人员以便在并行测试中运行
当我没有在参数中使用 -n 3 时，这意味着我没有使用任何工人，所有的测试都会一个接一个地依次运行。到现在为止还挺好。我还设法与工人一起运行（例如 -n 3），并且所有测试都按预期并行运行。我的问题开始了，当我添加到测试主体（例如，def test_1：）多处理（用于在测试运行时在后台运行屏幕帧捕获）现在，如果我不使用任何工作人员但测试使用多处理，则所有操作都按预期运行.当我使用工人时，我得到了这个异常（它只是由于我正在使用的测试中的多处理而发生

标签： python pytest xdist pytest-html

【解决方案1】：

几天前我遇到了同样的错误，我发现错误在record_property 夹具中

由于某种原因，record_property 夹具不接受接收某些类型的数据。就我而言，我试图将 API 请求中的 response 传递给 record_property 固定装置，例如：

record_property("api_res", res)

经过几次尝试，我创建了一个解决方法，通过record_property 夹具发送 API 请求的响应并存储我需要的信息：

@pytest.fixture
def store_api_response(record_property, request):
        record_property(
            'api_res', {
                'request_url': str(response.request.method + env + response.request.path_url),
                'headers': dict(response.request.headers),
                'request_body': response.request.body,
                'response_body': response_body,
                'status_code': str(response.status_code)
            }
        )
    return get_response

在您的情况下，错误恰好发生在这一行：

record_property("test_returns", {"screen_recorder": video_record})

record_property 夹具不接受存储在 video_record 变量中的数据

【讨论】：