TLA Line data Source code
1 : //
2 : // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3 : // Copyright (c) 2026 Steve Gerbino
4 : //
5 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
6 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7 : //
8 : // Official repository: https://github.com/cppalliance/corosio
9 : //
10 :
11 : #ifndef BOOST_COROSIO_TEST_MOCKET_HPP
12 : #define BOOST_COROSIO_TEST_MOCKET_HPP
13 :
14 : #include <boost/corosio/detail/except.hpp>
15 : #include <boost/corosio/io_context.hpp>
16 : #include <boost/corosio/socket_option.hpp>
17 : #include <boost/corosio/tcp_acceptor.hpp>
18 : #include <boost/corosio/tcp_socket.hpp>
19 : #include <boost/capy/buffers/buffer_copy.hpp>
20 : #include <boost/capy/buffers/make_buffer.hpp>
21 : #include <boost/capy/error.hpp>
22 : #include <boost/capy/ex/run_async.hpp>
23 : #include <boost/capy/io_result.hpp>
24 : #include <boost/capy/task.hpp>
25 : #include <boost/capy/test/fuse.hpp>
26 :
27 : #include <cstddef>
28 : #include <cstdio>
29 : #include <cstring>
30 : #include <stdexcept>
31 : #include <string>
32 : #include <system_error>
33 : #include <utility>
34 :
35 : namespace boost::corosio::test {
36 :
37 : /** A mock socket for testing I/O operations.
38 :
39 : This class provides a testable socket-like interface where data
40 : can be staged for reading and expected data can be validated on
41 : writes. A mocket is paired with a regular socket using
42 : @ref make_mocket_pair, allowing bidirectional communication testing.
43 :
44 : When reading, data comes from the `provide()` buffer first.
45 : When writing, data is validated against the `expect()` buffer.
46 : Once buffers are exhausted, I/O passes through to the underlying
47 : socket connection.
48 :
49 : Satisfies the `capy::Stream` concept.
50 :
51 : @tparam Socket The underlying socket type (default `tcp_socket`).
52 :
53 : @par Thread Safety
54 : Not thread-safe. All operations must occur on a single thread.
55 : All coroutines using the mocket must be suspended when calling
56 : `expect()` or `provide()`.
57 :
58 : @see make_mocket_pair
59 : */
60 : template<class Socket = tcp_socket>
61 : class basic_mocket
62 : {
63 : Socket sock_;
64 : std::string provide_;
65 : std::string expect_;
66 : capy::test::fuse fuse_;
67 : std::size_t max_read_size_;
68 : std::size_t max_write_size_;
69 :
70 : template<class MutableBufferSequence>
71 : std::size_t consume_provide(MutableBufferSequence const& buffers) noexcept;
72 :
73 : template<class ConstBufferSequence>
74 : bool validate_expect(
75 : ConstBufferSequence const& buffers, std::size_t& bytes_written);
76 :
77 : public:
78 : template<class MutableBufferSequence>
79 : class read_some_awaitable;
80 :
81 : template<class ConstBufferSequence>
82 : class write_some_awaitable;
83 :
84 : /** Destructor.
85 : */
86 HIT 36 : ~basic_mocket() = default;
87 :
88 : /** Construct a mocket.
89 :
90 : @param ctx The execution context for the socket.
91 : @param f The fuse for error injection testing.
92 : @param max_read_size Maximum bytes per read operation.
93 : @param max_write_size Maximum bytes per write operation.
94 : */
95 18 : basic_mocket(
96 : capy::execution_context& ctx,
97 : capy::test::fuse f = {},
98 : std::size_t max_read_size = std::size_t(-1),
99 : std::size_t max_write_size = std::size_t(-1))
100 18 : : sock_(ctx)
101 18 : , fuse_(std::move(f))
102 18 : , max_read_size_(max_read_size)
103 18 : , max_write_size_(max_write_size)
104 : {
105 18 : if (max_read_size == 0)
106 MIS 0 : detail::throw_logic_error("mocket: max_read_size cannot be 0");
107 HIT 18 : if (max_write_size == 0)
108 MIS 0 : detail::throw_logic_error("mocket: max_write_size cannot be 0");
109 HIT 18 : }
110 :
111 : /** Move constructor.
112 : */
113 18 : basic_mocket(basic_mocket&& other) noexcept
114 18 : : sock_(std::move(other.sock_))
115 18 : , provide_(std::move(other.provide_))
116 18 : , expect_(std::move(other.expect_))
117 18 : , fuse_(std::move(other.fuse_))
118 18 : , max_read_size_(other.max_read_size_)
119 18 : , max_write_size_(other.max_write_size_)
120 : {
121 18 : }
122 :
123 : /** Move assignment.
124 : */
125 : basic_mocket& operator=(basic_mocket&& other) noexcept
126 : {
127 : if (this != &other)
128 : {
129 : sock_ = std::move(other.sock_);
130 : provide_ = std::move(other.provide_);
131 : expect_ = std::move(other.expect_);
132 : fuse_ = other.fuse_;
133 : max_read_size_ = other.max_read_size_;
134 : max_write_size_ = other.max_write_size_;
135 : }
136 : return *this;
137 : }
138 :
139 : basic_mocket(basic_mocket const&) = delete;
140 : basic_mocket& operator=(basic_mocket const&) = delete;
141 :
142 : /** Return the execution context.
143 :
144 : @return Reference to the execution context that owns this mocket.
145 : */
146 : capy::execution_context& context() const noexcept
147 : {
148 : return sock_.context();
149 : }
150 :
151 : /** Return the underlying socket.
152 :
153 : @return Reference to the underlying socket.
154 : */
155 20 : Socket& socket() noexcept
156 : {
157 20 : return sock_;
158 : }
159 :
160 : /** Stage data for reads.
161 :
162 : Appends the given string to this mocket's provide buffer.
163 : When `read_some` is called, it will receive this data first
164 : before reading from the underlying socket.
165 :
166 : @param s The data to provide.
167 :
168 : @pre All coroutines using this mocket must be suspended.
169 : */
170 9 : void provide(std::string const& s)
171 : {
172 9 : provide_.append(s);
173 9 : }
174 :
175 : /** Set expected data for writes.
176 :
177 : Appends the given string to this mocket's expect buffer.
178 : When the caller writes to this mocket, the written data
179 : must match the expected data. On mismatch, `fuse::fail()`
180 : is called.
181 :
182 : @param s The expected data.
183 :
184 : @pre All coroutines using this mocket must be suspended.
185 : */
186 8 : void expect(std::string const& s)
187 : {
188 8 : expect_.append(s);
189 8 : }
190 :
191 : /** Close the mocket and verify test expectations.
192 :
193 : Closes the underlying socket and verifies that both the
194 : `expect()` and `provide()` buffers are empty. If either
195 : buffer contains unconsumed data, returns `test_failure`
196 : and calls `fuse::fail()`.
197 :
198 : @return An error code indicating success or failure.
199 : Returns `error::test_failure` if buffers are not empty.
200 : */
201 18 : std::error_code close()
202 : {
203 18 : if (!sock_.is_open())
204 MIS 0 : return {};
205 :
206 HIT 18 : if (!expect_.empty())
207 : {
208 2 : fuse_.fail();
209 2 : sock_.close();
210 2 : return capy::error::test_failure;
211 : }
212 16 : if (!provide_.empty())
213 : {
214 2 : fuse_.fail();
215 2 : sock_.close();
216 2 : return capy::error::test_failure;
217 : }
218 :
219 14 : sock_.close();
220 14 : return {};
221 : }
222 :
223 : /** Cancel pending I/O operations.
224 :
225 : Cancels any pending asynchronous operations on the underlying
226 : socket. Outstanding operations complete with `cond::canceled`.
227 : */
228 : void cancel()
229 : {
230 : sock_.cancel();
231 : }
232 :
233 : /** Check if the mocket is open.
234 :
235 : @return `true` if the mocket is open.
236 : */
237 5 : bool is_open() const noexcept
238 : {
239 5 : return sock_.is_open();
240 : }
241 :
242 : /** Initiate an asynchronous read operation.
243 :
244 : Reads available data into the provided buffer sequence. If the
245 : provide buffer has data, it is consumed first. Otherwise, the
246 : operation delegates to the underlying socket.
247 :
248 : @param buffers The buffer sequence to read data into.
249 :
250 : @return An awaitable yielding `(error_code, std::size_t)`.
251 : */
252 : template<class MutableBufferSequence>
253 11 : auto read_some(MutableBufferSequence const& buffers)
254 : {
255 11 : return read_some_awaitable<MutableBufferSequence>(*this, buffers);
256 : }
257 :
258 : /** Initiate an asynchronous write operation.
259 :
260 : Writes data from the provided buffer sequence. If the expect
261 : buffer has data, it is validated. Otherwise, the operation
262 : delegates to the underlying socket.
263 :
264 : @param buffers The buffer sequence containing data to write.
265 :
266 : @return An awaitable yielding `(error_code, std::size_t)`.
267 : */
268 : template<class ConstBufferSequence>
269 8 : auto write_some(ConstBufferSequence const& buffers)
270 : {
271 8 : return write_some_awaitable<ConstBufferSequence>(*this, buffers);
272 : }
273 : };
274 :
275 : /// Default mocket type using `tcp_socket`.
276 : using mocket = basic_mocket<>;
277 :
278 : template<class Socket>
279 : template<class MutableBufferSequence>
280 : std::size_t
281 10 : basic_mocket<Socket>::consume_provide(
282 : MutableBufferSequence const& buffers) noexcept
283 : {
284 : auto n =
285 10 : capy::buffer_copy(buffers, capy::make_buffer(provide_), max_read_size_);
286 10 : provide_.erase(0, n);
287 10 : return n;
288 : }
289 :
290 : template<class Socket>
291 : template<class ConstBufferSequence>
292 : bool
293 7 : basic_mocket<Socket>::validate_expect(
294 : ConstBufferSequence const& buffers, std::size_t& bytes_written)
295 : {
296 7 : if (expect_.empty())
297 MIS 0 : return true;
298 :
299 : // Build the write data up to max_write_size_
300 HIT 7 : std::string written;
301 7 : auto total = capy::buffer_size(buffers);
302 7 : if (total > max_write_size_)
303 1 : total = max_write_size_;
304 7 : written.resize(total);
305 7 : capy::buffer_copy(capy::make_buffer(written), buffers, max_write_size_);
306 :
307 : // Check if written data matches expect prefix
308 7 : auto const match_size = (std::min)(written.size(), expect_.size());
309 7 : if (std::memcmp(written.data(), expect_.data(), match_size) != 0)
310 : {
311 MIS 0 : fuse_.fail();
312 0 : bytes_written = 0;
313 0 : return false;
314 : }
315 :
316 : // Consume matched portion
317 HIT 7 : expect_.erase(0, match_size);
318 7 : bytes_written = written.size();
319 7 : return true;
320 7 : }
321 :
322 : template<class Socket>
323 : template<class MutableBufferSequence>
324 : class basic_mocket<Socket>::read_some_awaitable
325 : {
326 : using sock_awaitable = decltype(std::declval<Socket&>().read_some(
327 : std::declval<MutableBufferSequence>()));
328 :
329 : basic_mocket* m_;
330 : MutableBufferSequence buffers_;
331 : std::size_t n_ = 0;
332 : std::error_code ec_;
333 : union
334 : {
335 : char dummy_;
336 : sock_awaitable underlying_;
337 : };
338 : bool sync_ = true;
339 :
340 : public:
341 11 : read_some_awaitable(basic_mocket& m, MutableBufferSequence buffers) noexcept
342 11 : : m_(&m)
343 11 : , buffers_(std::move(buffers))
344 : {
345 11 : }
346 :
347 22 : ~read_some_awaitable()
348 : {
349 22 : if (!sync_)
350 1 : underlying_.~sock_awaitable();
351 22 : }
352 :
353 11 : read_some_awaitable(read_some_awaitable&& other) noexcept
354 11 : : m_(other.m_)
355 11 : , buffers_(std::move(other.buffers_))
356 11 : , n_(other.n_)
357 11 : , ec_(other.ec_)
358 11 : , sync_(other.sync_)
359 : {
360 11 : if (!sync_)
361 : {
362 MIS 0 : new (&underlying_) sock_awaitable(std::move(other.underlying_));
363 0 : other.underlying_.~sock_awaitable();
364 0 : other.sync_ = true;
365 : }
366 HIT 11 : }
367 :
368 : read_some_awaitable(read_some_awaitable const&) = delete;
369 : read_some_awaitable& operator=(read_some_awaitable const&) = delete;
370 : read_some_awaitable& operator=(read_some_awaitable&&) = delete;
371 :
372 11 : bool await_ready()
373 : {
374 : // Fuse injection point: an armed fuse fails this read as if the
375 : // transport did, so a fault-injection sweep exercises the error
376 : // path of every read the caller issues. Inert outside armed().
377 : // A transport reports failure through the result, never by
378 : // throwing from read_some, so the fuse's exception phase is
379 : // converted to the same error code its error-code phase yields.
380 11 : std::error_code fec;
381 : try
382 : {
383 11 : fec = m_->fuse_.maybe_fail();
384 : }
385 MIS 0 : catch (std::system_error const& e)
386 : {
387 0 : fec = e.code();
388 : }
389 HIT 11 : if (fec)
390 : {
391 MIS 0 : ec_ = fec;
392 0 : n_ = 0;
393 0 : return true;
394 : }
395 HIT 11 : if (!m_->provide_.empty())
396 : {
397 10 : n_ = m_->consume_provide(buffers_);
398 10 : return true;
399 : }
400 1 : new (&underlying_) sock_awaitable(m_->sock_.read_some(buffers_));
401 1 : sync_ = false;
402 1 : return underlying_.await_ready();
403 : }
404 :
405 : template<class... Args>
406 1 : auto await_suspend(Args&&... args)
407 : {
408 1 : return underlying_.await_suspend(std::forward<Args>(args)...);
409 : }
410 :
411 11 : capy::io_result<std::size_t> await_resume()
412 : {
413 11 : if (sync_)
414 10 : return {ec_, n_};
415 1 : return underlying_.await_resume();
416 : }
417 : };
418 :
419 : template<class Socket>
420 : template<class ConstBufferSequence>
421 : class basic_mocket<Socket>::write_some_awaitable
422 : {
423 : using sock_awaitable = decltype(std::declval<Socket&>().write_some(
424 : std::declval<ConstBufferSequence>()));
425 :
426 : basic_mocket* m_;
427 : ConstBufferSequence buffers_;
428 : std::size_t n_ = 0;
429 : std::error_code ec_;
430 : union
431 : {
432 : char dummy_;
433 : sock_awaitable underlying_;
434 : };
435 : bool sync_ = true;
436 :
437 : public:
438 8 : write_some_awaitable(basic_mocket& m, ConstBufferSequence buffers) noexcept
439 8 : : m_(&m)
440 8 : , buffers_(std::move(buffers))
441 : {
442 8 : }
443 :
444 16 : ~write_some_awaitable()
445 : {
446 16 : if (!sync_)
447 1 : underlying_.~sock_awaitable();
448 16 : }
449 :
450 8 : write_some_awaitable(write_some_awaitable&& other) noexcept
451 8 : : m_(other.m_)
452 8 : , buffers_(std::move(other.buffers_))
453 8 : , n_(other.n_)
454 8 : , ec_(other.ec_)
455 8 : , sync_(other.sync_)
456 : {
457 8 : if (!sync_)
458 : {
459 MIS 0 : new (&underlying_) sock_awaitable(std::move(other.underlying_));
460 0 : other.underlying_.~sock_awaitable();
461 0 : other.sync_ = true;
462 : }
463 HIT 8 : }
464 :
465 : write_some_awaitable(write_some_awaitable const&) = delete;
466 : write_some_awaitable& operator=(write_some_awaitable const&) = delete;
467 : write_some_awaitable& operator=(write_some_awaitable&&) = delete;
468 :
469 8 : bool await_ready()
470 : {
471 : // Fuse injection point: an armed fuse fails this write as if the
472 : // transport did, so a fault-injection sweep exercises the error
473 : // path of every write the caller issues. Inert outside armed().
474 : // A transport reports failure through the result, never by
475 : // throwing from write_some, so the fuse's exception phase is
476 : // converted to the same error code its error-code phase yields.
477 8 : std::error_code fec;
478 : try
479 : {
480 8 : fec = m_->fuse_.maybe_fail();
481 : }
482 MIS 0 : catch (std::system_error const& e)
483 : {
484 0 : fec = e.code();
485 : }
486 HIT 8 : if (fec)
487 : {
488 MIS 0 : ec_ = fec;
489 0 : n_ = 0;
490 0 : return true;
491 : }
492 HIT 8 : if (!m_->expect_.empty())
493 : {
494 7 : if (!m_->validate_expect(buffers_, n_))
495 : {
496 MIS 0 : ec_ = capy::error::test_failure;
497 0 : n_ = 0;
498 : }
499 HIT 7 : return true;
500 : }
501 1 : new (&underlying_) sock_awaitable(m_->sock_.write_some(buffers_));
502 1 : sync_ = false;
503 1 : return underlying_.await_ready();
504 : }
505 :
506 : template<class... Args>
507 1 : auto await_suspend(Args&&... args)
508 : {
509 1 : return underlying_.await_suspend(std::forward<Args>(args)...);
510 : }
511 :
512 8 : capy::io_result<std::size_t> await_resume()
513 : {
514 8 : if (sync_)
515 7 : return {ec_, n_};
516 1 : return underlying_.await_resume();
517 : }
518 : };
519 :
520 : /** Create a mocket paired with a socket.
521 :
522 : Creates a mocket and a socket connected via loopback.
523 : Data written to one can be read from the other.
524 :
525 : The mocket has fuse checks enabled via `maybe_fail()` and
526 : supports provide/expect buffers for test instrumentation.
527 : The socket is the "peer" end with no test instrumentation.
528 :
529 : Optional max_read_size and max_write_size parameters limit the
530 : number of bytes transferred per I/O operation on the mocket,
531 : simulating chunked network delivery for testing purposes.
532 :
533 : @tparam Socket The socket type (default `tcp_socket`).
534 : @tparam Acceptor The acceptor type (default `tcp_acceptor`).
535 :
536 : @param ctx The I/O context for the sockets.
537 : @param f The fuse for error injection testing.
538 : @param max_read_size Maximum bytes per read operation (default unlimited).
539 : @param max_write_size Maximum bytes per write operation (default unlimited).
540 :
541 : @return A pair of (mocket, socket).
542 :
543 : @note Mockets are not thread-safe and must be used in a
544 : single-threaded, deterministic context.
545 : */
546 : template<class Socket = tcp_socket, class Acceptor = tcp_acceptor>
547 : std::pair<basic_mocket<Socket>, Socket>
548 18 : make_mocket_pair(
549 : io_context& ctx,
550 : capy::test::fuse f = {},
551 : std::size_t max_read_size = std::size_t(-1),
552 : std::size_t max_write_size = std::size_t(-1))
553 : {
554 18 : auto ex = ctx.get_executor();
555 :
556 18 : basic_mocket<Socket> m(ctx, std::move(f), max_read_size, max_write_size);
557 :
558 18 : Socket peer(ctx);
559 :
560 18 : std::error_code accept_ec;
561 18 : std::error_code connect_ec;
562 18 : bool accept_done = false;
563 18 : bool connect_done = false;
564 :
565 18 : Acceptor acc(ctx);
566 18 : acc.open();
567 18 : acc.set_option(socket_option::reuse_address(true));
568 18 : if (auto bind_ec = acc.bind(endpoint(ipv4_address::loopback(), 0)))
569 MIS 0 : throw std::runtime_error("mocket bind failed: " + bind_ec.message());
570 HIT 18 : if (auto listen_ec = acc.listen())
571 MIS 0 : throw std::runtime_error(
572 : "mocket listen failed: " + listen_ec.message());
573 HIT 18 : auto port = acc.local_endpoint().port();
574 :
575 18 : peer.open();
576 :
577 18 : Socket accepted_socket(ctx);
578 :
579 18 : capy::run_async(ex)(
580 36 : [](Acceptor& a, Socket& s, std::error_code& ec_out,
581 : bool& done_out) -> capy::task<> {
582 : auto [ec] = co_await a.accept(s);
583 : ec_out = ec;
584 : done_out = true;
585 : }(acc, accepted_socket, accept_ec, accept_done));
586 :
587 18 : capy::run_async(ex)(
588 36 : [](Socket& s, endpoint ep, std::error_code& ec_out,
589 : bool& done_out) -> capy::task<> {
590 : auto [ec] = co_await s.connect(ep);
591 : ec_out = ec;
592 : done_out = true;
593 : }(peer, endpoint(ipv4_address::loopback(), port), connect_ec,
594 : connect_done));
595 :
596 18 : ctx.run();
597 18 : ctx.restart();
598 :
599 18 : if (!accept_done || accept_ec)
600 : {
601 MIS 0 : std::fprintf(
602 : stderr, "make_mocket_pair: accept failed (done=%d, ec=%s)\n",
603 : accept_done, accept_ec.message().c_str());
604 0 : acc.close();
605 0 : throw std::runtime_error("mocket accept failed");
606 : }
607 :
608 HIT 18 : if (!connect_done || connect_ec)
609 : {
610 MIS 0 : std::fprintf(
611 : stderr, "make_mocket_pair: connect failed (done=%d, ec=%s)\n",
612 : connect_done, connect_ec.message().c_str());
613 0 : acc.close();
614 0 : accepted_socket.close();
615 0 : throw std::runtime_error("mocket connect failed");
616 : }
617 :
618 HIT 18 : m.socket() = std::move(accepted_socket);
619 :
620 18 : acc.close();
621 :
622 36 : return {std::move(m), std::move(peer)};
623 18 : }
624 :
625 : } // namespace boost::corosio::test
626 :
627 : #endif
|
