如何从Python文件调用Pytest BDD文件并记录执行结果

我需要实现从另一个Python文件调用指定的Pytest BDD测试文件（steps.py），同时记录测试的执行结果。目前我通过以下命令行指令执行测试：

py -m pytest -vvv -q steps.py -o log_cli=true --log-cli-level=DEBUG --capture=tee-sys --feature=Test --alluredir allure-results

对应的steps.py代码如下：

import teradatasql,pytest,logging,re,warnings,sys,allure

from typing import List, Dict
from ApplicationProperties import ApplicationProperties as AP
import StringBuilder,AllureReports

from pytest_bdd import given,when,then,scenarios, parsers
import pandas as pd

warnings.filterwarnings('ignore') 

COL_DELIM = "|"
logging.basicConfig(filename='example.log', encoding='utf-8', level=logging.DEBUG)
LOG = logging.getLogger(__name__)
ROW_MAX_LENGTH = 100

props = AP()
shost = props.getDbHost()
suser = props.getDbUser()

spassword = props.getDbPass()
slogmech = props.getDbLogMech()

environment=props.getEnvironment()

featurefilename = "../features/"

n = len(sys.argv)
for i in range(1, n):
    if "--feature" in sys.argv[i]:
        args = sys.argv[i]
        argRemoveParameter = args.split("=")
        featurefilename = featurefilename +argRemoveParameter[1]

LOG.debug(featurefilename)
scenarios(featurefilename)

actusult = None

@pytest.fixture(autouse=True)
def setUp():
    conn = getConnection()
    yield conn
    #def tearDown():
        #global conn
    #close(conn)
    #request.addfinalizer(tearDown)
    
#problems here not parses examples table
@given(parsers.parse("{parameters}"), target_fixture="parameters",converters={"parameters": str})
def ParseParmeters(parameters):
    return parameters

@when(parsers.parse("I execute the query:\n{queryTemplate}"), target_fixture="actualResult")
@when(parsers.parse("SQL to execute:\n{queryTemplate}"), target_fixture="actualResult")
def whenIExecuteTheQuery(setUp,parameters,queryTemplate):
    global LOG
    LOG.debug("QueryTemplate=" + queryTemplate + "/n Parameters: " + parameters)
    query = populateParameters(parameters, queryTemplate)
    print(query)
    addSqlAsAttachment("query", query)
    LOG.debug("Executing query:" + query)
    try:
        stat = setUp
        #rs = stat.execute(query)
        rs = pd.read_sql(query, stat)
        LOG.debug("Got result: \n" + rs.to_string())
        #actualResult =asMaps(rs)
        #LOG.debug("Got result: \n" + actualResult)
        return rs
    except Exception as error:
        LOG.error("Error executing query: " + query, error)
        raise Exception(error)
    finally:
        close(stat)
        
    

@then(parsers.parse("EXPECTED_RESULT={expectedResult}"))
def actiontaken(actualResult, expectedResult):
    #used to be called then
    global LOG
    LOG.debug("expectedResult=" + expectedResult + "/n text: " + actualResult.to_string())
    if ("NO_ROWS_RETURNED".upper() == expectedResult.upper()):
        assert actualResult is not None, "Result set is not empty: " + actualResult.to_string()
    else:
        AssertionError("Unsupported EXPECTED_RESULT token!")
    

@then("I will get empty result set")
def actiontaken(actualResult):
    #used to be called then
    global LOG
    LOG.debug("Result" + actualResult.to_string())
    assert actualResult is not None, "Result set is not empty: " + actualResult.to_string()

@then(parsers.parse("I will get the following result set: {result}"))
def thenIWillGetTheFollowingResult(actualResult,expectedTable):
    ##expect a parser of table to be here ##
    expectedTableString = createTable(expectedTable.getRows())
    actualTableString = createTable(actualResult)
    expectedResult = [expectedTable.getRows()]
    for expected in expectedResult:
        for actual in actualResult:
            if expected == actual:
                actualResult.remove(actual)
                expectedResult.remove(expected)    
    tablereturn = StringBuilder()
    tablereturn.append("Expected:\n" + expectedTableString);
    tablereturn.append("\nActual:\n" + actualTableString);
    if actualResult.size() > 0:
        unexpectedTableString = createTable(actualResult);
        tablereturn.append("\nUnexpected:\n" + unexpectedTableString);
        
    if expectedResult.size() > 0:
        notFound = createTable(expectedResult);
        tablereturn.append("\nNot Found:\n" + notFound);
    #send to allure reports
    allure.attach( str(tablereturn),"ResultSets",attachment_type=allure.attachment_type.TXT);
    #ExamplesAwareReporter.addTextAttachment("ResultSets.txt", str(tablereturn));
    assert actualResult is None and expectedResult is None ,"Unexpected rows count:" + actualResult.size() + "; Not found rows count: " + expectedResult.size()

实现方案 #

方法一：用Pytest Python API直接调用 #

通过Pytest的内置API执行测试，可直接获取测试结果，无需依赖命令行。新建run_tests.py文件：

import pytest

def run_pytest_bdd_tests():
    # 映射原命令行参数
    pytest_args = [
        "-vvv",
        "-q",
        "steps.py",
        "-o", "log_cli=true",
        "--log-cli-level=DEBUG",
        "--capture=tee-sys",
        "--feature=Test",
        "--alluredir", "allure-results"
    ]
    
    # 执行测试并获取返回码
    result_code = pytest.main(pytest_args)
    
    # 输出结果
    if result_code == 0:
        print("所有测试通过")
    else:
        print(f"测试失败，返回码：{result_code}")
    
    # 获取详细测试统计
    reporter = pytest.config.pluginmanager.getplugin('_terminalreporter')
    if reporter:
        print("\n测试统计：")
        print(f"通过用例数：{len(reporter.stats.get('passed', []))}")
        print(f"失败用例数：{len(reporter.stats.get('failed', []))}")
        print(f"跳过用例数：{len(reporter.stats.get('skipped', []))}")

if __name__ == "__main__":
    run_pytest_bdd_tests()

方法二：用subprocess执行命令行并捕获输出 #

保留原命令行执行逻辑，通过subprocess调用并捕获输出。新建run_tests.py文件：

import subprocess
import sys

def run_via_subprocess():
    cmd = [
        sys.executable, "-m", "pytest",
        "-vvv", "-q", "steps.py",
        "-o", "log_cli=true",
        "--log-cli-level=DEBUG",
        "--capture=tee-sys",
        "--feature=Test",
        "--alluredir", "allure-results"
    ]
    
    # 执行命令并捕获所有输出
    exec_result = subprocess.run(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        encoding="utf-8"
    )
    
    # 打印输出并写入日志文件
    print("测试执行输出：")
    print(exec_result.stdout)
    with open("test_execution.log", "w", encoding="utf-8") as f:
        f.write(exec_result.stdout)
    
    # 判断执行结果
    if exec_result.returncode == 0:
        print("\n测试执行成功")
    else:
        print(f"\n测试执行失败，返回码：{exec_result.returncode}")

if __name__ == "__main__":
    run_via_subprocess()

对现有steps.py的优化建议 #

1. 规范参数解析：替换手动遍历sys.argv的方式，改用argparse模块：

import argparse

parser = argparse.ArgumentParser()
parser.add_argument("--feature", default="Test", help="指定feature文件名")
args = parser.parse_args()
featurefilename = f"../features/{args.feature}"

2. 完善数据库连接管理：在setUp fixture中添加连接关闭逻辑：

@pytest.fixture(autouse=True)
def setUp():
    conn = getConnection()
    try:
        yield conn
    finally:
        close(conn)

3. 修复断言错误抛出：在EXPECTED_RESULT分支中补充raise关键字：

else:
    raise AssertionError("不支持的EXPECTED_RESULT标识！")

内容的提问来源于stack exchange，提问作者h ca