#include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include 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 { auto threads = std::vector{}; threads.emplace_back([&] { io.run(); }); threads.emplace_back([&] { io.run(); }); return threads; } auto join_io_threads(std::vector &threads) -> void { for (auto &thread : threads) { thread.join(); } } auto delayed_echo(std::chrono::milliseconds delay, boost::json::value payload) -> boost::asio::awaitable> { 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> { 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> { 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 request, ligo::http::client_options options) -> boost::asio::awaitable> { 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> { return http_client_call_selected_and_stop_impl(server, std::make_shared(request), options); } auto http_async_stop_after(ligo::http::server &server, std::chrono::milliseconds delay, std::chrono::milliseconds timeout) -> boost::asio::awaitable> { 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> { 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(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(status); } auto request = boost::beast::http::request{ 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(written); if (write_error) { auto status = test_network_failure(write_error); co_return ligo::detail::fail(status); } auto buffer = boost::beast::flat_buffer{}; auto response = boost::beast::http::response{}; auto [read_error, read] = co_await boost::beast::http::async_read(socket, buffer, response, boost::asio::as_tuple(boost::asio::use_awaitable)); static_cast(read); if (read_error) { auto status = test_network_failure(read_error); co_return ligo::detail::fail(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(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> { 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, 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>(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>(status); } auto request = boost::beast::http::request{ 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(written); if (write_error) { auto status = test_network_failure(write_error); co_return ligo::detail::fail>(status); } auto buffer = boost::beast::flat_buffer{}; auto response = boost::beast::http::response{}; auto [read_error, read] = co_await boost::beast::http::async_read(socket, buffer, response, boost::asio::as_tuple(boost::asio::use_awaitable)); static_cast(read); if (read_error) { auto status = test_network_failure(read_error); co_return ligo::detail::fail>(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, 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> { 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> { 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 request, ligo::tcp::client_options options) -> boost::asio::awaitable> { 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> { return tcp_client_call_selected_and_stop_impl(server, std::make_shared(request), options); } auto tcp_async_stop_after(ligo::tcp::server &server, std::chrono::milliseconds delay, std::chrono::milliseconds timeout) -> boost::asio::awaitable> { 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 acceptor, std::shared_ptr socket, std::chrono::milliseconds duration) -> boost::asio::awaitable { 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 { return std::make_unique(io, boost::asio::ip::udp::endpoint{ boost::asio::ip::udp::v4(), 0, }); } auto udp_client_adopt(std::unique_ptr socket, boost::asio::ip::udp::endpoint peer, ligo::udp_reliability_config config = {}, std::uint64_t session_id = 1) -> std::expected { 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 { 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> { auto tcp_payload = boost::json::object{}; tcp_payload["ok"] = true; auto tcp_request = std::make_shared(); 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(co_await tcp_server.async_stop(std::chrono::milliseconds{1000})); static_cast(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(); 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(co_await tcp_server.async_stop(std::chrono::milliseconds{1000})); static_cast(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> { 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> { 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> { 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> { co_return ligo::text_response("text/html; charset=utf-8", "" + context.route + ""); }); 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()) == ("/setup")); } 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> { 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::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> { 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> { co_return ligo::detail::fail(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> { 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> { 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> { 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>>{}; 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> { 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> { 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> { 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> { 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> { 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::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> { co_return ligo::detail::fail(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(io, boost::asio::ip::tcp::endpoint{ boost::asio::ip::tcp::v4(), 0, }); auto held_socket = std::make_shared(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> { 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>>{}; 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> { 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(io, boost::asio::ip::tcp::endpoint{ boost::asio::ip::tcp::v4(), 0, }); auto held_socket = std::make_shared(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(io, boost::asio::ip::tcp::endpoint{ boost::asio::ip::tcp::v4(), 0, }); auto held_socket = std::make_shared(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); static_assert(!std::is_copy_assignable_v); static_assert(std::is_move_constructible_v); static_assert(std::is_move_assignable_v); 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); static_assert(!std::is_copy_assignable_v); static_assert(std::is_move_constructible_v); static_assert(std::is_move_assignable_v); 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(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()); } BOOST_AUTO_TEST_CASE(ligo_response_metadata_is_preserved_by_router) { auto io = boost::asio::io_context{}; auto router = ligo::service_router{}; router.add_response_route("GET", "/metadata", [](ligo::request_context context, boost::json::value) -> boost::asio::awaitable> { BOOST_CHECK_EQUAL(context.headers.at("cookie"), "session=example"); BOOST_CHECK_EQUAL(context.query.at("search"), "two words"); auto response = ligo::text_response("text/plain", "accepted"); response.status = 201; response.headers.emplace_back("Set-Cookie", "session=replacement; HttpOnly"); response.headers.emplace_back("Cache-Control", "no-store"); co_return response; }); auto future = boost::asio::co_spawn(io, router.call_response({.method = "GET", .route = "/metadata", .headers = {{"cookie", "session=example"}}, .query = {{"search", "two words"}}}, {}), boost::asio::use_future); io.run(); auto result = future.get(); BOOST_REQUIRE(result); BOOST_CHECK_EQUAL(result->status, 201); BOOST_CHECK_EQUAL(result->body, "accepted"); BOOST_CHECK_EQUAL(result->headers.size(), 2); // The old aggregate initialization remains valid. auto options = ligo::http::server_options{std::chrono::milliseconds{1000}, 1024, {}, ligo::security_config{}}; BOOST_CHECK(!options.parse_query); } BOOST_AUTO_TEST_CASE(ligo_http_raw_exchange_headers_query_and_status) { auto io = boost::asio::io_context{}; auto router = ligo::service_router{}; router.add_response_route("GET", "/raw", [](ligo::request_context context, boost::json::value) -> boost::asio::awaitable> { BOOST_CHECK_EQUAL(context.headers.at("if-none-match"), "revision-one"); BOOST_CHECK_EQUAL(context.query.at("search"), "two words+"); auto response = ligo::text_response("text/plain", "not a JSON envelope"); response.status = 202; response.headers = {{"ETag", "revision-two"}, {"Set-Cookie", "sample=value; HttpOnly"}}; co_return response; }); auto options = ligo::http::server_options{}; options.parse_query = true; auto server = ligo::http::server{io, {"127.0.0.1", 0}, std::move(router), options}; BOOST_REQUIRE(server.start()); auto exercise = [&]() -> boost::asio::awaitable { auto client = ligo::http::client{{"127.0.0.1", server.selected_port()}}; // GCC 12 can ICE on initializer-list temporaries in co_await expressions. auto request = ligo::http::request{.method = ligo::http::method::GET, .target = "/raw?search=two+words%2B"}; request.headers.emplace("If-None-Match", "revision-one"); auto result = co_await client.exchange(request); BOOST_REQUIRE(result); BOOST_CHECK_EQUAL(result->status, 202); BOOST_CHECK_EQUAL(result->body, "not a JSON envelope"); BOOST_CHECK_EQUAL(result->headers.at("etag"), "revision-two"); auto invalid_request = ligo::http::request{.method = ligo::http::method::GET, .target = "/raw?search=%ZZ"}; auto invalid = co_await client.exchange(invalid_request); BOOST_REQUIRE(invalid); BOOST_CHECK_EQUAL(invalid->status, 400); auto duplicate_request = ligo::http::request{.method = ligo::http::method::GET, .target = "/raw?x=1&x=2"}; auto duplicate = co_await client.exchange(duplicate_request); BOOST_REQUIRE(duplicate); BOOST_CHECK_EQUAL(duplicate->status, 400); server.stop(); }; auto future = boost::asio::co_spawn(io, exercise(), boost::asio::use_future); io.run(); future.get(); }