Microservices Communication Patterns (Saga, Circuit Breaker) · 课时

测试编排式 Saga

学习如何测试基于编排的 Saga，包括补偿路径的单元测试、模拟服务故障的集成测试，以及对 Saga 完成情况的端到端验证。

测试编排式 Saga 是 CoddyKit 上的免费 Microservices Communication Patterns (Saga, Circuit Breaker) 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Microservices Communication Patterns (Saga, Circuit Breaker) 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Microservices Communication Patterns (Saga, Circuit Breaker) 课程共包含 4 节课。

本课时的部分内容尚未翻译，以英文显示。

Why Test Sagas?

An orchestration saga coordinates multiple services through a central orchestrator. A bug in the orchestrator can leave the system in an inconsistent state across services.

Testing sagas is harder than testing a single function because you must verify both the happy path and every compensation path.

Did each step execute in order?
Did failures trigger the right compensations?
Is the final state consistent?

The Testing Pyramid for Sagas

Apply the classic testing pyramid:

Unit tests — the orchestrator's decision logic in isolation.
Integration tests — the orchestrator talking to real or stubbed services.
End-to-end tests — the full saga across all services.

Most tests should be unit tests because they are fast and deterministic.

Unit Testing the Orchestrator

The orchestrator is often a state machine. You can unit test its transitions by feeding events and asserting the next command.

def next_command(state, event):
    if state == 'STARTED' and event == 'PAYMENT_OK':
        return 'RESERVE_STOCK'
    if state == 'STARTED' and event == 'PAYMENT_FAILED':
        return 'ABORT'
    return 'NOOP'

print(next_command('STARTED', 'PAYMENT_OK'))
print(next_command('STARTED', 'PAYMENT_FAILED'))

Mocking Service Calls

In unit tests, replace real service calls with mocks or stubs. This lets you control exactly what each step returns.

For example, force the payment service to return a failure so you can assert the orchestrator issues a compensation command.

class FakePayment:
    def __init__(self, ok):
        self.ok = ok
    def charge(self, amount):
        return 'PAYMENT_OK' if self.ok else 'PAYMENT_FAILED'

print(FakePayment(False).charge(100))

Testing Compensation Paths

The riskiest part of any saga is compensation. For each forward step, write a test that:

Executes steps up to a chosen point.
Forces a failure.
Asserts that every prior step was compensated in reverse order.

executed = ['reserve_stock', 'charge_card']
compensations = list(reversed(executed))
print('Compensate in order:', compensations)

Integration Tests with Test Containers

Integration tests run the orchestrator against real dependencies in disposable containers (databases, message brokers).

Tools like Testcontainers spin up a real broker (e.g. Kafka or RabbitMQ) so message flow is exercised exactly as in production.

Simulating Service Failures

To test resilience, inject failures: make a downstream service return errors, time out, or crash mid-saga.

HTTP 500 from a step
Timeout (no response)
Duplicate event delivery

Each scenario should leave the system consistent.

Testing Timeouts

Orchestrators often set a timeout for each step. If a step does not respond, the saga should trigger compensation.

import time

def wait_for_step(timeout, elapsed):
    if elapsed > timeout:
        return 'TIMEOUT -> COMPENSATE'
    return 'OK'

print(wait_for_step(5, 7))
print(wait_for_step(5, 3))

Asserting Final State Consistency

The most important assertion is that the system ends in a valid state. After a failed saga, no money should be charged and no stock reserved.

Query each service and verify the invariants hold across them.

Replaying Events in Tests

Because messages can be delivered more than once, replay the same event twice in your tests and assert the saga state is unchanged the second time. This verifies idempotency at the orchestrator level.

processed = set()

def handle(event_id):
    if event_id in processed:
        return 'IGNORED (duplicate)'
    processed.add(event_id)
    return 'PROCESSED'

print(handle('e1'))
print(handle('e1'))

Observability in Tests

Add assertions on emitted logs, metrics, and trace spans. A well-tested saga should produce a clear audit trail so that, when a real failure happens in production, you can reconstruct exactly what occurred.

Quick Check

Which test type is the best place to verify the orchestrator's transition logic quickly and deterministically?

Recap

You learned how to test orchestrated sagas:

Use the testing pyramid: many unit tests, fewer integration and E2E tests.
Mock services to force happy and failure paths.
Always test compensation order and final state consistency.
Replay events to confirm idempotency, and inject timeouts and failures.

Thorough saga testing is what keeps distributed transactions trustworthy.

常见问题解答

「测试编排式 Saga」课时是免费的吗？

是的 — 「测试编排式 Saga」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Microservices Communication Patterns (Saga, Circuit Breaker) 课程的其余内容，请升级到 CoddyKit PRO。 Microservices Communication Patterns (Saga, Circuit Breaker) 课程共包含 4 节课。

「测试编排式 Saga」这节课中我会学到什么？

学习如何测试基于编排的 Saga，包括补偿路径的单元测试、模拟服务故障的集成测试，以及对 Saga 完成情况的端到端验证。你通过在浏览器中直接运行的动手代码来练习 Microservices Communication Patterns (Saga, Circuit Breaker)，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Microservices Communication Patterns (Saga, Circuit Breaker) 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Microservices Communication Patterns (Saga, Circuit Breaker) 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「测试编排式 Saga」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Microservices Communication Patterns (Saga, Circuit Breaker) 课中编写并运行代码吗？

能。每节 Microservices Communication Patterns (Saga, Circuit Breaker) 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 Microservices Communication Patterns (Saga, Circuit Breaker)