测试

# 安装 pytest 和常用插件
pip install pytest pytest-cov pytest-mock pytest-asyncio

# 运行测试
pytest                        # 自动发现 test_*.py / *_test.py
pytest -v                     # 详细输出
pytest -x                     # 遇到第一个失败就停止
pytest -k "test_discount"     # 只运行名字匹配的测试
pytest --cov=src --cov-report=html  # 生成覆盖率报告

# src/discount.py — 被测代码
def apply_discount(price: float, rate: float) -> float:
    """应用折扣率计算折后价"""
    if not 0 <= rate <= 1:
        raise ValueError("rate must be between 0 and 1")
    return round(price * (1 - rate), 2)


# tests/test_discount.py — 测试代码
def test_apply_discount_returns_discounted_price():
    """正常路径：合法折扣率返回正确折后价"""
    assert apply_discount(100.0, 0.2) == 80.0

def test_apply_discount_rejects_invalid_rate():
    """异常路径：非法折扣率抛出 ValueError"""
    import pytest
    with pytest.raises(ValueError):
        apply_discount(100.0, 1.5)

import pytest


def test_assertion_examples():
    # 基础断言
    assert 1 + 1 == 2
    assert "hello" in "hello world"
    assert [1, 2, 3]  # 非空判断

    # 浮点数比较
    assert 0.1 + 0.2 == pytest.approx(0.3)

    # 异常断言
    with pytest.raises(ValueError, match="must be between"):
        apply_discount(100, -0.1)

    # 异常断言 + 检查异常信息
    with pytest.raises(ValueError) as exc_info:
        apply_discount(100, 2.0)
    assert "rate must be" in str(exc_info.value)

    # 警告断言
    with pytest.warns(DeprecationWarning):
        import warnings
        warnings.warn("这个功能已弃用", DeprecationWarning)

import pytest
from pathlib import Path
import tempfile
import shutil


@pytest.fixture
def temp_dir():
    """创建临时目录，测试后自动清理"""
    dir_path = Path(tempfile.mkdtemp())
    yield dir_path  # yield 之前的代码是 setup，之后是 teardown
    shutil.rmtree(dir_path, ignore_errors=True)


def test_write_file(temp_dir):
    """fixture 通过参数名自动注入"""
    test_file = temp_dir / "test.txt"
    test_file.write_text("hello")
    assert test_file.read_text() == "hello"

# tests/conftest.py — pytest 会自动加载这个文件中的 fixture

import pytest
from dataclasses import dataclass


@dataclass
class MockUser:
    id: int
    name: str
    role: str = "user"


@pytest.fixture
def admin_user():
    """所有测试文件都可使用这个 fixture"""
    return MockUser(id=1, name="Admin", role="admin")


@pytest.fixture
def normal_user():
    return MockUser(id=2, name="Alice", role="user")


@pytest.fixture
def user_list(admin_user, normal_user):
    """fixture 可以组合其他 fixture"""
    return [admin_user, normal_user]

import pytest


@pytest.fixture(scope="session")
def db_connection():
    """整个测试会话只创建一次数据库连接"""
    import sqlite3
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE users (id INTEGER, name TEXT)")
    yield conn
    conn.close()


@pytest.fixture(scope="function")
def db_with_seed(db_connection):
    """每个测试函数获得干净的数据库状态"""
    db_connection.execute("DELETE FROM users")
    db_connection.execute("INSERT INTO users VALUES (1, 'Alice')")
    db_connection.commit()
    yield db_connection
    # 测试后清空（虽然 scope="function" 的 fixture 会重新执行，但显式清理是好习惯）
    db_connection.execute("DELETE FROM users")
    db_connection.commit()


def test_user_exists(db_with_seed):
    cursor = db_with_seed.execute("SELECT name FROM users WHERE id = 1")
    assert cursor.fetchone()[0] == "Alice"

import pytest


def fibonacci(n: int) -> int:
    """计算第 n 个斐波那契数"""
    if n < 0:
        raise ValueError("n 不能为负数")
    if n <= 1:
        return n
    a, b = 0, 1
    for _ in range(2, n + 1):
        a, b = b, a + b
    return b


@pytest.mark.parametrize("n, expected", [
    (0, 0),
    (1, 1),
    (2, 1),
    (5, 5),
    (10, 55),
    (20, 6765),
])
def test_fibonacci_values(n, expected):
    """参数化测试：多组输入验证同一个函数"""
    assert fibonacci(n) == expected


@pytest.mark.parametrize("invalid_input", [-1, -100])
def test_fibonacci_rejects_negative(invalid_input):
    """参数化测试：验证异常路径"""
    with pytest.raises(ValueError):
        fibonacci(invalid_input)

@pytest.mark.parametrize("price, rate, expected", [
    (100.0, 0.0, 100.0),    # 无折扣
    (100.0, 0.5, 50.0),     # 半价
    (100.0, 1.0, 0.0),      # 免费赠送
    (99.99, 0.15, 84.99),    # 精度测试
])
def test_discount_combinations(price, rate, expected):
    assert apply_discount(price, rate) == expected

@pytest.mark.parametrize("input_str, expected", [
    pytest.param("hello", "HELLO", id="小写转大写"),
    pytest.param("WORLD", "WORLD", id="已是大写"),
    pytest.param("HeLLo", "HELLO", id="混合大小写"),
    pytest.param("", "", id="空字符串"),
    pytest.param("123", "123", marks=pytest.mark.xfail(reason="数字不需要转换"), id="数字输入"),
])
def test_upper_variants(input_str, expected):
    assert input_str.upper() == expected

# src/notification.py — 被测代码
from typing import Protocol
from dataclasses import dataclass


class EmailSender(Protocol):
    def send(self, to: str, subject: str, body: str) -> None: ...


@dataclass
class NotificationService:
    email_sender: EmailSender

    def notify_user(self, email: str, message: str) -> None:
        if not email or "@" not in email:
            raise ValueError("无效的邮箱地址")
        self.email_sender.send(
            to=email,
            subject="系统通知",
            body=message,
        )


# tests/test_notification.py — 测试代码
class FakeEmailSender:
    """轻量级替身：记录发送的邮件，不真正发送"""
    def __init__(self) -> None:
        self.sent: list[tuple[str, str, str]] = []

    def send(self, to: str, subject: str, body: str) -> None:
        self.sent.append((to, subject, body))


def test_notify_user_sends_email():
    """正常路径：通知用户时发送邮件"""
    fake = FakeEmailSender()
    service = NotificationService(email_sender=fake)

    service.notify_user("user@example.com", "订单已发货")

    assert len(fake.sent) == 1
    assert fake.sent[0] == ("user@example.com", "系统通知", "订单已发货")


def test_notify_user_rejects_invalid_email():
    """异常路径：无效邮箱地址抛出 ValueError"""
    fake = FakeEmailSender()
    service = NotificationService(email_sender=fake)

    import pytest
    with pytest.raises(ValueError, match="无效的邮箱"):
        service.notify_user("", "消息")

    # 确认没有发送邮件
    assert len(fake.sent) == 0

# src/weather.py — 被测代码
import requests


def get_temperature(city: str) -> float:
    """获取城市温度（依赖外部 API）"""
    response = requests.get(f"https://api.weather.com/{city}")
    data = response.json()
    return data["temperature"]


# tests/test_weather.py — 测试代码
from unittest.mock import patch


def test_get_temperature():
    """用 mock 替换 requests.get，返回预设数据"""
    mock_response = type("MockResponse", (), {
        "json": lambda self: {"temperature": 25.5},
    })()

    with patch("src.weather.requests.get", return_value=mock_response):
        temp = get_temperature("beijing")
        assert temp == 25.5


def test_get_temperature_timeout():
    """模拟请求超时"""
    with patch("src.weather.requests.get", side_effect=requests.Timeout("超时")):
        import pytest
        with pytest.raises(requests.Timeout):
            get_temperature("beijing")

# pip install pytest-mock
# pytest-mock 提供 mocker fixture，比 unittest.mock 更简洁


def test_get_temperature_with_mocker(mocker):
    """使用 pytest-mock 的 mocker fixture"""
    mock_response = mocker.MagicMock()
    mock_response.json.return_value = {"temperature": 30.0}

    mocker.patch("src.weather.requests.get", return_value=mock_response)

    temp = get_temperature("shanghai")
    assert temp == 30.0

    # 验证调用参数
    requests.get.assert_called_once_with("https://api.weather.com/shanghai")

# pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-v --tb=short"

[tool.coverage.run]
source = ["src"]
omit = ["tests/*", "*/__pycache__/*"]

[tool.coverage.report]
fail_under = 80       # 覆盖率低于 80% 构建失败
show_missing = true   # 显示未覆盖的行号
exclude_lines = [
    "pragma: no cover",
    "if TYPE_CHECKING:",
    "raise NotImplementedError",
    "if __name__ == .__main__.:",
]

# 终端报告
pytest --cov=src --cov-report=term-missing

# HTML 报告（可点击查看未覆盖的代码）
pytest --cov=src --cov-report=html
open htmlcov/index.html

# 只检查关键模块
pytest --cov=src.core --cov=src.service

# src/validator.py — 被测代码
import re


def validate_email(email: str) -> bool:
    """验证邮箱格式"""
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))


def validate_password(password: str) -> tuple[bool, str]:
    """验证密码强度：至少8位，包含大小写和数字"""
    if len(password) < 8:
        return False, "密码至少8位"
    if not re.search(r'[A-Z]', password):
        return False, "密码必须包含大写字母"
    if not re.search(r'[a-z]', password):
        return False, "密码必须包含小写字母"
    if not re.search(r'\d', password):
        return False, "密码必须包含数字"
    return True, "密码强度合格"


# tests/test_validator.py — 测试代码
import pytest


class TestValidateEmail:
    """邮箱验证测试组"""

    @pytest.mark.parametrize("email", [
        "user@example.com",
        "user.name+tag@domain.co",
        "a@b.cc",
    ])
    def test_valid_emails(self, email):
        assert validate_email(email) is True

    @pytest.mark.parametrize("email", [
        "",
        "no-at-sign",
        "@domain.com",
        "user@",
        "user@.com",
        "user@domain",
    ])
    def test_invalid_emails(self, email):
        assert validate_email(email) is False


class TestValidatePassword:
    """密码验证测试组"""

    def test_strong_password(self):
        ok, msg = validate_password("Abc12345")
        assert ok is True
        assert "合格" in msg

    @pytest.mark.parametrize("password, expected_msg", [
        ("Ab1!", "至少8位"),
        ("abcdefgh", "大写字母"),
        ("ABCDEFGH", "小写字母"),
        ("Abcdefgh", "数字"),
    ])
    def test_weak_passwords(self, password, expected_msg):
        ok, msg = validate_password(password)
        assert ok is False
        assert expected_msg in msg

# src/user_repository.py — 被测代码
import sqlite3
from dataclasses import dataclass


@dataclass
class User:
    id: int
    name: str
    email: str


class UserRepository:
    def __init__(self, db_path: str) -> None:
        self.conn = sqlite3.connect(db_path)
        self.conn.row_factory = sqlite3.Row
        self._init_db()

    def _init_db(self) -> None:
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                email TEXT NOT NULL UNIQUE
            )
        """)
        self.conn.commit()

    def create_user(self, name: str, email: str) -> User:
        cursor = self.conn.execute(
            "INSERT INTO users (name, email) VALUES (?, ?)",
            (name, email),
        )
        self.conn.commit()
        return User(id=cursor.lastrowid, name=name, email=email)

    def get_user(self, user_id: int) -> User | None:
        row = self.conn.execute(
            "SELECT * FROM users WHERE id = ?", (user_id,)
        ).fetchone()
        if row is None:
            return None
        return User(id=row["id"], name=row["name"], email=row["email"])

    def close(self) -> None:
        self.conn.close()


# tests/test_user_repository.py — 测试代码
import pytest
from pathlib import Path


@pytest.fixture
def repo():
    """创建内存数据库仓库，测试后自动关闭"""
    repository = UserRepository(":memory:")
    yield repository
    repository.close()


def test_create_and_get_user(repo):
    """创建用户后可以查询到"""
    user = repo.create_user("Alice", "alice@example.com")
    assert user.id is not None
    assert user.name == "Alice"

    found = repo.get_user(user.id)
    assert found is not None
    assert found.email == "alice@example.com"


def test_get_nonexistent_user(repo):
    """查询不存在的用户返回 None"""
    assert repo.get_user(999) is None


def test_create_duplicate_email(repo):
    """重复邮箱应该抛出异常"""
    repo.create_user("Alice", "alice@example.com")
    with pytest.raises(sqlite3.IntegrityError):
        repo.create_user("Bob", "alice@example.com")

# src/async_service.py — 被测代码
import asyncio


async def fetch_data(url: str, delay: float = 0.1) -> dict[str, str]:
    """模拟异步网络请求"""
    await asyncio.sleep(delay)
    if "error" in url:
        raise ConnectionError(f"请求失败: {url}")
    return {"url": url, "status": "ok"}


async def fetch_multiple(urls: list[str]) -> list[dict[str, str]]:
    """并发获取多个 URL"""
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [r for r in results if not isinstance(r, Exception)]


# tests/test_async_service.py — 测试代码
import pytest


@pytest.mark.asyncio
async def test_fetch_data_success():
    """异步请求成功路径"""
    result = await fetch_data("https://example.com")
    assert result["status"] == "ok"


@pytest.mark.asyncio
async def test_fetch_data_error():
    """异步请求失败路径"""
    with pytest.raises(ConnectionError, match="请求失败"):
        await fetch_data("https://error.example.com")


@pytest.mark.asyncio
async def test_fetch_multiple():
    """并发请求：部分成功部分失败"""
    urls = [
        "https://a.com",
        "https://error.b.com",
        "https://c.com",
    ]
    results = await fetch_multiple(urls)
    # 失败的请求被过滤掉
    assert len(results) == 2
    assert all(r["status"] == "ok" for r in results)

# tests/test_temperature.py
def test_celsius_to_fahrenheit():
    from src.temperature import celsius_to_fahrenheit  # 还没实现
    assert celsius_to_fahrenheit(0) == 32
    assert celsius_to_fahrenheit(100) == 212

# src/temperature.py
def celsius_to_fahrenheit(celsius: float) -> float:
    return celsius * 9 / 5 + 32

def test_absolute_zero():
    assert celsius_to_fahrenheit(-273.15) == pytest.approx(-459.67, abs=0.01)

def test_invalid_temperature():
    with pytest.raises(ValueError):
        celsius_to_fahrenheit(-300)  # 低于绝对零度

def celsius_to_fahrenheit(celsius: float) -> float:
    if celsius < -273.15:
        raise ValueError("温度不能低于绝对零度")
    return celsius * 9 / 5 + 32

ABSOLUTE_ZERO_C = -273.15

def celsius_to_fahrenheit(celsius: float) -> float:
    _validate_celsius(celsius)
    return celsius * 9 / 5 + 32

def _validate_celsius(celsius: float) -> None:
    if celsius < ABSOLUTE_ZERO_C:
        raise ValueError(f"温度不能低于绝对零度 ({ABSOLUTE_ZERO_C}°C)")

# ❌ 反面：断言实现细节——重构后测试全废
def test_process_order_bad(mocker):
    mock_validate = mocker.patch("src.order.validate_order")
    mock_save = mocker.patch("src.order.save_order")
    mock_notify = mocker.patch("src.order.send_notification")

    process_order({"id": 1, "items": []})

    mock_validate.assert_called_once()  # 如果改了内部流程，测试就废了
    mock_save.assert_called_once()
    mock_notify.assert_called_once()

# ✅ 正面：断言可观察行为——重构不影响测试
def test_process_order_good():
    """处理有效订单后，订单状态应为 completed"""
    order = create_test_order(items=[{"sku": "A1", "qty": 2}])
    result = process_order(order)
    assert result.status == "completed"
    assert result.total == 200.0

# ❌ 反面：共享可变状态
_shared_cart: list[str] = []

def test_add_item():
    _shared_cart.append("apple")
    assert "apple" in _shared_cart  # 依赖之前的状态

def test_remove_item():
    _shared_cart.remove("apple")  # 如果 test_add_item 没先跑，这里就报错
    assert "apple" not in _shared_cart

# ✅ 正面：每个测试获得独立数据
@pytest.fixture
def cart():
    return ["apple"]  # 每次都创建新的

def test_add_item(cart):
    cart.append("banana")
    assert "banana" in cart

def test_remove_item(cart):
    cart.remove("apple")
    assert "apple" not in cart

[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
    "-v",
    "--tb=short",
    "--strict-markers",
]
markers = [
    "slow: 标记慢测试（deselect with '-m \"not slow\"')",
    "integration: 标记集成测试",
    "e2e: 标记端到端测试",
]

[tool.coverage.run]
source = ["src"]
omit = ["tests/*"]

[tool.coverage.report]
fail_under = 80
show_missing = true

project/
├── src/
│   ├── __init__.py
│   ├── validator.py
│   └── service.py
├── tests/
│   ├── conftest.py          ← 共享 fixture
│   ├── test_validator.py    ← 对应 src/validator.py
│   └── test_service.py      ← 对应 src/service.py
└── pyproject.toml

# .github/workflows/test.yml
name: Tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -e ".[dev]"
      - run: pytest --cov=src --cov-report=xml
      - uses: codecov/codecov-action@v4

# unittest 风格
import unittest

class TestDiscount(unittest.TestCase):
    def test_apply_discount(self):
        self.assertEqual(apply_discount(100, 0.2), 80.0)

# 等价的 pytest 风格（更简洁）
def test_apply_discount():
    assert apply_discount(100, 0.2) == 80.0