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