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/io_env.hpp>
23 : #include <boost/capy/ex/run_async.hpp>
24 : #include <boost/capy/io_result.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 40 : ~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 20 : 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 20 : : sock_(ctx)
105 20 : , fuse_(std::move(f))
106 20 : , max_read_size_(max_read_size)
107 20 : , max_write_size_(max_write_size)
108 : {
109 20 : if (max_read_size == 0)
110 MIS 0 : detail::throw_logic_error("mocket: max_read_size cannot be 0");
111 HIT 20 : if (max_write_size == 0)
112 MIS 0 : detail::throw_logic_error("mocket: max_write_size cannot be 0");
113 HIT 20 : }
114 :
115 : /** Move constructor.
116 : */
117 20 : basic_mocket(basic_mocket&& other) noexcept
118 20 : : sock_(std::move(other.sock_))
119 20 : , provide_(std::move(other.provide_))
120 20 : , expect_(std::move(other.expect_))
121 20 : , fuse_(std::move(other.fuse_))
122 20 : , max_read_size_(other.max_read_size_)
123 20 : , max_write_size_(other.max_write_size_)
124 : {
125 20 : }
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 22 : Socket& socket() noexcept
160 : {
161 22 : 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 5 : bool is_open() const noexcept
243 : {
244 5 : 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 io_env (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::io_env 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::io_result<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 io_env (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::io_env 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::io_result<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 : Optional `max_read_size` and `max_write_size` parameters limit the
561 : number of bytes transferred per I/O operation on the mocket,
562 : simulating chunked network delivery for testing purposes.
563 :
564 : @tparam Socket The socket type (default `tcp_socket`).
565 : @tparam Acceptor The acceptor type (default `tcp_acceptor`).
566 :
567 : @param ctx The I/O context for the sockets.
568 : @param f The fuse for error injection testing.
569 : @param max_read_size Maximum bytes per read operation (default unlimited).
570 : @param max_write_size Maximum bytes per write operation (default unlimited).
571 :
572 : @return A pair of (mocket, socket).
573 :
574 : @throws std::runtime_error if opening, binding, listening, accepting,
575 : or connecting fails.
576 :
577 : @note Mockets are not thread-safe and must be used in a
578 : single-threaded, deterministic context.
579 : */
580 : template<class Socket = tcp_socket, class Acceptor = tcp_acceptor>
581 : std::pair<basic_mocket<Socket>, Socket>
582 20 : make_mocket_pair(
583 : io_context& ctx,
584 : capy::test::fuse f = {},
585 : std::size_t max_read_size = std::size_t(-1),
586 : std::size_t max_write_size = std::size_t(-1))
587 : {
588 20 : auto ex = ctx.get_executor();
589 :
590 20 : basic_mocket<Socket> m(ctx, std::move(f), max_read_size, max_write_size);
591 :
592 20 : Socket peer(ctx);
593 :
594 20 : std::error_code accept_ec;
595 20 : std::error_code connect_ec;
596 20 : bool accept_done = false;
597 20 : bool connect_done = false;
598 :
599 20 : Acceptor acc(ctx);
600 20 : if (auto open_ec = acc.open())
601 MIS 0 : throw std::runtime_error("mocket open failed: " + open_ec.message());
602 HIT 20 : acc.set_option(socket_option::reuse_address(true));
603 20 : if (auto bind_ec = acc.bind(endpoint(ipv4_address::loopback(), 0)))
604 MIS 0 : throw std::runtime_error("mocket bind failed: " + bind_ec.message());
605 HIT 20 : if (auto listen_ec = acc.listen())
606 MIS 0 : throw std::runtime_error(
607 : "mocket listen failed: " + listen_ec.message());
608 HIT 20 : auto port = acc.local_endpoint().port();
609 :
610 20 : if (auto open_ec = peer.open())
611 MIS 0 : throw std::runtime_error("mocket open failed: " + open_ec.message());
612 :
613 HIT 20 : Socket accepted_socket(ctx);
614 :
615 20 : capy::run_async(ex)(
616 40 : [](Acceptor& a, Socket& s, std::error_code& ec_out,
617 : bool& done_out) -> capy::task<> {
618 : auto [ec] = co_await a.accept(s);
619 : ec_out = ec;
620 : done_out = true;
621 : }(acc, accepted_socket, accept_ec, accept_done));
622 :
623 40 : capy::run_async(ex)(
624 20 : [](Socket& s, endpoint ep, std::error_code& ec_out,
625 : bool& done_out) -> capy::task<> {
626 : auto [ec] = co_await s.connect(ep);
627 : ec_out = ec;
628 : done_out = true;
629 40 : }(peer, endpoint(ipv4_address::loopback(), port), connect_ec,
630 : connect_done));
631 :
632 20 : ctx.run();
633 20 : ctx.restart();
634 :
635 20 : if (!accept_done || accept_ec)
636 : {
637 MIS 0 : std::fprintf(
638 : stderr, "make_mocket_pair: accept failed (done=%d, ec=%s)\n",
639 : accept_done, accept_ec.message().c_str());
640 0 : acc.close();
641 0 : throw std::runtime_error("mocket accept failed");
642 : }
643 :
644 HIT 20 : if (!connect_done || connect_ec)
645 : {
646 MIS 0 : std::fprintf(
647 : stderr, "make_mocket_pair: connect failed (done=%d, ec=%s)\n",
648 : connect_done, connect_ec.message().c_str());
649 0 : acc.close();
650 0 : accepted_socket.close();
651 0 : throw std::runtime_error("mocket connect failed");
652 : }
653 :
654 HIT 20 : m.socket() = std::move(accepted_socket);
655 :
656 20 : acc.close();
657 :
658 40 : return {std::move(m), std::move(peer)};
659 20 : }
660 :
661 : } // namespace boost::corosio::test
662 :
663 : #endif
|
