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