pytest 9 实战（fixtures 与参数化与插件）

# 创建并激活虚拟环境
python3.12 -m venv .venv
source .venv/bin/activate

# 安装最新 pytest 9
pip install "pytest>=9.1.1"

# 校验版本
pytest --version
# pytest 9.1.1

pip install pytest-asyncio pytest-xdist pytest-html pytest-cov pytest-mock pytest-rerunfailures

project/
├── pyproject.toml          # 项目配置（pytest 9 推荐配置位置）
├── conftest.py             # 顶层 fixture 与 hook
├── src/
│   └── mypkg/
│       └── calc.py
└── tests/
    ├── conftest.py         # 测试级 fixture
    ├── unit/
    │   ├── conftest.py
    │   └── test_calc.py
    └── integration/
        └── test_api.py

[tool.pytest.ini_options]
minversion = "9.1"
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = "-ra --strict-markers --strict-config"
markers = [
    "slow: 标记慢用例，可用 -m 'not slow' 跳过",
    "smoke: 冒烟用例",
    "integration: 集成测试用例",
]

import pytest

@pytest.fixture
def sample_data():
    """提供一个简单的列表数据，函数返回值即注入值"""
    return [1, 2, 3]

def test_sum(sample_data):           # 参数名与 fixture 名一致即自动注入
    assert sum(sample_data) == 6

@pytest.fixture
def db_connection():
    # setup：建立连接
    conn = connect("sqlite:///:memory:")
    yield conn                       # yield 之前的代码为 setup
    # teardown：yield 之后为清理逻辑，无论测试是否失败都会执行
    conn.close()

@pytest.fixture(scope="session")
def app_container():
    """会话级 fixture：仅启动一次容器"""
    container = start_docker("myapp:latest")
    yield container
    container.stop()

@pytest.fixture(params=[1, 2, 3], ids=["一", "二", "三"])
def number(request):
    """request.param 即当前参数值，ids 控制测试 ID 显示"""
    return request.param

def test_positive(number):
    assert number > 0
# 等价生成 3 个用例：test_positive[一] / [二] / [三]

@pytest.fixture(autouse=True)
def reset_env(monkeypatch):
    """每个测试函数运行前自动清空环境变量"""
    for key in list(os.environ):
        monkeypatch.delenv(key, raising=False)

@pytest.mark.parametrize("input,expected", [
    (1, 2),
    (2, 4),
    (3, 6),
    (10, 20),
])
def test_double(input, expected):
    assert input * 2 == expected

@pytest.mark.parametrize(
    "user,role",
    [("alice", "admin"), ("bob", "guest")],
    ids=["管理员alice", "访客bob"],
)
def test_permission(user, role):
    assert has_permission(user, role) is True

def idfn(val):
    if isinstance(val, (list, tuple)):
        return "-".join(map(str, val))
    return repr(val)

@pytest.mark.parametrize("data", [(1, 2), (3, 4)], ids=idfn)
def test_data(data):
    assert len(data) == 2

@pytest.fixture
def db(request):
    """根据 param 中的 dsn 创建不同后端的连接"""
    return connect(request.param)

@pytest.mark.parametrize("db", ["sqlite", "postgres", "mysql"], indirect=True)
def test_query(db):
    assert db.execute("SELECT 1") == 1

# 笛卡尔积：3 browser × 2 env = 6 个用例
@pytest.mark.parametrize("browser", ["chromium", "firefox", "webkit"])
@pytest.mark.parametrize("env", ["dev", "prod"])
def test_runner(browser, env):
    ...

import pytest

@pytest.mark.smoke
def test_homepage():
    assert True

@pytest.mark.slow
def test_full_e2e():
    ...

[tool.pytest.ini_options]
addopts = "--strict-markers"
markers = [
    "smoke: 冒烟测试",
    "slow: 慢用例",
]

pytest -m "smoke and not slow"
pytest -m "integration"

@pytest.mark.xfail(reason="bug #1234 待修复", strict=True)
def test_known_bug():
    assert buggy_func() == 42

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"

# 测试代码
async def fetch_data():
    await asyncio.sleep(0.01)
    return {"status": "ok"}

async def test_fetch():          # auto 模式下无需 @pytest.mark.asyncio
    data = await fetch_data()
    assert data["status"] == "ok"

# 自动按 CPU 核数并行
pytest -n auto

# 指定 8 个 worker，按测试文件分发
pytest -n 8 --dist loadfile

def test_http_call(mocker):
    # 用 mocker.patch 替换 requests.get，避免真实网络调用
    mock_get = mocker.patch("requests.get")
    mock_get.return_value.json.return_value = {"code": 0}

    result = call_api()
    assert result["code"] == 0
    mock_get.assert_called_once_with("https://api.example.com")

# conftest.py
import time
import pytest

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item):
    start = time.perf_counter()
    outcome = yield                  # 让测试函数执行
    elapsed = time.perf_counter() - start
    if elapsed > 1.0:
        print(f"\n[慢用例] {item.nodeid}: {elapsed:.2f}s")

def pytest_addoption(parser):
    parser.addoption(
        "--env", default="dev", choices=["dev", "staging", "prod"],
        help="指定被测环境",
    )

@pytest.fixture
def env(request):
    return request.config.getoption("--env")

# pytest 8：静默跳过 await，结果不可预期
# pytest 9：直接报错
@pytest_asyncio.fixture
async def async_db():
    return await create_async_db()

def test_query(async_db):      # ❌ pytest 9 抛错
    assert async_db.query("SELECT 1") == 1

# 正确做法：测试也声明为 async
async def test_query(async_db):
    assert await async_db.query("SELECT 1") == 1

# 旧写法（pytest 8 及之前）
def pytest_ignore_collect(path, config):
    if path.basename.startswith("_"):
        return True

# pytest 9 新写法
def pytest_ignore_collect(collection_path, config):
    if collection_path.name.startswith("_"):
        return True

# tests/conftest.py
import pytest
from mypkg.db import Database
from mypkg.client import ApiClient

def pytest_addoption(parser):
    parser.addoption("--env", default="dev", choices=["dev", "staging", "prod"])

@pytest.fixture(scope="session")
def env(request):
    return request.config.getoption("--env")

@pytest.fixture(scope="session")
def db(env):
    """会话级数据库连接，整个测试会话共用一个连接"""
    conn = Database(env)
    yield conn
    conn.close()

@pytest.fixture(scope="function")
def api_client(db, mocker):
    """每个测试函数独立的 API 客户端，自动 mock 外部 HTTP"""
    client = ApiClient(db)
    mocker.patch.object(client, "_http_post")    # 隔离外部副作用
    return client

@pytest.fixture(autouse=True)
def _reset_db(db):
    """每个用例前自动清空数据表，确保用例隔离"""
    db.truncate_all()
    yield
    db.truncate_all()