测试数据与依赖管理：工厂模式、Mock 到 Testcontainers 的完整链路

┌─────────────────────────────────────────────────────────────┐
│                     测试数据管理困境                          │
├─────────────────────────────────────────────────────────────┤
│  环境漂移          数据污染           敏感泄露           维护成本 │
│    │                 │                  │                │   │
│    ▼                 ▼                  ▼                ▼   │
│  "在我机            测试 A            生产数据           线上 │
│   器能过"           改了用户          明文存在            DB  │
│                    ID，测试 B          测试环境           变更 │
│                    直接跪了                             同步  │
└─────────────────────────────────────────────────────────────┘

静态数据（固定值）     动态数据（每次生成）       合成数据（算法生成）
    │                     │                       │
    ▼                     ▼                       ▼
硬编码常量           faker / factory          基于原型的
（简单但脆弱）         （灵活但可控）            数据生成器
                                             （保持统计特征）

// Object Mother：集中管理所有测试对象的创建
public class ObjectMothers {

    public static User aStandardUser() {
        return User.builder()
            .id("user-001")
            .name("Alice")
            .email("alice@example.com")
            .role(Role.USER)
            .createdAt(LocalDateTime.of(2024, 1, 1, 0, 0))
            .build();
    }

    public static User anAdminUser() {
        return User.builder()
            .id("admin-001")
            .name("Bob")
            .email("bob@example.com")
            .role(Role.ADMIN)
            .build();
    }

    public static Order aPendingOrder() {
        return Order.builder()
            .id("order-001")
            .status(OrderStatus.PENDING)
            .items(List.of(
                OrderItem.builder().productId("prod-1").quantity(2).price(100).build()
            ))
            .total(200)
            .build();
    }
}

// 测试中使用
@Test
void shouldSendEmailWhenOrderPlaced() {
    User user = ObjectMothers.aStandardUser();
    Order order = ObjectMothers.aPendingOrder();
    // ...
}

// Builder：链式调用，灵活覆盖默认值
public class UserBuilder {
    private String id = UUID.randomUUID().toString();
    private String name = "Test User";
    private String email = "test@example.com";
    private Role role = Role.USER;

    public UserBuilder withId(String id) { this.id = id; return this; }
    public UserBuilder withName(String name) { this.name = name; return this; }
    public UserBuilder withRole(Role role) { this.role = role; return this; }
    public UserBuilder asAdmin() { this.role = Role.ADMIN; return this; }

    public User build() {
        return new User(id, name, email, role);
    }
}

// 测试中使用
@Test
void adminShouldAccessAdminPanel() {
    User admin = new UserBuilder().asAdmin().withName("Admin Bob").build();
    assertTrue(accessControl.canAccessAdminPanel(admin));
}

// 组合 Builder 创建复杂对象
@Test
void shouldApplyVIPDiscount() {
    User vip = new UserBuilder().withRole(Role.VIP).build();
    Order order = new OrderBuilder()
        .forUser(vip)
        .withItem("laptop", 1, 9999)
        .withItem("mouse", 2, 199)
        .build();

    assertEquals(9398, order.getTotal());  // VIP 95折
}

import factory
from factory import Faker
from models import User, Order, OrderItem

class UserFactory(factory.Factory):
    class Meta:
        model = User

    id = factory.Sequence(lambda n: f"user-{n:03d}")
    name = Faker("name")
    email = Faker("email")
    role = "user"
    created_at = factory.LazyFunction(datetime.now)

    # Trait：预设组合属性
    class Params:
        admin = factory.Trait(role="admin")
        vip = factory.Trait(role="vip")
        inactive = factory.Trait(is_active=False)

class OrderItemFactory(factory.Factory):
    class Meta:
        model = OrderItem

    product_id = factory.Sequence(lambda n: f"prod-{n}")
    quantity = 1
    price = factory.Faker("pyint", min_value=10, max_value=1000)

class OrderFactory(factory.Factory):
    class Meta:
        model = Order

    id = factory.Sequence(lambda n: f"order-{n:03d}")
    user = factory.SubFactory(UserFactory)
    status = "pending"

    @factory.post_generation
    def items(obj, create, extracted, **kwargs):
        if not create:
            return
        if extracted:
            obj.items = extracted
        else:
            obj.items = OrderItemFactory.build_batch(3)
        obj.calculate_total()

# 测试中使用
def test_vip_discount():
    vip_user = UserFactory(vip=True)
    order = OrderFactory(user=vip_user, items=[
        OrderItemFactory(product_id="laptop", price=9999, quantity=1),
    ])
    assert order.total == int(9999 * 0.95)

def test_admin_can_view_all_orders():
    admin = UserFactory(admin=True)
    orders = OrderFactory.create_batch(5)
    assert order_service.get_all_orders(admin) == 5

// Go 没有泛型工厂库，匿名 struct 是最地道的做法
func NewUser(overrides ...func(*User)) *User {
    u := &User{
        ID:        uuid.New().String(),
        Name:      "Test User",
        Email:     "test@example.com",
        Role:      RoleUser,
        CreatedAt: time.Now(),
    }
    for _, o := range overrides {
        o(u)
    }
    return u
}

// 使用
func TestAdminAccess(t *testing.T) {
    admin := NewUser(
        func(u *User) { u.Role = RoleAdmin },
        func(u *User) { u.Name = "Admin Bob" },
    )
    assert.True(t, CanAccessAdminPanel(admin))
}

// 更优雅的 Builder 风格
type UserOption func(*User)

func WithRole(r Role) UserOption {
    return func(u *User) { u.Role = r }
}

func WithName(name string) UserOption {
    return func(u *User) { u.Name = name }
}

func NewUserV2(opts ...UserOption) *User {
    u := &User{ /* defaults */ }
    for _, opt := range opts {
        opt(u)
    }
    return u
}

// 使用
vip := NewUserV2(WithRole(RoleVIP), WithName("Alice"))

# conftest.py
import pytest
from testcontainers.mysql import MySqlContainer

@pytest.fixture(scope="session")
def mysql():
    """整个测试会话共享一个 MySQL 容器"""
    with MySqlContainer("mysql:8.0") as mysql:
        yield mysql

@pytest.fixture(scope="function")
def db_session(mysql):
    """每个测试用例获取新的数据库 session"""
    engine = create_engine(mysql.get_connection_url())
    Session = sessionmaker(bind=engine)
    session = Session()

    # 创建表
    Base.metadata.create_all(engine)

    yield session

    # 清理
    session.rollback()
    session.close()
    Base.metadata.drop_all(engine)

@pytest.fixture
def sample_user(db_session):
    """每个测试自动有一个标准用户"""
    user = UserFactory()
    db_session.add(user)
    db_session.commit()
    return user

# 测试中使用 —— 自动获得干净的数据库和用户
def test_user_profile(sample_user, db_session):
    profile = get_profile(sample_user.id, db_session)
    assert profile.name == sample_user.name

@TestInstance(TestInstance.Lifecycle.PER_CLASS)  // 默认 PER_METHOD
class OrderRepositoryTest {

    private static MySQLContainer<?> mysql;
    private OrderRepository repository;

    @BeforeAll  // 整个类一次
    static void setUpDatabase() {
        mysql = new MySQLContainer<>("mysql:8.0");
        mysql.start();
    }

    @BeforeEach  // 每个测试方法
    void setUp() {
        repository = new OrderRepository(mysql.getJdbcUrl());
        TestDataSeeder.seed(repository);  // 插入基础数据
    }

    @AfterEach
    void tearDown() {
        repository.truncateAll();  // 清理数据
    }

    @AfterAll
    static void tearDownDatabase() {
        mysql.stop();
    }
}

// 使用 @Transactional 的回滚模式（Spring 测试）
@SpringBootTest
@Transactional  // 每个测试后自动回滚
class ServiceTest {
    @Autowired OrderRepository repo;

    @Test
    void testOrderCreation() {
        repo.save(new Order("ORD-001", 100));
        assertEquals(1, repo.count());
        // 测试结束后自动回滚，数据库恢复干净
    }
}

func TestMain(m *testing.M) {
    // 测试开始前：启动共享资源
    mysql := testcontainers.NewMySQL("mysql:8.0")
    mysql.Start()

    // 设置全局变量或环境变量
    os.Setenv("TEST_DB_URL", mysql.GetConnectionURL())

    // 运行所有测试
    code := m.Run()

    // 测试结束后：清理
    mysql.Stop()

    os.Exit(code)
}

// 每个测试的清理
type CleanupFunc func()

func setupTestDB(t *testing.T) (*sql.DB, CleanupFunc) {
    db, _ := sql.Open("mysql", os.Getenv("TEST_DB_URL"))
    db.Exec("CREATE TABLE IF NOT EXISTS orders ...")

    cleanup := func() {
        db.Exec("TRUNCATE TABLE orders")
        db.Close()
    }

    t.Cleanup(cleanup)  // 测试结束时自动执行
    return db, cleanup
}

from unittest.mock import Mock, patch, MagicMock
from datetime import datetime

def test_send_notification():
    notifier = EmailNotifier()

    # Mock SMTP 连接
    with patch('notifications.smtplib.SMTP') as mock_smtp:
        mock_server = MagicMock()
        mock_smtp.return_value.__enter__.return_value = mock_server

        notifier.send("user@example.com", "Hello")

        # 验证调用
        mock_server.send_message.assert_called_once()
        call_args = mock_server.send_message.call_args[0][0]
        assert call_args['To'] == "user@example.com"
        assert "Hello" in call_args['Subject']

def test_fetch_user_with_mocked_api():
    with patch('users.client.requests.get') as mock_get:
        mock_get.return_value.status_code = 200
        mock_get.return_value.json.return_value = {
            "id": "123",
            "name": "Alice"
        }

        user = fetch_user("123")
        assert user.name == "Alice"

        # 验证 URL 是否正确构造
        mock_get.assert_called_with("https://api.example.com/users/123", timeout=5)

@ExtendWith(MockitoExtension.class)
class OrderServiceTest {

    @Mock private PaymentGateway paymentGateway;
    @Mock private InventoryService inventoryService;
    @InjectMocks private OrderService orderService;

    @Test
    @DisplayName("库存不足时应拒绝下单")
    void shouldRejectWhenInventoryInsufficient() {
        when(inventoryService.checkStock("SKU-001", 10))
            .thenReturn(false);

        var result = orderService.placeOrder("SKU-001", 10);

        assertFalse(result.isSuccess());
        assertEquals("INSUFFICIENT_STOCK", result.getErrorCode());
        verify(paymentGateway, never()).charge(any(), any());
    }

    @Test
    @DisplayName("支付失败时应释放库存")
    void shouldReleaseStockWhenPaymentFails() {
        when(inventoryService.checkStock("SKU-001", 10)).thenReturn(true);
        when(inventoryService.reserve("SKU-001", 10)).thenReturn("RES-001");
        when(paymentGateway.charge(eq(1000.0), any()))
            .thenThrow(new PaymentException("DECLINED"));

        assertThrows(PaymentException.class, () ->
            orderService.placeOrder("SKU-001", 10)
        );

        verify(inventoryService).releaseReservation("RES-001");
    }

    @Test
    @DisplayName("应该使用正确的金额调用支付网关")
    void shouldChargeCorrectAmount() {
        when(inventoryService.checkStock(any(), anyInt())).thenReturn(true);
        when(inventoryService.reserve(any(), anyInt())).thenReturn("RES-001");
        when(paymentGateway.charge(anyDouble(), any()))
            .thenReturn("CHARGE-001");

        orderService.placeOrder("SKU-001", 5);  // 单价 200

        // 验证金额 = 5 * 200 = 1000
        ArgumentCaptor<Double> amountCaptor = ArgumentCaptor.forClass(Double.class);
        verify(paymentGateway).charge(amountCaptor.capture(), any());
        assertEquals(1000.0, amountCaptor.getValue());
    }
}

传统替代方案 vs Testcontainers

┌─────────────┐    ┌─────────────┐
│  H2 内存DB   │    │  Testcontainers│
│  快但不真实   │ vs │  真实 MySQL    │
│  语法不兼容   │    │  完全一致     │
│  不支持特性   │    │  稍慢但可靠   │
└─────────────┘    └─────────────┘

@Testcontainers
class OrderRepositoryIntegrationTest {

    @Container
    static MySQLContainer<?> mysql = new MySQLContainer<>("mysql:8.0")
        .withDatabaseName("test_db")
        .withUsername("test")
        .withPassword("test");

    @DynamicPropertySource
    static void configureProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.datasource.url", mysql::getJdbcUrl);
        registry.add("spring.datasource.username", mysql::getUsername);
        registry.add("spring.datasource.password", mysql::getPassword);
    }

    @Autowired OrderRepository repository;

    @Test
    @DisplayName("应能插入并查询订单")
    void shouldInsertAndFindOrder() {
        Order order = new Order("ORD-001", "USER-1", 9999);
        repository.save(order);

        Optional<Order> found = repository.findById("ORD-001");
        assertTrue(found.isPresent());
        assertEquals(9999, found.get().getAmount());
    }

    @Test
    @DisplayName("复杂查询应正确使用 JSON 字段")
    void shouldQueryJsonField() {
        // 这是 H2 无法测试的场景！
        repository.saveWithMetadata("ORD-002", Map.of("source", "mobile", "campaign", "summer"));

        List<Order> mobileOrders = repository.findByMetadataSource("mobile");
        assertEquals(1, mobileOrders.size());
    }
}

// 多个容器组合
@Testcontainers
class FullStackTest {
    @Container
    static MySQLContainer<?> mysql = new MySQLContainer<>("mysql:8.0");

    @Container
    static GenericContainer<?> redis = new GenericContainer<>("redis:7-alpine")
        .withExposedPorts(6379);

    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.5.0")
    );
}

import pytest
from testcontainers.mysql import MySqlContainer
from testcontainers.redis import RedisContainer
from sqlalchemy import create_engine, text

@pytest.fixture(scope="session")
def mysql_engine():
    with MySqlContainer("mysql:8.0") as mysql:
        engine = create_engine(mysql.get_connection_url())
        yield engine

@pytest.fixture
def db_session(mysql_engine):
    from sqlalchemy.orm import sessionmaker
    Session = sessionmaker(bind=mysql_engine)
    session = Session()

    # 执行迁移
    with open("migrations/001_init.sql") as f:
        session.execute(text(f.read()))
    session.commit()

    yield session

    # 清理
    session.execute(text("SET FOREIGN_KEY_CHECKS=0"))
    for table in ["orders", "users", "products"]:
        session.execute(text(f"TRUNCATE TABLE {table}"))
    session.execute(text("SET FOREIGN_KEY_CHECKS=1"))
    session.commit()
    session.close()

# 缓存测试
def test_redis_cache():
    with RedisContainer("redis:7-alpine") as redis:
        import redis as redis_lib
        client = redis_lib.Redis(host=redis.get_container_host_ip(),
                                  port=redis.get_exposed_port(6379))

        client.set("key", "value", ex=300)
        assert client.get("key") == b"value"

@SpringBootTest
@TestPropertySource(properties = "spring.flyway.clean-disabled=false")
class MigrationTest {
    @Autowired Flyway flyway;

    @BeforeEach
    void clean() {
        flyway.clean();      // 清空数据库
        flyway.migrate();    // 重新执行所有迁移
    }

    @Test
    void migrationV3_shouldAddIndexOnOrders() {
        // 验证 V3 迁移正确执行
        List<Map<String, Object>> indexes = jdbcTemplate.queryForList(
            "SHOW INDEX FROM orders"
        );
        assertTrue(indexes.stream()
            .anyMatch(i -> "idx_orders_user_id".equals(i.get("Key_name"))));
    }
}

微服务 A          API 契约            微服务 B
───────       ────────────       ─────────
Consumer  ──►  "GET /users/:id   Provider
  消费        返回 {id, name}"    提供

问题：B 改了字段名，A 的集成测试只能在部署后才发现
解决：契约测试在代码提交时就验证双方一致性

# consumer_test.py —— 消费者端定义期望
import pytest
from pact import Consumer, Provider

@pytest.fixture
def pact():
    return Consumer("order-service").has_pact_with(
        Provider("user-service"),
        pact_dir="./pacts"
    )

def test_get_user_by_id(pact):
    expected = {"id": "123", "name": "Alice", "email": "alice@example.com"}

    (pact
     .given("user with id 123 exists")
     .upon_receiving("a request for user 123")
     .with_request("GET", "/users/123")
     .will_respond_with(200, body=expected))

    with pact:
        user = user_client.get_user("123")
        assert user.name == "Alice"

# 生成的 pact 文件：order-service-user-service.json
# 提交到 Pact Broker，Provider 端验证

// ProviderTest.java —— 提供者端验证契约
@PactTestFor(providerName = "user-service")
class UserServicePactTest {

    @Pact(consumer = "order-service")
    RequestResponsePact getUserPact(PactDslWithProvider builder) {
        return builder
            .given("user with id 123 exists")
            .uponReceiving("a request for user 123")
            .path("/users/123")
            .method("GET")
            .willRespondWith()
            .status(200)
            .body(new PactDslJsonBody()
                .stringType("id", "123")
                .stringType("name", "Alice")
                .stringType("email", "alice@example.com"))
            .toPact();
    }

    @Test
    @PactTestFor(pactMethod = "getUserPact")
    void shouldReturnUserDetails(MockServer mockServer) {
        var user = userServiceClient
            .baseUrl(mockServer.getUrl())
            .getUser("123");
        assertNotNull(user);
    }
}

from faker import Faker

fake = Faker("zh_CN")  # 中文本地化

def generate_test_user():
    return {
        "id": fake.uuid4(),
        "name": fake.name(),
        "phone": fake.phone_number(),
        "id_card": fake.ssn(min_age=18, max_age=65),  # 合成身份证号
        "address": fake.address(),
        "company": fake.company(),
        "email": fake.email(),
        "created_at": fake.date_time_between(start_date="-2y"),
    }

# 生成 1000 条测试数据
users = [generate_test_user() for _ in range(1000)]