1
tests/
2
├── conftest.py          # 放通用 fixture
3
├── test_user.py
4
├── test_order.py
5
└── api/
6
    ├── conftest.py      # API 专用 fixture，覆盖上层
7
    └── test_endpoints.py

1
def get_api_key():
2
    return os.environ.get('API_KEY')
3

4

5
def test_get_api_key(monkeypatch):
6
    # 替换 os.environ 中的值
7
    monkeypatch.setenv('API_KEY', 'test-key-123')
8
    assert get_api_key() == 'test-key-123'
9

10
    # monkeypatch 自动在测试结束后恢复原值

1
# ❌ unittest.mock.patch — 参数顺序反直觉
2
@patch('myapp.requests.get')
3
def test_something(mock_get):
4
    pass
5

6
# ✅ pytest-mock — fixture 注入，顺序自然
7
def test_something(mocker):
8
    mocker.patch('myapp.requests.get')

1
def test_fetch_timeout(mocker):
2
    mock_get = mocker.patch('myapp.requests.get')
3
    mock_get.side_effect = requests.exceptions.Timeout('Connection timed out')
4

5
    with pytest.raises(requests.exceptions.Timeout):
6
        fetch_user_data(user_id=1)

1
# 模拟重试场景：第一次失败，第二次成功
2
def test_retry_logic(mocker):
3
    mock_get = mocker.patch('myapp.requests.get')
4
    mock_get.side_effect = [
5
        requests.exceptions.ConnectionError(),
6
        mocker.Mock(status_code=200, json=lambda: {'id': 1})
7
    ]
8

9
    result = fetch_with_retry(user_id=1)
10
    assert result['id'] == 1
11
    assert mock_get.call_count == 2  # 验证重试了一次

1
# ❌ 错误：mock 了 requests 的内部实现
2
mocker.patch('requests.adapters.HTTPAdapter.send')
3

4
# ✅ 正确：mock 了 requests.get 这个公共 API
5
mocker.patch('myapp.requests.get')
6

7
# ✅ 更好：通过接口抽象，mock 你自己的抽象层
8
mocker.patch('myapp.http_client.get')

1
@pytest.mark.parametrize('user_role', ['admin', 'editor', 'viewer'])
2
@pytest.mark.parametrize('endpoint', ['/users', '/posts', '/settings'])
3
def test_access_control(user_role, endpoint):
4
    """自动生成 3 × 3 = 9 个测试用例"""
5
    client = create_client(role=user_role)
6
    response = client.get(endpoint)
7
    # 根据 role 和 endpoint 检查权限

1
@pytest.mark.parametrize('url, expected_status', [
2
    ('https://httpbin.org/get', 200),
3
    pytest.param('https://slow-server.com', 200, marks=pytest.mark.slow),
4
    pytest.param('https://invalid-url', 404, marks=pytest.mark.xfail(reason='DNS 解析失败')),
5
])
6
def test_http_request(url, expected_status):
7
    response = requests.get(url, timeout=5)
8
    assert response.status_code == expected_status

1
import pytest
2
import httpx
3

4
@pytest.mark.asyncio
5
async def test_fetch_async():
6
    async with httpx.AsyncClient() as client:
7
        response = await client.get('https://httpbin.org/get')
8
        assert response.status_code == 200
9
        data = response.json()
10
        assert 'url' in data

1
@pytest.fixture
2
async def async_db():
3
    """异步数据库连接"""
4
    conn = await asyncpg.connect('postgresql://localhost/test')
5
    yield conn
6
    await conn.close()
7

8

9
@pytest.mark.asyncio
10
async def test_async_query(async_db):
11
    rows = await async_db.fetch('SELECT 1 AS value')
12
    assert rows[0]['value'] == 1

1
@pytest.mark.asyncio
2
async def test_async_fetch_with_mock(mocker):
3
    # mock 异步函数
4
    mock_get = mocker.AsyncMock()
5
    mock_get.return_value.json.return_value = {'status': 'ok'}
6
    mocker.patch('httpx.AsyncClient.get', mock_get)
7

8
    async with httpx.AsyncClient() as client:
9
        result = await client.get('https://api.example.com/health')
10
        data = result.json()
11
        assert data['status'] == 'ok'

1
[pytest]
2
testpaths = tests
3
python_files = test_*.py
4
python_classes = Test*
5
python_functions = test_*
6
addopts = -v --tb=short --strict-markers
7
markers =
8
    slow: marks tests as slow (deselect with '-m "not slow"')
9
    integration: marks tests as integration tests
10
    e2e: marks tests as end-to-end tests

1
# ❌ 危险：列表在测试间共享
2
@pytest.fixture(scope='module')
3
def shared_list():
4
    return []
5

6
def test_a(shared_list):
7
    shared_list.append(1)
8
    assert len(shared_list) == 1
9

10
def test_b(shared_list):
11
    # 如果 test_a 先跑，这里会失败
12
    assert len(shared_list) == 0  # 实际是 1！

1
# ❌ mock 了内置函数
2
mocker.patch('builtins.open', ...)  # 会导致所有文件操作失效
3

4
# ✅ 只 mock 特定路径
5
mocker.patch('myapp.config.open', ...)

1
# ❌ 失败时只知道 "AssertionError"
2
assert len(users) == expected_count
3

4
# ✅ 失败时显示实际值和期望值
5
assert len(users) == expected_count, f"Expected {expected_count} users, got {len(users)}"
6

7
# ✅✅ 用 pytest 的 assert 就够了——它自动解析表达式
8
assert len(users) == expected_count  # pytest 失败时显示: assert 5 == 3

1
def test_complex_assertion():
2
    result = calculate_metrics(data)
3
    # 用 pytest.approx 处理浮点数比较
4
    assert result['accuracy'] == pytest.approx(0.95, rel=1e-2)
5
    assert result['loss'] < pytest.approx(0.1, abs=1e-3)

1
# 用 pytest_generate_tests 动态生成参数
2
def pytest_generate_tests(metafunc):
3
    if 'csv_data' in metafunc.fixturenames:
4
        # 从 CSV 文件读取测试数据
5
        import csv
6
        with open('test_data.csv') as f:
7
            reader = csv.DictReader(f)
8
            metafunc.parametrize('csv_data', list(reader))