跳到主要内容

RPC 单元测试（离线 mock）

src/tools/rpc-unit-test 是本工程的 RPC 级单元测试支持库：在普通进程内启动一个最小 atapp runtime，用内存 mock 引擎替代框架已支持的外部依赖路径（SS/DNS/CS/DB/UUID/resource/HPA/telemetry），让业务代码的真实 RPC 路径（dispatcher、 task、协程、生成 API）可以在单测中被驱动和断言，无需 Redis、etcd、atbus 或系统 DNS。库 API 命名空间是 atframework::testing。

快速上手​

  1. 在已有构建目录启用 PROJECT_ENABLE_UNITTEST 和 PROJECT_SERVER_FRAME_ENABLE_UNIT_TEST_HOOKS， 按测试指南重新配置、构建并运行目标测试。
  2. 新增 case 时，参照所属服务或组件的现有 test/，使用 project_add_rpc_unit_test 声明目标， 沿用 runtime 初始化和 mock 引擎装配。
  3. 设置确定的 mock 响应，再通过真实生成 API 或 dispatcher 驱动业务，断言结果与通知。 外部依赖由测试引擎替代，不需要启动 Redis、etcd 或网关。下面给出目标声明与 API 的详细用法。

适用范围与构建开关​

  • 适用：service/component 逻辑的离线单元测试、router/SS/CS/DB 生成 API 的契约测试。
  • 非目标：真实网络/存储的集成测试、性能测试、跨进程测试。
  • 所有测试 executable 只使用 atframe_utils 私有测试框架（CASE_TEST/CASE_EXPECT_*），不引入 GTest。

单元测试 seam 由 PROJECT_SERVER_FRAME_ENABLE_UNIT_TEST_HOOKS 整体门控（cmake_dependent_option，默认跟随 PROJECT_ENABLE_UNITTEST）。hooks-off 的生产构建中：无测试状态、无热路径分支、无 mock 符号、无测试库依赖；生产 fallback 一直保留，只有 fixture 显式安装 hook 后才接管。所有 mock 子命名空间接口 （生成 SS/DB mock、HPA 功能 mock）同样被宏整体裁剪。

工作原理​

  • 发现与传输在 atapp 层注入：mock discovery 节点写入 global discovery，注册 mock:// connector 捕获出站 字节；不修改 logic_server_setup，从而同时验证按 id/name/type-zone/metadata/broadcast 的真实选择逻辑。
  • seam 与策略分离：server_frame 只加最小、无策略的 hook（registry + generation token），匹配规则、内存 数据、响应脚本、历史与诊断全部在测试库；生产逻辑不依赖测试库。每个 seam 的 registry 只在其拥有代码的 production DLL 的一个 .cpp 中定义（核心在 server_frame、Excel provider 在 server_frame-config、 Orbit client 在其 SDK），不做 header-local static/跨 DLL 复制。
  • 生成 mock 类型擦除桥接：<service>::mock、<db>::mock 编译进 server_frame/服务自身 TU，不链接 测试库，经 src/server_frame/rpc/unit_test/mock_engine_bridge.h 的槽转发；桥接为空时降级为 no-op。
  • 事件泵与 generation 屏障：runtime::pump_once() 依次做 time_utility::update() → run_noblock() → ++pump_generation → 各引擎 deliver_pending() → 再一次 run_noblock()。出站记录在第 N 代捕获、第 N+2 代才可被消费，保证调用 task 先注册 waiter，避免“响应早于 waiter”竞态。

默认行为（未注册规则时）：

  • 单向通知类（SS stream/no-wait/broadcast、CS 下行 data/kickoff/set-router）：记录并成功（结果丢弃）。
  • 需响应类（SS unary/wait 变体、DNS lookup、HPA pull）：立即返回明确错误码并输出诊断，不静默等待超时。
  • DB：默认空的内存 backend（读取返回项目 record-not-found 码）；所有已支持操作都有真实内存语义。

快速开始​

CMake（src/tools/rpc-unit-test/cmake/ProjectRpcUnitTest.cmake，私有框架说明见单元测试）：

project_add_rpc_unit_test(
  TARGET ${PROJECT_NAME}-my-component-unit-test
  COMPONENT my-component
  CATEGORY component          # component (default) | sdk | service
  SOURCES "my_test.cpp"
  LINK_LIBRARIES my-component-lib
  FEATURES SS DNS DB
  LABELS fast
  TIMEOUT 120)

测试服务可执行文件内部代码（CATEGORY service）时，不要在 SOURCES 里重复编译服务源码：服务声明 （project_service_declare_instance/project_component_declare_service）通过 MAIN_SOURCES 传入入口文件后， 其余代码只编译一次进静态库，测试链接别名 service::<name>::private 或 components::<name>::private 即可继承 include 目录、SDK/protocol 依赖与生成代码构建顺序（示例：src/lobbysvr/test/CMakeLists.txt）。生成的注册 TU （handle_*_rpc_*.atfw.gen.*）随入口目录（MAIN_DIRECTORIES，默认 app/）归入可执行文件 （MAIN_SOURCES/MAIN_HEADERS），不进静态库；用例需要直接调用 register_handles_for_<service>() 时， 把这一个生成文件列入测试自己的 SOURCES，路径用 project_service_get_target_root_dir() 从私有库别名取 根目录拼接，不写 ${CMAKE_CURRENT_LIST_DIR}/.. 相对路径（示例： src/component/distributed_transaction/test/CMakeLists.txt）。

最小用例（src/tools/rpc-unit-test/test/example_readme.cpp 逐字编译执行，保证与真实 API 一致）：

CASE_TEST(my_component, hello) {
  atfw::testing::runtime test;
  atfw::testing::runtime_options options;
  options.features = {atfw::testing::feature::ss};
  CASE_EXPECT_EQ(0, test.start(options));
  if (!test.is_running()) {
    return;  // CASE_EXPECT_* 非致命：前置失败后必须显式 return。
  }

  auto task = test.run_task("hello", std::chrono::seconds{2},
                            [](rpc::context &) -> rpc::result_code_type { RPC_RETURN_CODE(0); });
  if (!task.empty()) {
    auto result = test.wait(task, std::chrono::seconds{5});
    CASE_EXPECT_TRUE(result.task_exited);
    CASE_EXPECT_EQ(0, result.result_code);
  }
  CASE_EXPECT_EQ(0, test.stop());  // stop() 幂等；非零即期望校验/残留失败。
}

要点：单进程同时只允许一个 active runtime（app/dispatcher/task 单例）；run_task 体用 RPC_AWAIT_*/RPC_RETURN_CODE 跑真实 task；断言放在 wait() 之后。

直接驱动服务端 SS action​

测试某个已知的 task_action_ss_rpc_base 实现时，使用 <atframework/testing/ss_action.h>，不要在测试中 重复拼装 SSMsg 和调用 task_manager：

atfw::testing::ss_action_invoke_options invoke_options{
    rpc::my_service::packer::get_full_name_of_my_method()};
invoke_options.source.node_id = source_node_id;
invoke_options.source.node_name = "my-test-source";
invoke_options.source.source_task_id = source_task_id;
invoke_options.source.sequence = source_sequence;

auto task = test.run_task(
    "my_inbound_action", std::chrono::seconds{2},
    [request, invoke_options](rpc::context &ctx) -> rpc::result_code_type {
      RPC_RETURN_CODE(RPC_AWAIT_CODE_RESULT(
          atfw::testing::invoke_ss_action<task_action_my_method>(ctx, request, invoke_options)));
    });

该接口走真实 SSMsg 序列化/反序列化和 action create/start/wait；runtime 必须启用 feature::ss。请求类型由 TAction::rpc_request_type 固定，RPC 名必须使用生成的 packer::get_full_name_of_<rpc>()，运行时还会用 protobuf method descriptor 同时校验 action 的请求和响应类型。请求与选项按值保存在协程帧中，临时对象也 不会悬空。

返回值是 action 的最终 task result，不是响应 protobuf 内的业务码。source 的零 ID 表示匿名/系统来源； 测试 forward_rpc、链路追踪、回包或来源鉴权时应显式填写相关字段。此接口直接创建模板指定的 action， 不覆盖 dispatcher 的 RPC→action 注册查找；注册表、未知 RPC、非法 type URL/body/envelope 应走 raw transport/dispatcher 接口。可编译示例和契约自测分别见 test/example_readme.cpp、 test/rpc_unit_test_ss_action.cpp，DTMQ 转发案例见 src/component/dtmq/test/dtmq_test_task_forward.cpp。

使用指南​

runtime_options.features 决定模块/hook 集合（ss/dns/cs/db/uuid/resource/router/orbit/hpa/telemetry）；CMake FEATURES 只负责校验/链接/打标签，不隐式决定每个 case 行为。

服务发现 + SS​

atfw::testing::mock_node node;
node.set_id(0x130001).set_name("remote").set_type_id(4097).set_type_name("remote-type").set_zone_id(1);
node.add_label("hpa_scaling_ready", "1");  // 经 logic_hpa_discovery_select 选节点的消费者需要同名标签
auto remote = test.discovery().add_node(node);
logic_server_last_common_module()->reload();  // 注入后 reload 一次，回放节点进 discovery index

auto rule = test.ss().mock(
    rpc::my_service::packer::get_full_name_of_my_method(), Req::descriptor()->full_name(),
    Rsp::descriptor()->full_name(),
    [](const atfw::testing::ss_request_view &request, google::protobuf::Message &response)
        -> rpc::result_code_type {
      const auto &req = static_cast<const Req &>(request.body);
      static_cast<Rsp &>(response).set_echo("hello " + req.payload());
      RPC_RETURN_CODE(0);
    });
// RPC 全名用生成接口 rpc::<module>::packer::get_full_name_of_<rpc>()（gsl::string_view，声明在
// <service>.atfw.gen.h），不要硬编码 "pkg.MyService/my_method" 字符串。
// 生成层等价：<service>::mock::my_method(handler)，typed 首参 rpc::context&。
test.ss().expect(rpc::my_service::packer::get_full_name_of_my_method()).times(1).to_node(0x130001);

SS handler 返回 rpc::result_code_type，可以是协程并用 RPC_AWAIT_CODE_RESULT 等待嵌套 RPC。规则选项 ss_rule_options：match_node_id、times（FIFO 脚本）、delay_generations、no_response、 malformed_type_url/malformed_body。

rpc::<module>::oneway::<rpc>() 可用 atfw::testing::make_context() 在任务外调用，直接返回整数发送结果， 复用同名 mock 规则。请求在发送时序列化；runtime::pump_once() 后 handler 在独立任务中执行，可等待嵌套 RPC，但响应和返回码被丢弃。测试应等待可观察的 handler 结果并检查发送历史，完整用例见 src/tools/rpc-unit-test/test/rpc_unit_test_ss.cpp。

Router / DNS / UUID / resource / CS / raw transport​

  • Router：test.router_manager() + router_test_object 把 router object 寻址到 mock 节点；router RPC 与 unary SS 同链（mock:// connector），不改 Mako。
  • DNS：test.dns().mock_a(domain, ip) / mock(records) / mock_error(domain)；hook 在 uv_getaddrinfo 之前，测试不产生系统 DNS 请求。
  • DB：默认内存 backend 支持字段 presence 合并、CAS、KL 和 TTL，无需注册规则；但当前生产 KV set/insert 已改为全字段覆盖，mock 与契约测试尚未对齐。不要用该内存 backend 验证生产的全字段覆盖语义； 可使用 typed handler 显式固定当前业务预期。生成层 per-table typed handler（rpc::db::<ns>::mock::<interface>）与引擎层 test.db().mock_table(name) 回调 可覆盖单个接口，未覆盖的回落内存 backend。raw entry API（set_raw_kv/append_raw_kl/set_raw_ttl）用于 CAS 版本种子/精确 bytes。
  • UUID：generate_global_increase_id/generate_global_unique_id 经 DB inc_field 流入 DB mock （号段缓存跨 case 延续，不要断言绝对 id 值）；generate_standard_uuid*/generate_short_uuid 是纯本地函数、 无需 mock。feature::uuid 当前不做额外 runtime 装配。
  • resource：test.resource() 提供 path→bytes/version/version_error 内存 loader，reload() 驱动真实 manager 完整 reload/index。bind() 时自动快照真实生成的 excel bindir 全部 *.bytes，测试只需用 set_file() 覆盖自己关心的表、remove_file() 制造缺失；excel 表集合新增/删除都不需要改动既有测试。
  • CS：test.cs() / mock_client 模拟 gateway 客户端（add/post/remove/set-router-rsp），上行走真实 dispatch/action，下行 data/kick/set-router/broadcast 全部可捕获，无 atbus。
  • raw transport：test.transport() 提供 id/name/discovery/consistent-hash/random/round-robin/metadata/ broadcast 全维度规则与调用历史。

覆盖优先级与严格默认​

DB 覆盖顺序：生成层 typed handler → 引擎层 per-table 回调 → 内存 backend；默认不允许真实 Redis passthrough。 两类默认（单向记录+丢弃 / 需响应快速失败）都可用 per-rule 选项覆盖（如把 stream 显式设为 fail 做负向断言）。

关键语义与不变量​

  • delay_generations：0 表示在观察到请求的那个 pump 完成；N 表示额外延迟 N 个完整 pump（SS/DNS 一致）。
  • 引擎投递按到期顺序而非 FIFO；投递会同步恢复协程，任务可重入引擎（两阶段 drain，见 src/tools/rpc-unit-test/src/detail/pending_drain.h）。
  • DB 的 CAS set 接受无版本记录、相等期望版本或期望版本 0（忽略版本比较强制覆盖，存储版本仍按真实版本 +1 递增），否则返回 EN_DB_OLD_VERSION 并回写存储版本；insert 拒绝已有版本记录；KL 索引单调不复用， inc_field 从 0 创建且不动 CAS 版本。server_frame_test_db_script_contract.cpp 真实执行内嵌 Lua，但当前仍 断言所有写路径只合并 present 字段；生产 set/insert 已改为全字段覆盖，mock 与测试需另行修正。
  • 修改引擎/hook seam 的更深入不变量见 .agents/skills/rpc-unit-test/references/engine-invariants.md。

外部依赖离线矩阵​

runtime 生成的配置保证离线：禁止 etcd、bus.proxy、主机名 listen（用数字 IP 或 shm/unix）、OTLP gRPC/HTTP exporter、prometheus push/pull exporter、固定端口（配置即 fail-fast）。telemetry 默认 noop 零网络，可选 otlp_file 导出；HPA 默认零网络，启用时自动安装 prometheus pull hook，预置指标经公开回调链注入，未配置答案报错。

生命周期 / Timeout / 隔离​

  • 每个 CASE_TEST 独立 runtime：start() 建隔离 app/task/引擎，stop() 校验期望并按 deadline teardown； 单进程内 fixture 串行，连续 fixture 不得残留 app/task/session/静态状态（引擎规则 RAII + unbind 统一清理）。
  • Timeout 四层：RPC timeout < task timeout（run_task 参数）< runtime hard timeout（wait 参数）< CTest timeout（TIMEOUT）。hard timeout 会 poison 当前 runtime 并 kill-all 任务，此后只能 stop()，不能复用。
  • telemetry/冲突配置用例必须独占 executable（进程生命周期状态），用 project_add_rpc_unit_test 多 target 拆分。

常见失败诊断​

  • 未匹配 RPC：快速失败，日志列出调用名与已注册规则；检查 RPC 全名与 match_node_id。
  • response sequence 错误/不唤醒：确认 handler 返回 0 且响应类型与注册一致；stream/no-wait 不回响应。
  • task 未进入 wait：run_task 返回空 handle 时打印 task.get_diagnostic()。
  • 残留 session/task：stop() 诊断会列出；检查是否在 task 内遗漏 RPC_RETURN_CODE。
  • mock 注册返回空 handle：get_diagnostic()（未绑引擎、类型名错误、RPC 名格式错误）。

构建与运行​

cmake --build build_jobs_cmake_tools --target atsf4g-co-rpc-unit-test-selftest --parallel 12
ctest --test-dir build_jobs_cmake_tools -L rpc-unit-test --output-on-failure
# Filter one case: build_jobs_cmake_tools/test/atsf4g-co-rpc-unit-test-selftest.exe -r "rpc_unit_test.<case>"

参考​

  • 完整 API 与示例：src/tools/rpc-unit-test/README.md
  • AI agent 流程化指引：.agents/skills/rpc-unit-test/SKILL.md
  • 引擎/hook seam 不变量：.agents/skills/rpc-unit-test/references/engine-invariants.md