用单元测试保护微服务

构建自动化防护流程：检查微服务端点，在失败时向 Slack 告警，并让后续任务只在单元测试通过后运行。该功能可用于 Enterprise Edition 和 Cloud；单元测试与 RunTest 需要对应版本能力。

现代微服务和 API 后端通常暴露健康检查端点。利用 Kestra，可以监控这些端点、编写测试验证监控流程，再按测试结果控制下游自动化。本指南依次创建 HTTP 检查与 Slack 告警流程，覆盖正常和异常路径，并在测试套件通过后触发下游逻辑。

先决条件

• Kestra Enterprise Edition：Unit Tests 与 RunTest 任务需要此能力。
• Slack Incoming Webhook URL，或 Notifications 插件支持的其他通知渠道。
• 作为 KESTRA_API_TOKEN 保存的 Kestra API token，供测试运行流程使用。

步骤 1：监控 API 端点

在命名空间中创建下面的流程。当目标端点返回非 200 状态时，执行告警任务。

id: microservices-and-apis

namespace: tutorial

description: Microservices and APIs




inputs:

  - id: server_uri

    type: URI

    defaults: https://kestra.io




  - id: slack_webhook_uri

    type: URI

    defaults: https://kestra.io/api/mock




tasks:
  - id: http_request

    type: io.kestra.plugin.core.http.Request

    uri: "{{ inputs.server_uri }}"

    options:

      allowFailed: true




  - id: check_status

    type: io.kestra.plugin.core.flow.If

    condition: "{{ outputs.http_request.code != 200 }}"

    then:

      - id: server_unreachable_alert

        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ inputs.slack_webhook_uri }}"

        payload: |

          {

            "channel": "#alerts",

            "text": "The server {{ inputs.server_uri }} is down!"

          }

    else:

      - id: healthy

        type: io.kestra.plugin.core.log.Log

        message: Everything is fine!

该流程发出 HTTP 请求，配置 allowFailed: true，然后依据响应码发送 Slack 告警，或记录健康状态。

组件说明

• server_uri 将目标端点参数化，便于复用于预发布、生产或其他健康检查端点。
• slack_webhook_uri 接收告警地址，避免把真实密钥硬编码在流程中。也可直接在通知任务的 url 属性中引用 KV Store 或 secret。
• http_request 请求端点并获取 HTTP 状态码。allowFailed 允许处理失败的 HTTP 响应。
• check_status 按响应分支：非 200 时执行告警；200 时写入“Everything is fine!”日志。

步骤 2：添加单元测试

在同一命名空间中，将下面的片段保存为测试资源，以覆盖两种结果。

id: test_microservices_and_apis

flowId: microservices-and-apis

namespace: tutorial

testCases:

  - id: server_should_be_reachable

    type: io.kestra.core.tests.flow.UnitTest

    fixtures:

      inputs:

        server_uri: https://kestra.io

    assertions:
      - value: "{{ outputs.http_request.code }}"

        equalTo: 200




  - id: server_should_be_unreachable

    type: io.kestra.core.tests.flow.UnitTest

    fixtures:

      inputs:

        server_uri: https://kestra.io/bad-url

      tasks:

        - id: server_unreachable_alert

          description: no Slack message from tests

    assertions:

      - value: "{{ outputs.http_request.code }}"

        notEqualTo: 200

每个测试用例提供 fixtures（输入及可选任务覆盖）与 assertions。第二个用例替换 Slack 告警任务，避免测试发送 Slack 消息，同时检查 HTTP 响应不是 200。

组件说明

• 测试的 id、flowId 与 namespace 将套件关联到步骤 1 的流程。
• server_should_be_reachable 传入有效的 server_uri，并断言状态码等于 200。
• server_should_be_unreachable 传入坏路径，以任务替身抑制告警消息，并断言状态码不等于 200，减少测试消息对频道的干扰。

步骤 3：只在测试通过后运行下游逻辑

创建控制流程，执行测试套件，并依据 outputs.run_test.result.state 布尔值决定是否继续。

id: run_if_tests_pass

namespace: tutorial




tasks:

  - id: run_test

    type: io.kestra.plugin.kestra.ee.tests.RunTest

    auth:

      apiToken: "{{ secret('KESTRA_API_TOKEN') }}"

    namespace: tutorial
    testId: test_microservices_and_apis




  - id: run_if_tests_pass

    type: io.kestra.plugin.core.flow.If

    condition: "{{ outputs.run_test.result.state }}"

    then:

      - id: log

        type: io.kestra.plugin.core.log.Log

        message: hello

可把最后的 log 任务替换为部署、升级处理或其他只有测试成功才应执行的自动化。

组件说明

• run_test 使用 API token、命名空间和测试 ID 调用 Enterprise Edition 的 RunTest 插件，返回包含 state 布尔值的结果。
• run_if_tests_pass 在条件判断中检查该结果，通过后执行 then 分支。

步骤 4：执行测试

从 Kestra UI 或 CLI 运行测试，检查两条断言是否通过。原文展示了测试断言结果截图，说明监控流程应能在不产生 Slack 测试噪声的情况下被验证。这是教程的操作要求与示例结果；本稿没有运行测试，也没有制作通过截图。

后续步骤

• 遍历输入，或使用命名空间文件，让监控流程覆盖多个端点。
• 替换 Slack 任务，改为 PagerDuty、Teams 或邮件等 Notifications 插件支持的渠道。
• 把受测试结果门控的流程接入 CI/CD，使每次部署在上线前验证关键监控。