137 lines
5.2 KiB
C++
137 lines
5.2 KiB
C++
#include <catch2/catch_test_macros.hpp>
|
|
|
|
#include <algorithm>
|
|
#include <filesystem>
|
|
#include <string>
|
|
#include <vector>
|
|
|
|
#include <boost/asio/co_spawn.hpp>
|
|
#include <boost/asio/io_context.hpp>
|
|
#include <boost/asio/use_future.hpp>
|
|
|
|
#include <yams/daemon/client/in_process_transport.h>
|
|
#include <yams/daemon/embedded_service_host.h>
|
|
|
|
#include "../../common/test_helpers_catch2.h"
|
|
|
|
namespace fs = std::filesystem;
|
|
|
|
namespace {
|
|
|
|
yams::Result<void> run_streaming(yams::daemon::InProcessTransport& transport,
|
|
yams::daemon::Request request,
|
|
yams::daemon::IClientTransport::HeaderCallback onHeader,
|
|
yams::daemon::IClientTransport::ChunkCallback onChunk,
|
|
yams::daemon::IClientTransport::ErrorCallback onError,
|
|
yams::daemon::IClientTransport::CompleteCallback onComplete) {
|
|
boost::asio::io_context io;
|
|
auto fut =
|
|
boost::asio::co_spawn(io,
|
|
transport.send_request_streaming(
|
|
std::move(request), std::move(onHeader), std::move(onChunk),
|
|
std::move(onError), std::move(onComplete)),
|
|
boost::asio::use_future);
|
|
io.run();
|
|
return fut.get();
|
|
}
|
|
|
|
std::shared_ptr<yams::daemon::EmbeddedServiceHost> make_host_for_test(const fs::path& dataDir) {
|
|
yams::daemon::EmbeddedServiceHost::Options options;
|
|
options.dataDir = dataDir;
|
|
options.ioThreads = 1;
|
|
options.enableAutoRepair = false;
|
|
options.autoLoadPlugins = false;
|
|
options.enableModelProvider = false;
|
|
options.initTimeoutSeconds = 30;
|
|
|
|
auto host = yams::daemon::EmbeddedServiceHost::getOrCreate(options);
|
|
REQUIRE(host.has_value());
|
|
return host.value();
|
|
}
|
|
|
|
} // namespace
|
|
|
|
TEST_CASE("InProcessTransport orders callbacks for stream-capable requests",
|
|
"[daemon][streaming][inprocess]") {
|
|
#ifdef _WIN32
|
|
SKIP("InProcessTransport embedded host tests are unstable on Windows");
|
|
#endif
|
|
const auto dataDir = yams::test::make_temp_dir("yams_in_process_order_stream_");
|
|
|
|
yams::test::ScopedEnvVar dataEnv("YAMS_DATA_DIR", dataDir.string());
|
|
yams::test::ScopedEnvVar storageEnv("YAMS_STORAGE", dataDir.string());
|
|
yams::test::ScopedEnvVar disableVectors("YAMS_DISABLE_VECTORS", std::string("1"));
|
|
yams::test::ScopedEnvVar skipModelLoading("YAMS_SKIP_MODEL_LOADING", std::string("1"));
|
|
yams::test::ScopedEnvVar disableWatcher("YAMS_DISABLE_SESSION_WATCHER", std::string("1"));
|
|
|
|
auto host = make_host_for_test(dataDir);
|
|
yams::daemon::InProcessTransport transport(host);
|
|
|
|
std::vector<std::string> events;
|
|
yams::daemon::Request req = yams::daemon::ListRequest{};
|
|
|
|
auto result = run_streaming(
|
|
transport, req, [&](const yams::daemon::Response&) { events.push_back("header"); },
|
|
[&](const yams::daemon::Response&, bool) {
|
|
events.push_back("chunk");
|
|
return true;
|
|
},
|
|
[&](const yams::Error&) { events.push_back("error"); },
|
|
[&]() { events.push_back("complete"); });
|
|
|
|
REQUIRE(result.has_value());
|
|
REQUIRE_FALSE(events.empty());
|
|
REQUIRE(events.front() == "header");
|
|
REQUIRE(events.back() == "complete");
|
|
|
|
const auto completePos = std::find(events.begin(), events.end(), "complete");
|
|
REQUIRE(completePos != events.end());
|
|
REQUIRE(std::find(events.begin(), completePos, "chunk") != completePos);
|
|
REQUIRE(std::find(events.begin(), events.end(), "error") == events.end());
|
|
|
|
auto shutdownResult = host->shutdown();
|
|
REQUIRE(shutdownResult.has_value());
|
|
|
|
std::error_code ec;
|
|
fs::remove_all(dataDir, ec);
|
|
}
|
|
|
|
TEST_CASE("InProcessTransport orders callbacks for unary requests",
|
|
"[daemon][streaming][inprocess]") {
|
|
#ifdef _WIN32
|
|
SKIP("InProcessTransport embedded host tests are unstable on Windows");
|
|
#endif
|
|
const auto dataDir = yams::test::make_temp_dir("yams_in_process_order_unary_");
|
|
|
|
yams::test::ScopedEnvVar dataEnv("YAMS_DATA_DIR", dataDir.string());
|
|
yams::test::ScopedEnvVar storageEnv("YAMS_STORAGE", dataDir.string());
|
|
yams::test::ScopedEnvVar disableVectors("YAMS_DISABLE_VECTORS", std::string("1"));
|
|
yams::test::ScopedEnvVar skipModelLoading("YAMS_SKIP_MODEL_LOADING", std::string("1"));
|
|
yams::test::ScopedEnvVar disableWatcher("YAMS_DISABLE_SESSION_WATCHER", std::string("1"));
|
|
|
|
auto host = make_host_for_test(dataDir);
|
|
yams::daemon::InProcessTransport transport(host);
|
|
|
|
std::vector<std::string> events;
|
|
yams::daemon::Request req = yams::daemon::StatusRequest{};
|
|
|
|
auto result = run_streaming(
|
|
transport, req, [&](const yams::daemon::Response&) { events.push_back("header"); },
|
|
[&](const yams::daemon::Response&, bool) {
|
|
events.push_back("chunk");
|
|
return true;
|
|
},
|
|
[&](const yams::Error&) { events.push_back("error"); },
|
|
[&]() { events.push_back("complete"); });
|
|
|
|
REQUIRE(result.has_value());
|
|
REQUIRE(events.size() == 2);
|
|
REQUIRE(events[0] == "header");
|
|
REQUIRE(events[1] == "complete");
|
|
|
|
auto shutdownResult = host->shutdown();
|
|
REQUIRE(shutdownResult.has_value());
|
|
|
|
std::error_code ec;
|
|
fs::remove_all(dataDir, ec);
|
|
}
|