测试金字塔：多语言混合平台的质量保障体系

“

系列：S2 架构全景 · 第 14 篇 | 难度：中级 | 阅读时间：18 分钟

TL;DR

coomia-dip 是一个 Java + Python 混合技术栈的平台，测试策略必须同时覆盖两种语言的单元测试、集成测试和端到端测试。测试金字塔确保 80% 的测试在毫秒级完成，20% 的高价值测试验证跨 Layer 集成。
gRPC 接口是平台的内部通信骨干，也是测试的核心边界。每个 Layer 的 gRPC 服务必须有独立的契约测试，确保跨语言（Java ↔ Python）的接口兼容性。
场景化测试（Scenario Testing）是 coomia-dip 质量保障的特色方法论。它不是测试单个函数或接口，而是模拟完整的业务场景（如"创建对象 → 触发派生属性 → 验证级联更新"），验证多个 Layer 协同工作的正确性。

#1. 引言：为什么传统测试策略不够用

#1.1 多语言平台的测试挑战

coomia-dip 不是一个单一技术栈的项目。它的 8 个 Layer 跨越两种主要语言：

Code

Java Layers:
  Control Layer (Control Layer)     → Spring Boot 3.x, Java 21
  Data Layer (Data Layer)        → Quarkus 3.x, Java 21

Python Layers:
  Reasoning & Decision Layer (Reasoning)         → FastAPI, Python 3.x
  Agent Runtime Layer (Agent Runtime)     → FastAPI, Temporal, Python 3.x

跨语言接口:
  SDK & Developer Experience Layer (SDK)               → Python SDK + TypeScript SDK
  Deployment & Operations Layer (Deployment)        → Docker Compose, Python 脚本

传统的测试金字塔假设测试对象是同构的——所有代码使用同一种语言、同一个构建工具、同一个测试框架。但在 coomia-dip 中：

Java 测试用 JUnit 5 + Mockito + Testcontainers
Python 测试用 pytest + unittest.mock
跨语言集成测试需要同时启动 Java 和 Python 服务
gRPC 接口需要跨语言兼容性验证

#1.2 coomia-dip 的测试金字塔

我们对经典测试金字塔进行了适配：

Code

                    ╱╲
                   ╱  ╲
                  ╱ E2E╲         ← 5% | 场景化端到端测试
                 ╱──────╲          跨 Layer 全链路验证
                ╱ 集成测试╲      ← 15% | gRPC 契约测试 + DB 集成
               ╱──────────╲        单 Layer 内部集成
              ╱  单元测试    ╲   ← 80% | 领域逻辑、工具函数
             ╱────────────────╲    纯函数、无外部依赖

#2. 第一层：单元测试

#2.1 Java 单元测试（Control Layer + Data Layer）

Java Layer 的单元测试使用 JUnit 5 + Mockito：

Java

// Control Layer 单元测试示例
@ExtendWith(MockitoExtension.class)
class OntologyTypeServiceTest {

    @Mock
    private OntologyTypeRepository repository;

    @Mock
    private EventPublisher eventPublisher;

    @InjectMocks
    private OntologyTypeService service;

    @Test
    @DisplayName("创建 ObjectType 时应验证属性约束")
    void shouldValidatePropertyConstraints() {
        // Given
        var request = CreateObjectTypeRequest.newBuilder()
            .setName("Supplier")
            .addProperties(PropertyDef.newBuilder()
                .setName("name")
                .setType(PropertyType.STRING)
                .setRequired(true)
                .build())
            .build();

        when(repository.existsByName("Supplier")).thenReturn(false);

        // When
        var result = service.createObjectType(request);

        // Then
        assertThat(result.getName()).isEqualTo("Supplier");
        verify(eventPublisher).publish(any(ObjectTypeCreatedEvent.class));
    }

    @Test
    @DisplayName("重复名称应抛出异常")
    void shouldRejectDuplicateName() {
        var request = CreateObjectTypeRequest.newBuilder()
            .setName("ExistingType")
            .build();

        when(repository.existsByName("ExistingType")).thenReturn(true);

        assertThrows(DuplicateTypeException.class,
            () -> service.createObjectType(request));
    }
}

#2.2 Python 单元测试（Reasoning & Decision Layer + Agent Runtime Layer）

Python Layer 的单元测试使用 pytest：

Python

# Intelligence Layer 单元测试示例
import pytest
from unittest.mock import AsyncMock, patch
from decimal import Decimal

from reasoning_engine.core.rule_evaluator import RuleEvaluator
from reasoning_engine.models.rule import Rule, Condition, Action


class TestRuleEvaluator:
    """规则评估器单元测试"""

    @pytest.fixture
    def evaluator(self):
        return RuleEvaluator()

    @pytest.fixture
    def simple_rule(self):
        return Rule(
            name="high_risk_supplier",
            conditions=[
                Condition(field="risk_score", operator=">=", value=0.8),
                Condition(field="active", operator="==", value=True),
            ],
            actions=[
                Action(type="set_label", params={"label": "HIGH_RISK"}),
            ],
        )

    def test_rule_matches_when_all_conditions_met(
        self, evaluator, simple_rule
    ):
        context = {"risk_score": 0.85, "active": True}
        result = evaluator.evaluate(simple_rule, context)

        assert result.matched is True
        assert result.actions[0].type == "set_label"

    def test_rule_not_matched_when_condition_fails(
        self, evaluator, simple_rule
    ):
        context = {"risk_score": 0.5, "active": True}
        result = evaluator.evaluate(simple_rule, context)

        assert result.matched is False
        assert result.actions == []

    def test_missing_field_raises_evaluation_error(
        self, evaluator, simple_rule
    ):
        context = {"active": True}  # missing risk_score

        with pytest.raises(EvaluationError, match="Missing field: risk_score"):
            evaluator.evaluate(simple_rule, context)

#2.3 单元测试规范

规则	Java	Python
覆盖率要求	>= 80%	>= 80%
命名规范	`shouldXxxWhenYyy`	`test_xxx_when_yyy`
Mock 框架	Mockito	unittest.mock / pytest-mock
断言库	AssertJ	pytest 原生 assert
测试数据	Builder 模式	pytest.fixture
执行时间限制	单个 < 100ms	单个 < 100ms

#2.4 禁止行为

项目 CLAUDE.md 明确列出了测试相关的技术红线：

Code

❌ 删除失败的测试来"通过"测试
❌ 空的 catch 块 catch(e) {}

这两条红线的共同目标是禁止掩盖问题。如果测试失败，正确的做法是修复代码或更新测试预期——绝不是删除测试。空的 catch 块同理：吞掉异常会导致调试时缺少关键信息。

#3. 第二层：集成测试

#3.1 gRPC 契约测试

gRPC 接口是 coomia-dip 各 Layer 之间的通信协议。契约测试确保：

Protobuf 消息定义在生产者和消费者之间保持兼容
新增字段不会破坏已有消费者
删除字段会在编译期被发现

Python

# gRPC 契约测试示例 (Python SDK 调用 Java Control Layer)
import pytest
import grpc
from ontology_sdk.grpc_client import OntologyClient


class TestOntologyGrpcContract:
    """验证 Python SDK 与 Control Layer gRPC 接口的兼容性"""

    @pytest.fixture
    def client(self, grpc_channel):
        return OntologyClient(channel=grpc_channel)

    def test_create_object_type_contract(self, client):
        """验证 CreateObjectType 请求/响应格式"""
        response = client.create_object_type(
            name="TestType",
            properties=[
                {"name": "id", "type": "STRING", "required": True},
                {"name": "value", "type": "DECIMAL", "required": False},
            ],
        )

        # 验证响应结构符合 proto 定义
        assert hasattr(response, "type_id")
        assert hasattr(response, "name")
        assert hasattr(response, "version")
        assert response.name == "TestType"

    def test_backward_compatibility(self, client):
        """验证旧版本请求仍然有效"""
        # 不带可选字段的请求应该成功
        response = client.create_object_type(
            name="MinimalType",
            properties=[],  # 空属性列表
        )
        assert response.type_id is not None

#3.2 数据库集成测试

Java Layer 使用 Testcontainers 进行数据库集成测试：

Java

@Testcontainers
@SpringBootTest
class OntologyRepositoryIntegrationTest {

    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16")
        .withDatabaseName("ontology_test")
        .withUsername("test")
        .withPassword("test");

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

    @Autowired
    private OntologyTypeRepository repository;

    @Test
    @DisplayName("保存和查询 ObjectType 应保持数据完整")
    void shouldPersistAndRetrieveObjectType() {
        // Given
        var objectType = new ObjectTypeEntity();
        objectType.setName("Order");
        objectType.setVersion(1);

        // When
        var saved = repository.save(objectType);
        var retrieved = repository.findById(saved.getId());

        // Then
        assertThat(retrieved).isPresent();
        assertThat(retrieved.get().getName()).isEqualTo("Order");
    }
}

#3.3 Kafka 集成测试

事件驱动架构需要验证 Kafka 消息的正确性：

Java

@EmbeddedKafka(topics = {"ontology.cdc.changes"})
@SpringBootTest
class EventPublishingIntegrationTest {

    @Autowired
    private EventPublisher publisher;

    @Autowired
    private KafkaTemplate<String, byte[]> kafkaTemplate;

    @Test
    @DisplayName("对象变更应发布 CDC 事件到正确的 Topic")
    void shouldPublishCdcEventOnObjectChange() throws Exception {
        // Given
        var event = ObjectChangedEvent.newBuilder()
            .setObjectId("obj-001")
            .setObjectType("Supplier")
            .setChangeType(ChangeType.UPDATED)
            .build();

        // When
        publisher.publish(event);

        // Then
        var records = KafkaTestUtils.getRecords(consumer, Duration.ofSeconds(5));
        assertThat(records).hasSize(1);

        var published = ObjectChangedEvent.parseFrom(
            records.iterator().next().value()
        );
        assertThat(published.getObjectId()).isEqualTo("obj-001");
        assertThat(published.getChangeType()).isEqualTo(ChangeType.UPDATED);
    }
}

#4. 第三层：端到端测试

#4.1 场景化测试方法论

coomia-dip 的 E2E 测试采用场景化测试方法论（详见 /scenario-testing skill）。核心理念是：测试完整的业务场景，而不是孤立的接口调用。

一个典型的场景测试：

Code

场景: 创建供应商并验证风险评分自动计算

前置条件:
  - Ontology 已定义 "Supplier" ObjectType
  - "risk_score" 属性已配置为 DerivedProperty
  - 计算规则已注册到 Intelligence Layer

步骤:
  1. 通过 SDK 创建 Supplier 对象 (name="Acme Corp")
  2. 设置基础属性 (delivery_rate=0.75, complaint_count=3)
  3. 等待派生属性计算完成 (最长 5 秒)
  4. 查询 risk_score 属性值

预期结果:
  - Supplier 对象创建成功
  - risk_score 已自动计算
  - risk_score 值在 [0, 1] 范围内
  - 审计日志记录了完整的操作链

#4.2 Python SDK 场景测试实现

Python

# tests/scenarios/test_derived_property_lifecycle.py
import pytest
import time
from ontology_sdk import OntoPlatform


class TestDerivedPropertyLifecycle:
    """派生属性全生命周期场景测试"""

    @pytest.fixture
    def platform(self):
        return OntoPlatform(
            host="localhost",
            port=8080,
            tenant_id="test-tenant",
        )

    def test_create_object_triggers_derived_computation(self, platform):
        """场景：创建对象后派生属性自动计算"""
        # Step 1: 创建对象
        supplier = platform.ontology.create_object(
            type_name="Supplier",
            properties={
                "name": "Acme Corp",
                "delivery_rate": 0.75,
                "complaint_count": 3,
            },
        )
        assert supplier.object_id is not None

        # Step 2: 等待派生属性计算（轮询）
        max_wait = 5.0
        interval = 0.5
        elapsed = 0.0
        risk_score = None

        while elapsed < max_wait:
            obj = platform.ontology.get_object(
                type_name="Supplier",
                object_id=supplier.object_id,
                include_derived=True,
            )
            risk_score = obj.properties.get("risk_score")
            if risk_score is not None:
                break
            time.sleep(interval)
            elapsed += interval

        # Step 3: 验证结果
        assert risk_score is not None, (
            f"Derived property not computed within {max_wait}s"
        )
        assert 0.0 <= float(risk_score) <= 1.0

        # Step 4: 验证审计日志
        audit_logs = platform.audit.query(
            object_id=supplier.object_id,
            limit=10,
        )
        action_types = [log.action_type for log in audit_logs]
        assert "OBJECT_CREATED" in action_types
        assert "DERIVED_PROPERTY_COMPUTED" in action_types

    def test_update_triggers_cascade_recomputation(self, platform):
        """场景：更新源属性后派生属性级联重算"""
        # 创建并等待初始计算
        supplier = platform.ontology.create_object(
            type_name="Supplier",
            properties={"name": "Beta Inc", "delivery_rate": 0.9},
        )
        time.sleep(2)

        initial_score = platform.ontology.get_property(
            "Supplier", supplier.object_id, "risk_score"
        )

        # 更新源属性
        platform.ontology.update_object(
            type_name="Supplier",
            object_id=supplier.object_id,
            properties={"delivery_rate": 0.5},  # 大幅下降
        )
        time.sleep(2)

        # 验证派生属性已重新计算
        updated_score = platform.ontology.get_property(
            "Supplier", supplier.object_id, "risk_score"
        )

        assert updated_score != initial_score, (
            "Derived property should be recomputed after source change"
        )

#4.3 跨 Layer 测试编排

E2E 测试需要多个 Layer 同时运行。coomia-dip 使用 Docker Compose 编排测试环境：

YAML

# docker-compose.test.yml
services:
  control-Layer:
    build: ./control-Layer
    ports: ["8080:8080", "9090:9090"]
    depends_on:
      - postgres
      - kafka

  intelligence-Layer:
    build: ./intelligence-Layer
    ports: ["8081:8081", "9091:9091"]
    depends_on:
      - control-Layer
      - kafka

  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: ontology_test

  kafka:
    image: confluentinc/cp-kafka:7.5.0

  redis:
    image: redis:7

#5. 跨语言测试的特殊挑战

#5.1 Protobuf 版本兼容

Java 和 Python 共享同一套 .proto 文件，但编译产物需要保持同步：

Code

proto/
├── ontology_service.proto    ← 单一来源（Single Source of Truth）
├── action_service.proto
├── reasoning_service.proto
└── event_types.proto

编译流程:
  proto/ ──protoc──> control-Layer/src/gen/   (Java 生成代码)
  proto/ ──protoc──> python-sdk/ontology_sdk/proto/  (Python 生成代码)

风险: 如果 Java 侧更新了 proto 文件但忘记重新编译 Python 侧，就会出现运行时不兼容。

解决方案: CI 流水线在每次 proto 文件变更时自动触发双语言编译和契约测试。

#5.2 数据类型映射验证

Java 和 Python 的基本类型不完全一致，需要额外验证：

Protobuf 类型	Java 类型	Python 类型	需验证
int64	long	int	大数边界
double	double	float	精度损失
bytes	ByteString	bytes	编码格式
Timestamp	Instant	datetime	时区处理
Decimal (自定义)	BigDecimal	Decimal	精度和舍入

Python

# 跨语言类型兼容性测试
class TestCrossLanguageTypeCompatibility:

    def test_large_int64_roundtrip(self, grpc_client):
        """验证 int64 大数在 Java ↔ Python 间往返不丢精度"""
        large_value = 2**53 - 1  # JavaScript 安全整数上限
        response = grpc_client.echo_int64(large_value)
        assert response.value == large_value

    def test_decimal_precision(self, grpc_client):
        """验证 Decimal 在跨语言传输中保持精度"""
        from decimal import Decimal
        value = Decimal("123456789.123456789")
        response = grpc_client.echo_decimal(str(value))
        assert Decimal(response.value) == value

    def test_timestamp_timezone(self, grpc_client):
        """验证 Timestamp 在跨语言传输中时区一致"""
        from datetime import datetime, timezone
        now = datetime.now(timezone.utc)
        response = grpc_client.echo_timestamp(now)
        # 允许 1 秒误差（序列化精度）
        assert abs((response.value - now).total_seconds()) < 1.0

#6. 测试数据管理

#6.1 测试数据策略

coomia-dip 测试使用三种数据策略：

策略	适用层级	工具
Builder/Factory	单元测试	Java Builder / Python fixture
种子数据	集成测试	SQL 脚本 / Flyway migration
快照数据	E2E 测试	Docker volume snapshot

#6.2 测试数据隔离

多租户架构让测试数据隔离变得简单——每个测试用例使用独立的 tenant_id：

Python

@pytest.fixture
def isolated_tenant():
    """为每个测试用例创建独立租户"""
    tenant_id = f"test-{uuid.uuid4().hex[:8]}"
    yield tenant_id
    # 测试后清理（由 Testcontainers 自动处理）

#6.3 敏感数据处理

测试数据中绝不包含真实的用户数据或凭证。项目规范明确：

Code

❌ 直接 push 共享文件到 develop（必须走 PR）

这条规则也适用于测试数据——包含测试凭证的文件必须在 .gitignore 中排除，绝不提交到版本控制。

#7. 持续集成中的测试编排

#7.1 CI 流水线结构

Code

┌──────────────────────────────────────────────────────┐
│                    CI Pipeline                       │
│                                                      │
│  Stage 1: 静态检查 (并行)                              │
│  ├─ Java: gradle build (编译 + 单元测试)              │
│  ├─ Python: ruff check + black --check + mypy         │
│  └─ Proto: protoc --lint                              │
│                                                      │
│  Stage 2: 单元测试 (并行)                              │
│  ├─ Java: gradle test (JUnit 5)                       │
│  └─ Python: pytest -m "not integration"               │
│                                                      │
│  Stage 3: 集成测试 (并行)                              │
│  ├─ Java: gradle integrationTest (Testcontainers)     │
│  ├─ Python: pytest -m integration                     │
│  └─ gRPC: 契约测试                                    │
│                                                      │
│  Stage 4: E2E 测试 (串行)                              │
│  ├─ docker compose up -d                              │
│  ├─ pytest tests/scenarios/                           │
│  └─ docker compose down                              │
│                                                      │
│  Stage 5: 质量门                                      │
│  ├─ 覆盖率 >= 80%                                     │
│  ├─ 0 个 Critical 漏洞                                │
│  └─ 性能基准无退化                                     │
└──────────────────────────────────────────────────────┘

#7.2 测试分层执行策略

层级	执行频率	最大耗时	失败影响
单元测试	每次提交	2 分钟	阻塞合并
集成测试	每次 PR	10 分钟	阻塞合并
E2E 测试	每日 / PR 合并后	30 分钟	通知团队
性能测试	每周 / 发版前	2 小时	通知团队

#8. 测试覆盖率策略

#8.1 覆盖率目标

coomia-dip 项目规定所有模块覆盖率 >= 80%，但不同类型的代码有不同的侧重：

代码类型	目标覆盖率	重点测试内容
领域逻辑	>= 90%	边界条件、异常路径
gRPC Service	>= 85%	请求验证、错误处理
Repository	>= 70%	查询正确性（集成测试）
配置类	>= 50%	默认值、环境变量
生成代码 (proto)	不计入	由契约测试覆盖

#8.2 覆盖率工具

语言	工具	报告格式
Java	JaCoCo	HTML + XML
Python	coverage.py + pytest-cov	HTML + XML

#8.3 有意义的覆盖率

覆盖率数字本身不是目标——有意义的测试才是。以下是反模式：

Python

# ❌ 反模式：提高覆盖率但无断言
def test_create_object_runs_without_error():
    service.create_object({"name": "test"})
    # 没有任何 assert —— 这个测试毫无价值

# ✅ 正确做法：有明确的断言和预期
def test_create_object_returns_valid_id():
    result = service.create_object({"name": "test"})
    assert result.object_id is not None
    assert len(result.object_id) == 36  # UUID 格式

#9. 性能测试

#9.1 性能基准测试

coomia-dip 维护关键操作的性能基准：

操作	P50 基准	P99 基准	测试工具
创建 ObjectType	< 10ms	< 50ms	JMH (Java)
查询单个对象	< 5ms	< 20ms	JMH (Java)
派生属性计算	< 100ms	< 500ms	pytest-benchmark
规则评估	< 10ms	< 50ms	pytest-benchmark
SDK 端到端调用	< 50ms	< 200ms	k6

#9.2 性能回归检测

CI 流水线中的性能测试会与基准值比较，超过 10% 退化触发告警：

Python

# 性能基准测试示例
def test_rule_evaluation_performance(benchmark):
    evaluator = RuleEvaluator()
    rule = create_complex_rule(conditions=10)
    context = create_test_context()

    result = benchmark(evaluator.evaluate, rule, context)

    # 基准断言
    assert benchmark.stats["mean"] < 0.01  # 平均 < 10ms
    assert benchmark.stats["max"] < 0.05   # 最大 < 50ms

#10. 混沌测试与韧性验证

#10.1 故障注入场景

coomia-dip 的韧性测试模拟以下故障场景：

故障类型	注入方式	验证目标
Kafka 不可用	停止 Kafka 容器	事件缓冲、重试
gRPC 延迟	注入网络延迟	超时处理、熔断
数据库连接耗尽	限制连接池	优雅降级
内存压力	cgroup 限制内存	OOM 防护
Reasoning & Decision Layer 宕机	停止 Intelligence 容器	计算路由降级

#10.2 韧性测试示例

Python

class TestPlaneDFailure:
    """验证 Intelligence Layer 故障时的降级行为"""

    def test_derived_property_fallback_when_function_unavailable(
        self, platform, docker_compose
    ):
        """当函数计算不可用时，应回退到 SQL 计算"""
        # 创建对象（正常状态）
        supplier = platform.ontology.create_object(
            type_name="Supplier",
            properties={"name": "Test", "delivery_rate": 0.8},
        )
        time.sleep(2)

        # 停止 Intelligence Layer
        docker_compose.stop("intelligence-Layer")

        # 查询派生属性——应通过 SQL 降级计算
        score = platform.ontology.get_property(
            "Supplier", supplier.object_id, "risk_score"
        )

        # 降级计算应仍返回有效结果
        assert score is not None
        assert 0.0 <= float(score) <= 1.0

        # 恢复 Intelligence Layer
        docker_compose.start("intelligence-Layer")

#11. 测试环境管理

#11.1 环境分层

环境	用途	数据	生命周期
本地开发	单元测试 + 快速集成	Mock / 内存数据库	开发者管理
CI 环境	全量测试	Testcontainers	每次运行创建/销毁
Staging	E2E + 性能测试	脱敏生产数据	持续运行
生产	金丝雀测试	真实数据（只读）	持续运行

#11.2 本地开发测试加速

为了让开发者快速获得反馈，coomia-dip 提供以下加速措施：

增量测试：只运行受变更影响的测试
并行执行：单元测试支持并行执行（pytest -n auto）
热重载：Python 测试支持文件变更自动重跑（pytest-watch）
共享 Testcontainers：同一测试会话复用数据库容器

#12. 质量门与提交前检查

#12.1 提交前必须通过的检查

项目 CLAUDE.md 定义了完整的质量门：

语言	检查命令	作用
Python lint	`ruff check`	代码规范
Python format	`black --check && isort --check`	格式统一
Python typecheck	`mypy`	类型安全
Python test	`pytest`	功能正确
Java build	`gradle build`	编译 + 测试
Java test	`gradle test`	功能正确

#12.2 Git Hook 集成

coomia-dip 使用 pre-commit hook 在提交前自动执行检查：

Bash

# .pre-commit-config.yaml
repos:
  - repo: local
    hooks:
      - id: python-quality
        name: Python Quality Gate
        entry: bash -c 'ruff check && black --check . && mypy .'
        language: system
        types: [python]

      - id: java-build
        name: Java Build
        entry: bash -c 'cd control-Layer && gradle build'
        language: system
        types: [java]

#13. 与 Palantir Foundry 的测试对比

维度	Palantir Foundry	coomia-dip
测试框架	内部框架 (闭源)	JUnit 5 + pytest (开源)
集成测试	Foundry Test Runtime	Testcontainers + Docker Compose
契约测试	内部 API 兼容性套件	Protobuf 编译时 + gRPC 运行时
E2E 测试	Foundry Scenario Runner	场景化 pytest
性能测试	内部基准套件	JMH + pytest-benchmark + k6
覆盖率	内部标准	>= 80% (JaCoCo + coverage.py)

#Key Takeaways

多语言平台需要分层测试策略而非统一的测试框架。 Java 和 Python 各有成熟的测试生态（JUnit 5 和 pytest），强行统一反而增加复杂度。关键在于用 gRPC 契约测试作为跨语言的"胶水层"，确保 Protobuf 接口在编译时和运行时都保持兼容。
场景化测试是验证多 Layer 协作正确性的最有效手段。 单元测试保证单个函数正确，集成测试保证单个 Layer 内部正确，但只有场景化测试才能验证"创建对象 → 事件发布 → 派生计算 → 级联更新"这样跨越 4 个 Layer 的完整业务链路。投入 5% 的测试在场景化 E2E 上，能发现 50% 的生产环境 Bug。
测试是代码质量的最后一道防线，禁止任何形式的"掩盖"。 删除失败的测试、空 catch 块、无断言的测试——这些反模式看似提高了通过率，实则隐藏了问题。coomia-dip 的技术红线将这些行为列为禁止事项，配合 80% 覆盖率门槛和 CI 质量门，构建起完整的质量保障体系。

“

下一篇预告: [S2-15] 架构决策记录：用 ADR 追踪每一个关键技术选型——深入理解 coomia-dip 如何用结构化的方式记录"为什么选 gRPC 不选 REST"、"为什么用 Kafka 不用 RabbitMQ"等关键决策。

Tags: #testing #test-pyramid #unit-test #integration-test #e2e #scenario-testing #grpc-contract #testcontainers #coverage #performance-testing #chaos-testing #coomia-dip #智策平台