如何基于Boost实现Mock AsyncReadStream以单元测试HTTP会话类？

核心思路 #

要适配Boost.Asio的异步读写接口，Fake Stream需要满足AsyncReadStream的概念要求：提供async_read_some和async_write_some异步成员函数，同时支持asio的io_context调度。

极简Fake Stream实现 #

#include <boost/asio.hpp>
#include <queue>
#include <string>
#include <functional>

class FakeAsyncStream {
public:
    // 必须的类型定义，适配asio的异步操作要求
    using executor_type = boost::asio::io_context::executor_type;
    executor_type get_executor() noexcept { return io_context_.get_executor(); }

    explicit FakeAsyncStream(boost::asio::io_context& io_context)
        : io_context_(io_context) {}

    // 模拟异步读取：从预存的输入队列中取数据
    template <typename MutableBufferSequence, typename ReadHandler>
    void async_read_some(const MutableBufferSequence& buffers, ReadHandler&& handler) {
        if (input_buffers_.empty()) {
            // 无数据时，post一个EOF错误
            io_context_.post(std::bind(std::forward<ReadHandler>(handler),
                boost::asio::error::eof, 0));
            return;
        }

        auto& data = input_buffers_.front();
        size_t bytes_transferred = boost::asio::buffer_copy(buffers, boost::asio::buffer(data));
        
        if (bytes_transferred == data.size()) {
            input_buffers_.pop();
        } else {
            data = data.substr(bytes_transferred);
        }

        // 异步触发完成回调
        io_context_.post(std::bind(std::forward<ReadHandler>(handler),
            boost::system::error_code(), bytes_transferred));
    }

    // 模拟异步写入：把数据存到输出队列，供测试断言用
    template <typename ConstBufferSequence, typename WriteHandler>
    void async_write_some(const ConstBufferSequence& buffers, WriteHandler&& handler) {
        std::string data;
        data.reserve(boost::asio::buffer_size(buffers));
        boost::asio::buffer_copy(boost::asio::buffer(data), buffers);
        output_buffers_.push(data);

        // 异步触发完成回调
        io_context_.post(std::bind(std::forward<WriteHandler>(handler),
            boost::system::error_code(), data.size()));
    }

    // 测试辅助方法：添加模拟的输入数据（供http::async_read读取）
    void add_input_data(const std::string& data) {
        input_buffers_.push(data);
    }

    // 测试辅助方法：获取已写入的数据（断言http::async_write的输出）
    std::string get_output_data() {
        if (output_buffers_.empty()) return {};
        auto data = output_buffers_.front();
        output_buffers_.pop();
        return data;
    }

    bool has_output_data() const { return !output_buffers_.empty(); }

private:
    boost::asio::io_context& io_context_;
    std::queue<std::string> input_buffers_;  // 模拟网络输入数据
    std::queue<std::string> output_buffers_; // 记录发送出去的数据
};

单元测试使用示例 #

#include <boost/test/unit_test.hpp>
#include "your_http_session.h" // 引入你的HTTP会话类

BOOST_AUTO_TEST_CASE(test_http_session_basic) {
    boost::asio::io_context io_context;
    FakeAsyncStream fake_stream(io_context);

    // 初始化HTTP会话，传入fake_stream替代真实Socket
    YourHttpSession session(std::move(fake_stream));

    // 注入模拟的HTTP请求数据
    fake_stream.add_input_data("GET /api/test HTTP/1.1\r\nHost: localhost\r\n\r\n");

    // 运行io_context触发所有异步操作
    io_context.run();

    // 断言会话处理结果
    BOOST_CHECK(fake_stream.has_output_data());
    std::string response = fake_stream.get_output_data();
    BOOST_CHECK(response.starts_with("HTTP/1.1 200 OK"));
}

关键说明 #

• 适配AsyncReadStream概念：必须提供get_executor()以及符合签名要求的async_read_some、async_write_some，这是Boost.Asio异步操作能识别的核心条件。
• 模拟数据流转：用队列存储输入/输出数据，异步操作通过io_context.post触发回调，完全模拟真实Socket的异步调度逻辑，无需真实网络端口。
• 测试扩展性：可根据需求扩展FakeStream，比如模拟网络错误（在回调中传入非空error_code）、延迟响应（通过asio::steady_timer调度回调）等场景。

内容的提问来源于stack exchange，提问作者user129393192