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