当测试还包含多处理时，使用xdist对worker进行pytest（例如n1）

Question

我正在使用

• 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

为了在selenium期间捕获屏幕，我使用multprocessing， 开始捕获，然后继续测试

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

如果我正在使用-n1（意味着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]'

谷歌搜索了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)

Answer 1

几天前我遇到了同样的错误，我发现这个bug在record_property装置中

出于某种原因，record_property装置不接受接收某些类型的数据。在我的例子中，我试图将response从API请求传递到record_propertyfixture，类似于：

record_property("api_res", res)

经过几次尝试，我创建了一个变通方法，通过record_propertyfixture从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变量中的数据