在本篇文章中，我们将介绍如何通过Python实现一个增强版的HttpRunner框架。以下是主要的功能改进点：

• 增加了基于CSV文件的数据驱动支持。
• 在测试用例中引入parameters参数时，使用ddt库生成多条用例。
• 实现了setup_hooks和teardown_hooks钩子功能。
• 将打印输出替换为日志记录。

配置文件

data.yaml

- config:
    name: 示例测试用例
    request:
      base_url: https://httpbin.org
      headers:
        token: abc123
    variables:
      x: 50
      y: 100
      z: ${sum($x, $y)}

- test:
    name: 步骤1 - 发送GET请求
    setup_hooks:
      - ${before('start')}
    teardown_hooks:
      - ${after('end')}
    request:
      url: /get
      method: GET
    extract:
      - u: content.url
    validate:
      - eq: [status_code, 200]
      - comparator: eq
        check: content.url
        expect: https://httpbin.org/get

- test:
    name: 步骤2 - 发送POST请求
    request:
      url: /post
      method: POST
      data:
        x: $x
        y: 2
        z: $z

- test:
    name: 步骤3 - 数据驱动测试
    parameters:
      - m-n: ${P(data.csv)}
    request:
      url: /get
      method: GET
      params:
        m: $m
        n: $n

data.csv

m,n
7,8
9,10
11,12

辅助函数模块

debugtalk.py

def sum(a, b):
    return a + b

def before(message):
    print(f"Before hook: {message}")

def after(message):
    print(f"After hook: {message}")

核心逻辑实现

runner.py

import re
import json
import requests
import logging
from string import Template
from functools import reduce
from operator import eq, gt, lt, ge, le
import unittest
import ddt
import importlib
from logz import log  # pip install logz
from filez import file  # pip install filez

# 定义比较函数集合
COMPARE_FUNCS = {
    "eq": eq,
    "gt": gt,
    "lt": lt,
    "ge": ge,
    "le": le,
    "len_eq": lambda x, y: len(x) == len(y),
    "str_eq": lambda x, y: str(x) == str(y),
    "type_match": lambda x, y: isinstance(x, y),
    "regex_match": lambda x, y: re.match(y, x)
}

FUNCTION_REGEX = re.compile(r'\${(?P<func>.*?)}')
CSV_REGEX = re.compile(r'\${P\((?P<csv>.*?)\)}')

def fetch_value(item, key):
    """从嵌套结构中获取值"""
    if hasattr(item, key):
        return getattr(item, key)
    if key.isdigit():
        key = int(key)
    try:
        return item[key]
    except Exception as ex:
        logging.exception(ex)
        return key

def resolve_path(context, expr):
    """解析路径表达式"""
    if '.' in expr:
        parts = expr.split('.')
        value = context.get(parts[0])
        return reduce(fetch_value, parts[1:], value)
    return context.get(expr)

def complete_request(base_url, request):
    """补充完整的请求信息"""
    if base_url and not request['url'].startswith('http'):
        request['url'] = base_url + request['url']
    return request

def execute_request(context, session, request):
    """发送HTTP请求"""
    response = session.request(**request)
    log.info("Request Data: %s", request)
    log.info("Response Data: %s", response.text)
    try:
        content = response.json()
    except json.decoder.JSONDecodeError:
        content = {}

    context.update(
        response=response,
        request=response.request,
        content=content,
        status_code=response.status_code,
        headers=response.headers,
        ok=response.ok,
        reason=response.reason,
        elapsed_seconds=response.elapsed.seconds
    )

def handle_extraction(context, extract_list):
    """处理变量提取"""
    for entry in extract_list:
        key, path = next(iter(entry.items()))
        context[key] = resolve_path(context, path)

def handle_validation(context, validations):
    """执行断言验证"""
    for rule in validations:
        if 'comparator' in rule:
            comparator = rule['comparator']
            check = rule['check']
            expect = rule['expect']
        else:
            comparator, pair = next(iter(rule.items()))
            check, expect = pair
        compare_func = COMPARE_FUNCS.get(comparator)
        field = resolve_path(context, check)
        assert compare_func(field, expect)

def load_functions():
    """加载自定义函数"""
    module = importlib.import_module('debugtalk')
    functions = {
        key: value for key, value in module.__dict__.items()
        if not key.startswith('__') and callable(value)
    }
    return functions

def substitute_variables(context, data):
    """替换模板中的变量"""
    data_str = json.dumps(data)
    if '$' in data_str:
        data_str = Template(data_str).safe_substitute(context)
        return json.loads(data_str)
    return data

def process_functions(context, functions, data):
    """解析并执行函数调用"""
    data_str = json.dumps(data)
    if '$' in data_str:
        data_str = Template(data_str).safe_substitute(context)

        def replace_function(match):
            if not match:
                return
            return str(eval(match.group(1), {}, functions))

        data_str = re.sub(FUNCTION_REGEX, replace_function, data_str)
    return json.loads(data_str)

def parse_parameters(params):
    """解析参数化数据"""
    key_data_pair = next(iter(params[0].items()))
    keys, source = key_data_pair
    if isinstance(source, str):
        csv_match = re.match(CSV_REGEX, source)
        if not csv_match:
            raise ValueError(f"不支持的参数化格式: {source}")
        csv_file = csv_match.group('csv')
        source = file.load(csv_file)
        source = source[1:]  # 忽略标题行
    keys = keys.split('-')
    return keys, source

def create_test_method(index, test_case):
    """动态生成测试方法"""
    params = test_case.get('parameters')
    if params:
        keys, data = parse_parameters(params)

        def test_with_ddt(self, *args):
            param_dict = dict(zip(keys, args))
            self.context.update(param_dict)
            run_test(self, test_case)

        test_func = ddt.data(*data)(test_with_ddt)
    else:
        def test_default(self):
            run_test(self, test_case)

        test_func = test_default

    test_func.__name__ = f'test_api_{index + 1}'
    test_func.__doc__ = test_case.get("name")
    return test_func

def run_test(self, test_case):
    """运行单个测试用例"""
    if test_case.get('skip'):
        raise unittest.SkipTest
    setup_hooks = test_case.get('setup_hooks')
    teardown_hooks = test_case.get('teardown_hooks')
    resolved_test = substitute_variables(self.context, test_case)
    request = complete_request(self.base_url, resolved_test.get('request'))
    self.context['request'] = request

    if setup_hooks:
        process_functions(self.context, self.functions, setup_hooks)
    if teardown_hooks:
        self.addCleanup(process_functions, self.context, self.functions, teardown_hooks)

    execute_request(self.context, self.session, request)
    handle_extraction(self.context, resolved_test.get('extract', []))
    handle_validation(self.context, resolved_test.get('validate', []))

def build_test_suite(test_data):
    """构建测试套件"""
    class TestApi(unittest.TestCase):
        @classmethod
        def setUpClass(cls):
            cls.functions = load_functions()
            cls.session = requests.session()

            config = test_data[0].get('config')
            context = config.get('variables', {})
            processed_config = process_functions(context, cls.functions, config)
            config_request = processed_config.get('request', {})
            cls.base_url = config_request.pop('base_url', None)
            cls.context = processed_config.get('variables', {})
            for key, value in config_request.items():
                setattr(cls.session, key, value)

    for idx, test_case in enumerate(test_data[1:]):
        test_case = test_case.get('test')
        test_method = create_test_method(idx, test_case)
        setattr(TestApi, f'test_api_{idx + 1}', test_method)

    TestApi = ddt.ddt(TestApi)
    TestApi.__doc__ = test_data[0].get('config').get('name')

    suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestApi)
    return suite

def execute_suite(suite):
    """运行测试套件"""
    runner = unittest.TextTestRunner(verbosity=2)
    return runner.run(suite)

if __name__ == '__main__':
    test_data = file.load('/path/to/data.yaml')
    suite = build_test_suite(test_data)
    execute_suite(suite)