Unit testing

pw_rpc2: Next gen, zero-copy async RPC with end-to-end backpressure

pw::rpc2::test::TestHarness is a lightweight dependency-injection fixture for unit testing pw_rpc2 services and clients with minimal boilerplate.

Constructing TestHarness<MyService> instantiates MyService (forwarding any constructor arguments) and connects a generated service client to it. You can use the harness from either side:

• Testing a service: Call RPCs through tester.service_client() (or pwpb_client() / raw_client()) and assert on the responses or on tester.service().

  TestHarness<MyService> tester(arg1, arg2);
  auto resp = tester.RunToCompletion(tester.service_client().DoSomething(req));
  PW_TEST_ASSERT_OK(resp);
  EXPECT_EQ(tester.service().call_count(), 1u);
  

• Testing client code: Pass tester.service_client() or tester.client() to the client code under test and use a test ServiceImpl to verify how the client interacts with the service.

  TestHarness<FakeMyService> tester;
  MyClientComponent component(tester.service_client());
  PW_TEST_ASSERT_OK(tester.RunToCompletion(component.FetchStatus()));
  EXPECT_TRUE(tester.service().was_queried());
  

Setup

Add the test harness and generated RPC library to your unit test target:

Bazel

load("//pw_unit_test:pw_cc_test.bzl", "pw_cc_test")

pw_cc_test(
    name = "my_service_test",
    srcs = ["my_service_test.cc"],
    deps = [
        ":my_service",
        ":my_service_pwpb_rpc2",
        "@pigweed//pw_rpc2/cpp:test_harness",
    ],
)

GN

import("$dir_pw_unit_test/test.gni")

pw_test("my_service_test") {
  sources = [ "my_service_test.cc" ]
  deps = [
    ":my_service",
    ":my_service_proto.pwpb_rpc2",
    "$dir_pw_rpc2/cpp:test_harness",
  ]
}

CMake

pw_add_test(my_service_test
  SOURCES
    my_service_test.cc
  PRIVATE_DEPS
    my_service
    my_service_proto.pwpb_rpc2
    pw_rpc2.cpp.test_harness
)

Include pw_rpc2/test_harness.h alongside your service implementation and generated RPC header:

#include "my_service.h"
#include "my_service.pwpb.rpc2.h"
#include "pw_rpc2/test_harness.h"
#include "pw_unit_test/framework.h"

Testing with coroutines

Passing a coroutine lambda to RunToCompletion() lets a test co_await unary and streaming RPCs sequentially:

TEST(TestHarnessTest, UnaryCoroClient) {
  TestHarness<EchoFutureService> tester;

  tester.RunToCompletion([&](CoroContext) -> Coro<void> {
    RpcResult<EchoResponse> resp =
        co_await tester.service_client().EchoUnary(EchoRequest{.val = 21});
    PW_TEST_EXPECT_OK(resp);
    if (resp.ok()) {
      EXPECT_EQ(resp->val, 42u);
    }
  });
}

TEST(TestHarnessTest, ServerStreamingCoroClient) {
  TestHarness<EchoFutureService> tester;

  tester.RunToCompletion([&](CoroContext) -> Coro<void> {
    RpcResult<Reader<EchoResponse>> reader =
        co_await tester.service_client().CountUpServerStream(
            EchoRequest{.val = 100});
    PW_TEST_EXPECT_OK(reader);
    if (!reader.ok()) {
      co_return;
    }

    for (uint32_t expected = 101; expected <= 103; ++expected) {
      RpcResult<EchoResponse> msg = co_await reader->Read();
      PW_TEST_EXPECT_OK(msg);
      if (msg.ok()) {
        EXPECT_EQ(msg->val, expected);
      }
    }
    EXPECT_EQ((co_await reader->Read()).status(), RpcStatus::EndOfStream());
  });
}

TEST(TestHarnessTest, ClientStreamingCoroClient) {
  TestHarness<EchoFutureService> tester;

  tester.RunToCompletion([&](CoroContext) -> Coro<void> {
    auto call = co_await tester.service_client().AccumulateClientStream();
    PW_TEST_EXPECT_OK(call);
    if (!call.ok()) {
      co_return;
    }

    PW_TEST_EXPECT_OK(co_await call->writer().Write(EchoRequest{.val = 10}));
    PW_TEST_EXPECT_OK(co_await call->writer().Write(EchoRequest{.val = 20}));
    PW_TEST_EXPECT_OK(co_await call->writer().Write(EchoRequest{.val = 30}));
    PW_TEST_EXPECT_OK(co_await call->writer().Finish());

    RpcResult<EchoResponse> resp = co_await call->response();
    PW_TEST_EXPECT_OK(resp);
    if (resp.ok()) {
      EXPECT_EQ(resp->val, 60u);
    }
  });
}

TEST(TestHarnessTest, BidiStreamingCoroClient) {
  TestHarness<EchoFutureService> tester;

  tester.RunToCompletion([&](CoroContext) -> Coro<void> {
    auto call = co_await tester.service_client().EchoBidiStream();
    PW_TEST_EXPECT_OK(call);
    if (!call.ok()) {
      co_return;
    }

    PW_TEST_EXPECT_OK(co_await call->writer().Write(EchoRequest{.val = 5}));
    RpcResult<EchoResponse> first = co_await call->reader().Read();
    PW_TEST_EXPECT_OK(first);
    if (first.ok()) {
      EXPECT_EQ(first->val, 10u);
    }

    PW_TEST_EXPECT_OK(co_await call->writer().Write(EchoRequest{.val = 15}));
    RpcResult<EchoResponse> second = co_await call->reader().Read();
    PW_TEST_EXPECT_OK(second);
    if (second.ok()) {
      EXPECT_EQ(second->val, 30u);
    }

    PW_TEST_EXPECT_OK(co_await call->writer().Finish());
    EXPECT_EQ((co_await call->reader().Read()).status(),
              RpcStatus::EndOfStream());
  });
}

Testing with futures and tasks

Pass an RPC future directly to RunToCompletion():

TEST(TestHarnessTest, UnaryDirectFuture) {
  TestHarness<EchoFutureService> tester;

  RpcResult<EchoResponse> resp = tester.RunToCompletion(
      tester.service_client().EchoUnary(EchoRequest{.val = 21}));
  PW_TEST_ASSERT_OK(resp);
  EXPECT_EQ(resp->val, 42u);
  EXPECT_EQ(tester.service().calls(), 1u);
}

To drive a hand-written pw::async2::Task, pass it to RunToCompletion() as an lvalue. The task stays in scope, so you can inspect its results afterwards. Here, EchoUnaryClientTask is a task that makes one EchoUnary call and stores the response:

TEST(TestHarnessTest, UnaryTask) {
  TestHarness<EchoFutureService> tester;

  EchoUnaryClientTask task(tester.service_client(), 21);
  tester.RunToCompletion(task);

  PW_TEST_ASSERT_OK(task.result());
  EXPECT_EQ(task.result()->val, 42u);
}

When an RPC must pause on an external event or another stream operation before completing, use RunInTaskUntilStalled() to advance the call and inspect its intermediate state:

TEST(TestHarnessTest, StalledCallResumedWhenStreamFinishes) {
  TestHarness<EchoFutureService> tester;

  auto call =
      tester.RunToCompletion(tester.service_client().AccumulateClientStream());
  PW_TEST_ASSERT_OK(call);

  auto& response_fut = call->response();
  // First run stalls because the client has not finished the request stream.
  EXPECT_TRUE(tester.RunInTaskUntilStalled(response_fut).IsPending());

  PW_TEST_EXPECT_OK(
      tester.RunToCompletion(call->writer().Write(EchoRequest{.val = 42})));
  PW_TEST_EXPECT_OK(tester.RunToCompletion(call->writer().Finish()));

  auto poll = tester.RunInTaskUntilStalled(response_fut);
  ASSERT_TRUE(poll.IsReady());
  PW_TEST_ASSERT_OK(*poll);
  EXPECT_EQ((*poll)->val, 42u);
}