diff --git a/project.cpp b/project.cpp index 69c9b21..be39bc3 100644 --- a/project.cpp +++ b/project.cpp @@ -232,6 +232,7 @@ extern "C" Configuration CrafterBuildProject(std::span a for (const std::string_view test : coreTests) cfg.AddTest(test).Dependencies({ &core }); for (const std::string_view test : libraryTests) cfg.AddTest(test).Dependencies({ &cfg }); cfg.AddTest("ShouldRunClientCommands").Dependencies({ &cfg, &cli }); + cfg.AddTest("ShouldServeOverQuic").Dependencies({ &cfg, &cli }); } return cfg; } diff --git a/tests/ShouldServeOverQuic/main.cpp b/tests/ShouldServeOverQuic/main.cpp new file mode 100644 index 0000000..e9d6096 --- /dev/null +++ b/tests/ShouldServeOverQuic/main.cpp @@ -0,0 +1,264 @@ +// SPDX-License-Identifier: LGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// The QUIC transport end to end: a Session that listens for real, reached +// over loopback by the client's own code, as crafter-debug reaches it. What +// the in-process tests can't see: +// - discovery lists the instance with its port, token, protocol, source and +// client, and drops it once the Session is gone; +// - status, call, get and set work through the client, and log --follow +// streams lines as they are logged, then ends; +// - a wrong token is refused, a request that isn't JSON is answered, and +// one over 1 MiB is answered with an error instead of being buffered; +// - requests in parallel, on streams of one connection and on separate +// connections, each get their own reply; +// - destroying the Session answers a waiting follower and closes promptly +// with a client still connected, and a client then finds the port dead +// within seconds. +#include +#include +import std; +import Crafter.Thread; +import Crafter.Network; +import Crafter.Debug; +import Crafter.Debug.Cli; +using namespace Crafter::Debug; +namespace fs = std::filesystem; +using namespace std::chrono_literals; + +namespace { + std::int32_t failures = 0; + + void Check(bool ok, std::string_view what) { + std::println("{} {}", ok ? "PASS" : "FAIL", what); + if (!ok) ++failures; + } + + bool Has(const std::string& text, std::string_view part) { + return text.find(part) != std::string::npos; + } + + struct AddArgs { + Required a; + std::int32_t b = 0; + }; + + struct Ran { + std::int32_t code = 0; + std::string out; + std::string err; + }; + + // crafter-debug, with its real transport, against the discovery + // directory the test set. + Ran Run(std::vector args) { + std::ostringstream out; + std::ostringstream err; + Cli::Environment environment; + environment.connect = Cli::ConnectQuic; + environment.out = &out; + environment.err = &err; + environment.haveLldb = [] { return false; }; + const std::int32_t code = Cli::Run(args, environment); + return { code, out.str(), err.str() }; + } + + // A request sent by hand, on its own stream: what came back, and how + // the stream ended. + struct Received { + std::mutex mutex; + std::condition_variable changed; + std::string bytes; + bool ended = false; + bool clean = false; + }; + + struct Raw { + std::shared_ptr received = std::make_shared(); + Crafter::QUICStream stream; + + bool Wait(std::chrono::milliseconds limit) { + std::unique_lock lock(received->mutex); + return received->changed.wait_for(lock, limit, [&] { return received->ended; }); + } + + std::string Bytes() { + std::lock_guard lock(received->mutex); + return received->bytes; + } + }; + + std::unique_ptr Send(Crafter::ClientQUIC& client, std::string request) { + auto raw = std::make_unique(); + raw->stream = client.OpenStream(); + const std::shared_ptr received = raw->received; + raw->stream.OnRecieve( + [received](std::span data) { + { + std::lock_guard lock(received->mutex); + received->bytes.append(data.data(), data.size()); + } + received->changed.notify_all(); + }, + [received](bool clean) { + { + std::lock_guard lock(received->mutex); + received->ended = true; + received->clean = clean; + } + received->changed.notify_all(); + }); + raw->stream.SendAsync(request.data(), static_cast(request.size()), true, {}); + return raw; + } + + std::string Request(std::string_view token, std::string_view method, std::string_view params = "{}") { + return std::format(R"({{"token":"{}","method":"{}","params":{},"timeoutMs":5000}})", token, method, params) + "\n"; + } + + Crafter::QUICClientCredentials Insecure() { + Crafter::QUICClientCredentials credentials; + credentials.insecureNoServerValidation = true; + return credentials; + } +} + +int main() { + std::thread([] { + std::this_thread::sleep_for(50s); + std::println("FAIL the test hung"); + std::fflush(stdout); + std::_Exit(1); + }).detach(); + + const fs::path dir = fs::temp_directory_path() / "crafter-debug-ShouldServeOverQuic"; + fs::remove_all(dir); + fs::create_directories(dir); + ::setenv("CRAFTER_DEBUG_DIR", dir.c_str(), 1); + + Crafter::ThreadPool::Start(4); + + Command("Math_Add", "Add two numbers.", [](const AddArgs& args) { return *args.a + args.b; }); + float speed = 1.5f; + Watch("speed", speed); + + auto session = std::make_unique(SessionOptions{ .name = "quic-test" }); + + std::atomic pumping = true; + std::thread app([&] { + std::uint32_t frame = 0; + while (pumping) { + Pump(); + if (++frame % 10 == 0) Log("tick", "tick {}", frame); + std::this_thread::sleep_for(2ms); + } + }); + + // ---- discovery ---- + const std::vector live = ListInstances(); + Check(live.size() == 1, "discovery lists the listening session"); + if (live.size() != 1) { + pumping = false; + app.join(); + std::fflush(stdout); + std::_Exit(1); + } + const InstanceRecord record = live.front().record; + Check(record.name == "quic-test" && record.port != 0 && record.token == session->Token(), "the record names it, its port and its token"); + Check(record.protocol == protocolVersion && record.sourceId == Detail::SourceId() && record.client == Detail::ClientPath(), + "the record carries the protocol, source id and client path"); + + // ---- the client, over QUIC ---- + Ran r = Run({ "status" }); + Check(r.code == 0 && Has(r.out, "quic-test frame"), std::format("status ({})", r.out + r.err)); + r = Run({ "call", "Math_Add", "a=2", "b=3" }); + Check(r.code == 0 && Has(r.out, "5"), std::format("call ({})", r.out + r.err)); + r = Run({ "call", "Math_Ad", "a=2" }); + Check(r.code == 6 && Has(r.err, "Math_Add"), std::format("an unknown command suggests the right one ({})", r.err)); + r = Run({ "get", "speed" }); + Check(r.code == 0 && Has(r.out, "1.5"), std::format("get ({})", r.out + r.err)); + r = Run({ "set", "speed", "4" }); + Check(r.code == 0, std::format("set ({})", r.out + r.err)); + r = Run({ "get", "speed" }); + Check(r.code == 0 && Has(r.out, "4"), std::format("get after set ({})", r.out + r.err)); + + const auto followStart = std::chrono::steady_clock::now(); + r = Run({ "log", "--follow", "--for", "1s", "--cat", "tick" }); + const auto followTook = std::chrono::steady_clock::now() - followStart; + Check(r.code == 0 && Has(r.out, "tick") && Has(r.err, "(follow ended)"), std::format("log --follow streams, then ends ({})", r.err)); + Check(followTook >= 900ms && followTook < 5s, std::format("the follow lasted its --for ({} ms)", std::chrono::duration_cast(followTook).count())); + + // ---- in parallel, on separate connections ---- + { + std::vector callers; + std::vector results(6); + for (std::size_t i = 0; i < results.size(); ++i) { + callers.emplace_back([&results, i] { + const std::string a = std::format("a={}", 100 * (i + 1)); + results[i] = Run({ "call", "Math_Add", a, "b=1" }); + }); + } + for (std::thread& caller : callers) caller.join(); + bool all = true; + for (std::size_t i = 0; i < results.size(); ++i) { + all = all && results[i].code == 0 && Has(results[i].out, std::format("{}", 100 * (i + 1) + 1)); + } + Check(all, "six clients at once each get their own answer"); + } + + // ---- by hand: what the client never sends ---- + auto client = std::make_unique(std::string("127.0.0.1"), record.port, std::string(alpn), Insecure()); + { + std::unique_ptr wrong = Send(*client, Request("not-the-token", "debug.status")); + Check(wrong->Wait(5s) && Has(wrong->Bytes(), "Unauthorized"), std::format("a wrong token is refused ({})", wrong->Bytes())); + + std::unique_ptr garbage = Send(*client, "this is not json\n"); + Check(garbage->Wait(5s) && Has(garbage->Bytes(), "InvalidRequest"), std::format("a request that isn't JSON is answered ({})", garbage->Bytes())); + + std::unique_ptr huge = Send(*client, std::string(std::size_t{ 2 } << 20, ' ')); + Check(huge->Wait(10s) && Has(huge->Bytes(), "InvalidRequest") && Has(huge->Bytes(), "at most"), + std::format("a request over 1 MiB is answered with an error ({})", huge->Bytes().substr(0, 200))); + + std::vector> parallel; + for (std::int32_t i = 0; i < 8; ++i) { + parallel.push_back(Send(*client, Request(record.token, "Math_Add", std::format(R"({{"a":{},"b":1000}})", i)))); + } + bool all = true; + for (std::int32_t i = 0; i < 8; ++i) { + all = all && parallel[i]->Wait(5s) && Has(parallel[i]->Bytes(), std::format(R"({{"result":{}}})", 1000 + i)); + } + Check(all, "eight streams of one connection at once each get their own answer"); + } + + // ---- the end, with a follower waiting ---- + std::unique_ptr follower = Send(*client, Request(record.token, "debug.log", R"({"follow":true,"forMs":60000,"last":0})")); + bool following = false; + for (std::int32_t attempt = 0; attempt < 100 && !following; ++attempt) { + std::unique_ptr status = Send(*client, Request(record.token, "debug.status")); + following = status->Wait(5s) && Has(status->Bytes(), R"("following":1)"); + if (!following) std::this_thread::sleep_for(20ms); + } + Check(following, "the follower is registered"); + pumping = false; + app.join(); + + const auto endStart = std::chrono::steady_clock::now(); + session.reset(); + const auto endTook = std::chrono::steady_clock::now() - endStart; + Check(endTook < 3s, std::format("the Session ends promptly with a client connected ({} ms)", std::chrono::duration_cast(endTook).count())); + Check(follower->Wait(5s) && Has(follower->Bytes(), "ShuttingDown") && !Has(follower->Bytes(), "Math_Add"), + std::format("the follower, with no backlog, is told the app is shutting down ({})", follower->Bytes().substr(0, 400))); + Check(ListInstances().empty(), "discovery forgets the session"); + follower.reset(); + client.reset(); + + const auto deadStart = std::chrono::steady_clock::now(); + const std::unique_ptr dead = Cli::ConnectQuic(record); + const auto reply = dead->Call(Value::Parse(Request(record.token, "debug.status")), 5s, {}); + const auto deadTook = std::chrono::steady_clock::now() - deadStart; + Check(!reply && reply.error().kind == Cli::CallError::Kind::Unreachable && deadTook < 5s, + std::format("a client finds the port dead in {} ms", std::chrono::duration_cast(deadTook).count())); + + std::fflush(stdout); + std::_Exit(failures == 0 ? 0 : 1); +}