Files
khodowany e4125d4c4b
publish / publish (push) Successful in 26m13s
Fix
2026-10-06 04:12:27 -05:00

2361 lines
98 KiB
C++

#include <atomic>
#include <chrono>
#include <cstdint>
#include <exception>
#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());
}
BOOST_AUTO_TEST_CASE(ligo_http_client_operations_capture_inputs_before_await) {
auto io = boost::asio::io_context{};
auto client = ligo::http::client{{"127.0.0.1", 1}};
auto request = ligo::http::request{.target = ""};
auto call = client.call(request);
auto exchange = client.exchange(request);
client = ligo::http::client{{"127.0.0.1", 0}};
request.target = "/";
auto called = boost::asio::co_spawn(io, std::move(call), boost::asio::use_future);
auto exchanged = boost::asio::co_spawn(io, std::move(exchange), boost::asio::use_future);
io.run();
auto call_result = called.get();
BOOST_REQUIRE(!call_result);
BOOST_CHECK_EQUAL(call_result.error().code(), ligo::detail::error_invalid_argument);
BOOST_CHECK_EQUAL(call_result.error().message(), "HTTP target must not be empty");
auto exchange_result = exchanged.get();
BOOST_REQUIRE(!exchange_result);
BOOST_CHECK_EQUAL(exchange_result.error().code(), ligo::detail::error_invalid_argument);
BOOST_CHECK_EQUAL(exchange_result.error().message(), "HTTP target must not be empty");
}
BOOST_AUTO_TEST_CASE(ligo_http_client_call_outlives_client_and_request) {
auto io = boost::asio::io_context{};
// Both the helper's client and this temporary request expire before io.run().
auto future = boost::asio::co_spawn(io, http_client_call("127.0.0.1", 1, {.target = ""}), boost::asio::use_future);
io.run();
auto result = future.get();
BOOST_REQUIRE(!result);
BOOST_CHECK_EQUAL(result.error().code(), ligo::detail::error_invalid_argument);
BOOST_CHECK_EQUAL(result.error().message(), "HTTP target must not be empty");
}
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<std::expected<ligo::service_response, modulus::error>> {
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<std::expected<ligo::service_response, modulus::error>> {
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<void> {
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);
};
auto failure = std::exception_ptr{};
boost::asio::co_spawn(io, exercise(), [&](std::exception_ptr error) {
failure = error;
server.stop();
});
io.run();
if (failure) {
std::rethrow_exception(failure);
}
}