• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1 //
2 //
3 // Copyright 2015 gRPC authors.
4 //
5 // Licensed under the Apache License, Version 2.0 (the "License");
6 // you may not use this file except in compliance with the License.
7 // You may obtain a copy of the License at
8 //
9 //     http://www.apache.org/licenses/LICENSE-2.0
10 //
11 // Unless required by applicable law or agreed to in writing, software
12 // distributed under the License is distributed on an "AS IS" BASIS,
13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 // See the License for the specific language governing permissions and
15 // limitations under the License.
16 //
17 //
18 
19 #include "test/core/end2end/cq_verifier.h"
20 
21 #include <grpc/byte_buffer.h>
22 #include <grpc/compression.h>
23 #include <grpc/grpc.h>
24 #include <grpc/slice.h>
25 #include <grpc/slice_buffer.h>
26 #include <grpc/support/time.h>
27 #include <inttypes.h>
28 #include <stdio.h>
29 #include <string.h>
30 
31 #include <algorithm>
32 #include <limits>
33 #include <string>
34 #include <utility>
35 #include <vector>
36 
37 #include "absl/log/check.h"
38 #include "absl/log/log.h"
39 #include "absl/strings/escaping.h"
40 #include "absl/strings/str_cat.h"
41 #include "absl/strings/str_format.h"
42 #include "absl/strings/str_join.h"
43 #include "absl/strings/string_view.h"
44 #include "gtest/gtest.h"
45 #include "src/core/lib/compression/message_compress.h"
46 #include "src/core/lib/surface/event_string.h"
47 #include "src/core/util/crash.h"
48 #include "src/core/util/match.h"
49 #include "test/core/test_util/build.h"
50 #include "test/core/test_util/test_config.h"
51 
52 // a set of metadata we expect to find on an event
53 typedef struct metadata {
54   size_t count;
55   size_t cap;
56   char** keys;
57   char** values;
58 } metadata;
59 
has_metadata(const grpc_metadata * md,size_t count,const char * key,const char * value)60 static int has_metadata(const grpc_metadata* md, size_t count, const char* key,
61                         const char* value) {
62   size_t i;
63   for (i = 0; i < count; i++) {
64     if (0 == grpc_slice_str_cmp(md[i].key, key) &&
65         0 == grpc_slice_str_cmp(md[i].value, value)) {
66       return 1;
67     }
68   }
69   return 0;
70 }
71 
contains_metadata(grpc_metadata_array * array,const char * key,const char * value)72 int contains_metadata(grpc_metadata_array* array, const char* key,
73                       const char* value) {
74   return has_metadata(array->metadata, array->count, key, value);
75 }
76 
has_metadata_slices(const grpc_metadata * md,size_t count,grpc_slice key,grpc_slice value)77 static int has_metadata_slices(const grpc_metadata* md, size_t count,
78                                grpc_slice key, grpc_slice value) {
79   size_t i;
80   for (i = 0; i < count; i++) {
81     if (grpc_slice_eq(md[i].key, key) && grpc_slice_eq(md[i].value, value)) {
82       return 1;
83     }
84   }
85   return 0;
86 }
87 
contains_metadata_slices(grpc_metadata_array * array,grpc_slice key,grpc_slice value)88 int contains_metadata_slices(grpc_metadata_array* array, grpc_slice key,
89                              grpc_slice value) {
90   return has_metadata_slices(array->metadata, array->count, key, value);
91 }
92 
merge_slices(grpc_slice * slices,size_t nslices)93 static grpc_slice merge_slices(grpc_slice* slices, size_t nslices) {
94   size_t i;
95   size_t len = 0;
96   uint8_t* cursor;
97   grpc_slice out;
98 
99   for (i = 0; i < nslices; i++) {
100     len += GRPC_SLICE_LENGTH(slices[i]);
101   }
102 
103   out = grpc_slice_malloc(len);
104   cursor = GRPC_SLICE_START_PTR(out);
105 
106   for (i = 0; i < nslices; i++) {
107     memcpy(cursor, GRPC_SLICE_START_PTR(slices[i]),
108            GRPC_SLICE_LENGTH(slices[i]));
109     cursor += GRPC_SLICE_LENGTH(slices[i]);
110   }
111 
112   return out;
113 }
114 
raw_byte_buffer_eq_slice(grpc_byte_buffer * rbb,grpc_slice b)115 int raw_byte_buffer_eq_slice(grpc_byte_buffer* rbb, grpc_slice b) {
116   grpc_slice a;
117   int ok;
118 
119   if (!rbb) return 0;
120 
121   a = merge_slices(rbb->data.raw.slice_buffer.slices,
122                    rbb->data.raw.slice_buffer.count);
123   ok = GRPC_SLICE_LENGTH(a) == GRPC_SLICE_LENGTH(b) &&
124        0 == memcmp(GRPC_SLICE_START_PTR(a), GRPC_SLICE_START_PTR(b),
125                    GRPC_SLICE_LENGTH(a));
126   if (!ok) {
127     LOG(ERROR) << "SLICE MISMATCH: left_length=" << GRPC_SLICE_LENGTH(a)
128                << " right_length=" << GRPC_SLICE_LENGTH(b);
129     std::string out;
130     const char* a_str = reinterpret_cast<const char*>(GRPC_SLICE_START_PTR(a));
131     const char* b_str = reinterpret_cast<const char*>(GRPC_SLICE_START_PTR(b));
132     for (size_t i = 0; i < std::max(GRPC_SLICE_LENGTH(a), GRPC_SLICE_LENGTH(b));
133          i++) {
134       if (i >= GRPC_SLICE_LENGTH(a)) {
135         absl::StrAppend(&out, "\u001b[36m",  // cyan
136                         absl::CEscape(absl::string_view(&b_str[i], 1)),
137                         "\u001b[0m");
138       } else if (i >= GRPC_SLICE_LENGTH(b)) {
139         absl::StrAppend(&out, "\u001b[35m",  // magenta
140                         absl::CEscape(absl::string_view(&a_str[i], 1)),
141                         "\u001b[0m");
142       } else if (a_str[i] == b_str[i]) {
143         absl::StrAppend(&out, absl::CEscape(absl::string_view(&a_str[i], 1)));
144       } else {
145         absl::StrAppend(&out, "\u001b[31m",  // red
146                         absl::CEscape(absl::string_view(&a_str[i], 1)),
147                         "\u001b[33m",  // yellow
148                         absl::CEscape(absl::string_view(&b_str[i], 1)),
149                         "\u001b[0m");
150       }
151       LOG(ERROR) << out;
152     }
153   }
154   grpc_slice_unref(a);
155   grpc_slice_unref(b);
156   return ok;
157 }
158 
byte_buffer_eq_slice(grpc_byte_buffer * bb,grpc_slice b)159 int byte_buffer_eq_slice(grpc_byte_buffer* bb, grpc_slice b) {
160   if (bb == nullptr) return 0;
161   if (bb->data.raw.compression > GRPC_COMPRESS_NONE) {
162     grpc_slice_buffer decompressed_buffer;
163     grpc_slice_buffer_init(&decompressed_buffer);
164     CHECK(grpc_msg_decompress(bb->data.raw.compression,
165                               &bb->data.raw.slice_buffer,
166                               &decompressed_buffer));
167     grpc_byte_buffer* rbb = grpc_raw_byte_buffer_create(
168         decompressed_buffer.slices, decompressed_buffer.count);
169     int ret_val = raw_byte_buffer_eq_slice(rbb, b);
170     grpc_byte_buffer_destroy(rbb);
171     grpc_slice_buffer_destroy(&decompressed_buffer);
172     return ret_val;
173   }
174   return raw_byte_buffer_eq_slice(bb, b);
175 }
176 
byte_buffer_eq_string(grpc_byte_buffer * bb,const char * str)177 int byte_buffer_eq_string(grpc_byte_buffer* bb, const char* str) {
178   return byte_buffer_eq_slice(bb, grpc_slice_from_copied_string(str));
179 }
180 
181 namespace {
IsProbablyInteger(void * p)182 bool IsProbablyInteger(void* p) {
183   return reinterpret_cast<uintptr_t>(p) < 1000000 ||
184          (reinterpret_cast<uintptr_t>(p) >
185           std::numeric_limits<uintptr_t>::max() - 10);
186 }
187 
TagStr(void * tag)188 std::string TagStr(void* tag) {
189   if (IsProbablyInteger(tag)) {
190     return absl::StrFormat("tag(%" PRIdPTR ")",
191                            reinterpret_cast<intptr_t>(tag));
192   } else {
193     return absl::StrFormat("%p", tag);
194   }
195 }
196 }  // namespace
197 
198 namespace grpc_core {
199 
CqVerifier(grpc_completion_queue * cq,absl::AnyInvocable<void (Failure)const> fail,absl::AnyInvocable<void (grpc_event_engine::experimental::EventEngine::Duration)const> step_fn)200 CqVerifier::CqVerifier(
201     grpc_completion_queue* cq, absl::AnyInvocable<void(Failure) const> fail,
202     absl::AnyInvocable<
203         void(grpc_event_engine::experimental::EventEngine::Duration) const>
204         step_fn)
205     : cq_(cq), fail_(std::move(fail)), step_fn_(std::move(step_fn)) {}
206 
~CqVerifier()207 CqVerifier::~CqVerifier() { Verify(); }
208 
ToString() const209 std::string CqVerifier::Expectation::ToString() const {
210   return absl::StrCat(
211       location.file(), ":", location.line(), ": ", TagStr(tag), " ",
212       Match(
213           result,
214           [](bool success) {
215             return absl::StrCat("success=", success ? "true" : "false");
216           },
217           [](Maybe) { return std::string("maybe"); },
218           [](AnyStatus) { return std::string("any success value"); },
219           [](const PerformAction&) {
220             return std::string("perform some action");
221           },
222           [](const MaybePerformAction&) {
223             return std::string("maybe perform action");
224           }));
225 }
226 
ToShortString() const227 std::string CqVerifier::Expectation::ToShortString() const {
228   return absl::StrCat(
229       TagStr(tag),
230       Match(
231           result,
232           [](bool success) -> std::string {
233             if (!success) return "-❌";
234             return "-✅";
235           },
236           [](Maybe) { return std::string("-❓"); },
237           [](AnyStatus) { return std::string("-��"); },
238           [](const PerformAction&) { return std::string("-��"); },
239           [](const MaybePerformAction&) { return std::string("-��❓"); }));
240 }
241 
ToStrings() const242 std::vector<std::string> CqVerifier::ToStrings() const {
243   std::vector<std::string> expectations;
244   expectations.reserve(expectations_.size());
245   for (const auto& e : expectations_) {
246     expectations.push_back(e.ToString());
247   }
248   return expectations;
249 }
250 
ToString() const251 std::string CqVerifier::ToString() const {
252   return absl::StrJoin(ToStrings(), "\n");
253 }
254 
ToShortStrings() const255 std::vector<std::string> CqVerifier::ToShortStrings() const {
256   std::vector<std::string> expectations;
257   expectations.reserve(expectations_.size());
258   for (const auto& e : expectations_) {
259     expectations.push_back(e.ToShortString());
260   }
261   return expectations;
262 }
263 
ToShortString() const264 std::string CqVerifier::ToShortString() const {
265   return absl::StrJoin(ToShortStrings(), " ");
266 }
267 
FailNoEventReceived(const SourceLocation & location) const268 void CqVerifier::FailNoEventReceived(const SourceLocation& location) const {
269   fail_(Failure{location, "No event received", ToStrings(), {}});
270 }
271 
FailUnexpectedEvent(grpc_event * ev,const SourceLocation & location) const272 void CqVerifier::FailUnexpectedEvent(grpc_event* ev,
273                                      const SourceLocation& location) const {
274   std::vector<std::string> message_details;
275   if (ev->type == GRPC_OP_COMPLETE && ev->success) {
276     auto successful_state_strings = successful_state_strings_.find(ev->tag);
277     if (successful_state_strings != successful_state_strings_.end()) {
278       for (SuccessfulStateString* sss : successful_state_strings->second) {
279         message_details.push_back(sss->GetSuccessfulStateString());
280       }
281     }
282   }
283   fail_(Failure{location,
284                 absl::StrCat("Unexpected event: ", grpc_event_string(ev)),
285                 ToStrings(), std::move(message_details)});
286 }
287 
288 namespace {
CrashMessage(const CqVerifier::Failure & failure)289 std::string CrashMessage(const CqVerifier::Failure& failure) {
290   std::string message = failure.message;
291   if (!failure.message_details.empty()) {
292     absl::StrAppend(&message, "\nwith:");
293     for (const auto& detail : failure.message_details) {
294       absl::StrAppend(&message, "\n  ", detail);
295     }
296   }
297   absl::StrAppend(&message, "\nchecked @ ", failure.location.file(), ":",
298                   failure.location.line());
299   if (!failure.expected.empty()) {
300     absl::StrAppend(&message, "\nexpected:\n");
301     for (const auto& line : failure.expected) {
302       absl::StrAppend(&message, "  ", line, "\n");
303     }
304   } else {
305     absl::StrAppend(&message, "\nexpected nothing");
306   }
307   return message;
308 }
309 }  // namespace
310 
FailUsingGprCrashWithStdio(const Failure & failure)311 void CqVerifier::FailUsingGprCrashWithStdio(const Failure& failure) {
312   CrashWithStdio(CrashMessage(failure));
313 }
314 
FailUsingGprCrash(const Failure & failure)315 void CqVerifier::FailUsingGprCrash(const Failure& failure) {
316   Crash(CrashMessage(failure));
317 }
318 
FailUsingGtestFail(const Failure & failure)319 void CqVerifier::FailUsingGtestFail(const Failure& failure) {
320   std::string message = absl::StrCat("  ", failure.message);
321   if (!failure.expected.empty()) {
322     absl::StrAppend(&message, "\n  expected:\n");
323     for (const auto& line : failure.expected) {
324       absl::StrAppend(&message, "    ", line, "\n");
325     }
326   } else {
327     absl::StrAppend(&message, "\n  expected nothing");
328   }
329   ADD_FAILURE_AT(failure.location.file(), failure.location.line()) << message;
330 }
331 
332 namespace {
IsMaybe(const CqVerifier::ExpectedResult & r)333 bool IsMaybe(const CqVerifier::ExpectedResult& r) {
334   return Match(
335       r, [](bool) { return false; }, [](CqVerifier::Maybe) { return true; },
336       [](CqVerifier::AnyStatus) { return false; },
337       [](const CqVerifier::PerformAction&) { return false; },
338       [](const CqVerifier::MaybePerformAction&) { return true; });
339 }
340 }  // namespace
341 
Step(gpr_timespec deadline)342 grpc_event CqVerifier::Step(gpr_timespec deadline) {
343   if (step_fn_ != nullptr) {
344     while (true) {
345       grpc_event r = grpc_completion_queue_next(
346           cq_, gpr_inf_past(deadline.clock_type), nullptr);
347       if (r.type != GRPC_QUEUE_TIMEOUT) return r;
348       auto now = gpr_now(deadline.clock_type);
349       if (gpr_time_cmp(deadline, now) < 0) break;
350       // Add a millisecond to ensure we overshoot the cq timeout if nothing is
351       // happening. Not doing so can lead to infinite loops in some tests.
352       // TODO(ctiller): see if there's a cleaner way to resolve this.
353       step_fn_(Timestamp::FromTimespecRoundDown(deadline) +
354                Duration::Milliseconds(1) - Timestamp::Now());
355     }
356     return grpc_event{GRPC_QUEUE_TIMEOUT, 0, nullptr};
357   }
358   return grpc_completion_queue_next(cq_, deadline, nullptr);
359 }
360 
Verify(Duration timeout,SourceLocation location)361 void CqVerifier::Verify(Duration timeout, SourceLocation location) {
362   if (expectations_.empty()) return;
363   bool must_log = true;
364   const gpr_timespec deadline =
365       grpc_timeout_milliseconds_to_deadline(timeout.millis());
366   while (!expectations_.empty()) {
367     must_log = std::exchange(added_expectations_, false) || must_log;
368     if (log_verifications_ && must_log) {
369       LOG(ERROR) << "Verify " << ToShortString() << " for " << timeout;
370     }
371     must_log = false;
372     grpc_event ev = Step(deadline);
373     if (ev.type == GRPC_QUEUE_TIMEOUT) break;
374     if (ev.type != GRPC_OP_COMPLETE) {
375       FailUnexpectedEvent(&ev, location);
376     }
377     bool found = false;
378     for (auto it = expectations_.begin(); it != expectations_.end(); ++it) {
379       if (it->tag != ev.tag) continue;
380       auto expectation = std::move(*it);
381       expectations_.erase(it);
382       const bool expected = Match(
383           expectation.result,
384           [ev](bool success) { return ev.success == success; },
385           [ev](Maybe m) {
386             if (m.seen != nullptr) *m.seen = true;
387             return ev.success != 0;
388           },
389           [ev](AnyStatus a) {
390             if (a.result != nullptr) *a.result = ev.success;
391             return true;
392           },
393           [ev](const PerformAction& action) {
394             action.action(ev.success);
395             return true;
396           },
397           [ev](const MaybePerformAction& action) {
398             action.action(ev.success);
399             return true;
400           });
401       if (!expected) {
402         FailUnexpectedEvent(&ev, location);
403       }
404       found = true;
405       break;
406     }
407     if (!found) FailUnexpectedEvent(&ev, location);
408     if (AllMaybes()) break;
409   }
410   expectations_.erase(
411       std::remove_if(expectations_.begin(), expectations_.end(),
412                      [](const Expectation& e) { return IsMaybe(e.result); }),
413       expectations_.end());
414   if (!expectations_.empty()) FailNoEventReceived(location);
415 }
416 
AllMaybes() const417 bool CqVerifier::AllMaybes() const {
418   for (const auto& e : expectations_) {
419     if (!IsMaybe(e.result)) return false;
420   }
421   return true;
422 }
423 
VerifyEmpty(Duration timeout,SourceLocation location)424 void CqVerifier::VerifyEmpty(Duration timeout, SourceLocation location) {
425   if (log_verifications_) {
426     LOG(ERROR) << "Verify empty completion queue for " << timeout;
427   }
428   const gpr_timespec deadline =
429       gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), timeout.as_timespec());
430   CHECK(expectations_.empty());
431   grpc_event ev = Step(deadline);
432   if (ev.type != GRPC_QUEUE_TIMEOUT) {
433     FailUnexpectedEvent(&ev, location);
434   }
435 }
436 
Expect(void * tag,ExpectedResult result,SourceLocation location)437 void CqVerifier::Expect(void* tag, ExpectedResult result,
438                         SourceLocation location) {
439   added_expectations_ = true;
440   expectations_.push_back(Expectation{location, tag, std::move(result)});
441 }
442 
AddSuccessfulStateString(void * tag,SuccessfulStateString * successful_state_string)443 void CqVerifier::AddSuccessfulStateString(
444     void* tag, SuccessfulStateString* successful_state_string) {
445   successful_state_strings_[tag].push_back(successful_state_string);
446 }
447 
ClearSuccessfulStateStrings(void * tag)448 void CqVerifier::ClearSuccessfulStateStrings(void* tag) {
449   successful_state_strings_.erase(tag);
450 }
451 
452 }  // namespace grpc_core
453