数据驱动与参数化框架

# pytest 参数化：登录场景的等价类覆盖
import pytest
 
@pytest.mark.parametrize(
    "username, password, expected_code",
    [
        ("valid_user",  "Valid@123", 200),  # 正常登录
        ("valid_user",  "wrong_pwd", 401),  # 密码错误
        ("nonexistent", "any_pwd",   404),  # 用户不存在
        ("",            "Valid@123", 400),  # 用户名为空
    ],
    ids=["正常登录", "密码错误", "用户不存在", "用户名为空"],
)
def test_login(username, password, expected_code):
    # 同一段测试逻辑，喂入不同数据
    resp = login(username, password)
    assert resp.status_code == expected_code

# 为部分用例附加 mark：仅 flaky 用例重试
@pytest.mark.parametrize(
    "payload, expected",
    [
        pytest.param({"q": "iphone"}, 200, id="正常搜索"),
        pytest.param({"q": ""},       400, id="空关键词"),
        pytest.param({"q": "x" * 999}, 414, id="超长关键词",
                     marks=pytest.mark.flaky(reruns=3)),  # 该用例允许重试 3 次
    ],
)
def test_search(payload, expected):
    assert search(payload).status_code == expected

// JUnit 5 参数化：@ValueSource / @CsvSource / @MethodSource 三种数据源
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.MethodSource;
import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
 
class LoginTest {
 
    // 简单 CSV 内联数据
    @ParameterizedTest(name = "登录用户={0} 应返回 {2}")
    @CsvSource({
        "valid_user, Valid@123, 200",
        "valid_user, wrong_pwd, 401",
        "nonexistent, any_pwd,   404"
    })
    void testLogin(String username, String password, int expectedCode) {
        Response resp = login(username, password);
        assertEquals(expectedCode, resp.getStatusCode());
    }
 
    // 复杂对象通过 MethodSource 提供
    static Stream<Arguments> orderScenarios() {
        return Stream.of(
            Arguments.of(new Order("VIP", 1000), 0.8),   // VIP 八折
            Arguments.of(new Order("NEW", 1000), 0.95),  // 新客九五折
            Arguments.of(new Order("NORMAL", 1000), 1.0)  // 普通用户无折扣
        );
    }
 
    @ParameterizedTest
    @MethodSource("orderScenarios")
    void testDiscount(Order order, double expectedDiscount) {
        assertEquals(expectedDiscount, order.calculateDiscount(), 0.001);
    }
}

// TestNG 数据驱动：@DataProvider 并行执行
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
 
public class CheckoutTest {
 
    // name 标识数据源，parallel = true 开启数据级并行
    @DataProvider(name = "checkoutData", parallel = true)
    public Object[][] provideCheckoutData() {
        return new Object[][] {
            {"VIP",     1000, true,  200},   // VIP 用户满 1000 可用券
            {"NEW",     1000, false, 403},   // 新客不可用券
            {"NORMAL",  500,  true,  400},   // 普通用户不满 1000
        };
    }
 
    @Test(dataProvider = "checkoutData")
    public void testCheckout(String userType, int amount,
                             boolean useCoupon, int expectedCode) {
        Response resp = checkout(userType, amount, useCoupon);
        assertEquals(resp.getStatusCode(), expectedCode);
    }
}

# tests/data/login_scenarios.yaml
- name: 正常登录
  username: valid_user
  password: Valid@123
  expected_code: 200
  tags: [smoke, regression]
 
- name: 密码错误
  username: valid_user
  password: wrong_pwd
  expected_code: 401
  tags: [regression]
 
- name: 用户不存在
  username: nonexistent
  password: any_pwd
  expected_code: 404
  tags: [regression]

# tests/conftest.py
import yaml
import pytest
from pathlib import Path
 
def load_scenarios(yaml_path: str) -> list:
    """加载 YAML 数据文件，返回场景列表"""
    with open(yaml_path, encoding="utf-8") as f:
        return yaml.safe_load(f)
 
# tests/test_login_ddt.py
SCENARIOS = load_scenarios("tests/data/login_scenarios.yaml")
 
@pytest.mark.parametrize(
    "scenario",
    SCENARIOS,
    ids=[s["name"] for s in SCENARIOS],  # 用例名取自 YAML 的 name 字段
)
def test_login(scenario):
    resp = login(scenario["username"], scenario["password"])
    assert resp.status_code == scenario["expected_code"]

import json
import pytest
 
def load_json(path: str):
    with open(path, encoding="utf-8") as f:
        return json.load(f)
 
# JSON 数据嵌套结构天然适配 API 用例
API_CASES = load_json("tests/data/api_cases.json")
 
@pytest.mark.parametrize("case", API_CASES, ids=lambda c: c["case_id"])
def test_api(case):
    resp = requests.request(
        method=case["method"],
        url=BASE_URL + case["path"],
        json=case["payload"],
        headers=case.get("headers", {}),
    )
    for assertion in case["assertions"]:
        assert eval(assertion["expression"], {"resp": resp}) == assertion["expected"]

import pandas as pd
import pytest
 
# 用 pandas 读取 Excel，明确指定 dtype 避免类型漂移
df = pd.read_excel("tests/data/login.xlsx", sheet_name="scenarios", dtype=str)
 
@pytest.mark.parametrize(
    "row",
    df.to_dict("records"),
    ids=df["用例名称"].tolist(),
)
def test_login_from_excel(row):
    resp = login(row["用户名"], row["密码"])
    assert resp.status_code == int(row["期望状态码"])

import pytest
import psycopg2
from psycopg2.extras import RealDictCursor
 
@pytest.fixture(scope="session")
def db_conn():
    """会话级 DB 连接，所有用例复用"""
    conn = psycopg2.connect(
        host="localhost", dbname="test_db",
        user="postgres", password="postgres",
    )
    yield conn
    conn.close()
 
def fetch_active_users(db_conn, limit=100):
    """查询活跃用户作为测试数据源"""
    with db_conn.cursor(cursor_factory=RealDictCursor) as cur:
        cur.execute(
            "SELECT user_id, username, status FROM users "
            "WHERE status = 'active' AND deleted_at IS NULL "
            "ORDER BY user_id LIMIT %s",
            (limit,),
        )
        return cur.fetchall()
 
# 数据库驱动参数化
ACTIVE_USERS = fetch_active_users(db_conn())
 
@pytest.mark.parametrize("user", ACTIVE_USERS, ids=lambda u: u["username"])
def test_user_profile(user, db_conn):
    resp = get_profile(user["user_id"])
    assert resp.status_code == 200
    assert resp.json()["username"] == user["username"]

# pip install testcontainers[postgresql]
import pytest
from testcontainers.postgres import PostgresContainer
import psycopg2
 
@pytest.fixture(scope="session")
def pg_container():
    """会话级启动 PostgreSQL 容器，会话结束自动销毁"""
    with PostgresContainer("postgres:16-alpine") as pg:
        yield pg
 
@pytest.fixture(scope="session")
def db_conn(pg_container):
    conn = psycopg2.connect(
        host=pg_container.get_container_host_ip(),
        port=pg_container.get_exposed_port(5432),
        user=pg_container.username,
        password=pg_container.password,
        dbname=pg_container.dbname,
    )
    # 预埋测试数据
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE users (
                user_id SERIAL PRIMARY KEY,
                username VARCHAR(64) UNIQUE,
                status VARCHAR(16)
            );
        """)
        cur.executemany(
            "INSERT INTO users (username, status) VALUES (%s, %s)",
            [("alice", "active"), ("bob", "inactive"), ("carol", "active")],
        )
    conn.commit()
    yield conn
    conn.close()
 
def test_active_users_only(db_conn):
    with db_conn.cursor() as cur:
        cur.execute("SELECT username FROM users WHERE status = 'active'")
        usernames = [r[0] for r in cur.fetchall()]
    assert set(usernames) == {"alice", "carol"}

// JUnit 5 + Testcontainers
import org.testcontainers.containers.PostgreSQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
 
@Testcontainers
class UserRepositoryIT {
 
    @Container
    static PostgreSQLContainer<?> pg = new PostgreSQLContainer<>("postgres:16-alpine")
        .withDatabaseName("test")
        .withUsername("test")
        .withPassword("test")
        .withInitScript("schema.sql");  // 类路径下初始化脚本
 
    @Test
    void shouldQueryActiveUsers() throws SQLException {
        try (Connection conn = DriverManager.getConnection(pg.getJdbcUrl(),
                pg.getUsername(), pg.getPassword());
             Statement stmt = conn.createStatement()) {
            ResultSet rs = stmt.executeQuery(
                "SELECT username FROM users WHERE status = 'active'");
            List<String> names = new ArrayList<>();
            while (rs.next()) names.add(rs.getString("username"));
            assertTrue(names.contains("alice"));
        }
    }
}

# 订单 Builder：链式组装，默认值兜底
from dataclasses import dataclass, field
from typing import Optional
 
@dataclass
class Order:
    user_id: str
    items: list = field(default_factory=list)
    coupon: Optional[str] = None
    address: Optional[str] = None
    payment: str = "ALIPAY"
 
class OrderBuilder:
    """订单构建器：链式调用，每个 with 方法返回 self"""
    def __init__(self):
        self._order = Order(user_id="default_user")
 
    def with_user(self, user_id: str) -> "OrderBuilder":
        self._order.user_id = user_id
        return self
 
    def with_item(self, sku: str, qty: int = 1) -> "OrderBuilder":
        self._order.items.append({"sku": sku, "qty": qty})
        return self
 
    def with_coupon(self, coupon: str) -> "OrderBuilder":
        self._order.coupon = coupon
        return self
 
    def with_address(self, addr: str) -> "OrderBuilder":
        self._order.address = addr
        return self
 
    def build(self) -> Order:
        return self._order
 
# 用例侧：意图清晰，只关注差异点
def test_vip_order_with_coupon():
    order = (OrderBuilder()
             .with_user("vip_001")
             .with_item("SKU-IPHONE-15", qty=1)
             .with_coupon("VIP_8折券")
             .build())
    assert checkout(order).status_code == 200
 
def test_empty_cart_rejected():
    order = OrderBuilder().with_user("new_user").build()
    assert checkout(order).status_code == 400

class OrderMother:
    """订单对象工厂：命名方法封装典型业务场景"""
    @staticmethod
    def vip_order_with_coupon() -> Order:
        return (OrderBuilder()
                .with_user("vip_001")
                .with_item("SKU-IPHONE-15")
                .with_coupon("VIP_8折券")
                .build())
 
    @staticmethod
    def new_user_first_order() -> Order:
        return (OrderBuilder()
                .with_user("new_user_001")
                .with_item("SKU-CASE")
                .build())
 
    @staticmethod
    def empty_cart_order() -> Order:
        return OrderBuilder().with_user("any_user").build()
 
# 用例侧极简
def test_vip_discount():
    order = OrderMother.vip_order_with_coupon()
    assert checkout(order).json()["discount"] == 0.8

from faker import Faker
 
# 固定 seed 保证用例可复现
Faker.seed(42)
fake = Faker("zh_CN")  # 中文 locale
 
class UserBuilder:
    def __init__(self):
        self._user = {
            "username": fake.user_name(),
            "phone": fake.phone_number(),
            "email": fake.email(),
            "address": fake.address(),
            "id_card": fake.ssn(),  # 中文身份证格式
        }
 
    def with_invalid_phone(self) -> "UserBuilder":
        self._user["phone"] = "00000000000"  # 故意构造非法值
        return self
 
    def build(self) -> dict:
        return self._user
 
# 生成 100 个真实形态的用户做压力回归
@pytest.mark.parametrize("user", [UserBuilder().build() for _ in range(100)])
def test_user_register(user):
    resp = register(user)
    assert resp.status_code in (200, 409)  # 409 表示偶发重复，可接受

# LLM 生成测试数据的工程化封装
from openai import OpenAI
from pydantic import BaseModel, Field, ValidationError
 
class OrderCase(BaseModel):
    """订单用例数据 schema"""
    case_id: str = Field(description="用例唯一标识")
    user_type: str = Field(pattern="^(VIP|NEW|NORMAL)$")
    amount: float = Field(gt=0)
    coupon_count: int = Field(ge=0, le=3)
    expected_code: int
 
client = OpenAI()
 
def generate_cases_via_llm(rule: str, n: int = 20) -> list[OrderCase]:
    """调用 LLM 生成测试用例数据，并做 schema 校验"""
    prompt = f"""
    你是测试数据生成专家。根据以下业务规则生成 {n} 条测试用例数据：
    规则：{rule}
    要求：覆盖等价类、边界值、异常场景。输出 JSON 数组。
    """
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    raw = resp.choices[0].message.content
    import json
    data = json.loads(raw)["cases"]
    # Schema 校验：非法数据直接丢弃，避免污染用例集
    valid = []
    for item in data:
        try:
            valid.append(OrderCase(**item))
        except ValidationError as e:
            print(f"丢弃非法用例 {item.get('case_id')}: {e}")
    return valid
 
cases = generate_cases_via_llm("VIP 用户单笔满 1000 可叠加优惠券，但不超过 3 张")
 
@pytest.mark.parametrize("case", cases, ids=lambda c: c.case_id)
def test_checkout_llm_generated(case):
    resp = checkout(case.user_type, case.amount, case.coupon_count)
    assert resp.status_code == case.expected_code