pytest — Python 测试框架

# 安装
pip install pytest

# 运行当前目录下所有测试
pytest

# 指定文件
pytest test_calculator.py

# 指定目录
pytest tests/

# 显示详细输出
pytest -v

# 遇到第一个失败就停止
pytest -x

# 只运行匹配关键字的测试
pytest -k "api"

# 只运行被标记为 slow 的测试
pytest -m slow

project/
  tests/
    __init__.py          # 可为空
    test_models.py       # 自动发现
    test_views.py        # 自动发现
    conftest.py          # 共享 fixture
    helpers/
      test_utils.py      # 自动发现
    integration/
      test_api.py        # 自动发现

# test_fixture_basic.py — 需要 Python 3.10+
import pytest


# 定义一个 fixture
@pytest.fixture
def sample_user() -> dict:
    """返回一个标准测试用户，供多个测试复用"""
    return {
        "id": 1,
        "name": "Alice",
        "email": "alice@example.com",
        "role": "admin",
    }


# 测试函数通过参数名注入 fixture
def test_user_name(sample_user: dict):
    assert sample_user["name"] == "Alice"


def test_user_role(sample_user: dict):
    assert sample_user["role"] == "admin"


def test_user_id(sample_user: dict):
    assert sample_user["id"] == 1

# test_scope.py — 需要 Python 3.10+
import pytest


@pytest.fixture(scope="function")
def function_fixture():
    """每个测试函数都会调用一次"""
    print("\n  [setup] function_fixture")
    yield {"data": "function-level"}
    print("\n  [teardown] function_fixture")


@pytest.fixture(scope="module")
def module_fixture():
    """同一个 .py 文件内只调用一次"""
    print("\n  [setup] module_fixture")
    yield {"data": "module-level"}
    print("\n  [teardown] module_fixture")


@pytest.fixture(scope="session")
def session_fixture():
    """整个测试运行期间只调用一次"""
    print("\n  [setup] session_fixture")
    yield {"data": "session-level"}
    print("\n  [teardown] session_fixture")


def test_a(function_fixture, module_fixture, session_fixture):
    assert function_fixture["data"] == "function-level"


def test_b(function_fixture, module_fixture, session_fixture):
    # module_fixture 和 session_fixture 的实例与 test_a 相同
    # function_fixture 是全新创建的
    assert module_fixture["data"] == "module-level"

# test_yield_fixture.py — 需要 Python 3.10+
import pytest
from pathlib import Path
import tempfile
import os


@pytest.fixture
def temp_file():
    """创建临时文件，测试结束后自动删除"""
    fd, path = tempfile.mkstemp(suffix=".txt")
    filepath = Path(path)
    # ===== setup 阶段 =====
    filepath.write_text("temp content")
    print(f"\n  [setup] 临时文件已创建: {filepath}")

    yield filepath  # 将路径注入测试函数

    # ===== teardown 阶段 =====
    os.close(fd)
    filepath.unlink(missing_ok=True)
    print(f"\n  [teardown] 临时文件已删除: {filepath}")


def test_read_temp_file(temp_file: Path):
    assert temp_file.read_text() == "temp content"
    # 测试结束后，temp_file 自动被清理

project/
  tests/
    conftest.py              # 全局 fixture，所有测试可用
    unit/
      conftest.py            # 仅 unit/ 目录下可见
      test_models.py
    integration/
      conftest.py            # 仅 integration/ 目录下可见
      test_api.py

# tests/conftest.py — 需要 Python 3.10+
import pytest
from typing import Generator


@pytest.fixture(scope="session")
def db_url() -> str:
    """全局数据库连接字符串"""
    return "postgresql://test:test@localhost:5432/testdb"


@pytest.fixture
def sample_users() -> list[dict]:
    """可复用的用户列表"""
    return [
        {"id": 1, "name": "Alice", "active": True},
        {"id": 2, "name": "Bob", "active": False},
        {"id": 3, "name": "Charlie", "active": True},
    ]

# test_fixture_param.py — 需要 Python 3.10+
import pytest


@pytest.fixture(params=["mysql", "postgresql", "sqlite"])
def db_backend(request):
    """fixture 参数化：依次测试三种数据库后端"""
    print(f"\n  [setup] 初始化 {request.param} 连接")
    # 实际项目中这里会建立真正的数据库连接
    yield request.param
    print(f"\n  [teardown] 关闭 {request.param} 连接")


def test_db_ping(db_backend: str):
    """这个测试会运行 3 次，分别对应 mysql/postgresql/sqlite"""
    assert db_backend in ("mysql", "postgresql", "sqlite")
    print(f"  测试 {db_backend} 连接成功")

# test_parametrize.py — 需要 Python 3.10+
import pytest


# 基础用法：逐个指定参数值
@pytest.mark.parametrize("a, b, expected", [
    (1, 2, 3),
    (0, 0, 0),
    (-1, 1, 0),
    (100, 200, 300),
])
def test_addition(a: int, b: int, expected: int):
    assert a + b == expected


# 使用 ids 参数给每组测试命名
@pytest.mark.parametrize("username, is_valid", [
    ("alice", True),
    ("a" * 100, False),      # 太长
    ("", False),              # 空字符串
    ("user@name", True),
], ids=["正常用户名", "超长用户名", "空用户名", "含特殊字符"])
def test_username_validation(username: str, is_valid: bool):
    result = (0 < len(username) <= 50)
    assert result == is_valid

# 多个 parametrize 装饰器叠加 = 笛卡尔积
@pytest.mark.parametrize("a", [1, 2])
@pytest.mark.parametrize("b", [10, 20])
@pytest.mark.parametrize("op", ["add", "sub"])
def test_math(a: int, b: int, op: str):
    """此测试运行 2 x 2 x 2 = 8 次"""
    if op == "add":
        assert a + b == a + b  # 简化为示例
    elif op == "sub":
        assert a - b == a - b

# test_exceptions.py — 需要 Python 3.10+
import pytest


def divide(a: float, b: float) -> float:
    if b == 0:
        raise ValueError("除数不能为零")
    return a / b


def test_divide_by_zero():
    # 验证抛出了指定异常
    with pytest.raises(ValueError, match="除数不能为零"):
        divide(10, 0)


def test_divide_success():
    assert divide(10, 2) == 5.0


def test_raises_captures_exception_info():
    """捕获异常对象，进一步验证异常属性"""
    with pytest.raises(ValueError) as exc_info:
        divide(10, 0)

    # 验证异常信息
    assert "除数不能为零" in str(exc_info.value)
    # exc_info.type 是异常类型
    assert exc_info.type is ValueError


def test_no_exception():
    """pytest.raises 的 '利器' 用法：验证某段代码不抛出异常"""
    # 如果不抛出异常，测试通过
    result = divide(10, 2)
    assert result == 5.0


# ========== 警告测试 ==========
import warnings


def deprecated_function():
    warnings.warn("此函数已废弃，请使用 new_function()", DeprecationWarning)
    return "result"


def test_deprecated_warning():
    with pytest.warns(DeprecationWarning, match="已废弃"):
        result = deprecated_function()
    assert result == "result"


def test_no_warning():
    """验证代码不产生警告"""
    with warnings.catch_warnings():
        warnings.simplefilter("error")  # 将警告转为异常
        result = divide(10, 2)
        assert result == 5.0

# test_monkeypatch.py — 需要 Python 3.10+
import pytest
import os
from pathlib import Path


# ========== 修改环境变量 ==========
def test_env_variable(monkeypatch):
    # 设置临时环境变量
    monkeypatch.setenv("DATABASE_URL", "sqlite:///:memory:")
    monkeypatch.setenv("DEBUG", "true")

    assert os.environ["DATABASE_URL"] == "sqlite:///:memory:"
    # 测试结束后环境变量自动恢复


# ========== 修改对象属性 ==========
class Config:
    base_url = "https://api.prod.example.com"
    timeout = 30


def test_config_override(monkeypatch):
    monkeypatch.setattr(Config, "base_url", "https://api.test.example.com")
    monkeypatch.setattr(Config, "timeout", 5)

    assert Config.base_url == "https://api.test.example.com"
    assert Config.timeout == 5
    # 测试结束后 Config 的属性自动恢复


# ========== 修改字典 ==========
def test_dict_override(monkeypatch):
    data = {"api_key": "real-key", "version": "1.0"}

    monkeypatch.setitem(data, "api_key", "test-key")
    monkeypatch.setitem(data, "version", "2.0-test")

    assert data["api_key"] == "test-key"
    # 测试结束后字典自动恢复


# ========== 替换函数 ==========
def call_external_api(url: str) -> dict:
    """实际的外部 API 调用（测试中需要 mock）"""
    import requests
    return requests.get(url).json()


def test_mock_api_call(monkeypatch):
    """用 monkeypatch 替换函数，避免真实网络请求"""

    def mock_get(url: str) -> dict:
        return {"status": "ok", "data": "mocked"}

    monkeypatch.setattr(
        "test_monkeypatch.call_external_api",  # 注意：要用模块路径
        mock_get,
    )

    result = call_external_api("https://api.example.com")
    assert result == {"status": "ok", "data": "mocked"}

# test_mock.py — 需要 Python 3.10+
from unittest.mock import Mock, patch, MagicMock
import pytest


# ========== Mock 对象基础 ==========
def test_mock_basic():
    mock = Mock(return_value=42)
    assert mock() == 42
    assert mock(1, 2, key="value") == 42

    # 验证调用
    mock.assert_called()          # 至少被调用一次
    assert mock.call_count == 2   # 被调用了 2 次
    mock.assert_called_with(1, 2, key="value")  # 最后一次调用参数


# ========== 验证调用顺序和参数 ==========
def test_mock_call_tracking():
    service = Mock()
    service.connect("db1")
    service.connect("db2")
    service.query("SELECT 1")

    assert service.connect.call_count == 2
    service.connect.assert_any_call("db1")
    service.connect.assert_any_call("db2")

    # 获取所有调用参数
    calls = service.connect.call_args_list
    assert calls[0].args == ("db1",)
    assert calls[1].args == ("db2",)


# ========== side_effect：模拟异常 ==========
def test_mock_side_effect():
    mock = Mock(side_effect=ConnectionError("连接超时"))

    with pytest.raises(ConnectionError, match="连接超时"):
        mock()


# ========== side_effect：序列返回值 ==========
def test_mock_side_effect_sequence():
    mock = Mock(side_effect=[1, 2, 3])

    assert mock() == 1
    assert mock() == 2
    assert mock() == 3
    # 第四次调用会抛出 StopIteration


# ========== patch 装饰器 ==========
class UserService:
    @staticmethod
    def get_user_count() -> int:
        """实际查询数据库获取用户数"""
        import random
        return random.randint(1, 1000)


@patch("test_mock.UserService.get_user_count", return_value=42)
def test_user_count_with_patch(mock_get_count):
    """使用 patch 装饰器替换方法"""
    assert UserService.get_user_count() == 42
    mock_get_count.assert_called_once()

# 安装
pip install pytest-cov

# 运行测试并生成覆盖率报告
pytest --cov=myapp tests/

# 生成终端报告（显示未覆盖的行）
pytest --cov=myapp --cov-report=term-missing tests/

# 生成 HTML 报告
pytest --cov=myapp --cov-report=html tests/
# 报告生成在 htmlcov/index.html

# pyproject.toml 中配置覆盖率
[tool.coverage.run]
source = ["myapp"]
omit = [
    "*/migrations/*",
    "*/tests/*",
    "*/__init__.py",
]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "if __name__ == .__main__.:",
    "raise NotImplementedError",
]

# 自动检测 CPU 核心数，并行执行测试
pytest -n auto

# 指定 4 个 worker 并行
pytest -n 4

# 按文件分发（每个 worker 执行一个文件，推荐）
pytest -n auto --dist loadfile

# 全局设置超时时间（秒）
pytest --timeout=30

# 或者通过装饰器对单个测试设置

import pytest


@pytest.mark.timeout(5)
def test_slow_operation():
    """此测试必须在 5 秒内完成，否则失败"""
    import time
    time.sleep(2)  # 模拟耗时操作
    assert True

# test_with_pytest_mock.py — 需要 Python 3.10+
import pytest
import os


class Database:
    def connect(self, url: str) -> bool:
        # 真实连接数据库
        return True


def test_mocker_fixture(mocker):
    """pytest-mock 提供的 mocker fixture 比 unittest.mock 更便捷"""
    # mocker.patch 自动在测试结束后撤销
    mock_connect = mocker.patch.object(Database, "connect", return_value=True)

    db = Database()
    result = db.connect("postgresql://...")

    assert result is True
    mock_connect.assert_called_once()

    # mocker 还提供了 spy 功能
    mocker.spy(os, "getenv")
    os.getenv("PATH")
    assert os.getenv.call_count == 1

# test_aaa_pattern.py — 需要 Python 3.10+
import pytest


class ShoppingCart:
    def __init__(self):
        self._items: list[dict] = []

    def add_item(self, name: str, price: float, quantity: int = 1):
        self._items.append({"name": name, "price": price, "quantity": quantity})

    def total(self) -> float:
        return sum(item["price"] * item["quantity"] for item in self._items)

    def clear(self):
        self._items.clear()


def test_cart_total_with_multiple_items():
    # ===== Arrange（准备）=====
    cart = ShoppingCart()
    cart.add_item("Python 编程", 59.0, 2)
    cart.add_item("数据结构", 45.0, 1)

    # ===== Act（执行）=====
    result = cart.total()

    # ===== Assert（断言）=====
    assert result == 163.0


def test_cart_total_empty():
    # Arrange
    cart = ShoppingCart()

    # Act
    result = cart.total()

    # Assert
    assert result == 0.0

# ❌ 不好：一个测试验证多个概念
def test_user_service():
    service = UserService()
    service.create_user("alice", "alice@example.com")
    user = service.get_user("alice")
    assert user.name == "alice"
    assert user.email == "alice@example.com"
    service.delete_user("alice")
    assert service.get_user("alice") is None
    # 这个测试同时验证了创建、查询、删除——任何一个失败都不好定位

# ✅ 好：拆分为独立的测试
def test_create_user():
    service = UserService()
    service.create_user("alice", "alice@example.com")
    user = service.get_user("alice")
    assert user.name == "alice"

def test_get_user_email():
    service = UserService()
    service.create_user("alice", "alice@example.com")
    user = service.get_user("alice")
    assert user.email == "alice@example.com"

def test_delete_user():
    service = UserService()
    service.create_user("alice", "alice@example.com")
    service.delete_user("alice")
    assert service.get_user("alice") is None

tests/
  conftest.py              # 全局 fixture
  unit/                    # 单元测试
    test_models.py
    test_services.py
    test_utils.py
  integration/             # 集成测试
    test_api.py
    test_database.py
  e2e/                     # 端到端测试
    test_user_flow.py
  fixtures/                # 测试用的静态数据
    users.json
    sample_data.csv

# 使用自定义标记对测试分类
@pytest.mark.slow
def test_large_dataset_processing():
    ...

@pytest.mark.integration
def test_full_pipeline():
    ...

@pytest.mark.smoke
def test_health_check():
    ...

# pyproject.toml 中注册自定义标记
[tool.pytest.ini_options]
markers = [
    "slow: 慢速测试（耗时长）",
    "integration: 集成测试（需要外部服务）",
    "smoke: 冒烟测试（核心功能快速验证）",
]

# 只运行冒烟测试
pytest -m smoke

# 跳过慢速测试
pytest -m "not slow"

# 运行集成测试但跳过慢速的
pytest -m "integration and not slow"

# test_calculator.py — 需要 Python 3.10+
# 一个完整的测试文件，涵盖所有核心概念
import pytest
from typing import Any


# ========== 被测代码 ==========
class Calculator:
    """一个简单的计算器类"""

    def add(self, a: float, b: float) -> float:
        return a + b

    def subtract(self, a: float, b: float) -> float:
        return a - b

    def multiply(self, a: float, b: float) -> float:
        return a * b

    def divide(self, a: float, b: float) -> float:
        if b == 0:
            raise ValueError("除数不能为零")
        return a / b

    def power(self, base: float, exponent: float) -> float:
        return base ** exponent


# ========== Fixture ==========
@pytest.fixture
def calc() -> Calculator:
    """每个测试获得一个全新的 Calculator 实例"""
    return Calculator()


@pytest.fixture
def large_numbers() -> list[tuple[float, float]]:
    """提供大数测试数据"""
    return [(1e10, 1e10), (1e100, 1e100)]


# ========== 基本断言 ==========
def test_add(calc: Calculator):
    assert calc.add(1, 2) == 3
    assert calc.add(-1, 1) == 0
    assert calc.add(0.1, 0.2) == pytest.approx(0.3)  # 浮点数比较


# ========== 参数化测试 ==========
@pytest.mark.parametrize("a, b, expected", [
    (10, 2, 5),
    (9, 3, 3),
    (0, 5, 0),
    (-10, 2, -5),
    (-10, -2, 5),
], ids=["正数除正数", "整除", "零除正数", "负数除正数", "负数除负数"])
def test_divide(calc: Calculator, a: float, b: float, expected: float):
    assert calc.divide(a, b) == expected


# ========== 异常测试 ==========
def test_divide_by_zero_raises(calc: Calculator):
    with pytest.raises(ValueError, match="除数不能为零"):
        calc.divide(10, 0)


# ========== 使用 monkeypatch ==========
def test_with_env_config(calc: Calculator, monkeypatch):
    """模拟环境变量影响计算精度"""
    monkeypatch.setenv("CALC_PRECISION", "2")

    # 如果项目中有精度配置逻辑，这里可以验证
    assert calc.add(1.234, 5.678) == pytest.approx(6.912, abs=0.001)


# ========== 跳过测试 ==========
@pytest.mark.skip(reason="指数运算待实现")
def test_power(calc: Calculator):
    assert calc.power(2, 3) == 8


@pytest.mark.skipif(
    "sys.version_info < (3, 11)",
    reason="需要 Python 3.11+ 的数学库特性",
)
def test_python311_feature(calc: Calculator):
    # 该测试仅在 Python 3.11+ 上运行
    assert calc.add(1, 1) == 2