AssertionError: assert 0 == 3
 +  where 0 = BatchResult(status=<Status.Failed: 'Failed'>, total_lines=0, completed_lines=0, failed_lines=0, node_status={}, start_time=datetime.datetime(2024, 5, 6, 19, 11, 6, 997510), end_time=datetime.datetime(2024, 5, 6, 19, 11, 22, 108065), metrics={}, system_metrics=SystemMetrics(total_tokens=0, prompt_tokens=0, completion_tokens=0, duration=15.110555), error_summary=ErrorSummary(failed_user_error_lines=0, failed_system_error_lines=0, error_list=[], aggr_error_dict={}, batch_error_dict={'message': 'The batch run failed due to timeout [5s]. Please adjust the timeout to a higher value.', 'messageFormat': 'The batch run failed due to timeout [{batch_timeout_sec}s]. Please adjust the timeout to a higher value.', 'messageParameters': {'batch_timeout_sec': '5'}, 'referenceCode': 'Batch', 'code': 'UserError', 'innerError': {'code': 'BatchRunTimeoutError', 'innerError': None}})).total_lines
self = <executor.e2etests.test_batch_timeout.TestBatchTimeout object at 0x7f5479b2d310>
flow_folder = 'one_line_of_bulktest_timeout', line_timeout_sec = 600
batch_timeout_sec = 5
expected_error_message = 'Line 2 execution timeout for exceeding'
batch_run_status = <Status.Failed: 'Failed'>

    @pytest.mark.parametrize(
        "flow_folder, line_timeout_sec, batch_timeout_sec, expected_error_message, batch_run_status",
        [
            # For the first case, the line timeout will not take effect
            # because the real line timeout is calculated according to batch run timeout.
            # So, both cases will raise LineExecutionTimeoutError
            (ONE_LINE_OF_BULK_TEST_TIMEOUT, 600, 5, "Line 2 execution timeout for exceeding", Status.Failed),
            (
                ONE_LINE_OF_BULK_TEST_TIMEOUT,
                3,
                600,
                "Line 2 execution timeout for exceeding 3 seconds",
                Status.Completed,
            ),
        ],
    )
    def test_batch_timeout(
        self, flow_folder, line_timeout_sec, batch_timeout_sec, expected_error_message, batch_run_status
    ):
        mem_run_storage = MemoryRunStorage()
        batch_engine = BatchEngine(
            get_yaml_file(flow_folder),
            get_flow_folder(flow_folder),
            connections={},
            storage=mem_run_storage,
            line_timeout_sec=line_timeout_sec,
            batch_timeout_sec=batch_timeout_sec,
        )
    
        input_dirs = {"data": get_flow_inputs_file(flow_folder, file_name="samples.json")}
        output_dir = Path(mkdtemp())
        inputs_mapping = {"idx": "${data.idx}"}
        batch_results = batch_engine.run(input_dirs, inputs_mapping, output_dir)
    
        assert isinstance(batch_results, BatchResult)
        # assert the line status in batch result
        assert batch_results.status == batch_run_status
>       assert batch_results.total_lines == 3
E       AssertionError: assert 0 == 3
E        +  where 0 = BatchResult(status=<Status.Failed: 'Failed'>, total_lines=0, completed_lines=0, failed_lines=0, node_status={}, start_time=datetime.datetime(2024, 5, 6, 19, 11, 6, 997510), end_time=datetime.datetime(2024, 5, 6, 19, 11, 22, 108065), metrics={}, system_metrics=SystemMetrics(total_tokens=0, prompt_tokens=0, completion_tokens=0, duration=15.110555), error_summary=ErrorSummary(failed_user_error_lines=0, failed_system_error_lines=0, error_list=[], aggr_error_dict={}, batch_error_dict={'message': 'The batch run failed due to timeout [5s]. Please adjust the timeout to a higher value.', 'messageFormat': 'The batch run failed due to timeout [{batch_timeout_sec}s]. Please adjust the timeout to a higher value.', 'messageParameters': {'batch_timeout_sec': '5'}, 'referenceCode': 'Batch', 'code': 'UserError', 'innerError': {'code': 'BatchRunTimeoutError', 'innerError': None}})).total_lines

/home/runner/work/promptflow/promptflow/src/promptflow/tests/executor/e2etests/test_batch_timeout.py:162: AssertionError