Python 项目工程化实战（二）：pytest 测试体系建设从 fixtures 到 CI 门槛 | Python 进阶

1
def test_assertion_intro():
2
    result = {"name": "Alice", "age": 30}
3
    assert result["age"] == 31

1
E       AssertionError: assert 30 == 31
2
E        +  where 30 = {'name': 'Alice', 'age': 30}['age']

1
# 相等 / 布尔
2
assert x == expected
3
assert isinstance(obj, MyClass)
4
assert "key" in my_dict
5
assert result is None
6

7
# 浮点数比较（不要直接用 ==）
8
import pytest
9
assert 0.1 + 0.2 == pytest.approx(0.3)
10
assert value == pytest.approx(3.14159, rel=1e-3)
11

12
# 断言异常
13
with pytest.raises(ValueError):
14
    int("not a number")
15

16
# 断言异常 + 检查消息
17
with pytest.raises(ValueError, match=r"invalid literal"):
18
    int("abc")
19

20
# 捕获异常对象做进一步断言
21
with pytest.raises(ValueError) as exc_info:
22
    raise ValueError("boom")
23
assert "boom" in str(exc_info.value)
24

25
# 断言警告
26
with pytest.warns(DeprecationWarning):
27
    old_function()

1
- 文件：test_*.py 或 *_test.py
2
- 类：  Test 开头（且不能有 __init__ 方法）
3
- 函数/方法：test 开头

1
import pytest
2

3
@pytest.fixture
4
def sample_user():
5
    """提供一个测试用的 User 对象。"""
6
    return User(name="Alice", age=30)
7

8
# 测试函数把 fixture 名字作为参数，pytest 自动注入
9
def test_user_name(sample_user):
10
    assert sample_user.name == "Alice"
11

12
def test_user_age(sample_user):
13
    # 每个测试都会拿到「全新」的 sample_user（function 作用域）
14
    assert sample_user.age == 30

1
@pytest.fixture
2
def db_connection():
3
    conn = create_connection("sqlite:///:memory:")   # setup
4
    conn.execute("CREATE TABLE users (id INT, name TEXT)")
5
    yield conn                                         # 把资源交给测试
6
    conn.close()                                       # teardown：一定会执行
7

8
def test_insert(db_connection):
9
    db_connection.execute("INSERT INTO users VALUES (1, 'bob')")
10
    rows = db_connection.execute("SELECT * FROM users").fetchall()
11
    assert len(rows) == 1

1
@pytest.fixture(scope="session")   # 整个测试会话只创建一次
2
def app_client():
3
    app = create_app(config="testing")
4
    client = app.test_client()
5
    yield client
6
    # 会话结束时清理
7

8
@pytest.fixture(scope="module")    # 每个测试文件创建一次
9
def large_dataset():
10
    return load_csv("big_file.csv")   # 加载大文件，多个测试共享

1
@pytest.fixture
2
def make_user():
3
    created = []
4

5
    def _make(name="test", age=20, role="guest"):
6
        user = User(name=name, age=age, role=role)
7
        created.append(user)
8
        return user
9

10
    yield _make
11
    # 清理所有创建的对象
12
    for user in created:
13
        user.delete()
14

15
def test_admin(make_user):
16
    admin = make_user(name="root", role="admin")   # 定制参数
17
    assert admin.role == "admin"
18

19
def test_multiple(make_user):
20
    u1 = make_user(name="a")
21
    u2 = make_user(name="b")     # 一个测试里造多个对象
22
    assert u1.name != u2.name

1
@pytest.fixture
2
def db_connection():
3
    conn = create_connection("sqlite:///:memory:")
4
    yield conn
5
    conn.close()
6

7
@pytest.fixture
8
def user_repo(db_connection):        # 依赖上面的 db_connection
9
    return UserRepository(db_connection)
10

11
@pytest.fixture
12
def saved_user(user_repo):           # 依赖 user_repo
13
    return user_repo.save(User("alice"))
14

15
def test_find(user_repo, saved_user):
16
    found = user_repo.find_by_id(saved_user.id)
17
    assert found.name == "alice"

1
@pytest.fixture(autouse=True)
2
def reset_singleton():
3
    """每个测试前自动重置全局单例，无需在测试函数里声明。"""
4
    GlobalConfig.reset()
5
    yield
6
    GlobalCache.clear()

1
def test_tmp_file(tmp_path):
2
    # tmp_path 是一个 pathlib.Path，指向本次测试专属的临时目录
3
    f = tmp_path / "config.json"
4
    f.write_text('{"key": "value"}')
5
    assert f.read_text() == '{"key": "value"}'
6

7
def test_env(monkeypatch):
8
    # monkeypatch：临时修改环境变量/属性，测试后自动还原
9
    monkeypatch.setenv("API_KEY", "test-key")
10
    monkeypatch.setattr("app.config.DEBUG", True)
11
    assert os.environ["API_KEY"] == "test-key"
12

13
def test_capture(capsys):
14
    # capsys：捕获 stdout/stderr
15
    print("hello")
16
    captured = capsys.readouterr()
17
    assert captured.out == "hello\n"
18

19
def test_logging(caplog):
20
    # caplog：捕获日志
21
    logging.getLogger().warning("something")
22
    assert "something" in caplog.text

1
@pytest.mark.parametrize("email, is_valid", [
2
    ("user@example.com", True),
3
    ("invalid.email", False),
4
    ("", False),
5
], ids=["正常邮箱", "缺少@符号", "空字符串"])
6
def test_validate_email(email, is_valid):
7
    assert validate_email(email) is is_valid

1
@pytest.mark.parametrize("value, expected", [
2
    (10, 100),
3
    (5, 25),
4
    pytest.param(0, 1, marks=pytest.mark.xfail(reason="0 的处理有已知 bug")),
5
])
6
def test_square(value, expected):
7
    assert square(value) == expected

1
@pytest.mark.parametrize("browser", ["chrome", "firefox"])
2
@pytest.mark.parametrize("os", ["windows", "macos", "linux"])
3
def test_matrix(browser, os):
4
    # 自动生成 2 × 3 = 6 个用例组合
5
    assert run_on(browser, os) is True

1
@pytest.fixture(params=["sqlite", "postgres", "mysql"])
2
def database(request):
3
    db = connect(request.param)      # request.param 拿到当前参数
4
    yield db
5
    db.close()
6

7
def test_query(database):
8
    # 这个测试会自动针对 3 种数据库各跑一次
9
    assert database.query("SELECT 1") is not None

1
# tests/test_service.py —— 正确 ✅
2
def test_get_user_count(mocker):
3
    # patch 'app.service.fetch_data'（使用处），不是 'app.utils.fetch_data'
4
    mock_fetch = mocker.patch("app.service.fetch_data")
5
    mock_fetch.return_value = [{"id": 1}, {"id": 2}, {"id": 3}]
6

7
    assert get_user_count() == 3
8
    mock_fetch.assert_called_once_with("https://api.example.com/users")

1
def test_mock_basics(mocker):
2
    m = mocker.Mock()
3

4
    # 固定返回值
5
    m.get_name.return_value = "Alice"
6
    assert m.get_name() == "Alice"
7

8
    # side_effect：抛异常
9
    m.connect.side_effect = ConnectionError("网络不可用")
10
    with pytest.raises(ConnectionError):
11
        m.connect()
12

13
    # side_effect：每次调用返回不同值
14
    m.next_id.side_effect = [1, 2, 3]
15
    assert m.next_id() == 1
16
    assert m.next_id() == 2
17

18
    # side_effect：用函数动态计算
19
    m.double.side_effect = lambda x: x * 2
20
    assert m.double(5) == 10

1
def test_call_assertions(mocker):
2
    mock_send = mocker.patch("app.notifier.send_email")
3

4
    notify_user("bob@example.com", "Hello")
5

6
    mock_send.assert_called_once()                          # 恰好调用一次
7
    mock_send.assert_called_once_with("bob@example.com", "Hello")  # 参数校验
8
    assert mock_send.call_count == 1
9
    # 查看具体调用参数
10
    args, kwargs = mock_send.call_args
11
    assert args[0] == "bob@example.com"

1
# ── mock HTTP 请求（requests）─────────────────────────────
2
def test_api_call(mocker):
3
    mock_get = mocker.patch("app.client.requests.get")
4
    mock_get.return_value.status_code = 200
5
    mock_get.return_value.json.return_value = {"result": "ok"}
6

7
    assert call_api() == {"result": "ok"}
8

9
# ── mock 时间 ─────────────────────────────────────────────
10
def test_time(mocker):
11
    mock_now = mocker.patch("app.service.datetime")
12
    mock_now.now.return_value = datetime(2026, 1, 1, 12, 0, 0)
13
    assert get_timestamp() == "2026-01-01 12:00:00"
14

15
# ── mock 文件读取 ─────────────────────────────────────────
16
def test_read_config(mocker):
17
    mocker.patch("builtins.open", mocker.mock_open(read_data='{"debug": true}'))
18
    assert load_config()["debug"] is True
19

20
# ── spy：真实调用 + 记录（不替换实现）───────────────────────
21
def test_spy(mocker):
22
    spy = mocker.spy(math, "sqrt")
23
    result = compute_something(16)
24
    spy.assert_called_once_with(16)
25
    assert spy.spy_return == 4.0

1
import sys, pytest
2

3
@pytest.mark.skip(reason="功能还没实现")
4
def test_future_feature():
5
    ...
6

7
@pytest.mark.skipif(sys.version_info < (3, 12), reason="需要 Python 3.12+")
8
def test_new_syntax():
9
    ...
10

11
@pytest.mark.xfail(reason="已知 bug，修复前预期失败")
12
def test_known_bug():
13
    assert buggy_function() == "correct"

1
# pyproject.toml —— 承接第一篇的配置
2
[tool.pytest.ini_options]
3
markers = [
4
    "slow: 运行较慢的测试（用 -m 'not slow' 跳过）",
5
    "integration: 集成测试，需要外部服务",
6
    "smoke: 冒烟测试，最核心的用例",
7
]

1
@pytest.mark.slow
2
def test_heavy_computation():
3
    ...
4

5
@pytest.mark.integration
6
def test_real_database():
7
    ...

1
def test_withdraw_reduces_balance():
2
    # Arrange：准备
3
    account = Account(balance=100)
4

5
    # Act：执行被测行为
6
    account.withdraw(30)
7

8
    # Assert：断言结果
9
    assert account.balance == 70