Python -- 基于 Jinja2 让测试用例“动”起来

简介：

Jinja2 是一个功能强大且灵活的 Python 模板引擎，用于动态生成文本内容，广泛应用于 Web 开发、配置文件生成、邮件模板、自动化报告等场景。

它允许开发者将 代码逻辑 与 内容展示 分离，通过在模板中嵌入变量和控制结构，将数据动态渲染为最终文本。
​ 

目的：

Jinja2（使用 ${} 作为变量分隔符）集成到 pytest 中，实现测试用例中的动态变量渲染。

​ 

准备：

安装依赖库

pip install pytest jinja2 jsonpath-ng

​ 

逻辑实现：

1、创建 Jinja2 自定义环境

config文件中新建 templates.py 文件，修改变量标识符。（用习惯了jmeter的变量符）

from jinja2 import Environment, BaseLoader


class DollarBraceEnvironment(Environment):
    """
    Jinja2 环境，使用 ${var} 而非 {{ var }}
    """
    def __init__(self, **options):
        super().__init__(
            variable_start_string="${",
            variable_end_string="}",
            autoescape=False,
            loader=BaseLoader(),  # 允许直接 from_string
            **options
        )


# 全局实例
jinja_env = DollarBraceEnvironment()

​ 

2、创建渲染器

config文件夹下新建 renderer.py 文件，递归处理yaml中的变量。

from .templates import jinja_env
from typing import Any, Dict, List, Union


def render(value: Any, context: Dict[str, Any]) -> Any:
    """
    递归渲染任意结构中的 ${} 变量
    支持 str, dict, list, nested
    """
    if isinstance(value, str):
        try:
            template = jinja_env.from_string(value)
            return template.render(**context)
        except Exception as e:
            print(f"[Render Error] '{value}' -> {e}")
            return value

    elif isinstance(value, dict):
        return {k: render(v, context) for k, v in value.items()}
    elif isinstance(value, list):
        return [render(item, context) for item in value]
    else:
        return value

​ 

3、在 conftest.py 中管理上下文

根目录下新建 conftest.py 文件，使用 fixture 装饰器来注入。

import pytest

from config.renderer import render


@pytest.fixture(scope="session")
def context():
    return {
        "base_url": "http://127.0.0.1:8080"
    }


@pytest.fixture
def render_context(context):
    """
    返回一个渲染函数，支持传入额外上下文（如 base_url）
    """
    def _render(value, extra_context=None):
        full_context = {**context}
        if extra_context:
            full_context.update(extra_context)
        return render(value, full_context)
    return _render

​ 

4、核心逻辑

从yaml中读取用例；先将data用例参数和request下的字段拿去渲染（如果存在 ${} 声明的变量）；基于httpx发起请求；将用例的response下的字段拿去渲染（如果存在 ${} 声明的变量）；对比实际响应结果，判断用例是否通过；提取用例中的response下的extract，并通过jsonpath_ng提取value，存到local_context这个上下文字典变量中，下个步骤的用例就可以调用它了。

from pathlib import Path

import httpx
import pytest
import yaml
from jsonpath_ng import parse


local_context = {}


def load_test_cases():
    """
    读取 YAML 测试数据
    :return:
    """
    # 获取当前文件所在目录
    current_dir = Path(__file__).parent.parent.parent
    yml_path = current_dir / "data/login.yml"
    with open(yml_path, 'r', encoding='utf-8') as file:
        data = yaml.safe_load(file)
        return data['test_cases']


def extract_by_jsonpath(data: dict, jsonpath: str):
    """
    从 data 这个字典中，根据 jsonpath 表达式提取值
    :param data:
    :param jsonpath:
    :return:
    """
    try:
        expr = parse(jsonpath)
        matches = [match.value for match in expr.find(data)]
        return matches[0] if matches else None
    except:
        return None


def is_equal(a, b):
    """增强比较：处理 str vs int 的数字"""
    if isinstance(a, str) and isinstance(b, int):
        return a.isdigit() and int(a) == b
    if isinstance(a, int) and isinstance(b, str):
        return b.isdigit() and a == int(b)
    return a == b


@pytest.mark.parametrize("test_case", load_test_cases())
def test_login_flow(test_case, render_context, context):
    case_id = test_case["id"]
    description = test_case["description"]
    data_list = test_case.get("data", [{}])
    steps = test_case["steps"]

    # 遍历yaml中的data测试数据组
    for idx, data_item in enumerate(data_list):
        print(f"\n [{case_id}] {description} | 数据 #{idx+1}")

        # 拷贝每组数据作为独立上下文，包含 username, password, expected_code
        global local_context
        local_context.update(data_item)

        for step in steps:
            step_id = step["id"]
            step_name = step["name"]
            req = step["request"]
            expected_resp = step["response"]  # 期望的响应字段

            print(f"  步骤:{step_id}_{step_name}")

            # --- 1. 渲染请求 ---
            rendered_req = render_context(req, extra_context=local_context)

            # --- 2. 发送请求 ---
            url = rendered_req["url"]
            method = rendered_req["method"].upper()
            json_body = rendered_req.get("json")
            headers = rendered_req.get("headers")

            base_url = context["base_url"]  # 或从 context 获取
            full_url = f"{base_url}{url}"

            # 在发送请求前加这几行
            print("发送请求前的上下文:", local_context)
            print("即将使用的 headers:", headers)
            try:
                resp = httpx.request(method, full_url, json=json_body, headers=headers, timeout=10)
                actual_json = resp.json()
            except Exception as e:
                pytest.fail(f"请求失败: {e}")

            # 构建实际响应数据
            actual_response = {
                "status": resp.status_code,
                "code": actual_json.get("code"),
                "msg": actual_json.get("msg"),
                "data": actual_json.get("data")
            }

            # --- 3. 渲染期望值（支持 ${expected_code}）---
            rendered_expected = render_context(expected_resp, extra_context=local_context)

            # 移除 extract 字段，只保留用于对比的字段
            compare_fields = {k: v for k, v in rendered_expected.items() if k != "extract"}

            # --- 4. 自动对比每个字段 ---
            for field, expected_value in compare_fields.items():
                actual_value = actual_response.get(field)
                if not is_equal(actual_value, expected_value):
                    pytest.fail(
                        f"\n"
                        f"字段 '{field}' 不匹配\n"
                        f"期望: {expected_value}\n"
                        f"实际: {actual_value}"
                    )
                print(f"  {field} = {expected_value}")

            # --- 5. 提取变量 ---
            extract_config = rendered_expected.get("extract", {})
            for var_name, jsonpath in extract_config.items():
                value = extract_by_jsonpath(actual_json, jsonpath)
                if value is None:
                    pytest.fail(f"提取失败: {var_name} <- {jsonpath}")
                local_context[var_name] = value  # 写入上下文，后续步骤可用
                print(f"  提取 {var_name} = {value}")

            print(f"  步骤通过: {step_name}")

​ 

​ 

示例：

1、实现“1个用例N组参数”

test_cases:
  - id: "login_smoke"
    description: "账密登录验证"
    data:
      - username: "xiaoming"
        password: "123456"
        expected_code: 200
      - username: "xiaohong"
        password: "123456"
        expected_code: 200

    steps:
      - id: "step1"
        name: "执行登录"
        request:
          url: "/login2"
          method: POST
          json:
            username: "${username}"
            password: "${password}"
        response:
          status: 200
          code: "${expected_code}"
          extract:
            token: $.data.token
            userId: $.data.userId

  - id: "login_password_error"
    description: "异常验证 -- 密码错误登录失败"
    data:
      - username: "xiaoming"
        password: "wrong_password"
        expected_code: 401

    steps:
      - id: "step1"
        name: "执行登录"
        request:
          url: "/login2"
          method: POST
          json:
            username: "${username}"
            password: "${password}"
        response:
          status: 200
          code: "${expected_code}"

正常读取动态参数并请求，显示两个用例均通过测试。

2、实现“登录+获取用户信息”接口联动。

两个用例之间进行token参数关联

test_cases:
  - id: "login_smoke"
    description: "账密登录验证"

    steps:
      - id: "step1"
        name: "执行登录"
        request:
          url: "/login2"
          method: POST
          json:
            username: "xiaoming"
            password: "123456"
        response:
          status: 200
          code: 200
          extract:
            token: $.data.token
            userId: $.data.userId

  - id: "getUserInfo_smoke"
    description: "获取刚才登录的用户信息"

    steps:
      - id: "step1"
        name: "获取用户登录信息"
        request:
          url: "/getUserInfo"
          method: GET
          headers:
            Authorization: "Bearer ${token}"
        response:
          status: 200
          code: 200

或者，一个用例的不同步骤中token参数关联

test_cases:
  - id: "login_smoke"
    description: "账密登录验证"

    steps:
      - id: "step1"
        name: "执行登录"
        request:
          url: "/login2"
          method: POST
          json:
            username: "xiaoming"
            password: "123456"
        response:
          status: 200
          code: 200
          extract:
            token: $.data.token
            userId: $.data.userId

      - id: "step2"
        name: "获取用户登录信息"
        request:
          url: "/getUserInfo"
          method: GET
          headers:
            Authorization: "Bearer ${token}"
        response:
          status: 200
          code: 200