ligo/tests/test.cpp

#include <atomic>
#include <chrono>
#include <cstdint>
#include <fstream>
#include <future>
#include <memory>
#include <string>
#include <thread>
#include <type_traits>
#include <utility>
#include <vector>
#include <expected>

#include <boost/asio/as_tuple.hpp>
#include <boost/asio/awaitable.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/executor_work_guard.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/ip/udp.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/this_coro.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/use_future.hpp>
#include <boost/asio/write.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <boost/json/object.hpp>
#include <boost/json/parse.hpp>
#include <boost/json/serialize.hpp>
#include <boost/json/value.hpp>

#include <modulus.hpp>

#include <ligo/config.hpp>
#include <ligo/http.hpp>
#include <ligo/net.hpp>
#include <ligo/security.hpp>
#include <ligo/service.hpp>
#include <ligo/detail/errors.hpp>
#include <ligo/tcp.hpp>
#include <ligo/udp.hpp>

#include <test.hpp>


BOOST_AUTO_TEST_CASE(ligo_asserts_follow_build_type) {
#if defined(LIGO_EXPECT_NDEBUG) && !defined(NDEBUG)
  BOOST_FAIL("release-style builds must define NDEBUG");
#endif
#if defined(LIGO_EXPECT_DEBUG_ASSERTS) && defined(NDEBUG)
  BOOST_FAIL("debug builds must not define NDEBUG");
#endif
}

namespace {

auto test_network_failure(boost::system::error_code const &error) -> modulus::error {
  return ligo::detail::make_error(ligo::detail::error_unavailable, error.message());
}

auto run_two_threads(boost::asio::io_context &io) -> void {
  auto first = std::thread{[&] {
    io.run();
  }};
  auto second = std::thread{[&] {
    io.run();
  }};
  first.join();
  second.join();
}

auto start_io_threads(boost::asio::io_context &io) -> std::vector<std::thread> {
  auto threads = std::vector<std::thread>{};
  threads.emplace_back([&] {
    io.run();
  });
  threads.emplace_back([&] {
    io.run();
  });
  return threads;
}

auto join_io_threads(std::vector<std::thread> &threads) -> void {
  for (auto &thread : threads) {
    thread.join();
  }
}

auto delayed_echo(std::chrono::milliseconds delay, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  auto timer = boost::asio::steady_timer{co_await boost::asio::this_coro::executor};
  timer.expires_after(delay);
  co_await timer.async_wait(boost::asio::as_tuple(boost::asio::use_awaitable));
  co_return modulus::success(std::move(payload));
}

auto test_certificate_pem() -> std::string_view {
  return R"pem(-----BEGIN CERTIFICATE-----
MIIDJzCCAg+gAwIBAgIUJxLnHId20YVw1DjhofsjGza7C3AwDQYJKoZIhvcNAQEL
BQAwFDESMBAGA1UEAwwJbG9jYWxob3N0MCAXDTI2MDcyNDAzMjcyNFoYDzIxMjYw
NjMwMDMyNzI0WjAUMRIwEAYDVQQDDAlsb2NhbGhvc3QwggEiMA0GCSqGSIb3DQEB
AQUAA4IBDwAwggEKAoIBAQC3hGPOsWT4xHja/FgIs/Bj/aIjT+vJiFddPbwtgf68
gm81cF6zq9oBVWqTXTLQPdURhLN30/k3W8miB4Z0TEfVjhUTwLpa9gPX56zccBaK
dmTE4Gv2+++vXtgDLi8Z8kv3hb1nFABuKtxhB+OYjcoqCEyhhLxrsT1p7WPB7MFQ
johh8mlLixIoIXHFFLuRgkXYN77NuXDv0HW0ur4/YwSyJSSdzpN+KmonrWWVZxgE
30HgZC2MTYStNT3ypXsy2qIcH9AryoshGtaDC0DE7MGv82e3p+8W9iSqPSfjYLWX
ajHszOFtQ0KVGVYoUL0gl+ZwEOn3ZgALJ8F56QzSZ5qJAgMBAAGjbzBtMB0GA1Ud
DgQWBBQ25CRCNG/7URDhEovtBOU0nclsmDAfBgNVHSMEGDAWgBQ25CRCNG/7URDh
EovtBOU0nclsmDAPBgNVHRMBAf8EBTADAQH/MBoGA1UdEQQTMBGCCWxvY2FsaG9z
dIcEfwAAATANBgkqhkiG9w0BAQsFAAOCAQEAMVC7GWYzvuFrvxjeRqgvZru/QMun
6f6j7xPbsM2qyX9kXbEYwvb0U6GgVaXDTYUdn3UJ+e2iXDyjFMHALfJQIAshy58H
h+KSyrPQ1ketj5sO0q0sMWB0cU0nxiPxpOK+HIasdDBGubeTkTwz5AxyGUY9JdPx
F2oRnG1lHNhp2GzdoBCtymer/tgDCxqkg/FAvSJe0lTNJks6AQ1AdjHmT8R0Vhav
dkWkf0MzZNqoT1pIrPssXcnOAeB4KXhtVIgE9XgMgcX3K620eMwt7f4Qv7I0Lb3E
omXFqnkuoidk9CvoXhBkq8o2cwJj4IQ6SrQ6RXvD3DiusGxOXP/GBLi1+Q==
-----END CERTIFICATE-----
)pem";
}

auto test_private_key_pem() -> std::string_view {
  return R"pem(-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC3hGPOsWT4xHja
/FgIs/Bj/aIjT+vJiFddPbwtgf68gm81cF6zq9oBVWqTXTLQPdURhLN30/k3W8mi
B4Z0TEfVjhUTwLpa9gPX56zccBaKdmTE4Gv2+++vXtgDLi8Z8kv3hb1nFABuKtxh
B+OYjcoqCEyhhLxrsT1p7WPB7MFQjohh8mlLixIoIXHFFLuRgkXYN77NuXDv0HW0
ur4/YwSyJSSdzpN+KmonrWWVZxgE30HgZC2MTYStNT3ypXsy2qIcH9AryoshGtaD
C0DE7MGv82e3p+8W9iSqPSfjYLWXajHszOFtQ0KVGVYoUL0gl+ZwEOn3ZgALJ8F5
6QzSZ5qJAgMBAAECggEADCJ7bQfq2s43BZfpBE5Np3b2eSpIN4/F4xr5bfLqErH3
Q4vWH0BOTodE4FCp0v47sAhby6EfGgt+NjIUMG141ISsHR3B1NDuqSCfHAZnXyl7
+U6zpOW80BZP50nvAGK2pjvYLdAzRRWdc+cPJB/ADe51jrqVsPNL0czKK2EJ2Vcu
qa3YGAgeU485BG5kj9hfSY7ISYdL53iVGEvBn6ZMAyf/VfgWISKt1wItb4zD9xpw
GoxHnbEoCFO0xbBuqGaSSJZUYXDqmWUXntmocDG0I5TfCHVGI7k3HWoxv3PdgwST
oQSF5HTn7ZkLCUqn1CEMAqIqR9qfgOEoKV3WRqOK+QKBgQDqc+QWK6zZSWQrXQez
c0aGbNwhnsmiQFav+NGHJs7X6mP+8LKtYpsy8u9JP5YEBCOdPPy6NhIHo7MVPqTr
fJGWInlu2RMCBKOY/gtz2fZLUGGa0vheH89LXoDw2De892Pof0fHYakXVF+BIgIu
ROs7rome/J9DJdG9SZS8hcAh1QKBgQDIYhuSIosczJIs76l9x97Bo4ijVpm1xQnp
IH6PiUxFgXwPUYSxBBas3yR197LrbSQEC3ECzSIw8zGTR60+4NXIABymRtrcH0pB
CbGoPb9bCNFboLPIyoeItdg08BFQW5xjB/YtCf1VIvQ4MPKKy2+xmRTe0BIvQ6Oj
VaKURCh75QKBgQCO5k4qOTqkQbhdOdm8xdFBk85ka40IaOCtZyIGl1/ReJq0Q+8U
ARsJZcr1QV1lW0R9zTNZKCW7JEq98y3AHm/6K08KgiK0wzEXyhge58XXx8mF/XrG
m+rUTGiY46Cte7gYeKO+YccXYiTH53ueygD6onpdPiqwWbfKALQ4Z+lgdQKBgByl
cd2VVJ/cWvVTmb5iiWTTVSBMrBLBRWZvf8hTT3UMSskB1V9c/0TdbZ/ztj9TzfdK
D3DgFIg87BRrBchcahN6eJ4IuRcI3huoIB62HRoNi6h2s9z7nx6bKHqA62i6gAPk
Snwr1myKX7F8xd+zr5cLFGG6Cy94kvcxsdlSWRtFAoGANZrGt1EWuVq5nBy3M5+M
PNntbt0H4u/fLbsvZ7n8wqLwbvlWpjdXaUHSGm86rhIdO1oUGpTespZZ9VhorWBi
sDPpFj7fJUR44k3t5Bmebe71mkbCEj2cRRO/3Auw0c/kfyE28wCRyZ/AuNGtm8xj
vTua3WDpZe9NKsuywyLWkcY=
-----END PRIVATE KEY-----
)pem";
}

auto test_tls_server_config() -> ligo::security_config {
  auto context = ligo::make_tls_server_context({
    .mode = ligo::security_mode::TLS,
  });
  BOOST_REQUIRE(context);
  auto certificate = test_certificate_pem();
  auto private_key = test_private_key_pem();
  auto error = boost::system::error_code{};
  (*context)->use_certificate_chain(boost::asio::buffer(certificate.data(), certificate.size()), error);
  BOOST_REQUIRE_MESSAGE(!error, error.message());
  (*context)->use_private_key(boost::asio::buffer(private_key.data(), private_key.size()), boost::asio::ssl::context::file_format::pem, error);
  BOOST_REQUIRE_MESSAGE(!error, error.message());
  return {
    .mode = ligo::security_mode::TLS,
    .verify = ligo::verify_mode::NONE,
    .tls_context = std::move(*context),
  };
}

auto test_tls_client_config() -> ligo::security_config {
  return {
    .mode = ligo::security_mode::TLS,
    .verify = ligo::verify_mode::NONE,
  };
}

auto write_test_pem_file(std::string name, std::string_view contents) -> std::string {
  static auto counter = std::atomic_uint{};
  auto path = "/tmp/ligo-" + name + "-" + std::to_string(counter.fetch_add(1)) + ".pem";
  auto output = std::ofstream{path};
  BOOST_REQUIRE(output);
  output << contents;
  return path;
}

auto test_tls_verified_client_config(std::string peer_name = "localhost") -> ligo::security_config {
  return {
    .mode = ligo::security_mode::TLS,
    .verify = ligo::verify_mode::PEER,
    .peer_name = std::move(peer_name),
    .ca_file = write_test_pem_file("tls-ca", test_certificate_pem()),
  };
}

auto test_dtls_server_config() -> ligo::security_config {
  return {
    .mode = ligo::security_mode::DTLS,
    .verify = ligo::verify_mode::NONE,
    .certificate_chain_file = write_test_pem_file("dtls-cert", test_certificate_pem()),
    .private_key_file = write_test_pem_file("dtls-key", test_private_key_pem()),
  };
}

auto test_dtls_client_config() -> ligo::security_config {
  return {
    .mode = ligo::security_mode::DTLS,
    .verify = ligo::verify_mode::NONE,
  };
}

auto test_dtls_verified_client_config(std::string peer_name = "localhost") -> ligo::security_config {
  return {
    .mode = ligo::security_mode::DTLS,
    .verify = ligo::verify_mode::PEER,
    .peer_name = std::move(peer_name),
    .ca_file = write_test_pem_file("dtls-ca", test_certificate_pem()),
  };
}

auto test_dtls_identified_client_config() -> ligo::security_config {
  return {
    .mode = ligo::security_mode::DTLS,
    .verify = ligo::verify_mode::NONE,
    .certificate_chain_file = write_test_pem_file("dtls-client-cert", test_certificate_pem()),
    .private_key_file = write_test_pem_file("dtls-client-key", test_private_key_pem()),
  };
}

auto test_dtls_server_verify_client_config() -> ligo::security_config {
  return {
    .mode = ligo::security_mode::DTLS,
    .verify = ligo::verify_mode::PEER,
    .ca_file = write_test_pem_file("dtls-client-ca", test_certificate_pem()),
    .certificate_chain_file = write_test_pem_file("dtls-server-cert", test_certificate_pem()),
    .private_key_file = write_test_pem_file("dtls-server-key", test_private_key_pem()),
  };
}

auto http_client_call(std::string host, std::uint16_t port, ligo::http::request const &request, ligo::http::client_options options = {}) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  auto client = ligo::http::client{
    {
      .host = std::move(host),
      .port = port,
    },
    options,
  };
  return client.call(request);
}

auto http_client_call_selected(ligo::http::server &server, ligo::http::request const &request, ligo::http::client_options options = {}) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  return http_client_call("127.0.0.1", server.selected_port(), request, options);
}

auto http_client_call_selected_and_stop_impl(ligo::http::server &server, std::shared_ptr<ligo::http::request const> request, ligo::http::client_options options) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  auto result = co_await http_client_call_selected(server, *request, options);
  server.stop();
  co_return result;
}

auto http_client_call_selected_and_stop(ligo::http::server &server, ligo::http::request const &request, ligo::http::client_options options = {}) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  return http_client_call_selected_and_stop_impl(server, std::make_shared<ligo::http::request const>(request), options);
}

auto http_async_stop_after(ligo::http::server &server, std::chrono::milliseconds delay, std::chrono::milliseconds timeout) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  auto timer = boost::asio::steady_timer{co_await boost::asio::this_coro::executor};
  timer.expires_after(delay);
  co_await timer.async_wait(boost::asio::as_tuple(boost::asio::use_awaitable));
  co_return co_await server.async_stop(timeout);
}

auto raw_http_post(std::uint16_t port, std::string target, std::string body) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  auto executor = co_await boost::asio::this_coro::executor;
  auto resolver = boost::asio::ip::tcp::resolver{executor};
  auto socket = boost::asio::ip::tcp::socket{executor};
  auto [resolve_error, endpoints] = co_await resolver.async_resolve("127.0.0.1", std::to_string(port), boost::asio::as_tuple(boost::asio::use_awaitable));
  if (resolve_error) {
    auto status = test_network_failure(resolve_error);
    co_return ligo::detail::fail<boost::json::value>(status);
  }
  auto [connect_error, _] = co_await boost::asio::async_connect(socket, endpoints, boost::asio::as_tuple(boost::asio::use_awaitable));
  if (connect_error) {
    auto status = test_network_failure(connect_error);
    co_return ligo::detail::fail<boost::json::value>(status);
  }
  auto request = boost::beast::http::request<boost::beast::http::string_body>{
    boost::beast::http::verb::post,
    target,
    11,
  };
  request.set(boost::beast::http::field::content_type, "application/json");
  request.body() = std::move(body);
  request.prepare_payload();
  auto [write_error, written] = co_await boost::beast::http::async_write(socket, request, boost::asio::as_tuple(boost::asio::use_awaitable));
  static_cast<void>(written);
  if (write_error) {
    auto status = test_network_failure(write_error);
    co_return ligo::detail::fail<boost::json::value>(status);
  }
  auto buffer = boost::beast::flat_buffer{};
  auto response = boost::beast::http::response<boost::beast::http::string_body>{};
  auto [read_error, read] = co_await boost::beast::http::async_read(socket, buffer, response, boost::asio::as_tuple(boost::asio::use_awaitable));
  static_cast<void>(read);
  if (read_error) {
    auto status = test_network_failure(read_error);
    co_return ligo::detail::fail<boost::json::value>(status);
  }
  auto parse_error = boost::system::error_code{};
  auto value = boost::json::parse(response.body(), parse_error);
  if (parse_error) {
    co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_invalid_argument, parse_error.message());
  }
  co_return modulus::success(std::move(value));
}

auto raw_http_post_selected_and_stop(ligo::http::server &server, std::string target, std::string body) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  auto result = co_await raw_http_post(server.selected_port(), std::move(target), std::move(body));
  server.stop();
  co_return result;
}

auto raw_http_get(std::uint16_t port, std::string target) -> boost::asio::awaitable<std::expected<boost::beast::http::response<boost::beast::http::string_body>, modulus::error>> {
  auto executor = co_await boost::asio::this_coro::executor;
  auto resolver = boost::asio::ip::tcp::resolver{executor};
  auto socket = boost::asio::ip::tcp::socket{executor};
  auto [resolve_error, endpoints] = co_await resolver.async_resolve("127.0.0.1", std::to_string(port), boost::asio::as_tuple(boost::asio::use_awaitable));
  if (resolve_error) {
    auto status = test_network_failure(resolve_error);
    co_return ligo::detail::fail<boost::beast::http::response<boost::beast::http::string_body>>(status);
  }
  auto [connect_error, _] = co_await boost::asio::async_connect(socket, endpoints, boost::asio::as_tuple(boost::asio::use_awaitable));
  if (connect_error) {
    auto status = test_network_failure(connect_error);
    co_return ligo::detail::fail<boost::beast::http::response<boost::beast::http::string_body>>(status);
  }
  auto request = boost::beast::http::request<boost::beast::http::empty_body>{
    boost::beast::http::verb::get,
    target,
    11,
  };
  auto [write_error, written] = co_await boost::beast::http::async_write(socket, request, boost::asio::as_tuple(boost::asio::use_awaitable));
  static_cast<void>(written);
  if (write_error) {
    auto status = test_network_failure(write_error);
    co_return ligo::detail::fail<boost::beast::http::response<boost::beast::http::string_body>>(status);
  }
  auto buffer = boost::beast::flat_buffer{};
  auto response = boost::beast::http::response<boost::beast::http::string_body>{};
  auto [read_error, read] = co_await boost::beast::http::async_read(socket, buffer, response, boost::asio::as_tuple(boost::asio::use_awaitable));
  static_cast<void>(read);
  if (read_error) {
    auto status = test_network_failure(read_error);
    co_return ligo::detail::fail<boost::beast::http::response<boost::beast::http::string_body>>(status);
  }
  co_return modulus::success(std::move(response));
}

auto raw_http_get_selected_and_stop(ligo::http::server &server, std::string target) -> boost::asio::awaitable<std::expected<boost::beast::http::response<boost::beast::http::string_body>, modulus::error>> {
  auto result = co_await raw_http_get(server.selected_port(), std::move(target));
  server.stop();
  co_return result;
}

auto tcp_client_call(std::string host, std::uint16_t port, ligo::tcp::request const &request, ligo::tcp::client_options options = {}) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  auto client = ligo::tcp::client{
    {
      .host = std::move(host),
      .port = port,
    },
    options,
  };
  return client.call(request);
}

auto tcp_client_call_selected(ligo::tcp::server &server, ligo::tcp::request const &request, ligo::tcp::client_options options = {}) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  return tcp_client_call("127.0.0.1", server.selected_port(), request, options);
}

auto tcp_client_call_selected_and_stop_impl(ligo::tcp::server &server, std::shared_ptr<ligo::tcp::request const> request, ligo::tcp::client_options options) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  auto result = co_await tcp_client_call_selected(server, *request, options);
  server.stop();
  co_return result;
}

auto tcp_client_call_selected_and_stop(ligo::tcp::server &server, ligo::tcp::request const &request, ligo::tcp::client_options options = {}) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  return tcp_client_call_selected_and_stop_impl(server, std::make_shared<ligo::tcp::request const>(request), options);
}

auto tcp_async_stop_after(ligo::tcp::server &server, std::chrono::milliseconds delay, std::chrono::milliseconds timeout) -> boost::asio::awaitable<std::expected<void, modulus::error>> {
  auto timer = boost::asio::steady_timer{co_await boost::asio::this_coro::executor};
  timer.expires_after(delay);
  co_await timer.async_wait(boost::asio::as_tuple(boost::asio::use_awaitable));
  co_return co_await server.async_stop(timeout);
}

auto accept_and_hold(std::shared_ptr<boost::asio::ip::tcp::acceptor> acceptor, std::shared_ptr<boost::asio::ip::tcp::socket> socket, std::chrono::milliseconds duration) -> boost::asio::awaitable<void> {
  auto [accept_error] = co_await acceptor->async_accept(*socket, boost::asio::as_tuple(boost::asio::use_awaitable));
  if (accept_error) {
    co_return;
  }
  auto timer = boost::asio::steady_timer{co_await boost::asio::this_coro::executor};
  timer.expires_after(duration);
  co_await timer.async_wait(boost::asio::as_tuple(boost::asio::use_awaitable));
}

auto make_udp_socket(boost::asio::io_context &io) -> std::unique_ptr<boost::asio::ip::udp::socket> {
  return std::make_unique<boost::asio::ip::udp::socket>(io, boost::asio::ip::udp::endpoint{
                                                              boost::asio::ip::udp::v4(),
                                                              0,
                                                            });
}

auto udp_client_adopt(std::unique_ptr<boost::asio::ip::udp::socket> socket, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config = {}, std::uint64_t session_id = 1) -> std::expected<ligo::udp::client, modulus::error> {
  return ligo::udp::client::adopt(std::move(socket), std::move(peer), {
                                                                        .reliability = config,
                                                                        .session_id = session_id,
                                                                      });
}

auto udp_client_bind(boost::asio::io_context &io, boost::asio::ip::udp::endpoint local, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config = {}, std::uint64_t session_id = 1) -> std::expected<ligo::udp::client, modulus::error> {
  return ligo::udp::client::bind(io, {
                                       .local = std::move(local),
                                       .peer = std::move(peer),
                                     },
                                 {
                                   .reliability = config,
                                   .session_id = session_id,
                                 });
}

auto config_http_tcp_calls_and_stop(ligo::http::server &http_server, ligo::tcp::server &tcp_server) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
  auto tcp_payload = boost::json::object{};
  tcp_payload["ok"] = true;
  auto tcp_request = std::make_shared<ligo::tcp::request>();
  tcp_request->route = "echo";
  tcp_request->payload = std::move(tcp_payload);
  auto tcp_options = ligo::tcp::client_options{};
  tcp_options.timeout = std::chrono::milliseconds{1000};
  auto tcp_result = co_await tcp_client_call_selected(tcp_server, *tcp_request, tcp_options);
  if (!tcp_result) {
    static_cast<void>(co_await tcp_server.async_stop(std::chrono::milliseconds{1000}));
    static_cast<void>(co_await http_server.async_stop(std::chrono::milliseconds{1000}));
    co_return tcp_result;
  }
  auto http_payload = boost::json::object{};
  http_payload["ok"] = true;
  auto http_request = std::make_shared<ligo::http::request>();
  http_request->target = "/echo";
  http_request->payload = std::move(http_payload);
  auto http_options = ligo::http::client_options{};
  http_options.timeout = std::chrono::milliseconds{1000};
  auto http_result = co_await http_client_call_selected(http_server, *http_request, http_options);
  static_cast<void>(co_await tcp_server.async_stop(std::chrono::milliseconds{1000}));
  static_cast<void>(co_await http_server.async_stop(std::chrono::milliseconds{1000}));
  co_return http_result;
}

}

BOOST_AUTO_TEST_CASE(ligo_config_drives_http_tcp_and_udp_setup) {
  auto io = boost::asio::io_context{};
  auto udp_client_socket = make_udp_socket(io);
  auto udp_server_socket = make_udp_socket(io);
  auto udp_client_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    udp_client_socket->local_endpoint().port(),
  };
  auto udp_server_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    udp_server_socket->local_endpoint().port(),
  };
  auto config_text = std::string{
                       "{\"endpoints\":[",
                     } +
                     "{\"name\":\"http\",\"protocol\":\"http\",\"bind\":\"127.0.0.1:0\",\"timeout_ms\":1000}," + "{\"name\":\"tcp\",\"protocol\":\"tcp\",\"bind\":\"127.0.0.1:0\",\"timeout_ms\":1000}," + "{\"name\":\"udp-client\",\"protocol\":\"udp\",\"bind\":\"127.0.0.1:" + std::to_string(udp_client_endpoint.port()) + "\",\"connect\":\"127.0.0.1:" + std::to_string(udp_server_endpoint.port()) + "\",\"udp\":{\"resend_interval_ms\":5,\"max_in_flight\":2,\"session_timeout_ms\":1000}}," + "{\"name\":\"udp-server\",\"protocol\":\"udp\",\"bind\":\"127.0.0.1:" + std::to_string(udp_server_endpoint.port()) + "\",\"connect\":\"127.0.0.1:" + std::to_string(udp_client_endpoint.port()) + "\",\"udp\":{\"resend_interval_ms\":5,\"max_in_flight\":2,\"session_timeout_ms\":1000}}" + "]}";
  auto configs = ligo::parse_endpoint_configs(boost::json::parse(config_text));
  BOOST_REQUIRE(configs);
  BOOST_REQUIRE((configs->size()) == (4u));
  auto http_router = ligo::service_router{};
  http_router.add_route("POST", "/echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto http_bind = ligo::http::server_endpoint_from_config((*configs)[0]);
  BOOST_REQUIRE(http_bind);
  auto http_server = ligo::http::server{
    io,
    *http_bind,
    std::move(http_router),
    ligo::http::server_options_from_config((*configs)[0]),
  };
  BOOST_REQUIRE(http_server.start());
  auto tcp_router = ligo::service_router{};
  tcp_router.add_route("TCP", "echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto tcp_bind = ligo::tcp::server_endpoint_from_config((*configs)[1]);
  BOOST_REQUIRE(tcp_bind);
  auto tcp_server = ligo::tcp::server{
    io,
    *tcp_bind,
    std::move(tcp_router),
    ligo::tcp::server_options_from_config((*configs)[1]),
  };
  BOOST_REQUIRE(tcp_server.start());
  auto udp_client_parts = ligo::udp::client_endpoint_from_config((*configs)[2]);
  auto udp_server_parts = ligo::udp::server_endpoint_from_config((*configs)[3]);
  BOOST_REQUIRE(udp_client_parts);
  BOOST_REQUIRE(udp_server_parts);
  auto udp_client_result = ligo::udp::client::adopt(std::move(udp_client_socket), udp_client_parts->peer, ligo::udp::client_options_from_config((*configs)[2]));
  auto udp_server_result = ligo::udp::server::adopt(std::move(udp_server_socket), udp_server_parts->peer, ligo::udp::server_options_from_config((*configs)[3]));
  BOOST_REQUIRE(udp_client_result);
  BOOST_REQUIRE(udp_server_result);
  auto udp_client = std::move(*udp_client_result);
  auto udp_server = std::move(*udp_server_result);
  auto udp_payload = boost::json::object{};
  udp_payload["ok"] = true;
  auto udp_sent = boost::asio::co_spawn(io, udp_client.send(std::move(udp_payload)), boost::asio::use_future);
  auto udp_received = boost::asio::co_spawn(io, udp_server.receive(), boost::asio::use_future);
  auto done = boost::asio::co_spawn(io, config_http_tcp_calls_and_stop(http_server, tcp_server), boost::asio::use_future);
  io.run();
  BOOST_CHECK(udp_sent.get());
  auto udp_message = udp_received.get();
  BOOST_REQUIRE(udp_message);
  BOOST_CHECK(udp_message->payload.as_object().at("ok").as_bool());
  auto result = done.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(result->as_object().at("ok").as_bool());
}

BOOST_AUTO_TEST_CASE(ligo_config_parses_structured_security) {
  auto value = boost::json::parse(R"json({"name": "mysql", "protocol": "tcp", "connect": "localhost:3306", "timeout_ms": 2500, "security": {"mode": "tls", "verify": "peer", "peer_name": "db.local", "ca_file": "/tmp/ca.pem", "certificate_chain_file": "/tmp/cert.pem", "private_key_file": "/tmp/key.pem"}, "http": {"max_body_size": 4096}})json");
  auto config = ligo::parse_endpoint_config(value);
  BOOST_REQUIRE(config);
  BOOST_CHECK((config->security.mode) == (ligo::security_mode::TLS));
  BOOST_CHECK((config->security.verify) == (ligo::verify_mode::PEER));
  BOOST_CHECK((config->security.peer_name) == ("db.local"));
  BOOST_CHECK((config->security.ca_file) == ("/tmp/ca.pem"));
  BOOST_CHECK((config->security.certificate_chain_file) == ("/tmp/cert.pem"));
  BOOST_CHECK((config->security.private_key_file) == ("/tmp/key.pem"));
  auto options = ligo::tcp::client_options_from_config(*config);
  BOOST_CHECK((options.timeout) == (std::chrono::milliseconds{2500}));
  BOOST_CHECK((options.security.mode) == (ligo::security_mode::TLS));
  auto http_options = ligo::http::client_options_from_config(*config);
  BOOST_CHECK((http_options.max_body_size) == (4096U));
  auto http_server_options = ligo::http::server_options_from_config(*config);
  BOOST_CHECK((http_server_options.max_body_size) == (4096U));
  auto round_trip = ligo::parse_endpoint_config(ligo::to_json(*config));
  BOOST_REQUIRE(round_trip);
  BOOST_CHECK((round_trip->security.mode) == (ligo::security_mode::TLS));
  BOOST_CHECK((round_trip->security.verify) == (ligo::verify_mode::PEER));
  BOOST_CHECK((round_trip->http.max_body_size) == (4096U));
}

BOOST_AUTO_TEST_CASE(ligo_config_rejects_invalid_http_body_limit) {
  auto bad_shape = ligo::parse_endpoint_config(boost::json::parse(R"json({"name": "http", "protocol": "http", "bind": "127.0.0.1:0", "http": 42})json"));
  BOOST_CHECK(!bad_shape);
  BOOST_CHECK((bad_shape.error().code()) == (ligo::detail::error_invalid_argument));
  auto zero_limit = ligo::parse_endpoint_config(boost::json::parse(R"json({"name": "http", "protocol": "http", "bind": "127.0.0.1:0", "http": {"max_body_size": 0}})json"));
  BOOST_CHECK(!zero_limit);
  BOOST_CHECK((zero_limit.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_config_rejects_unknown_keys_and_defaults_protocol) {
  auto defaults = ligo::parse_endpoint_config(boost::json::parse(R"json({"name": "default-tcp", "connect": "127.0.0.1:7"})json"));
  BOOST_REQUIRE(defaults);
  BOOST_CHECK((defaults->protocol) == (ligo::protocol::TCP));
  auto unknown_root = ligo::parse_endpoint_configs(boost::json::parse(R"json({"endpoints": [], "endpointz": []})json"));
  BOOST_REQUIRE(!unknown_root);
  BOOST_CHECK(unknown_root.error().describe().find("config root has unknown key: endpointz") != std::string::npos);
  auto unknown_endpoint = ligo::parse_endpoint_config(boost::json::parse(R"json({"name": "bad", "protocol": "tcp", "connect": "127.0.0.1:7", "timeot_ms": 1000})json"));
  BOOST_REQUIRE(!unknown_endpoint);
  BOOST_CHECK(unknown_endpoint.error().describe().find("endpoint has unknown key: timeot_ms") != std::string::npos);
  auto unknown_nested = ligo::parse_endpoint_config(boost::json::parse(R"json({"name": "bad-security", "protocol": "tcp", "security": {"mode": "tls", "verfy": "peer"}})json"));
  BOOST_REQUIRE(!unknown_nested);
  BOOST_CHECK(unknown_nested.error().describe().find("endpoint security has unknown key: verfy") != std::string::npos);
}

BOOST_AUTO_TEST_CASE(ligo_servers_bind_localhost_hostname) {
  auto io = boost::asio::io_context{};
  auto http_server = ligo::http::server{
    io,
    {
      .host = "localhost",
      .port = 0,
    },
    ligo::service_router{},
  };
  auto http_started = http_server.start();
  BOOST_REQUIRE(http_started);
  BOOST_CHECK((http_server.selected_port()) != (0));
  http_server.stop();
  auto tcp_server = ligo::tcp::server{
    io,
    {
      .host = "localhost",
      .port = 0,
    },
    ligo::service_router{},
  };
  auto tcp_started = tcp_server.start();
  BOOST_REQUIRE(tcp_started);
  BOOST_CHECK((tcp_server.selected_port()) != (0));
  tcp_server.stop();
}

BOOST_AUTO_TEST_CASE(ligo_http_post_round_trip_dispatches_json_handler) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("POST", "/echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto future = boost::asio::co_spawn(
    io,
    http_client_call_selected_and_stop(
      server,
      {
        .target = "/echo",
        .payload = boost::json::parse(R"json({"ok":true})json"),
      }),
    boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(result->as_object().at("ok").as_bool());
}

BOOST_AUTO_TEST_CASE(ligo_http_get_response_route_serves_text_body) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_response_route("GET", "/setup", [](ligo::request_context context, boost::json::value) -> boost::asio::awaitable<std::expected<ligo::service_response, modulus::error>> {
    co_return ligo::text_response("text/html; charset=utf-8", "<!doctype html><title>" + context.route + "</title>");
  });
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto future = boost::asio::co_spawn(io, raw_http_get_selected_and_stop(server, "/setup"), boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK((result->result()) == (boost::beast::http::status::ok));
  BOOST_CHECK((std::string{result->at(boost::beast::http::field::content_type)}) == ("text/html; charset=utf-8"));
  BOOST_CHECK((result->body()) == ("<!doctype html><title>/setup</title>"));
}

BOOST_AUTO_TEST_CASE(ligo_http_tls_post_round_trip_dispatches_json_handler) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("POST", "/echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
    {
      .timeout = std::chrono::milliseconds{1000},
      .on_error = {},
      .security = test_tls_server_config(),
    },
  };
  auto started = server.start();
  BOOST_REQUIRE(started);
  auto future = boost::asio::co_spawn(
    io,
    http_client_call_selected_and_stop(
      server,
      {
        .target = "/echo",
        .payload = boost::json::parse(R"json({"ok":true})json"),
      },
      {
        .timeout = std::chrono::milliseconds{1000},
        .security = test_tls_client_config(),
      }),
    boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(result->as_object().at("ok").as_bool());
}

BOOST_AUTO_TEST_CASE(ligo_http_tls_server_requires_certificate_context) {
  auto io = boost::asio::io_context{};
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    ligo::service_router{},
    {
      .timeout = std::chrono::milliseconds{1000},
      .on_error = {},
      .security = {
        .mode = ligo::security_mode::TLS,
      },
    },
  };
  auto status = server.start();
  BOOST_CHECK(!(status));
  BOOST_CHECK((status.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_http_tls_server_rejects_empty_provided_context) {
  auto io = boost::asio::io_context{};
  auto context = std::make_shared<boost::asio::ssl::context>(boost::asio::ssl::context::tls_server);
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    ligo::service_router{},
    {
      .timeout = std::chrono::milliseconds{1000},
      .on_error = {},
      .security = {
        .mode = ligo::security_mode::TLS,
        .tls_context = context,
      },
    },
  };
  auto status = server.start();
  BOOST_CHECK(!(status));
  BOOST_CHECK((status.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_http_get_round_trip_dispatches_method_and_route) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("GET", "/method", [](ligo::request_context context, boost::json::value) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    auto response = boost::json::object{};
    response["method"] = context.method;
    response["route"] = context.route;
    co_return modulus::success(std::move(response));
  });
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto future = boost::asio::co_spawn(io, http_client_call_selected_and_stop(server, {
                                                                                       .method = ligo::http::method::GET,
                                                                                       .target = "/method",
                                                                                     }),
                                      boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK((result->as_object().at("method").as_string()) == ("GET"));
  BOOST_CHECK((result->as_object().at("route").as_string()) == ("/method"));
}

BOOST_AUTO_TEST_CASE(ligo_http_missing_route_returns_status) {
  auto io = boost::asio::io_context{};
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    ligo::service_router{},
  };
  BOOST_REQUIRE(server.start());
  auto future = boost::asio::co_spawn(io, http_client_call_selected_and_stop(server, {
                                                                                       .target = "/missing",
                                                                                       .payload = boost::json::object{},
                                                                                     }),
                                      boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_CHECK(!(result));
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_not_found));
}

BOOST_AUTO_TEST_CASE(ligo_http_handler_failure_returns_status) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("POST", "/boom", [](ligo::request_context, boost::json::value) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_internal, "boom");
  });
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto future = boost::asio::co_spawn(io, http_client_call_selected_and_stop(server, {
                                                                                       .target = "/boom",
                                                                                       .payload = boost::json::object{},
                                                                                     }),
                                      boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_CHECK(!(result));
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_internal));
}

BOOST_AUTO_TEST_CASE(ligo_http_invalid_json_returns_envelope_status) {
  auto io = boost::asio::io_context{};
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    ligo::service_router{},
  };
  BOOST_REQUIRE(server.start());
  auto future = boost::asio::co_spawn(io, raw_http_post_selected_and_stop(server, "/missing", "{"), boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(!(result->as_object().at("ok").as_bool()));
  BOOST_CHECK((result->as_object().at("code").as_string()) == ("invalid_argument"));
}

BOOST_AUTO_TEST_CASE(ligo_http_server_rejects_oversized_request_body) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("POST", "/echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
    {
      .timeout = std::chrono::milliseconds{1000},
      .max_body_size = 32,
    },
  };
  BOOST_REQUIRE(server.start());
  auto body = std::string{R"json({"data":")json"} + std::string(64, 'x') + R"json("})json";
  auto future = boost::asio::co_spawn(io, raw_http_post_selected_and_stop(server, "/echo", std::move(body)), boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(!(result->as_object().at("ok").as_bool()));
  BOOST_CHECK((result->as_object().at("code").as_string()) == ("invalid_argument"));
}

BOOST_AUTO_TEST_CASE(ligo_http_client_rejects_oversized_response_body) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("POST", "/large", [](ligo::request_context, boost::json::value) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    auto response = boost::json::object{};
    response["data"] = std::string(128, 'x');
    co_return modulus::success(std::move(response));
  });
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto future = boost::asio::co_spawn(
    io,
    http_client_call_selected_and_stop(
      server,
      {
        .target = "/large",
        .payload = boost::json::object{},
      },
      {
        .timeout = std::chrono::milliseconds{1000},
        .max_body_size = 32,
      }),
    boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_CHECK(!result);
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_http_concurrent_client_calls_succeed) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("POST", "/echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto client = ligo::http::client{
    {
      .host = "127.0.0.1",
      .port = server.selected_port(),
    },
  };
  auto work = boost::asio::make_work_guard(io);
  auto runners = start_io_threads(io);
  auto futures = std::vector<std::future<std::expected<boost::json::value, modulus::error>>>{};
  for (auto index = 0; index < 16; ++index) {
    auto payload = boost::json::object{};
    payload["index"] = index;
    futures.push_back(boost::asio::co_spawn(io, client.call({
                                                  .target = "/echo",
                                                  .payload = std::move(payload),
                                                }),
                                            boost::asio::use_future));
  }
  for (auto &future : futures) {
    auto result = future.get();
    BOOST_REQUIRE(result);
    BOOST_CHECK(result->is_object());
  }
  server.stop();
  work.reset();
  join_io_threads(runners);
}

BOOST_AUTO_TEST_CASE(ligo_http_async_stop_cancels_active_session) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("POST", "/slow", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return co_await delayed_echo(std::chrono::milliseconds{50}, std::move(payload));
  });
  auto server = ligo::http::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto call = boost::asio::co_spawn(
    io,
    http_client_call_selected(
      server,
      {
        .target = "/slow",
        .payload = boost::json::object{},
      },
      {
        .timeout = std::chrono::milliseconds{200},
      }),
    boost::asio::use_future);
  auto stopped = boost::asio::co_spawn(io, http_async_stop_after(server, std::chrono::milliseconds{5}, std::chrono::milliseconds{200}), boost::asio::use_future);
  run_two_threads(io);
  BOOST_CHECK(!(call.get()));
  BOOST_CHECK(stopped.get());
}

BOOST_AUTO_TEST_CASE(ligo_tcp_round_trip_dispatches_json_handler) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("TCP", "echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto server = ligo::tcp::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto future = boost::asio::co_spawn(
    io,
    tcp_client_call_selected_and_stop(
      server,
      {
        .route = "echo",
        .payload = boost::json::parse(R"json({"ok":true})json"),
      }),
    boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(result->as_object().at("ok").as_bool());
}

BOOST_AUTO_TEST_CASE(ligo_tcp_tls_round_trip_dispatches_json_handler) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("TCP", "echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto server = ligo::tcp::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
    {
      .timeout = std::chrono::milliseconds{1000},
      .on_error = {},
      .security = test_tls_server_config(),
    },
  };
  auto started = server.start();
  BOOST_REQUIRE(started);
  auto future = boost::asio::co_spawn(
    io,
    tcp_client_call_selected_and_stop(
      server,
      {
        .route = "echo",
        .payload = boost::json::parse(R"json({"ok":true})json"),
      },
      {
        .timeout = std::chrono::milliseconds{1000},
        .security = test_tls_client_config(),
      }),
    boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(result->as_object().at("ok").as_bool());
}

BOOST_AUTO_TEST_CASE(ligo_tcp_tls_verified_peer_succeeds) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("TCP", "echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto server = ligo::tcp::server{
    io.get_executor(),
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
    {
      .timeout = std::chrono::milliseconds{1000},
      .on_error = {},
      .security = test_tls_server_config(),
    },
  };
  auto started = server.start();
  BOOST_REQUIRE(started);
  auto future = boost::asio::co_spawn(
    io,
    tcp_client_call(
      "127.0.0.1",
      server.selected_port(),
      {
        .route = "echo",
        .payload = boost::json::parse(R"json({"ok":true})json"),
      },
      {
        .timeout = std::chrono::milliseconds{1000},
        .security = test_tls_verified_client_config("localhost"),
      }),
    boost::asio::use_future);
  auto stopper = boost::asio::steady_timer{io};
  stopper.expires_after(std::chrono::milliseconds{200});
  stopper.async_wait([&server](boost::system::error_code const &) {
    server.stop();
  });
  io.run();
  auto result = future.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(result->as_object().at("ok").as_bool());
}

BOOST_AUTO_TEST_CASE(ligo_tcp_tls_hostname_mismatch_fails) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("TCP", "echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto server = ligo::tcp::server{
    io.get_executor(),
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
    {
      .timeout = std::chrono::milliseconds{1000},
      .on_error = {},
      .security = test_tls_server_config(),
    },
  };
  auto started = server.start();
  BOOST_REQUIRE(started);
  auto future = boost::asio::co_spawn(
    io,
    tcp_client_call(
      "127.0.0.1",
      server.selected_port(),
      {
        .route = "echo",
        .payload = boost::json::object{},
      },
      {
        .timeout = std::chrono::milliseconds{1000},
        .security = test_tls_verified_client_config("wrong.local"),
      }),
    boost::asio::use_future);
  auto stopper = boost::asio::steady_timer{io};
  stopper.expires_after(std::chrono::milliseconds{200});
  stopper.async_wait([&server](boost::system::error_code const &) {
    server.stop();
  });
  io.run();
  auto result = future.get();
  BOOST_CHECK(!result);
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_unavailable));
}

BOOST_AUTO_TEST_CASE(ligo_tcp_tls_server_requires_certificate_context) {
  auto io = boost::asio::io_context{};
  auto server = ligo::tcp::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    ligo::service_router{},
    {
      .timeout = std::chrono::milliseconds{1000},
      .on_error = {},
      .security = {
        .mode = ligo::security_mode::TLS,
      },
    },
  };
  auto status = server.start();
  BOOST_CHECK(!(status));
  BOOST_CHECK((status.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_tcp_tls_server_rejects_empty_provided_context) {
  auto io = boost::asio::io_context{};
  auto context = std::make_shared<boost::asio::ssl::context>(boost::asio::ssl::context::tls_server);
  auto server = ligo::tcp::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    ligo::service_router{},
    {
      .timeout = std::chrono::milliseconds{1000},
      .on_error = {},
      .security = {
        .mode = ligo::security_mode::TLS,
        .tls_context = context,
      },
    },
  };
  auto status = server.start();
  BOOST_CHECK(!(status));
  BOOST_CHECK((status.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_tcp_missing_route_returns_status) {
  auto io = boost::asio::io_context{};
  auto server = ligo::tcp::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    ligo::service_router{},
  };
  BOOST_REQUIRE(server.start());
  auto future = boost::asio::co_spawn(io, tcp_client_call_selected_and_stop(server, {
                                                                                      .route = "missing",
                                                                                      .payload = boost::json::object{},
                                                                                    }),
                                      boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_CHECK(!(result));
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_not_found));
}

BOOST_AUTO_TEST_CASE(ligo_tcp_handler_failure_returns_status) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("TCP", "boom", [](ligo::request_context, boost::json::value) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return ligo::detail::fail<boost::json::value>(ligo::detail::error_internal, "boom");
  });
  auto server = ligo::tcp::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto future = boost::asio::co_spawn(io, tcp_client_call_selected_and_stop(server, {
                                                                                      .route = "boom",
                                                                                      .payload = boost::json::object{},
                                                                                    }),
                                      boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_CHECK(!(result));
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_internal));
}

BOOST_AUTO_TEST_CASE(ligo_tcp_connection_refused_returns_unavailable) {
  auto io = boost::asio::io_context{};
  auto future = boost::asio::co_spawn(io, tcp_client_call("127.0.0.1", 9, {
                                                                            .route = "echo",
                                                                            .payload = boost::json::object{},
                                                                          }),
                                      boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_CHECK(!(result));
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_unavailable));
}

BOOST_AUTO_TEST_CASE(ligo_tcp_stalled_peer_times_out) {
  auto io = boost::asio::io_context{};
  auto acceptor = std::make_shared<boost::asio::ip::tcp::acceptor>(io, boost::asio::ip::tcp::endpoint{
                                                                         boost::asio::ip::tcp::v4(),
                                                                         0,
                                                                       });
  auto held_socket = std::make_shared<boost::asio::ip::tcp::socket>(io);
  boost::asio::co_spawn(io, accept_and_hold(acceptor, held_socket, std::chrono::milliseconds{50}), boost::asio::detached);
  auto future = boost::asio::co_spawn(
    io,
    tcp_client_call(
      "127.0.0.1",
      acceptor->local_endpoint().port(),
      {
        .route = "echo",
        .payload = boost::json::object{},
      },
      {
        .timeout = std::chrono::milliseconds{20},
      }),
    boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_CHECK(!(result));
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_timeout));
}

BOOST_AUTO_TEST_CASE(ligo_tcp_concurrent_client_calls_succeed) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("TCP", "echo", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return modulus::success(std::move(payload));
  });
  auto server = ligo::tcp::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto client = ligo::tcp::client{
    {
      .host = "127.0.0.1",
      .port = server.selected_port(),
    },
  };
  auto work = boost::asio::make_work_guard(io);
  auto runners = start_io_threads(io);
  auto futures = std::vector<std::future<std::expected<boost::json::value, modulus::error>>>{};
  for (auto index = 0; index < 16; ++index) {
    auto payload = boost::json::object{};
    payload["index"] = index;
    futures.push_back(boost::asio::co_spawn(io, client.call({
                                                  .route = "echo",
                                                  .payload = std::move(payload),
                                                }),
                                            boost::asio::use_future));
  }
  for (auto &future : futures) {
    auto result = future.get();
    BOOST_REQUIRE(result);
    BOOST_CHECK(result->is_object());
  }
  server.stop();
  work.reset();
  join_io_threads(runners);
}

BOOST_AUTO_TEST_CASE(ligo_tcp_async_stop_cancels_active_session) {
  auto io = boost::asio::io_context{};
  auto router = ligo::service_router{};
  router.add_route("TCP", "slow", [](ligo::request_context, boost::json::value payload) -> boost::asio::awaitable<std::expected<boost::json::value, modulus::error>> {
    co_return co_await delayed_echo(std::chrono::milliseconds{50}, std::move(payload));
  });
  auto server = ligo::tcp::server{
    io,
    {
      .host = "127.0.0.1",
      .port = 0,
    },
    std::move(router),
  };
  BOOST_REQUIRE(server.start());
  auto call = boost::asio::co_spawn(
    io,
    tcp_client_call_selected(
      server,
      {
        .route = "slow",
        .payload = boost::json::object{},
      },
      {
        .timeout = std::chrono::milliseconds{200},
      }),
    boost::asio::use_future);
  auto stopped = boost::asio::co_spawn(io, tcp_async_stop_after(server, std::chrono::milliseconds{5}, std::chrono::milliseconds{200}), boost::asio::use_future);
  run_two_threads(io);
  BOOST_CHECK(!(call.get()));
  BOOST_CHECK(stopped.get());
}

BOOST_AUTO_TEST_CASE(ligo_tcp_connect_returns_connected_socket) {
  auto io = boost::asio::io_context{};
  auto acceptor = std::make_shared<boost::asio::ip::tcp::acceptor>(io, boost::asio::ip::tcp::endpoint{
                                                                         boost::asio::ip::tcp::v4(),
                                                                         0,
                                                                       });
  auto held_socket = std::make_shared<boost::asio::ip::tcp::socket>(io);
  boost::asio::co_spawn(io, accept_and_hold(acceptor, held_socket, std::chrono::milliseconds{10}), boost::asio::detached);
  auto result = boost::asio::co_spawn(
    io,
    ligo::tcp::connect(
      io.get_executor(),
      {
        .host = "127.0.0.1",
        .port = acceptor->local_endpoint().port(),
      },
      {
        .timeout = std::chrono::milliseconds{1000},
      }),
    boost::asio::use_future);
  io.run();
  auto socket = result.get();
  BOOST_REQUIRE(socket);
  BOOST_TEST(socket->is_open());
  auto ignored = boost::system::error_code{};
  socket->close(ignored);
}

BOOST_AUTO_TEST_CASE(ligo_net_connect_returns_connected_socket) {
  auto io = boost::asio::io_context{};
  auto acceptor = std::make_shared<boost::asio::ip::tcp::acceptor>(io, boost::asio::ip::tcp::endpoint{
                                                                         boost::asio::ip::tcp::v4(),
                                                                         0,
                                                                       });
  auto held_socket = std::make_shared<boost::asio::ip::tcp::socket>(io);
  boost::asio::co_spawn(io, accept_and_hold(acceptor, held_socket, std::chrono::milliseconds{10}), boost::asio::detached);
  auto result = boost::asio::co_spawn(
    io,
    ligo::net::connect(
      io.get_executor(),
      {
        .host = "127.0.0.1",
        .port = acceptor->local_endpoint().port(),
      },
      {
        .timeout = std::chrono::milliseconds{1000},
      }),
    boost::asio::use_future);
  io.run();
  auto socket = result.get();
  BOOST_REQUIRE(socket);
  BOOST_TEST(socket->is_open());
  auto ignored = boost::system::error_code{};
  socket->close(ignored);
}

BOOST_AUTO_TEST_CASE(ligo_net_listener_accept_returns_socket) {
  auto io = boost::asio::io_context{};
  auto listener_result = ligo::net::listener::bind(io.get_executor(), {
                                                                        .host = "127.0.0.1",
                                                                        .port = 0,
                                                                      });
  BOOST_REQUIRE(listener_result);
  auto listener = std::move(*listener_result);
  BOOST_CHECK((listener.selected_port()) != (0));
  auto accepted = boost::asio::co_spawn(io, listener.async_accept(), boost::asio::use_future);
  auto connected = boost::asio::co_spawn(
    io,
    ligo::net::connect_tcp(
      io.get_executor(),
      {
        .host = "127.0.0.1",
        .port = listener.selected_port(),
      },
      {
        .timeout = std::chrono::milliseconds{1000},
      }),
    boost::asio::use_future);
  io.run();
  auto accepted_socket = accepted.get();
  auto connected_socket = connected.get();
  BOOST_REQUIRE(accepted_socket);
  BOOST_REQUIRE(connected_socket);
  BOOST_CHECK(accepted_socket->is_open());
  BOOST_CHECK(connected_socket->is_open());
  auto ignored = boost::system::error_code{};
  accepted_socket->close(ignored);
  connected_socket->close(ignored);
  listener.close();
}

BOOST_AUTO_TEST_CASE(ligo_net_byte_stream_round_trips_over_tcp) {
  auto io = boost::asio::io_context{};
  auto listener_result = ligo::net::listener::bind(io.get_executor(), {
                                                                        .host = "127.0.0.1",
                                                                        .port = 0,
                                                                      });
  BOOST_REQUIRE(listener_result);
  auto listener = std::move(*listener_result);
  auto accepted = boost::asio::co_spawn(io, listener.async_accept(), boost::asio::use_future);
  auto connected = boost::asio::co_spawn(
    io,
    ligo::net::connect_tcp(
      io.get_executor(),
      {
        .host = "127.0.0.1",
        .port = listener.selected_port(),
      },
      {
        .timeout = std::chrono::milliseconds{1000},
      }),
    boost::asio::use_future);
  io.run();
  auto accepted_socket = accepted.get();
  auto connected_socket = connected.get();
  BOOST_REQUIRE(accepted_socket);
  BOOST_REQUIRE(connected_socket);
  auto accepted_stream = ligo::net::byte_stream::from_socket(std::move(*accepted_socket));
  auto connected_stream = ligo::net::byte_stream::from_socket(std::move(*connected_socket));
  BOOST_REQUIRE(accepted_stream);
  BOOST_REQUIRE(connected_stream);
  auto payload = std::string{"hello"};
  auto buffer = std::string(payload.size(), '\0');
  io.restart();
  auto read = boost::asio::co_spawn(io, accepted_stream->async_read_some(boost::asio::buffer(buffer), std::chrono::milliseconds{1000}), boost::asio::use_future);
  auto written = boost::asio::co_spawn(io, connected_stream->async_write_all(boost::asio::buffer(payload), std::chrono::milliseconds{1000}), boost::asio::use_future);
  io.run();
  auto read_result = read.get();
  BOOST_REQUIRE(read_result);
  BOOST_CHECK((*read_result) == payload.size());
  BOOST_CHECK(buffer == payload);
  BOOST_CHECK(written.get());
  auto ignored = boost::system::error_code{};
  accepted_stream->close(ignored);
  connected_stream->close(ignored);
  listener.close();
}

BOOST_AUTO_TEST_CASE(ligo_net_listener_tls_accept_rejects_non_tls_security_modes) {
  auto io = boost::asio::io_context{};
  auto listener_result = ligo::net::listener::bind(io.get_executor(), {
                                                                        .host = "127.0.0.1",
                                                                        .port = 0,
                                                                      });
  BOOST_REQUIRE(listener_result);
  auto listener = std::move(*listener_result);
  auto none = boost::asio::co_spawn(io, listener.async_accept_tls_configured({
                                                                               .mode = ligo::security_mode::NONE,
                                                                             },
                                                                             std::chrono::milliseconds{1000}),
                                    boost::asio::use_future);
  io.run();
  auto result = none.get();
  BOOST_CHECK(!result);
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_invalid_argument));
  listener.close();
}

BOOST_AUTO_TEST_CASE(ligo_net_listener_tls_handshake_succeeds) {
  auto io = boost::asio::io_context{};
  auto listener_result = ligo::net::listener::bind(io.get_executor(), {
                                                                        .host = "127.0.0.1",
                                                                        .port = 0,
                                                                      });
  BOOST_REQUIRE(listener_result);
  auto listener = std::move(*listener_result);
  auto accepted = boost::asio::co_spawn(io, listener.async_accept_tls_handshaken(test_tls_server_config(), std::chrono::milliseconds{1000}), boost::asio::use_future);
  auto connected = boost::asio::co_spawn(
    io,
    ligo::net::connect_tls_handshaken(
      io.get_executor(),
      {
        .host = "127.0.0.1",
        .port = listener.selected_port(),
      },
      {
        .timeout = std::chrono::milliseconds{1000},
      },
      test_tls_client_config()),
    boost::asio::use_future);
  io.run();
  auto accepted_socket = accepted.get();
  auto connected_socket = connected.get();
  BOOST_REQUIRE(accepted_socket);
  BOOST_REQUIRE(connected_socket);
  BOOST_CHECK(accepted_socket->is_open());
  BOOST_CHECK(connected_socket->is_open());
  auto ignored = boost::system::error_code{};
  accepted_socket->close(ignored);
  connected_socket->close(ignored);
  listener.close();
}

BOOST_AUTO_TEST_CASE(ligo_tcp_raw_connect_rejects_secure_modes) {
  auto io = boost::asio::io_context{};
  auto future = boost::asio::co_spawn(
    io,
    ligo::tcp::connect(
      io.get_executor(),
      {
        .host = "127.0.0.1",
        .port = 3306,
      },
      {
        .timeout = std::chrono::milliseconds{1000},
        .security = {
          .mode = ligo::security_mode::TLS,
        },
      }),
    boost::asio::use_future);
  io.run();
  auto result = future.get();
  BOOST_CHECK(!result);
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_tcp_tls_connect_rejects_non_tls_security_modes) {
  auto io = boost::asio::io_context{};
  auto none = boost::asio::co_spawn(
    io,
    ligo::tcp::connect_tls_handshaken(
      io.get_executor(),
      {
        .host = "127.0.0.1",
        .port = 3306,
      },
      {
        .timeout = std::chrono::milliseconds{1000},
      },
      {
        .mode = ligo::security_mode::NONE,
      }),
    boost::asio::use_future);
  auto dtls = boost::asio::co_spawn(
    io,
    ligo::tcp::connect_tls_configured(
      io.get_executor(),
      {
        .host = "127.0.0.1",
        .port = 3306,
      },
      {
        .timeout = std::chrono::milliseconds{1000},
      },
      {
        .mode = ligo::security_mode::DTLS,
      }),
    boost::asio::use_future);
  io.run();
  auto none_result = none.get();
  BOOST_CHECK(!none_result);
  BOOST_CHECK((none_result.error().code()) == (ligo::detail::error_invalid_argument));
  auto dtls_result = dtls.get();
  BOOST_CHECK(!dtls_result);
  BOOST_CHECK((dtls_result.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_tcp_socket_handle_is_move_only_and_release_is_checked) {
  static_assert(!std::is_copy_constructible_v<ligo::tcp::socket_handle>);
  static_assert(!std::is_copy_assignable_v<ligo::tcp::socket_handle>);
  static_assert(std::is_move_constructible_v<ligo::tcp::socket_handle>);
  static_assert(std::is_move_assignable_v<ligo::tcp::socket_handle>);
  auto io = boost::asio::io_context{};
  auto handle = ligo::tcp::socket_handle{
    boost::asio::ip::tcp::socket{io},
  };
  auto socket = handle.socket();
  BOOST_REQUIRE(socket);
  auto released = handle.release_socket();
  BOOST_REQUIRE(released);
  BOOST_CHECK(!(handle.is_open()));
  auto released_again = handle.release_socket();
  BOOST_CHECK(!(released_again));
  BOOST_CHECK((released_again.error().code()) == (ligo::detail::error_unavailable));
  auto socket_again = handle.socket();
  BOOST_CHECK(!(socket_again));
  BOOST_CHECK((socket_again.error().code()) == (ligo::detail::error_unavailable));
}

BOOST_AUTO_TEST_CASE(ligo_tcp_tls_socket_release_is_checked) {
  static_assert(!std::is_copy_constructible_v<ligo::tcp::tls_socket>);
  static_assert(!std::is_copy_assignable_v<ligo::tcp::tls_socket>);
  static_assert(std::is_move_constructible_v<ligo::tcp::tls_socket>);
  static_assert(std::is_move_assignable_v<ligo::tcp::tls_socket>);
  auto io = boost::asio::io_context{};
  auto context = ligo::make_tls_client_context({
    .mode = ligo::security_mode::TLS,
    .verify = ligo::verify_mode::NONE,
  });
  BOOST_REQUIRE(context);
  auto handle = ligo::tcp::tls_socket{
    std::move(*context),
    boost::asio::ip::tcp::socket{io},
  };
  auto stream_ref = handle.stream();
  BOOST_REQUIRE(stream_ref);
  auto stream = handle.release_stream();
  BOOST_REQUIRE(stream);
  BOOST_REQUIRE(stream->context);
  auto stream_again = handle.release_stream();
  BOOST_CHECK(!(stream_again));
  BOOST_CHECK((stream_again.error().code()) == (ligo::detail::error_unavailable));
  auto stream_ref_again = handle.stream();
  BOOST_CHECK(!(stream_ref_again));
  BOOST_CHECK((stream_ref_again.error().code()) == (ligo::detail::error_unavailable));
}

BOOST_AUTO_TEST_CASE(ligo_udp_dtls_reliable_round_trip_succeeds) {
  auto io = boost::asio::io_context{};
  auto client_socket = make_udp_socket(io);
  auto server_socket = make_udp_socket(io);
  auto client_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    client_socket->local_endpoint().port(),
  };
  auto server_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    server_socket->local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 2,
    .session_timeout = std::chrono::milliseconds{1000},
  };
  auto client_result = ligo::udp::client::adopt(std::move(client_socket), server_endpoint, {
                                                                                             .reliability = config,
                                                                                             .security = test_dtls_client_config(),
                                                                                             .session_id = 142,
                                                                                           });
  auto server_result = ligo::udp::server::adopt(std::move(server_socket), client_endpoint, {
                                                                                             .reliability = config,
                                                                                             .security = test_dtls_server_config(),
                                                                                             .session_id = 142,
                                                                                           });
  BOOST_REQUIRE(client_result);
  BOOST_REQUIRE(server_result);
  auto client = std::move(*client_result);
  auto server = std::move(*server_result);
  auto sent = boost::asio::co_spawn(io, client.send(boost::json::parse(R"json({"secure":true})json")), boost::asio::use_future);
  auto received = boost::asio::co_spawn(io, server.receive(), boost::asio::use_future);
  io.run();
  auto sent_status = sent.get();
  BOOST_CHECK(sent_status);
  auto result = received.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(result->payload.as_object().at("secure").as_bool());
}

BOOST_AUTO_TEST_CASE(ligo_udp_dtls_verified_peer_succeeds) {
  auto io = boost::asio::io_context{};
  auto client_socket = make_udp_socket(io);
  auto server_socket = make_udp_socket(io);
  auto client_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    client_socket->local_endpoint().port(),
  };
  auto server_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    server_socket->local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 2,
    .session_timeout = std::chrono::milliseconds{1000},
  };
  auto client_result = ligo::udp::client::adopt(std::move(client_socket), server_endpoint, {
                                                                                             .reliability = config,
                                                                                             .security = test_dtls_verified_client_config("localhost"),
                                                                                             .session_id = 143,
                                                                                           });
  auto server_result = ligo::udp::server::adopt(std::move(server_socket), client_endpoint, {
                                                                                             .reliability = config,
                                                                                             .security = test_dtls_server_config(),
                                                                                             .session_id = 143,
                                                                                           });
  BOOST_REQUIRE(client_result);
  BOOST_REQUIRE(server_result);
  auto client = std::move(*client_result);
  auto server = std::move(*server_result);
  auto sent = boost::asio::co_spawn(io, client.send(boost::json::parse(R"json({"secure":true})json")), boost::asio::use_future);
  auto received = boost::asio::co_spawn(io, server.receive(), boost::asio::use_future);
  io.run();
  auto status = sent.get();
  BOOST_CHECK(status);
  auto result = received.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(result->payload.as_object().at("secure").as_bool());
}

BOOST_AUTO_TEST_CASE(ligo_udp_dtls_server_peer_verification_succeeds_with_client_certificate) {
  auto io = boost::asio::io_context{};
  auto client_socket = make_udp_socket(io);
  auto server_socket = make_udp_socket(io);
  auto client_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    client_socket->local_endpoint().port(),
  };
  auto server_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    server_socket->local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 2,
    .session_timeout = std::chrono::milliseconds{1000},
  };
  auto client_result = ligo::udp::client::adopt(std::move(client_socket), server_endpoint, {
                                                                                             .reliability = config,
                                                                                             .security = test_dtls_identified_client_config(),
                                                                                             .session_id = 145,
                                                                                           });
  auto server_result = ligo::udp::server::adopt(std::move(server_socket), client_endpoint, {
                                                                                             .reliability = config,
                                                                                             .security = test_dtls_server_verify_client_config(),
                                                                                             .session_id = 145,
                                                                                           });
  BOOST_REQUIRE(client_result);
  BOOST_REQUIRE(server_result);
  auto client = std::move(*client_result);
  auto server = std::move(*server_result);
  auto sent = boost::asio::co_spawn(io, client.send(boost::json::parse(R"json({"secure":true})json")), boost::asio::use_future);
  auto received = boost::asio::co_spawn(io, server.receive(), boost::asio::use_future);
  io.run();
  auto status = sent.get();
  BOOST_CHECK(status);
  auto result = received.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(result->payload.as_object().at("secure").as_bool());
}

BOOST_AUTO_TEST_CASE(ligo_udp_dtls_server_peer_verification_fails_without_client_certificate) {
  auto io = boost::asio::io_context{};
  auto client_socket = make_udp_socket(io);
  auto server_socket = make_udp_socket(io);
  auto client_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    client_socket->local_endpoint().port(),
  };
  auto server_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    server_socket->local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 2,
    .session_timeout = std::chrono::milliseconds{100},
  };
  auto client_result = ligo::udp::client::adopt(std::move(client_socket), server_endpoint, {
                                                                                             .reliability = config,
                                                                                             .security = test_dtls_client_config(),
                                                                                             .session_id = 146,
                                                                                           });
  auto server_result = ligo::udp::server::adopt(std::move(server_socket), client_endpoint, {
                                                                                             .reliability = config,
                                                                                             .security = test_dtls_server_verify_client_config(),
                                                                                             .session_id = 146,
                                                                                           });
  BOOST_REQUIRE(client_result);
  BOOST_REQUIRE(server_result);
  auto client = std::move(*client_result);
  auto server = std::move(*server_result);
  auto sent = boost::asio::co_spawn(io, client.send(boost::json::object{}), boost::asio::use_future);
  auto received = boost::asio::co_spawn(io, server.receive(), boost::asio::use_future);
  auto stopper = boost::asio::steady_timer{io};
  stopper.expires_after(std::chrono::milliseconds{200});
  stopper.async_wait([&client, &server](boost::system::error_code const &) {
    client.close();
    server.close();
  });
  io.run();
  auto status = sent.get();
  BOOST_CHECK(!status);
  auto result = received.get();
  BOOST_CHECK(!result);
}

BOOST_AUTO_TEST_CASE(ligo_udp_dtls_untrusted_peer_fails) {
  auto io = boost::asio::io_context{};
  auto client_socket = make_udp_socket(io);
  auto server_socket = make_udp_socket(io);
  auto client_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    client_socket->local_endpoint().port(),
  };
  auto server_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    server_socket->local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 2,
    .session_timeout = std::chrono::milliseconds{100},
  };
  auto client_result = ligo::udp::client::adopt(std::move(client_socket), server_endpoint, {
                                                                                             .reliability = config,
                                                                                             .security = {
                                                                                               .mode = ligo::security_mode::DTLS,
                                                                                               .verify = ligo::verify_mode::PEER,
                                                                                               .peer_name = "localhost",
                                                                                             },
                                                                                             .session_id = 144,
                                                                                           });
  auto server_result = ligo::udp::server::adopt(std::move(server_socket), client_endpoint, {
                                                                                             .reliability = config,
                                                                                             .security = test_dtls_server_config(),
                                                                                             .session_id = 144,
                                                                                           });
  BOOST_REQUIRE(client_result);
  BOOST_REQUIRE(server_result);
  auto client = std::move(*client_result);
  auto server = std::move(*server_result);
  auto sent = boost::asio::co_spawn(io, client.send(boost::json::object{}), boost::asio::use_future);
  auto received = boost::asio::co_spawn(io, server.receive(), boost::asio::use_future);
  auto stopper = boost::asio::steady_timer{io};
  stopper.expires_after(std::chrono::milliseconds{200});
  stopper.async_wait([&client, &server](boost::system::error_code const &) {
    client.close();
    server.close();
  });
  io.run();
  auto status = sent.get();
  BOOST_CHECK(!status);
  auto result = received.get();
  BOOST_CHECK(!result);
}

BOOST_AUTO_TEST_CASE(ligo_udp_dtls_server_requires_identity) {
  auto io = boost::asio::io_context{};
  auto peer = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    9,
  };
  auto server = ligo::udp::server::bind(io.get_executor(), {
                                                             .local = {
                                                               boost::asio::ip::udp::v4(),
                                                               0,
                                                             },
                                                             .peer = peer,
                                                           },
                                        {
                                          .security = {
                                            .mode = ligo::security_mode::DTLS,
                                          },
                                        });
  BOOST_CHECK(!server);
  BOOST_CHECK((server.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_udp_reliable_round_trip_succeeds) {
  auto io = boost::asio::io_context{};
  auto left_socket = make_udp_socket(io);
  auto right_socket = make_udp_socket(io);
  auto left_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    left_socket->local_endpoint().port(),
  };
  auto right_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    right_socket->local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 2,
    .session_timeout = std::chrono::milliseconds{100},
  };
  auto left_result = udp_client_adopt(std::move(left_socket), right_endpoint, config, 42);
  auto right_result = udp_client_adopt(std::move(right_socket), left_endpoint, config, 42);
  BOOST_REQUIRE(left_result);
  BOOST_REQUIRE(right_result);
  auto left = std::move(*left_result);
  auto right = std::move(*right_result);
  auto sent = boost::asio::co_spawn(io, left.send(boost::json::parse(R"json({"ok":true})json")), boost::asio::use_future);
  auto received = boost::asio::co_spawn(io, right.receive(), boost::asio::use_future);
  io.run();
  BOOST_CHECK(sent.get());
  auto result = received.get();
  BOOST_REQUIRE(result);
  BOOST_CHECK(result->payload.as_object().at("ok").as_bool());
}

BOOST_AUTO_TEST_CASE(ligo_udp_bidirectional_concurrent_send_receive_succeeds) {
  auto io = boost::asio::io_context{};
  auto left_socket = make_udp_socket(io);
  auto right_socket = make_udp_socket(io);
  auto left_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    left_socket->local_endpoint().port(),
  };
  auto right_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    right_socket->local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 2,
    .session_timeout = std::chrono::milliseconds{100},
  };
  auto left_result = udp_client_adopt(std::move(left_socket), right_endpoint, config, 84);
  auto right_result = udp_client_adopt(std::move(right_socket), left_endpoint, config, 84);
  BOOST_REQUIRE(left_result);
  BOOST_REQUIRE(right_result);
  auto left = std::move(*left_result);
  auto right = std::move(*right_result);
  auto left_sent = boost::asio::co_spawn(io, left.send(boost::json::parse(R"json({"from":"left"})json")), boost::asio::use_future);
  auto left_received = boost::asio::co_spawn(io, left.receive(), boost::asio::use_future);
  auto right_sent = boost::asio::co_spawn(io, right.send(boost::json::parse(R"json({"from":"right"})json")), boost::asio::use_future);
  auto right_received = boost::asio::co_spawn(io, right.receive(), boost::asio::use_future);
  io.run();
  BOOST_CHECK(left_sent.get());
  BOOST_CHECK(right_sent.get());
  auto left_payload = left_received.get();
  auto right_payload = right_received.get();
  BOOST_REQUIRE(left_payload);
  BOOST_REQUIRE(right_payload);
  BOOST_CHECK((left_payload->payload.as_object().at("from").as_string()) == ("right"));
  BOOST_CHECK((right_payload->payload.as_object().at("from").as_string()) == ("left"));
}

BOOST_AUTO_TEST_CASE(ligo_udp_wrong_session_times_out) {
  auto io = boost::asio::io_context{};
  auto receiver = boost::asio::ip::udp::socket{
    io,
    {
      boost::asio::ip::udp::v4(),
      0,
    },
  };
  auto receiver_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    receiver.local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 1,
    .session_timeout = std::chrono::milliseconds{20},
  };
  auto channel_result = udp_client_bind(io, boost::asio::ip::udp::endpoint{
                                              boost::asio::ip::udp::v4(),
                                              0,
                                            },
                                        receiver_endpoint, config, 7);
  BOOST_REQUIRE(channel_result);
  auto channel = std::move(*channel_result);
  auto future = boost::asio::co_spawn(io, channel.send(boost::json::object{}), boost::asio::use_future);
  io.run();
  auto status = future.get();
  BOOST_CHECK(!(status));
  BOOST_CHECK((status.error().code()) == (ligo::detail::error_timeout));
}

BOOST_AUTO_TEST_CASE(ligo_udp_oversized_payload_returns_invalid_argument) {
  auto io = boost::asio::io_context{};
  auto receiver = boost::asio::ip::udp::socket{
    io,
    {
      boost::asio::ip::udp::v4(),
      0,
    },
  };
  auto receiver_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    receiver.local_endpoint().port(),
  };
  auto channel_result = udp_client_bind(io, boost::asio::ip::udp::endpoint{
                                              boost::asio::ip::udp::v4(),
                                              0,
                                            },
                                        receiver_endpoint);
  BOOST_REQUIRE(channel_result);
  auto channel = std::move(*channel_result);
  auto payload = boost::json::value{
    boost::json::string{std::string(70000, 'x')},
  };
  auto future = boost::asio::co_spawn(io, channel.send(std::move(payload)), boost::asio::use_future);
  io.run();
  auto status = future.get();
  BOOST_CHECK(!(status));
  BOOST_CHECK((status.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_udp_close_wakes_pending_receive) {
  auto io = boost::asio::io_context{};
  auto right_socket = boost::asio::ip::udp::socket{
    io,
    {
      boost::asio::ip::udp::v4(),
      0,
    },
  };
  auto right_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    right_socket.local_endpoint().port(),
  };
  auto channel_result = udp_client_bind(io, boost::asio::ip::udp::endpoint{
                                              boost::asio::ip::udp::v4(),
                                              0,
                                            },
                                        right_endpoint);
  BOOST_REQUIRE(channel_result);
  auto channel = std::move(*channel_result);
  auto received = boost::asio::co_spawn(io, channel.receive(), boost::asio::use_future);
  auto timer = boost::asio::steady_timer{io};
  timer.expires_after(std::chrono::milliseconds{5});
  timer.async_wait([&channel](boost::system::error_code const &) {
    channel.close();
  });
  io.run();
  auto result = received.get();
  BOOST_CHECK(!(result));
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_unavailable));
}

BOOST_AUTO_TEST_CASE(ligo_udp_async_close_drains_pending_receive) {
  auto io = boost::asio::io_context{};
  auto right_socket = boost::asio::ip::udp::socket{
    io,
    {
      boost::asio::ip::udp::v4(),
      0,
    },
  };
  auto right_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    right_socket.local_endpoint().port(),
  };
  auto channel_result = udp_client_bind(io, boost::asio::ip::udp::endpoint{
                                              boost::asio::ip::udp::v4(),
                                              0,
                                            },
                                        right_endpoint);
  BOOST_REQUIRE(channel_result);
  auto channel = std::move(*channel_result);
  auto received = boost::asio::co_spawn(io, channel.receive(), boost::asio::use_future);
  auto closed = boost::asio::co_spawn(io, channel.async_close(std::chrono::milliseconds{1000}), boost::asio::use_future);
  io.run();
  BOOST_CHECK(closed.get());
  auto result = received.get();
  BOOST_CHECK(!(result));
  BOOST_CHECK((result.error().code()) == (ligo::detail::error_unavailable));
}

BOOST_AUTO_TEST_CASE(ligo_udp_async_close_drains_pending_send) {
  auto io = boost::asio::io_context{};
  auto receiver = boost::asio::ip::udp::socket{
    io,
    {
      boost::asio::ip::udp::v4(),
      0,
    },
  };
  auto receiver_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    receiver.local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 1,
    .session_timeout = std::chrono::milliseconds{1000},
  };
  auto channel_result = udp_client_bind(io, boost::asio::ip::udp::endpoint{
                                              boost::asio::ip::udp::v4(),
                                              0,
                                            },
                                        receiver_endpoint, config, 171);
  BOOST_REQUIRE(channel_result);
  auto channel = std::move(*channel_result);
  auto sent = boost::asio::co_spawn(io, channel.send(boost::json::object{}), boost::asio::use_future);
  auto closed = boost::asio::co_spawn(io, channel.async_close(std::chrono::milliseconds{1000}), boost::asio::use_future);
  io.run();
  BOOST_CHECK(closed.get());
  auto status = sent.get();
  BOOST_CHECK(!status);
  BOOST_CHECK((status.error().code()) == (ligo::detail::error_unavailable));
}

BOOST_AUTO_TEST_CASE(ligo_udp_async_close_moved_from_channel_is_ok) {
  auto io = boost::asio::io_context{};
  auto right_socket = boost::asio::ip::udp::socket{
    io,
    {
      boost::asio::ip::udp::v4(),
      0,
    },
  };
  auto right_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    right_socket.local_endpoint().port(),
  };
  auto channel_result = udp_client_bind(io, boost::asio::ip::udp::endpoint{
                                              boost::asio::ip::udp::v4(),
                                              0,
                                            },
                                        right_endpoint);
  BOOST_REQUIRE(channel_result);
  auto channel = std::move(*channel_result);
  auto moved = std::move(channel);
  static_cast<void>(moved);
  auto closed = boost::asio::co_spawn(io, channel.async_close(std::chrono::milliseconds{1000}), boost::asio::use_future);
  io.run();
  BOOST_CHECK(closed.get());
}

BOOST_AUTO_TEST_CASE(ligo_udp_async_close_rejects_invalid_timeout) {
  auto io = boost::asio::io_context{};
  auto right_socket = boost::asio::ip::udp::socket{
    io,
    {
      boost::asio::ip::udp::v4(),
      0,
    },
  };
  auto right_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    right_socket.local_endpoint().port(),
  };
  auto channel_result = udp_client_bind(io, boost::asio::ip::udp::endpoint{
                                              boost::asio::ip::udp::v4(),
                                              0,
                                            },
                                        right_endpoint);
  BOOST_REQUIRE(channel_result);
  auto channel = std::move(*channel_result);
  auto closed = boost::asio::co_spawn(io, channel.async_close(std::chrono::milliseconds{0}), boost::asio::use_future);
  io.run();
  auto status = closed.get();
  BOOST_CHECK(!status);
  BOOST_CHECK((status.error().code()) == (ligo::detail::error_invalid_argument));
}

BOOST_AUTO_TEST_CASE(ligo_udp_rapid_sequential_round_trips_succeed) {
  auto io = boost::asio::io_context{};
  auto left_socket = make_udp_socket(io);
  auto right_socket = make_udp_socket(io);
  auto left_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    left_socket->local_endpoint().port(),
  };
  auto right_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    right_socket->local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 2,
    .session_timeout = std::chrono::milliseconds{100},
  };
  auto left_result = udp_client_adopt(std::move(left_socket), right_endpoint, config, 242);
  auto right_result = udp_client_adopt(std::move(right_socket), left_endpoint, config, 242);
  BOOST_REQUIRE(left_result);
  BOOST_REQUIRE(right_result);
  auto left = std::move(*left_result);
  auto right = std::move(*right_result);
  auto work = boost::asio::make_work_guard(io);
  auto runners = start_io_threads(io);
  for (auto index = 0; index < 32; ++index) {
    auto payload = boost::json::object{};
    payload["index"] = index;
    auto sent = boost::asio::co_spawn(io, left.send(std::move(payload)), boost::asio::use_future);
    auto received = boost::asio::co_spawn(io, right.receive(), boost::asio::use_future);
    auto status = sent.get();
    auto message = received.get();
    BOOST_REQUIRE(status);
    BOOST_REQUIRE(message);
    BOOST_CHECK((message->payload.as_object().at("index").as_int64()) == (index));
  }
  left.close();
  right.close();
  work.reset();
  join_io_threads(runners);
}

BOOST_AUTO_TEST_CASE(ligo_udp_multithreaded_io_context_round_trip_succeeds) {
  auto io = boost::asio::io_context{};
  auto left_socket = make_udp_socket(io);
  auto right_socket = make_udp_socket(io);
  auto left_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    left_socket->local_endpoint().port(),
  };
  auto right_endpoint = boost::asio::ip::udp::endpoint{
    boost::asio::ip::make_address("127.0.0.1"),
    right_socket->local_endpoint().port(),
  };
  auto config = ligo::udp_reliability_config{
    .resend_interval = std::chrono::milliseconds{5},
    .max_in_flight = 4,
    .session_timeout = std::chrono::milliseconds{100},
  };
  auto left_result = udp_client_adopt(std::move(left_socket), right_endpoint, config, 128);
  auto right_result = udp_client_adopt(std::move(right_socket), left_endpoint, config, 128);
  BOOST_REQUIRE(left_result);
  BOOST_REQUIRE(right_result);
  auto left = std::move(*left_result);
  auto right = std::move(*right_result);
  auto sent = boost::asio::co_spawn(io, left.send(boost::json::parse(R"json({"ok":true})json")), boost::asio::use_future);
  auto received = boost::asio::co_spawn(io, right.receive(), boost::asio::use_future);
  run_two_threads(io);
  BOOST_CHECK(sent.get());
  auto message = received.get();
  BOOST_REQUIRE(message);
  BOOST_CHECK(message->payload.as_object().at("ok").as_bool());
}