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