当测试还包含多处理时,使用xdist与工作人员例如-n 1一起使用pytest

问题描述

我正在使用

  • 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中使用(表示没有工作人员)-test成功通过了测试(包括屏幕捕获)

如果我使用-n 1(表示1个工作人员)pytest,则出现以下错误

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,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,set_desired,owner_phone,real_unit,free_job_flow,set_crm_unit_status,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)

解决方法

几天前,我遇到了相同的错误,发现该错误位于record_property固定装置中

由于某些原因,record_property固定装置不接受接收某些类型的数据。就我而言,我试图将response从API请求传递到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变量中的数据