1 // Copyright 2022 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/synchronous_call.h"
16
17 #include <chrono>
18
19 #include "pw_chrono/system_clock.h"
20 #include "pw_rpc/channel.h"
21 #include "pw_rpc/internal/packet.h"
22 #include "pw_rpc/nanopb/fake_channel_output.h"
23 #include "pw_rpc_test_protos/test.rpc.pb.h"
24 #include "pw_status/status.h"
25 #include "pw_status/status_with_size.h"
26 #include "pw_thread/thread.h"
27 #include "pw_unit_test/framework.h"
28 #include "pw_work_queue/test_thread.h"
29 #include "pw_work_queue/work_queue.h"
30
31 namespace pw::rpc::test {
32 namespace {
33
34 using pw::rpc::test::pw_rpc::nanopb::TestService;
35 using MethodInfo = internal::MethodInfo<TestService::TestUnaryRpc>;
36
37 #if PW_THREAD_JOINING_ENABLED // join() is required
38
39 class SynchronousCallTest : public ::testing::Test {
40 public:
SynchronousCallTest()41 SynchronousCallTest()
42 : channels_({{Channel::Create<42>(&fake_output_)}}), client_(channels_) {}
43
SetUp()44 void SetUp() override {
45 work_thread_ =
46 Thread(work_queue::test::WorkQueueThreadOptions(), work_queue_);
47 }
48
TearDown()49 void TearDown() override {
50 work_queue_.RequestStop();
51 work_thread_.join();
52 }
53
54 protected:
55 using FakeChannelOutput = NanopbFakeChannelOutput<2>;
56
OnSend(span<const std::byte> buffer,Status status)57 void OnSend(span<const std::byte> buffer, Status status) {
58 if (!status.ok()) {
59 return;
60 }
61 auto result = internal::Packet::FromBuffer(buffer);
62 EXPECT_TRUE(result.ok());
63 request_packet_ = *result;
64
65 EXPECT_TRUE(work_queue_.PushWork([this]() { SendResponse(); }).ok());
66 }
67
SendResponse()68 void SendResponse() {
69 std::array<std::byte, 256> buffer;
70 std::array<std::byte, 32> payload_buffer;
71
72 StatusWithSize size_status =
73 MethodInfo::serde().response().Encode(&response_, payload_buffer);
74 EXPECT_TRUE(size_status.ok());
75
76 auto response =
77 internal::Packet::Response(request_packet_, response_status_);
78 response.set_payload({payload_buffer.data(), size_status.size()});
79 EXPECT_TRUE(client_.ProcessPacket(response.Encode(buffer).value()).ok());
80 }
81
set_response(const pw_rpc_test_TestResponse & response,Status response_status=OkStatus ())82 void set_response(const pw_rpc_test_TestResponse& response,
83 Status response_status = OkStatus()) {
84 response_ = response;
85 response_status_ = response_status;
86 output().set_on_send([this](span<const std::byte> buffer, Status status) {
87 OnSend(buffer, status);
88 });
89 }
90
generated_client()91 MethodInfo::GeneratedClient generated_client() {
92 return MethodInfo::GeneratedClient(client(), channel().id());
93 }
94
output()95 FakeChannelOutput& output() { return fake_output_; }
channel() const96 const Channel& channel() const { return channels_.front(); }
client()97 Client& client() { return client_; }
98
99 private:
100 FakeChannelOutput fake_output_;
101 std::array<Channel, 1> channels_;
102 Client client_;
103 Thread work_thread_;
104 work_queue::WorkQueueWithBuffer<1> work_queue_;
105 pw_rpc_test_TestResponse response_{};
106 Status response_status_ = OkStatus();
107 internal::Packet request_packet_;
108 };
109
TEST_F(SynchronousCallTest,SynchronousCallSuccess)110 TEST_F(SynchronousCallTest, SynchronousCallSuccess) {
111 pw_rpc_test_TestRequest request{.integer = 5, .status_code = 0};
112 pw_rpc_test_TestResponse response{.value = 42, .repeated_field{}};
113
114 set_response(response, OkStatus());
115
116 auto result = SynchronousCall<TestService::TestUnaryRpc>(
117 client(), channel().id(), request);
118 EXPECT_TRUE(result.ok());
119 EXPECT_EQ(result.response().value, 42);
120 }
121
TEST_F(SynchronousCallTest,SynchronousCallServerError)122 TEST_F(SynchronousCallTest, SynchronousCallServerError) {
123 pw_rpc_test_TestRequest request{.integer = 5, .status_code = 0};
124 pw_rpc_test_TestResponse response{.value = 42, .repeated_field{}};
125
126 set_response(response, Status::Internal());
127
128 auto result = SynchronousCall<TestService::TestUnaryRpc>(
129 client(), channel().id(), request);
130 EXPECT_TRUE(result.is_error());
131 EXPECT_EQ(result.status(), Status::Internal());
132
133 // We should still receive the response
134 EXPECT_TRUE(result.is_server_response());
135 EXPECT_EQ(result.response().value, 42);
136 }
137
TEST_F(SynchronousCallTest,SynchronousCallRpcError)138 TEST_F(SynchronousCallTest, SynchronousCallRpcError) {
139 pw_rpc_test_TestRequest request{.integer = 5, .status_code = 0};
140
141 // Internally, if Channel receives a non-ok status from the
142 // ChannelOutput::Send, it will always return Unknown.
143 output().set_send_status(Status::Unknown());
144
145 auto result = SynchronousCall<TestService::TestUnaryRpc>(
146 client(), channel().id(), request);
147 EXPECT_TRUE(result.is_rpc_error());
148 EXPECT_EQ(result.status(), Status::Unknown());
149 }
150
TEST_F(SynchronousCallTest,SynchronousCallForTimeoutError)151 TEST_F(SynchronousCallTest, SynchronousCallForTimeoutError) {
152 pw_rpc_test_TestRequest request{.integer = 5, .status_code = 0};
153
154 auto result = SynchronousCallFor<TestService::TestUnaryRpc>(
155 client(),
156 channel().id(),
157 request,
158 chrono::SystemClock::for_at_least(std::chrono::milliseconds(1)));
159
160 EXPECT_TRUE(result.is_timeout());
161 EXPECT_EQ(result.status(), Status::DeadlineExceeded());
162 }
163
TEST_F(SynchronousCallTest,SynchronousCallUntilTimeoutError)164 TEST_F(SynchronousCallTest, SynchronousCallUntilTimeoutError) {
165 pw_rpc_test_TestRequest request{.integer = 5, .status_code = 0};
166
167 auto result = SynchronousCallUntil<TestService::TestUnaryRpc>(
168 client(), channel().id(), request, chrono::SystemClock::now());
169
170 EXPECT_TRUE(result.is_timeout());
171 EXPECT_EQ(result.status(), Status::DeadlineExceeded());
172 }
173
TEST_F(SynchronousCallTest,GeneratedClientSynchronousCallSuccess)174 TEST_F(SynchronousCallTest, GeneratedClientSynchronousCallSuccess) {
175 pw_rpc_test_TestRequest request{.integer = 5, .status_code = 0};
176 pw_rpc_test_TestResponse response{.value = 42, .repeated_field{}};
177
178 set_response(response, OkStatus());
179
180 auto result =
181 SynchronousCall<TestService::TestUnaryRpc>(generated_client(), request);
182 EXPECT_TRUE(result.ok());
183 EXPECT_EQ(result.response().value, 42);
184 }
185
TEST_F(SynchronousCallTest,GeneratedClientSynchronousCallServerError)186 TEST_F(SynchronousCallTest, GeneratedClientSynchronousCallServerError) {
187 pw_rpc_test_TestRequest request{.integer = 5, .status_code = 0};
188 pw_rpc_test_TestResponse response{.value = 42, .repeated_field{}};
189
190 set_response(response, Status::Internal());
191
192 auto result =
193 SynchronousCall<TestService::TestUnaryRpc>(generated_client(), request);
194 EXPECT_TRUE(result.is_error());
195 EXPECT_EQ(result.status(), Status::Internal());
196
197 // We should still receive the response
198 EXPECT_TRUE(result.is_server_response());
199 EXPECT_EQ(result.response().value, 42);
200 }
201
TEST_F(SynchronousCallTest,GeneratedClientSynchronousCallRpcError)202 TEST_F(SynchronousCallTest, GeneratedClientSynchronousCallRpcError) {
203 pw_rpc_test_TestRequest request{.integer = 5, .status_code = 0};
204
205 // Internally, if Channel receives a non-ok status from the
206 // ChannelOutput::Send, it will always return Unknown.
207 output().set_send_status(Status::Unknown());
208
209 auto result =
210 SynchronousCall<TestService::TestUnaryRpc>(generated_client(), request);
211 EXPECT_TRUE(result.is_rpc_error());
212 EXPECT_EQ(result.status(), Status::Unknown());
213 }
214
TEST_F(SynchronousCallTest,GeneratedClientSynchronousCallForTimeoutError)215 TEST_F(SynchronousCallTest, GeneratedClientSynchronousCallForTimeoutError) {
216 pw_rpc_test_TestRequest request{.integer = 5, .status_code = 0};
217
218 auto result = SynchronousCallFor<TestService::TestUnaryRpc>(
219 generated_client(),
220 request,
221 chrono::SystemClock::for_at_least(std::chrono::milliseconds(1)));
222
223 EXPECT_TRUE(result.is_timeout());
224 EXPECT_EQ(result.status(), Status::DeadlineExceeded());
225 }
226
TEST_F(SynchronousCallTest,GeneratedClientSynchronousCallUntilTimeoutError)227 TEST_F(SynchronousCallTest, GeneratedClientSynchronousCallUntilTimeoutError) {
228 pw_rpc_test_TestRequest request{.integer = 5, .status_code = 0};
229
230 auto result = SynchronousCallUntil<TestService::TestUnaryRpc>(
231 generated_client(), request, chrono::SystemClock::now());
232
233 EXPECT_TRUE(result.is_timeout());
234 EXPECT_EQ(result.status(), Status::DeadlineExceeded());
235 }
236
237 #endif // PW_THREAD_JOINING_ENABLED
238
239 } // namespace
240 } // namespace pw::rpc::test
241