1 // Copyright 2021 The Pigweed Authors
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License"); you may not
4 // use this file except in compliance with the License. You may obtain a copy of
5 // the License at
6 //
7 // https://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11 // WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12 // License for the specific language governing permissions and limitations under
13 // the License.
14
15 #include "pw_rpc/internal/call.h"
16
17 #include <algorithm>
18 #include <array>
19 #include <cstdint>
20 #include <cstring>
21 #include <optional>
22
23 #include "gtest/gtest.h"
24 #include "pw_rpc/internal/test_method.h"
25 #include "pw_rpc/internal/test_utils.h"
26 #include "pw_rpc/service.h"
27 #include "pw_rpc_private/fake_server_reader_writer.h"
28
29 namespace pw::rpc {
30
31 class TestService : public Service {
32 public:
TestService(uint32_t id)33 constexpr TestService(uint32_t id) : Service(id, method) {}
34
35 static constexpr internal::TestMethodUnion method = internal::TestMethod(8);
36 };
37
38 namespace internal {
39 namespace {
40
41 constexpr Packet kPacket(pwpb::PacketType::REQUEST, 99, 16, 8);
42
43 using ::pw::rpc::internal::test::FakeServerReader;
44 using ::pw::rpc::internal::test::FakeServerReaderWriter;
45 using ::pw::rpc::internal::test::FakeServerWriter;
46 using ::std::byte;
47 using ::testing::Test;
48
49 static_assert(sizeof(Call) ==
50 // IntrusiveList::Item pointer
51 sizeof(IntrusiveList<Call>::Item) +
52 // Endpoint pointer
53 sizeof(Endpoint*) +
54 // call_id, channel_id, service_id, method_id
55 4 * sizeof(uint32_t) +
56 // Packed state and properties
57 sizeof(void*) +
58 // on_error and on_next callbacks
59 2 * sizeof(Function<void(Status)>),
60 "Unexpected padding in Call!");
61
62 static_assert(sizeof(CallProperties) == sizeof(uint8_t));
63
TEST(CallProperties,ValuesMatch)64 TEST(CallProperties, ValuesMatch) {
65 constexpr CallProperties props_1(
66 MethodType::kBidirectionalStreaming, kClientCall, kRawProto);
67 static_assert(props_1.method_type() == MethodType::kBidirectionalStreaming);
68 static_assert(props_1.call_type() == kClientCall);
69 static_assert(props_1.callback_proto_type() == kRawProto);
70
71 constexpr CallProperties props_2(
72 MethodType::kClientStreaming, kServerCall, kProtoStruct);
73 static_assert(props_2.method_type() == MethodType::kClientStreaming);
74 static_assert(props_2.call_type() == kServerCall);
75 static_assert(props_2.callback_proto_type() == kProtoStruct);
76
77 constexpr CallProperties props_3(
78 MethodType::kUnary, kClientCall, kProtoStruct);
79 static_assert(props_3.method_type() == MethodType::kUnary);
80 static_assert(props_3.call_type() == kClientCall);
81 static_assert(props_3.callback_proto_type() == kProtoStruct);
82 }
83
84 class ServerWriterTest : public Test {
85 public:
ServerWriterTest()86 ServerWriterTest() : context_(TestService::method.method()) {
87 rpc_lock().lock();
88 FakeServerWriter writer_temp(context_.get().ClaimLocked());
89 rpc_lock().unlock();
90 writer_ = std::move(writer_temp);
91 }
92
93 ServerContextForTest<TestService> context_;
94 FakeServerWriter writer_;
95 };
96
TEST_F(ServerWriterTest,ConstructWithContext_StartsOpen)97 TEST_F(ServerWriterTest, ConstructWithContext_StartsOpen) {
98 EXPECT_TRUE(writer_.active());
99 }
100
TEST_F(ServerWriterTest,Move_ClosesOriginal)101 TEST_F(ServerWriterTest, Move_ClosesOriginal) {
102 FakeServerWriter moved(std::move(writer_));
103
104 #ifndef __clang_analyzer__
105 EXPECT_FALSE(writer_.active());
106 #endif // ignore use-after-move
107 EXPECT_TRUE(moved.active());
108 }
109
TEST_F(ServerWriterTest,DefaultConstruct_Closed)110 TEST_F(ServerWriterTest, DefaultConstruct_Closed) {
111 FakeServerWriter writer;
112 EXPECT_FALSE(writer.active());
113 }
114
TEST_F(ServerWriterTest,Construct_RegistersWithServer)115 TEST_F(ServerWriterTest, Construct_RegistersWithServer) {
116 RpcLockGuard lock;
117 IntrusiveList<Call>::iterator call = context_.server().FindCall(kPacket);
118 ASSERT_NE(call, context_.server().calls_end());
119 EXPECT_EQ(static_cast<void*>(&*call), static_cast<void*>(&writer_));
120 }
121
TEST_F(ServerWriterTest,Destruct_RemovesFromServer)122 TEST_F(ServerWriterTest, Destruct_RemovesFromServer) {
123 {
124 // Note `lock_guard` cannot be used here, because while the constructor
125 // of `FakeServerWriter` requires the lock be held, the destructor acquires
126 // it!
127 rpc_lock().lock();
128 FakeServerWriter writer(context_.get().ClaimLocked());
129 rpc_lock().unlock();
130 }
131
132 RpcLockGuard lock;
133 EXPECT_EQ(context_.server().FindCall(kPacket), context_.server().calls_end());
134 }
135
TEST_F(ServerWriterTest,Finish_RemovesFromServer)136 TEST_F(ServerWriterTest, Finish_RemovesFromServer) {
137 EXPECT_EQ(OkStatus(), writer_.Finish());
138 RpcLockGuard lock;
139 EXPECT_EQ(context_.server().FindCall(kPacket), context_.server().calls_end());
140 }
141
TEST_F(ServerWriterTest,Finish_SendsResponse)142 TEST_F(ServerWriterTest, Finish_SendsResponse) {
143 EXPECT_EQ(OkStatus(), writer_.Finish());
144
145 ASSERT_EQ(context_.output().total_packets(), 1u);
146 const Packet& packet = context_.output().last_packet();
147 EXPECT_EQ(packet.type(), pwpb::PacketType::RESPONSE);
148 EXPECT_EQ(packet.channel_id(), context_.channel_id());
149 EXPECT_EQ(packet.service_id(), context_.service_id());
150 EXPECT_EQ(packet.method_id(), context_.get().method().id());
151 EXPECT_TRUE(packet.payload().empty());
152 EXPECT_EQ(packet.status(), OkStatus());
153 }
154
TEST_F(ServerWriterTest,Finish_ReturnsStatusFromChannelSend)155 TEST_F(ServerWriterTest, Finish_ReturnsStatusFromChannelSend) {
156 context_.output().set_send_status(Status::Unauthenticated());
157
158 // All non-OK statuses are remapped to UNKNOWN.
159 EXPECT_EQ(Status::Unknown(), writer_.Finish());
160 }
161
TEST_F(ServerWriterTest,Finish)162 TEST_F(ServerWriterTest, Finish) {
163 ASSERT_TRUE(writer_.active());
164 EXPECT_EQ(OkStatus(), writer_.Finish());
165 EXPECT_FALSE(writer_.active());
166 EXPECT_EQ(Status::FailedPrecondition(), writer_.Finish());
167 }
168
TEST_F(ServerWriterTest,Open_SendsPacketWithPayload)169 TEST_F(ServerWriterTest, Open_SendsPacketWithPayload) {
170 constexpr byte data[] = {byte{0xf0}, byte{0x0d}};
171 ASSERT_EQ(OkStatus(), writer_.Write(data));
172
173 byte encoded[64];
174 auto result = context_.server_stream(data).Encode(encoded);
175 ASSERT_EQ(OkStatus(), result.status());
176
177 ConstByteSpan payload = context_.output().last_packet().payload();
178 EXPECT_EQ(sizeof(data), payload.size());
179 EXPECT_EQ(0, std::memcmp(data, payload.data(), sizeof(data)));
180 }
181
TEST_F(ServerWriterTest,Closed_IgnoresFinish)182 TEST_F(ServerWriterTest, Closed_IgnoresFinish) {
183 EXPECT_EQ(OkStatus(), writer_.Finish());
184 EXPECT_EQ(Status::FailedPrecondition(), writer_.Finish());
185 }
186
TEST_F(ServerWriterTest,DefaultConstructor_NoClientStream)187 TEST_F(ServerWriterTest, DefaultConstructor_NoClientStream) {
188 FakeServerWriter writer;
189 RpcLockGuard lock;
190 EXPECT_FALSE(writer.as_server_call().has_client_stream());
191 EXPECT_FALSE(writer.as_server_call().client_stream_open());
192 }
193
TEST_F(ServerWriterTest,Open_NoClientStream)194 TEST_F(ServerWriterTest, Open_NoClientStream) {
195 RpcLockGuard lock;
196 EXPECT_FALSE(writer_.as_server_call().has_client_stream());
197 EXPECT_FALSE(writer_.as_server_call().client_stream_open());
198 }
199
200 class ServerReaderTest : public Test {
201 public:
ServerReaderTest()202 ServerReaderTest() : context_(TestService::method.method()) {
203 rpc_lock().lock();
204 FakeServerReader reader_temp(context_.get().ClaimLocked());
205 rpc_lock().unlock();
206 reader_ = std::move(reader_temp);
207 }
208
209 ServerContextForTest<TestService> context_;
210 FakeServerReader reader_;
211 };
212
TEST_F(ServerReaderTest,DefaultConstructor_ClientStreamClosed)213 TEST_F(ServerReaderTest, DefaultConstructor_ClientStreamClosed) {
214 FakeServerReader reader;
215 EXPECT_FALSE(reader.as_server_call().active());
216 RpcLockGuard lock;
217 EXPECT_FALSE(reader.as_server_call().client_stream_open());
218 }
219
TEST_F(ServerReaderTest,Open_ClientStreamStartsOpen)220 TEST_F(ServerReaderTest, Open_ClientStreamStartsOpen) {
221 RpcLockGuard lock;
222 EXPECT_TRUE(reader_.as_server_call().has_client_stream());
223 EXPECT_TRUE(reader_.as_server_call().client_stream_open());
224 }
225
TEST_F(ServerReaderTest,Close_ClosesClientStream)226 TEST_F(ServerReaderTest, Close_ClosesClientStream) {
227 EXPECT_TRUE(reader_.as_server_call().active());
228 rpc_lock().lock();
229 EXPECT_TRUE(reader_.as_server_call().client_stream_open());
230 rpc_lock().unlock();
231 EXPECT_EQ(OkStatus(),
232 reader_.as_server_call().CloseAndSendResponse(OkStatus()));
233
234 EXPECT_FALSE(reader_.as_server_call().active());
235 RpcLockGuard lock;
236 EXPECT_FALSE(reader_.as_server_call().client_stream_open());
237 }
238
TEST_F(ServerReaderTest,EndClientStream_OnlyClosesClientStream)239 TEST_F(ServerReaderTest, EndClientStream_OnlyClosesClientStream) {
240 EXPECT_TRUE(reader_.active());
241 rpc_lock().lock();
242 EXPECT_TRUE(reader_.as_server_call().client_stream_open());
243 reader_.as_server_call().HandleClientStreamEnd();
244
245 EXPECT_TRUE(reader_.active());
246 RpcLockGuard lock;
247 EXPECT_FALSE(reader_.as_server_call().client_stream_open());
248 }
249
250 class ServerReaderWriterTest : public Test {
251 public:
ServerReaderWriterTest()252 ServerReaderWriterTest() : context_(TestService::method.method()) {
253 rpc_lock().lock();
254 FakeServerReaderWriter reader_writer_temp(context_.get().ClaimLocked());
255 rpc_lock().unlock();
256 reader_writer_ = std::move(reader_writer_temp);
257 }
258
259 ServerContextForTest<TestService> context_;
260 FakeServerReaderWriter reader_writer_;
261 };
262
TEST_F(ServerReaderWriterTest,Move_MaintainsClientStream)263 TEST_F(ServerReaderWriterTest, Move_MaintainsClientStream) {
264 FakeServerReaderWriter destination;
265
266 rpc_lock().lock();
267 EXPECT_FALSE(destination.as_server_call().client_stream_open());
268 rpc_lock().unlock();
269
270 destination = std::move(reader_writer_);
271 RpcLockGuard lock;
272 EXPECT_TRUE(destination.as_server_call().has_client_stream());
273 EXPECT_TRUE(destination.as_server_call().client_stream_open());
274 }
275
TEST_F(ServerReaderWriterTest,Move_MovesCallbacks)276 TEST_F(ServerReaderWriterTest, Move_MovesCallbacks) {
277 int calls = 0;
278 reader_writer_.set_on_error([&calls](Status) { calls += 1; });
279 reader_writer_.set_on_next([&calls](ConstByteSpan) { calls += 1; });
280
281 #if PW_RPC_CLIENT_STREAM_END_CALLBACK
282 reader_writer_.set_on_client_stream_end([&calls]() { calls += 1; });
283 #endif // PW_RPC_CLIENT_STREAM_END_CALLBACK
284
285 FakeServerReaderWriter destination(std::move(reader_writer_));
286 rpc_lock().lock();
287 destination.as_server_call().HandlePayload({});
288 rpc_lock().lock();
289 destination.as_server_call().HandleClientStreamEnd();
290 rpc_lock().lock();
291 destination.as_server_call().HandleError(Status::Unknown());
292
293 EXPECT_EQ(calls, 2 + PW_RPC_CLIENT_STREAM_END_CALLBACK);
294 }
295
TEST_F(ServerReaderWriterTest,Move_ClearsCallAndChannelId)296 TEST_F(ServerReaderWriterTest, Move_ClearsCallAndChannelId) {
297 rpc_lock().lock();
298 reader_writer_.set_id(999);
299 EXPECT_NE(reader_writer_.channel_id_locked(), 0u);
300 rpc_lock().unlock();
301
302 FakeServerReaderWriter destination(std::move(reader_writer_));
303
304 RpcLockGuard lock;
305 EXPECT_EQ(reader_writer_.id(), 0u);
306 EXPECT_EQ(reader_writer_.channel_id_locked(), 0u);
307 }
308
TEST_F(ServerReaderWriterTest,Move_SourceAwaitingCleanup_CleansUpCalls)309 TEST_F(ServerReaderWriterTest, Move_SourceAwaitingCleanup_CleansUpCalls) {
310 std::optional<Status> on_error_cb;
311 reader_writer_.set_on_error([&on_error_cb](Status error) {
312 ASSERT_FALSE(on_error_cb.has_value());
313 on_error_cb = error;
314 });
315
316 rpc_lock().lock();
317 context_.server().CloseCallAndMarkForCleanup(reader_writer_.as_server_call(),
318 Status::NotFound());
319 rpc_lock().unlock();
320
321 FakeServerReaderWriter destination(std::move(reader_writer_));
322
323 EXPECT_EQ(Status::NotFound(), on_error_cb);
324 }
325
TEST_F(ServerReaderWriterTest,Move_BothAwaitingCleanup_CleansUpCalls)326 TEST_F(ServerReaderWriterTest, Move_BothAwaitingCleanup_CleansUpCalls) {
327 rpc_lock().lock();
328 // Use call ID 123 so this call is distinct from the other.
329 FakeServerReaderWriter destination(context_.get(123).ClaimLocked());
330 rpc_lock().unlock();
331
332 std::optional<Status> destination_on_error_cb;
333 destination.set_on_error([&destination_on_error_cb](Status error) {
334 ASSERT_FALSE(destination_on_error_cb.has_value());
335 destination_on_error_cb = error;
336 });
337
338 std::optional<Status> source_on_error_cb;
339 reader_writer_.set_on_error([&source_on_error_cb](Status error) {
340 ASSERT_FALSE(source_on_error_cb.has_value());
341 source_on_error_cb = error;
342 });
343
344 // Simulate these two calls being closed by another thread.
345 rpc_lock().lock();
346 context_.server().CloseCallAndMarkForCleanup(destination.as_server_call(),
347 Status::NotFound());
348 context_.server().CloseCallAndMarkForCleanup(reader_writer_.as_server_call(),
349 Status::Unauthenticated());
350 rpc_lock().unlock();
351
352 destination = std::move(reader_writer_);
353
354 EXPECT_EQ(Status::NotFound(), destination_on_error_cb);
355 EXPECT_EQ(Status::Unauthenticated(), source_on_error_cb);
356 }
357
TEST_F(ServerReaderWriterTest,Close_ClearsCallAndChannelId)358 TEST_F(ServerReaderWriterTest, Close_ClearsCallAndChannelId) {
359 rpc_lock().lock();
360 reader_writer_.set_id(999);
361 EXPECT_NE(reader_writer_.channel_id_locked(), 0u);
362 rpc_lock().unlock();
363
364 EXPECT_EQ(OkStatus(), reader_writer_.Finish());
365
366 RpcLockGuard lock;
367 EXPECT_EQ(reader_writer_.id(), 0u);
368 EXPECT_EQ(reader_writer_.channel_id_locked(), 0u);
369 }
370
371 } // namespace
372 } // namespace internal
373 } // namespace pw::rpc
374