127 lines
3.9 KiB
C++
127 lines
3.9 KiB
C++
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
#include <catch2/catch_test_macros.hpp>
|
|
|
|
#include <yams/daemon/ipc/message_framing.h>
|
|
#include <yams/daemon/ipc/request_handler.h>
|
|
|
|
#include <boost/asio.hpp>
|
|
|
|
#include <array>
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <cstdint>
|
|
#include <memory>
|
|
#include <optional>
|
|
#include <span>
|
|
#include <thread>
|
|
#include <vector>
|
|
|
|
using namespace yams::daemon;
|
|
|
|
namespace {
|
|
|
|
class CountingProcessor final : public RequestProcessor {
|
|
public:
|
|
boost::asio::awaitable<Response> process(const Request&) override {
|
|
unary_calls_++;
|
|
StatusResponse s{};
|
|
s.running = true;
|
|
s.ready = false;
|
|
s.version = "test";
|
|
s.overallStatus = "initializing";
|
|
co_return Response{std::in_place_type<StatusResponse>, s};
|
|
}
|
|
|
|
boost::asio::awaitable<std::optional<Response>> process_streaming(const Request&) override {
|
|
streaming_calls_++;
|
|
co_return std::optional<Response>{std::nullopt};
|
|
}
|
|
|
|
bool supports_streaming(const Request&) const override { return true; }
|
|
|
|
int unary_calls() const { return unary_calls_.load(); }
|
|
int streaming_calls() const { return streaming_calls_.load(); }
|
|
|
|
private:
|
|
std::atomic<int> unary_calls_{0};
|
|
std::atomic<int> streaming_calls_{0};
|
|
};
|
|
|
|
} // namespace
|
|
|
|
TEST_CASE("RequestHandler: StatusRequest forces unary even when client expects streaming",
|
|
"[daemon][ipc][status][unary]") {
|
|
#ifdef _WIN32
|
|
SKIP("Unix domain socket tests skipped on Windows");
|
|
#endif
|
|
|
|
boost::asio::io_context io;
|
|
|
|
boost::asio::local::stream_protocol::socket client_sock(io);
|
|
boost::asio::local::stream_protocol::socket server_sock(io);
|
|
boost::asio::local::connect_pair(client_sock, server_sock);
|
|
|
|
RequestHandler::Config cfg;
|
|
cfg.enable_multiplexing = false;
|
|
cfg.enable_streaming = true;
|
|
|
|
auto processor = std::make_shared<CountingProcessor>();
|
|
auto handler = std::make_shared<RequestHandler>(processor, cfg);
|
|
|
|
// Run the handler on the server socket.
|
|
yams::compat::stop_source stop_source;
|
|
auto server_sock_ptr =
|
|
std::make_shared<boost::asio::local::stream_protocol::socket>(std::move(server_sock));
|
|
boost::asio::co_spawn(
|
|
io,
|
|
[handler, server_sock_ptr,
|
|
token = stop_source.get_token()]() -> boost::asio::awaitable<void> {
|
|
co_await handler->handle_connection(server_sock_ptr, token, 1);
|
|
co_return;
|
|
},
|
|
boost::asio::detached);
|
|
|
|
// Drive the async server while the client uses sync I/O.
|
|
auto work = boost::asio::make_work_guard(io);
|
|
std::thread io_thread([&io]() { io.run(); });
|
|
|
|
// Send StatusRequest that asks for streaming.
|
|
Message msg;
|
|
msg.version = 1;
|
|
msg.requestId = 7;
|
|
msg.expectsStreamingResponse = true;
|
|
StatusRequest sreq;
|
|
sreq.detailed = true;
|
|
msg.payload = Request{std::in_place_type<StatusRequest>, sreq};
|
|
|
|
MessageFramer framer(1024 * 1024);
|
|
std::vector<uint8_t> frame;
|
|
auto fr = framer.frame_message_into(msg, frame);
|
|
REQUIRE(fr);
|
|
boost::asio::write(client_sock, boost::asio::buffer(frame));
|
|
|
|
// Read one response frame and verify it's not chunked.
|
|
std::array<uint8_t, MessageFramer::HEADER_SIZE> hdr_buf{};
|
|
boost::asio::read(client_sock, boost::asio::buffer(hdr_buf));
|
|
auto hdr_res = framer.parse_header(std::span<const uint8_t>(hdr_buf.data(), hdr_buf.size()));
|
|
REQUIRE(hdr_res);
|
|
const auto hdr = hdr_res.value();
|
|
REQUIRE(hdr.is_chunked() == false);
|
|
|
|
std::vector<uint8_t> payload(hdr.payload_size);
|
|
boost::asio::read(client_sock, boost::asio::buffer(payload));
|
|
|
|
// Confirm handler used the unary processor path.
|
|
REQUIRE(processor->unary_calls() == 1);
|
|
REQUIRE(processor->streaming_calls() == 0);
|
|
|
|
// Stop the connection loop.
|
|
stop_source.request_stop();
|
|
{
|
|
boost::system::error_code ec;
|
|
client_sock.close(ec);
|
|
}
|
|
work.reset();
|
|
io_thread.join();
|
|
}
|