#FastAPI Pytest单元测试完全指南

# 无测试 vs 有测试的开发体验
"""
无测试场景：
- 修改代码 → 手动测试 → 上线 → 发现bug → 回滚 → 用户投诉 😱
- 重构代码 → 担心破坏功能 → 代码腐化 → 技术债务堆积

有测试场景：
- 修改代码 → 运行测试 → 快速反馈 → 自信上线 🚀
- 重构代码 → 测试保证 → 持续优化 → 代码质量提升
"""

# tests/conftest.py - Pytest配置文件
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from fastapi.testclient import TestClient

from app.main import app
from app.database import get_db, Base

TEST_DATABASE_URL = "sqlite:///./test.db"

@pytest.fixture(scope="function")
def test_engine():
    """创建测试数据库引擎"""
    engine = create_engine(
        TEST_DATABASE_URL,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    yield engine
    Base.metadata.drop_all(bind=engine)
    engine.dispose()

@pytest.fixture(scope="function")
def test_session(test_engine):
    """创建测试数据库会话"""
    Session = sessionmaker(bind=test_engine)
    session = Session()
    yield session
    session.rollback()
    session.close()

@pytest.fixture(scope="function")
def client(test_session):
    """同步测试客户端"""
    def override_get_db():
        try:
            yield test_session
        finally:
            pass
    
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.clear()

@pytest.fixture(scope="function")
async def async_client(test_session):
    """异步测试客户端"""
    def override_get_db():
        try:
            yield test_session
        finally:
            pass
    
    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()

# tests/test_basic.py
import pytest
from fastapi.testclient import TestClient
from app.main import app

def test_root_endpoint(client):
    """测试根端点"""
    response = client.get("/")
    assert response.status_code == 200
    assert "message" in response.json()

def test_post_request(client):
    """测试POST请求"""
    test_data = {"name": "Test User", "email": "test@example.com"}
    response = client.post("/users/", json=test_data)
    assert response.status_code == 201
    assert response.json()["name"] == "Test User"

def test_error_handling(client):
    """测试错误处理"""
    response = client.get("/nonexistent-endpoint")
    assert response.status_code == 404

# tests/test_async.py
import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_async_endpoint(async_client):
    """异步端点测试"""
    response = await async_client.get("/")
    assert response.status_code == 200
    assert "message" in response.json()

@pytest.mark.asyncio
async def test_async_post(async_client):
    """异步POST请求测试"""
    data = {"name": "Async User", "email": "async@example.com"}
    response = await async_client.post("/users/", json=data)
    assert response.status_code == 201

# tests/test_fixtures.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from app.services.user_service import UserService

@pytest.fixture
def mock_user_service():
    """模拟用户服务"""
    mock = AsyncMock(spec=UserService)
    mock.get_user_by_id.return_value = {
        "id": 1,
        "email": "mock@example.com",
        "name": "Mock User"
    }
    return mock

@pytest.fixture
def sample_user_data():
    """样本用户数据"""
    return {
        "email": "test@example.com",
        "password": "TestPassword123!",
        "name": "Test User"
    }

def test_with_mocked_service(mock_user_service):
    """使用模拟服务的测试"""
    user = mock_user_service.get_user_by_id(1)
    assert user["email"] == "mock@example.com"

def test_with_sample_data(client, sample_user_data):
    """使用样本数据的测试"""
    response = client.post("/users/", json=sample_user_data)
    assert response.status_code == 201
    assert response.json()["email"] == sample_user_data["email"]

# tests/test_database.py
import pytest
from sqlalchemy import text
from app.models import User
from app.schemas import UserCreate

class TestDatabase:
    """数据库测试"""
    
    async def test_create_user(self, test_session):
        """测试创建用户"""
        from app.crud import create_user
        
        user_data = UserCreate(
            email="test@example.com",
            password="Password123!",
            name="Test User"
        )
        
        user = create_user(test_session, user_data)
        assert user.email == user_data.email
        
        result = test_session.execute(
            text("SELECT COUNT(*) FROM users WHERE email = :email"),
            {"email": user_data.email}
        )
        count = result.scalar()
        assert count == 1

# tests/test_user_api.py
import pytest

class TestUserAPI:
    """用户API测试"""
    
    @pytest.mark.asyncio
    async def test_create_user(self, async_client):
        """测试创建用户"""
        user_data = {
            "email": "newuser@example.com",
            "password": "SecurePassword123!",
            "name": "New User"
        }
        
        response = await async_client.post("/users/", json=user_data)
        assert response.status_code == 201
        
        data = response.json()
        assert data["email"] == user_data["email"]
        assert data["name"] == user_data["name"]
    
    @pytest.mark.asyncio
    async def test_get_user(self, async_client):
        """测试获取用户"""
        user_data = {
            "email": "getuser@example.com",
            "password": "Password123!",
            "name": "Get User"
        }
        create_response = await async_client.post("/users/", json=user_data)
        user_id = create_response.json()["id"]
        
        response = await async_client.get(f"/users/{user_id}")
        assert response.status_code == 200
        assert response.json()["id"] == user_id

# tests/test_parameterized.py
import pytest

class TestParameterized:
    """参数化测试"""
    
    @pytest.mark.parametrize("email,is_valid", [
        ("valid@example.com", True),
        ("user.name@example.com", True),
        ("invalid", False),
        ("@example.com", False),
    ])
    @pytest.mark.asyncio
    async def test_email_validation(self, async_client, email, is_valid):
        """参数化测试邮箱验证"""
        user_data = {
            "email": email,
            "password": "Password123!",
            "name": "Test User"
        }
        
        response = await async_client.post("/users/", json=user_data)
        
        if is_valid:
            assert response.status_code in [201, 409]
        else:
            assert response.status_code == 422

"""
TDD (Test-Driven Development) 开发实践：

1. RED - 编写失败的测试
2. GREEN - 编写最少的代码使测试通过
3. REFACTOR - 重构代码，保持测试通过
"""

# 1. RED: 先写测试
class TestShoppingCartTDD:
    """购物车功能的TDD开发示例"""
    
    def test_cart_starts_empty(self):
        """测试购物车初始化时空的 - RED"""
        cart = ShoppingCart()
        assert len(cart.items) == 0
        assert cart.total_price == 0

# 2. GREEN: 实现最简单的代码使测试通过
class ShoppingCart:
    def __init__(self):
        self.items = []
    
    @property
    def total_price(self):
        return sum(item["price"] * item["quantity"] for item in self.items)
    
    def add_item(self, item):
        self.items.append(item)

# 3. REFACTOR: 重构代码
class ShoppingCart:
    def __init__(self):
        self._items = {}
    
    def add_item(self, item):
        item_id = item["id"]
        if item_id in self._items:
            self._items[item_id]["quantity"] += item["quantity"]
        else:
            self._items[item_id] = item.copy()
    
    @property
    def items(self):
        return list(self._items.values())
    
    @property
    def total_price(self):
        return sum(item["price"] * item["quantity"] 
                  for item in self._items.values())