diff --git a/CMakeLists.txt b/CMakeLists.txt index 0fb4871c..d07fa058 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -486,7 +486,7 @@ if (BUILD_TESTING) set_tests_properties(quote_tracker_test PROPERTIES LABELS "cpu;server;api") add_executable(trace_test tests/cli/trace_test.cpp) - target_link_libraries(trace_test PRIVATE gufo_logging) + target_link_libraries(trace_test PRIVATE gufo_logging gufo_platform) add_test(NAME trace_test COMMAND trace_test) set_tests_properties(trace_test PROPERTIES LABELS "cpu;server;trace") @@ -494,6 +494,11 @@ if (BUILD_TESTING) target_link_libraries(http_server_test PRIVATE gufo_http) add_test(NAME http_server_test COMMAND http_server_test) set_tests_properties(http_server_test PROPERTIES LABELS "cpu;server;http" TIMEOUT 30) + add_executable(http_server_signals_test tests/cli/http_server_signals_test.cpp) + target_link_libraries(http_server_signals_test PRIVATE gufo_http) + add_test(NAME http_server_signals_test COMMAND http_server_signals_test) + set_tests_properties(http_server_signals_test + PROPERTIES LABELS "cpu;server;http" TIMEOUT 60) add_executable(audio_websocket_test tests/cli/audio_websocket_test.cpp) target_link_libraries(audio_websocket_test PRIVATE gufo_http) add_test(NAME audio_websocket_test COMMAND audio_websocket_test) @@ -725,7 +730,7 @@ if (BUILD_TESTING) LABELS "cpu;tools;profiling") add_executable(gpu_queues_test tests/diagnostics/gpu_queues_test.cpp) - target_link_libraries(gpu_queues_test PRIVATE gufo_diagnostics) + target_link_libraries(gpu_queues_test PRIVATE gufo_diagnostics gufo_platform) add_test(NAME gpu_queues_test COMMAND gpu_queues_test) set_tests_properties(gpu_queues_test PROPERTIES LABELS "cpu;diagnostics;serving") diff --git a/cmake/Checks.cmake b/cmake/Checks.cmake index ef248ab2..e2909c16 100644 --- a/cmake/Checks.cmake +++ b/cmake/Checks.cmake @@ -19,7 +19,8 @@ set(gufo_pr_targets platform_fs_test logit_sampler_test ggml_dequant_test quote_tracker_test - openai_chat_test http_server_test trace_test audio_websocket_test + openai_chat_test http_server_test http_server_signals_test trace_test + audio_websocket_test text_generation_scheduler_test component_text_cache_test text_model_runner_test continuation_disk_store_test bench_cli_test prompt_cli_test eval_http_client_test diff --git a/src/core/platform/CMakeLists.txt b/src/core/platform/CMakeLists.txt index 5fa9a154..9e2efc08 100644 --- a/src/core/platform/CMakeLists.txt +++ b/src/core/platform/CMakeLists.txt @@ -1,6 +1,7 @@ # The OS leaves. The Linux build and the Windows platform-only build # (GUFO_PLATFORM_ONLY) share this list. add_library(gufo_platform STATIC + env.cpp mapped_file.cpp net.cpp shutdown_signal.cpp diff --git a/src/core/platform/env.cpp b/src/core/platform/env.cpp new file mode 100644 index 00000000..27c4e415 --- /dev/null +++ b/src/core/platform/env.cpp @@ -0,0 +1,31 @@ +#include "src/core/platform/env.hpp" + +#include + +#if defined(_WIN32) +#include +#else +#include +#endif + +namespace gufo::platform { + +bool SetEnv(std::string_view name, std::string_view value) { +#if defined(_WIN32) + return ::_putenv_s(std::string(name).c_str(), std::string(value).c_str()) == + 0; +#else + return ::setenv(std::string(name).c_str(), std::string(value).c_str(), 1) == + 0; +#endif +} + +bool UnsetEnv(std::string_view name) { +#if defined(_WIN32) + return ::_putenv(std::string(name).append("=").c_str()) == 0; +#else + return ::unsetenv(std::string(name).c_str()) == 0; +#endif +} + +} // namespace gufo::platform diff --git a/src/core/platform/env.hpp b/src/core/platform/env.hpp new file mode 100644 index 00000000..7d49344b --- /dev/null +++ b/src/core/platform/env.hpp @@ -0,0 +1,20 @@ +#ifndef GUFO_CORE_PLATFORM_ENV_HPP_ +#define GUFO_CORE_PLATFORM_ENV_HPP_ + +#include + +namespace gufo::platform { + +/// Sets an environment variable in this process, overwriting any previous +/// value. Returns false when the OS rejects the name or value. An empty +/// value removes the variable on Windows (the CRT cannot store one); +/// POSIX stores it empty. +bool SetEnv(std::string_view name, std::string_view value); + +/// Removes an environment variable from this process. Removing an absent +/// variable succeeds. Returns false when the OS rejects the name. +bool UnsetEnv(std::string_view name); + +} // namespace gufo::platform + +#endif // GUFO_CORE_PLATFORM_ENV_HPP_ diff --git a/src/core/platform/fs.hpp b/src/core/platform/fs.hpp index 6f1e7968..e2f20b65 100644 --- a/src/core/platform/fs.hpp +++ b/src/core/platform/fs.hpp @@ -211,6 +211,12 @@ std::size_t ReadAt(PlatformFile file, std::uint64_t offset, void* output, /// writes). Always true on Windows until ACL checks exist. bool OwnerOnly(const FileInfo& info, std::uint32_t forbidden) noexcept; +/// Makes an existing directory owner-only: chmod(path, 0700) on POSIX, so +/// its entries in a shared temporary root cannot be listed or entered by +/// other users. A no-op that reports success on Windows, where access is +/// left to the directory's inherited ACLs. Returns false on failure. +bool RestrictToOwner(const std::filesystem::path& path) noexcept; + } // namespace gufo::platform #endif // GUFO_CORE_PLATFORM_FS_HPP_ diff --git a/src/core/platform/fs_posix.cpp b/src/core/platform/fs_posix.cpp index e0545e94..43c7eb34 100644 --- a/src/core/platform/fs_posix.cpp +++ b/src/core/platform/fs_posix.cpp @@ -379,4 +379,8 @@ bool OwnerOnly(const FileInfo& info, std::uint32_t forbidden) noexcept { return info.user_id == ::geteuid() && (info.permissions & forbidden) == 0; } +bool RestrictToOwner(const std::filesystem::path& path) noexcept { + return ::chmod(path.c_str(), 0700) == 0; +} + } // namespace gufo::platform diff --git a/src/core/platform/fs_windows.cpp b/src/core/platform/fs_windows.cpp index 9e415df3..758d8725 100644 --- a/src/core/platform/fs_windows.cpp +++ b/src/core/platform/fs_windows.cpp @@ -466,4 +466,9 @@ bool OwnerOnly(const FileInfo&, std::uint32_t) noexcept { return true; } +bool RestrictToOwner(const std::filesystem::path&) noexcept { + // No-op: access follows the directory's inherited ACLs. + return true; +} + } // namespace gufo::platform diff --git a/src/core/platform/net.cpp b/src/core/platform/net.cpp index 588a2b20..41ce03b8 100644 --- a/src/core/platform/net.cpp +++ b/src/core/platform/net.cpp @@ -35,6 +35,10 @@ void CloseSocket(Socket socket) noexcept { #endif } +Socket CreateStreamSocket(int address_family) noexcept { + return ::socket(address_family, SOCK_STREAM, 0); +} + int PollSocket(void* descriptors, std::uint32_t count, int timeout_ms) { #if defined(_WIN32) return WSAPoll(static_cast(descriptors), count, timeout_ms); @@ -71,6 +75,29 @@ bool SetNonBlocking(Socket socket) noexcept { #endif } +std::intptr_t Receive(Socket socket, void* data, std::size_t bytes, + int flags) noexcept { +#if defined(_WIN32) + // Winsock's recv takes an int length; clamp each call so oversized reads + // narrow safely. + const int chunk = static_cast(std::min( + bytes, static_cast(std::numeric_limits::max()))); + return ::recv(socket, static_cast(data), chunk, flags); +#else + return ::recv(socket, data, bytes, flags); +#endif +} + +bool ShutdownSocket(Socket socket, ShutdownHow how) noexcept { +#if defined(_WIN32) + return ::shutdown(socket, how == ShutdownHow::kWrite ? SD_SEND : SD_BOTH) == + 0; +#else + return ::shutdown(socket, how == ShutdownHow::kWrite ? SHUT_WR : SHUT_RDWR) == + 0; +#endif +} + bool SetBlocking(Socket socket) noexcept { #if defined(_WIN32) u_long mode = 0; diff --git a/src/core/platform/net.hpp b/src/core/platform/net.hpp index 81679b39..451c8121 100644 --- a/src/core/platform/net.hpp +++ b/src/core/platform/net.hpp @@ -7,7 +7,14 @@ #if defined(_WIN32) // clang-format off // winsock2.h must precede windows.h (which would pull in winsock 1), and -// iphlpapi.h needs windows.h first; keep the chain unsorted. +// iphlpapi.h needs windows.h first; keep the chain unsorted. The macro +// guards keep windows.h's min/max macros out of std:: calls downstream. +#ifndef WIN32_LEAN_AND_MEAN +#define WIN32_LEAN_AND_MEAN +#endif +#ifndef NOMINMAX +#define NOMINMAX +#endif #include #include #include @@ -29,8 +36,21 @@ namespace gufo::net { #if defined(_WIN32) using Socket = SOCKET; + +/// Length type for socket address calls: int on Windows. +using SocklenT = int; #else using Socket = int; + +/// Length type for socket address calls: socklen_t on POSIX. +using SocklenT = socklen_t; +#endif + +/// Returned by CreateStreamSocket on failure. +#if defined(_WIN32) +inline constexpr Socket kInvalidSocket = INVALID_SOCKET; +#else +inline constexpr Socket kInvalidSocket = -1; #endif /// The one-time WinSock startup the HTTP server performs before opening its @@ -40,12 +60,28 @@ void EnsureSocketRuntimeStarted(); /// Close a socket: closesocket on Windows, close on POSIX. void CloseSocket(Socket socket) noexcept; +/// Create a blocking TCP stream socket for the address family (AF_INET or +/// AF_INET6). Returns kInvalidSocket on failure. +Socket CreateStreamSocket(int address_family) noexcept; + /// poll()/WSAPoll() with the platform's native timeout unit. int PollSocket(void* descriptors, std::uint32_t count, int timeout_ms); /// Send that never raises SIGPIPE (MSG_NOSIGNAL on POSIX; default on Winsock). std::intptr_t SendNoSignal(Socket socket, const void* data, std::size_t bytes); +/// Receive up to `bytes` from the socket into `data`, with optional message +/// flags. Returns the byte count, 0 when the peer closed, -1 on error. +std::intptr_t Receive(Socket socket, void* data, std::size_t bytes, + int flags = 0) noexcept; + +/// How ShutdownSocket half-closes a connected socket: stop sending, or stop +/// both directions. +enum class ShutdownHow { kWrite, kReadWrite }; + +/// Half-close a connected socket; returns false on failure. +bool ShutdownSocket(Socket socket, ShutdownHow how) noexcept; + /// Mark a socket non-blocking; returns false on failure. bool SetNonBlocking(Socket socket) noexcept; diff --git a/tests/cache/idle_spill_test.cpp b/tests/cache/idle_spill_test.cpp index 488e806c..6a28e8ac 100644 --- a/tests/cache/idle_spill_test.cpp +++ b/tests/cache/idle_spill_test.cpp @@ -172,7 +172,9 @@ void IdleAndZeroIdle() { f.Verify(); const auto metrics = f.slot.Metrics(); assert(metrics.reassignments == before.reassignments + 1); - assert(metrics.residual_wait_ns > before.residual_wait_ns); + // A coarse steady_clock (QueryPerformanceCounter on Windows, typically + // 100 ns ticks) can measure a fast reassignment as 0 ns. + assert(metrics.residual_wait_ns >= before.residual_wait_ns); assert(metrics.idle_bytes == (idle ? sizeof(f.tokens) : 0)); assert(metrics.foreground_bytes == (idle ? 0 : sizeof(f.tokens))); assert(f.acquired == 3); // Fresh stream lease for every bounded piece. diff --git a/tests/cli/component_text_cache_test.cpp b/tests/cli/component_text_cache_test.cpp index 3074843b..b4a5e521 100644 --- a/tests/cli/component_text_cache_test.cpp +++ b/tests/cli/component_text_cache_test.cpp @@ -12,17 +12,20 @@ #include #include #include +#include #include #include "src/cache/retention.hpp" #include "src/cli/serve/logging.hpp" #include "src/cli/serve/text_model_runner.hpp" #include "tests/cache/fake_adapter.hpp" +#include "tests/support/temporary_directory.hpp" using namespace gufo; using namespace gufo::server; using namespace gufo::cache::testing; namespace { + struct State final : TextRunnerState { FakeAdapter& adapter; cache::Slot& slot; @@ -985,8 +988,8 @@ void ReusedConversationOutlivesOneUseRequests() { assert(turn(conversation) == 16); } void RestartAndCorruption() { - char pattern[] = "/tmp/gufo-component-serving-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-serving-test"); + const std::filesystem::path directory = temp.path(); const TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1 << 20, .staging_capacity_bytes = 7}; @@ -1213,8 +1216,8 @@ void FragmentedBackingKeepsCompletePrompt() { runner->adapter->Append(restored_state.slot, {&next, 1}, {&next, 1}); } void DiskBudgetEvictsAnUnpinnedConversation() { - char pattern[] = "/tmp/gufo-component-budget-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-budget-test"); + const std::filesystem::path directory = temp.path(); const TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1024, .staging_capacity_bytes = 7}; @@ -1250,8 +1253,8 @@ std::uintmax_t ManagedBytes(const std::filesystem::path& directory) { return bytes; } void LoweredDiskBudgetEvictsOldestAtStartup() { - char pattern[] = "/tmp/gufo-component-lowered-budget-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-lowered-budget-test"); + const std::filesystem::path directory = temp.path(); TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1 << 20, .staging_capacity_bytes = 7}; @@ -1283,8 +1286,8 @@ void LoweredDiskBudgetEvictsOldestAtStartup() { std::filesystem::remove_all(directory); } void OrphansLeaveBeforeCheckpoints() { - char pattern[] = "/tmp/gufo-component-orphan-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-orphan-test"); + const std::filesystem::path directory = temp.path(); TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1 << 20, .staging_capacity_bytes = 7}; @@ -1316,8 +1319,8 @@ void OrphansLeaveBeforeCheckpoints() { std::filesystem::remove_all(directory); } void StagingIsInternal() { - char pattern[] = "/tmp/gufo-component-staging-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-staging-test"); + const std::filesystem::path directory = temp.path(); // The legacy option is ignored: even a value below one four-byte row // still writes and restores through the internal staging buffer. TextRunnerDiskCacheOptions disk{.directory = directory.string(), @@ -1358,8 +1361,8 @@ void StagingCountsAgainstRamCeiling() { return resources; } }; - char pattern[] = "/tmp/gufo-component-ceiling-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-ceiling-test"); + const std::filesystem::path directory = temp.path(); // Pinned staging and RAM share the ceiling: an explicit 4 MiB beside the // internal 1 MiB of staging commits 3 MiB, whatever the ignored option says. for (const std::size_t option : {std::size_t{0}, std::size_t{4} << 20}) { @@ -1431,8 +1434,8 @@ void PinnedSlotBytesCountOnce() { } assert(refused); // Staging shares the ceiling: the limit is 4 MiB less 1 MiB of staging. - char pattern[] = "/tmp/gufo-component-pinned-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-pinned-test"); + const std::filesystem::path directory = temp.path(); { const TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1 << 20}; @@ -1494,8 +1497,8 @@ void DiskDirectoryHasOneOwner() { return resources; } }; - char pattern[] = "/tmp/gufo-component-owner-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-owner-test"); + const std::filesystem::path directory = temp.path(); const TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1 << 20, .staging_capacity_bytes = 7}; @@ -1522,8 +1525,8 @@ void DiskDirectoryHasOneOwner() { std::filesystem::remove_all(directory); } void GridDoesNotSuppressFirstDiskPrompt() { - char pattern[] = "/tmp/gufo-component-spacing-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-spacing-test"); + const std::filesystem::path directory = temp.path(); const TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1 << 20, .staging_capacity_bytes = 7}; @@ -1566,8 +1569,8 @@ void PublishedResidentCheckpointKeepsDiskStep() { // A published checkpoint that is still resident restores from RAM, so its // lookup candidate carries no disk description. The next capture within // the step must still be refused. - char pattern[] = "/tmp/gufo-component-step-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-step-test"); + const std::filesystem::path directory = temp.path(); const TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1 << 20, .staging_capacity_bytes = 7, @@ -1607,8 +1610,8 @@ void PublishedResidentCheckpointKeepsDiskStep() { std::filesystem::remove_all(directory); } void NearTailDiskBranchSurvivesRestart() { - char pattern[] = "/tmp/gufo-component-branch-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-branch-test"); + const std::filesystem::path directory = temp.path(); const TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1 << 20, .staging_capacity_bytes = 7}; @@ -1667,8 +1670,8 @@ void UnbackedDeepRowsSurviveDiskRestart() { return resources; } }; - char pattern[] = "/tmp/gufo-component-deep-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-deep-test"); + const std::filesystem::path directory = temp.path(); const TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1 << 20, .staging_capacity_bytes = 8, @@ -1784,8 +1787,8 @@ void OverRamCaptureStagesPrivateStateUnderPressure(bool protect) { // from the live slot, but its private state is staged in RAM backing, and // two retained checkpoints fill that backing. Wide chunks keep the capture // metadata within RAM. - char pattern[] = "/tmp/gufo-component-over-ram-test-XXXXXX"; - const std::filesystem::path directory = mkdtemp(pattern); + const TemporaryDirectory temp("gufo-component-over-ram-test"); + const std::filesystem::path directory = temp.path(); const TextRunnerDiskCacheOptions disk{.directory = directory.string(), .capacity_bytes = 1 << 20, .min_checkpoint_step_tokens = 1}; diff --git a/tests/cli/continuation_disk_store_test.cpp b/tests/cli/continuation_disk_store_test.cpp index 1b20000b..a56ed739 100644 --- a/tests/cli/continuation_disk_store_test.cpp +++ b/tests/cli/continuation_disk_store_test.cpp @@ -25,6 +25,7 @@ #include #include "src/core/crypto/sha256.hpp" +#include "tests/support/temporary_directory.hpp" namespace { @@ -57,35 +58,6 @@ void Expect(bool condition, std::string_view message) { } } -class TemporaryDirectory { -public: - TemporaryDirectory() { - std::string pattern = (std::filesystem::temp_directory_path() / - "gufo-continuation-disk-XXXXXX") - .string(); - char* created = ::mkdtemp(pattern.data()); - if (created == nullptr) { - throw std::runtime_error("failed to create temporary directory"); - } - path_ = created; - } - - ~TemporaryDirectory() { - std::error_code error; - std::filesystem::remove_all(path_, error); - } - - TemporaryDirectory(const TemporaryDirectory&) = delete; - TemporaryDirectory& operator=(const TemporaryDirectory&) = delete; - - [[nodiscard]] const std::filesystem::path& path() const noexcept { - return path_; - } - -private: - std::filesystem::path path_; -}; - template void PutLittleEndian(std::span destination, std::size_t offset, Integer value) { @@ -367,7 +339,7 @@ void TestSha256KnownVector() { Expect(complete == incremental.Finish(), "reading a prefix digest does not consume incremental state"); - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const auto path = directory.path() / "artifact.gguf"; { std::ofstream output(path, std::ios::binary); @@ -379,7 +351,7 @@ void TestSha256KnownVector() { } void TestRestartRestoreAndCompatibilityIdentity() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const FakeRunner writer("artifact-A", "fake-old-executable"); auto snapshot = MakeSnapshot(writer, 0x123456789ABCDEF0ULL, 3); { @@ -412,7 +384,7 @@ void TestRestartRestoreAndCompatibilityIdentity() { } void TestLongestPrefixAndForcedHashCollision() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const FakeRunner runner("collision-model"); const auto forced_collision = [](std::span) { return std::string(64, '0'); @@ -448,7 +420,7 @@ void TestLongestPrefixAndForcedHashCollision() { } void TestCorruptionBecomesDeterministicMissAndRemoval() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const FakeRunner runner("corruption-model"); std::vector events; ContinuationDiskStore store( @@ -501,7 +473,7 @@ void TestCorruptionBecomesDeterministicMissAndRemoval() { } void TestByteAndStagingLimits() { - TemporaryDirectory staging_directory; + TemporaryDirectory staging_directory("gufo-continuation-disk"); const FakeRunner runner("limits"); const std::size_t file_bytes = ExpectedFileBytes(6, 2); auto snapshot = MakeSnapshot(runner, 1, 2); @@ -512,7 +484,7 @@ void TestByteAndStagingLimits() { staging_limited.entry_count() == 0, "full file image must fit the bounded staging allocation"); - TemporaryDirectory byte_directory; + TemporaryDirectory byte_directory("gufo-continuation-disk"); ContinuationDiskStore byte_limited( StoreOptions(byte_directory.path(), file_bytes - 1, 4096)); Expect(!SaveTokens(byte_limited, runner, {1, 2}, *snapshot).stored && @@ -521,14 +493,14 @@ void TestByteAndStagingLimits() { } void TestAutomaticStagingAndAdmissionDiagnostics() { - TemporaryDirectory default_directory; + TemporaryDirectory default_directory("gufo-continuation-disk"); ContinuationDiskStore defaults({.directory = default_directory.path()}); Expect(defaults.capacity_bytes() == std::size_t{8} * 1024U * 1024U * 1024U && defaults.staging_capacity_bytes() > kDiskHeaderBytes && defaults.staging_capacity_bytes() <= defaults.capacity_bytes(), "defaults bound RAM staging by available RAM and disk retention"); - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); std::vector events; const FakeRunner runner("automatic"); ContinuationDiskStore store( @@ -558,7 +530,7 @@ void TestAutomaticStagingAndAdmissionDiagnostics() { events.back().reason == ContinuationDiskEventReason::kByteCapacity, "disk retention rejection is distinguished from staging pressure"); - TemporaryDirectory explicit_directory; + TemporaryDirectory explicit_directory("gufo-continuation-disk"); ContinuationDiskStore explicit_limit( StoreOptions(explicit_directory.path(), 4096, 256), [&](const auto& event) { events.push_back(event); }); @@ -571,7 +543,7 @@ void TestAutomaticStagingAndAdmissionDiagnostics() { } void TestSmallerStagingPreservesExistingFiles() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const FakeRunner runner("staging-restart"); const auto file_bytes = ExpectedFileBytes(15, 2); { @@ -618,7 +590,7 @@ void TestSmallerStagingPreservesExistingFiles() { } void TestLruEvictionUsesActualFileBytes() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const FakeRunner runner("12345678"); const std::size_t entry_bytes = ExpectedFileBytes(8, 2); ContinuationDiskStore store( @@ -652,7 +624,7 @@ void TestLruEvictionUsesActualFileBytes() { } void TestAtomicPublicationAndPrivatePermissions() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const auto orphan = directory.path() / ".tmp-orphan"; { std::ofstream output(orphan, std::ios::binary); @@ -683,7 +655,7 @@ void TestAtomicPublicationAndPrivatePermissions() { } void TestStartupRejectsUnsafeAndInvalidFiles() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const FakeRunner runner("startup-valid"); const FakeRunner incompatible("startup-incompatible"); std::filesystem::path truncated_file; @@ -786,7 +758,7 @@ void TestStartupRejectsUnsafeAndInvalidFiles() { } void TestRootSymlinkIsRejected() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const auto target = directory.path() / "target"; const auto link = directory.path() / "cache-link"; std::filesystem::create_directory(target); @@ -801,7 +773,7 @@ void TestRootSymlinkIsRejected() { } void TestConcurrentCallersRemainBoundedAndExact() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const FakeRunner runner("concurrent"); constexpr std::size_t kThreadCount = 8; ContinuationDiskStore store( @@ -835,7 +807,7 @@ void TestConcurrentCallersRemainBoundedAndExact() { } void TestConcurrentStoreInstancesPublishSafely() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const FakeRunner runner("multi-process"); { ContinuationDiskStore first(StoreOptions(directory.path())); @@ -868,7 +840,7 @@ void TestConcurrentStoreInstancesPublishSafely() { } void TestStartupPreservesActivePublisher() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); auto runner = std::make_shared("startup-publisher"); std::binary_semaphore entered(0), release(0); const auto orphan = directory.path() / ".tmp-orphan"; @@ -913,7 +885,7 @@ void TestStartupPreservesActivePublisher() { } void TestSharedPrefixBoundariesAndExactDedup() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); const FakeRunner runner("shared-prefix-model"); ContinuationDiskStore store(StoreOptions(directory.path()), {}); @@ -973,7 +945,7 @@ void TestSharedPrefixBoundariesAndExactDedup() { } void TestBoundedAsyncPersistenceDoesNotBlockLookup() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); auto runner = std::make_shared("async-store"); std::binary_semaphore entered(0), release(0); const auto bytes = ExpectedFileBytes( @@ -1043,7 +1015,7 @@ void TestBoundedAsyncPersistenceDoesNotBlockLookup() { } void TestIndexedPrefixLookup() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); FakeRunner runner("prefix-index"); ContinuationDiskStore store(StoreOptions(directory.path())); // Many independent branches, with terminal prefixes inside compressed edges. @@ -1073,7 +1045,7 @@ void TestIndexedPrefixLookup() { } // namespace void TestImageIdentitySurvivesRestart() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); FakeRunner runner("image-cache-test"); const std::vector prompt{1, 248056, 3}; const std::vector image_a{1, 2}, image_b{1, 3}; @@ -1105,7 +1077,7 @@ void TestImageIdentitySurvivesRestart() { void TestAppendedImagePrefixesSurviveRestart() { using gufo::server::ContinuationInputPrefix; - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); FakeRunner runner("image-prefix-cache"); const std::vector prompt{1, 2, 3, 248056, 4, 5, 248056, 6}; const auto tokens = std::span(prompt); @@ -1149,7 +1121,7 @@ void TestAppendedImagePrefixesSurviveRestart() { } void TestMinimumCheckpointStep() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-continuation-disk"); auto runner = std::make_shared("min-step-model"); std::vector events; auto options = StoreOptions(directory.path()); @@ -1191,7 +1163,7 @@ void TestMinimumCheckpointStep() { "shared prefixes are exempt from the step"); options.min_checkpoint_step_tokens = 0; - TemporaryDirectory ungated_directory; + TemporaryDirectory ungated_directory("gufo-continuation-disk"); options.directory = ungated_directory.path(); ContinuationDiskStore ungated(options); Expect( diff --git a/tests/cli/http_server_signals_test.cpp b/tests/cli/http_server_signals_test.cpp new file mode 100644 index 00000000..53b1d680 --- /dev/null +++ b/tests/cli/http_server_signals_test.cpp @@ -0,0 +1,61 @@ +// Process signals must never terminate the test runner itself; this suite +// forks itself to prove graceful shutdown, so it stays POSIX-only. +#include +#include + +#include +#include +#include + +#include "tests/cli/http_server_test_support.hpp" + +namespace { + +void TestSignalShutdown() { + // Process signals must never terminate the test runner itself. Prove that + // both idle listeners and active generation return through normal cleanup. + for (const int signal : {SIGINT, SIGTERM}) { + for (const bool active : {false, true}) { + const pid_t child = ::fork(); + assert(child >= 0); + if (child == 0) { + ::alarm(5); + { + RunningServer server({}, true); + // Accepting a request proves run() installed its handlers. + ExpectStatus(server.Post("/echo", "ready"), 200); + int fd = -1; + if (active) { + server.backend->wait_for_disconnect = true; + fd = server.Connect(); + const std::string body = R"({"prompt":"hello"})"; + const std::string request = + "POST /v1/completions HTTP/1.1\r\nContent-Length: " + + std::to_string(body.size()) + "\r\n\r\n" + body; + assert(net::SendNoSignal(fd, request.data(), request.size()) == + static_cast(request.size())); + assert(server.backend->entered.try_acquire_for( + std::chrono::seconds(2))); + } + assert(::kill(::getpid(), signal) == 0); + assert(server.run_finished.try_acquire_for(std::chrono::seconds(2))); + if (active) { + assert(server.backend->disconnected); + net::CloseSocket(fd); + } + } + ::_exit(0); + } + int status = 0; + assert(::waitpid(child, &status, 0) == child); + assert(WIFEXITED(status) && WEXITSTATUS(status) == 0); + } + } +} +} // namespace + +int main() { + TestSignalShutdown(); + std::cout << "HTTP signal shutdown checks passed.\n"; + return 0; +} diff --git a/tests/cli/http_server_test.cpp b/tests/cli/http_server_test.cpp index 1fb80b02..267fcd60 100644 --- a/tests/cli/http_server_test.cpp +++ b/tests/cli/http_server_test.cpp @@ -1,15 +1,8 @@ #include "src/cli/serve/http_server.hpp" -#include -#include -#include -#include -#include - #include #include #include -#include #include #include #include @@ -29,376 +22,16 @@ #include "src/cli/serve/embedding_service.hpp" #include "src/cli/serve/logging.hpp" #include "src/cli/serve/trace.hpp" +#include "src/core/platform/env.hpp" #include "src/core/platform/net.hpp" +#include "src/core/platform/os_info.hpp" +#include "tests/cli/http_server_test_support.hpp" namespace { -namespace net = gufo::net; - -using gufo::server::HttpServer; using gufo::server::Logger; using gufo::server::LogLevel; using gufo::server::LogLevelFromName; using gufo::server::LogLevelName; -using gufo::server::TextGenerationBackend; - -/// Reports an optional admission and fixed prompt progress before delegating -/// token generation. -class ProgressRequest final : public TextGenerationBackend::GenerationRequest { -public: - ProgressRequest(std::shared_ptr inner, - std::vector progress, - bool admit = false) - : inner_(std::move(inner)), - progress_(std::move(progress)), - admit_(admit) {} - - TextGenerationBackend::Result Wait( - const TextGenerationBackend::TokenCallback& on_token, - const TextGenerationBackend::ProgressCallback& on_progress, - const TextGenerationBackend::StartCallback& on_start) override { - if (admit_ && on_start && !on_start()) - inner_->Cancel(); - for (const auto& value : progress_) { - if (on_progress && !on_progress(value)) { - inner_->Cancel(); - break; - } - } - return inner_->Wait(on_token, on_progress, on_start); - } - void Cancel() noexcept override { inner_->Cancel(); } - -private: - std::shared_ptr inner_; - std::vector progress_; - bool admit_; -}; - -class FakeBackend final : public TextGenerationBackend { -public: - struct Call { - std::string prompt; - gufo::server::ChatRequest chat; - std::size_t max_tokens = 0; - gufo::sampling::SamplingConfig sampling; - std::string client_id; - std::vector stop_sequences; - std::string trace_request; - }; - Call LastCall() { - const std::lock_guard lock(mutex_); - return last_; - } - void SetOutput(std::string text) { - const std::lock_guard lock(mutex_); - output_ = std::move(text); - } - std::string model_id() const override { return "test"; } - std::vector model_ids() const override { return names; } - std::vector names{"test"}; - bool ready() const override { return true; } - bool device_lost() const override { return lost.load(); } - bool supports_images() const override { return image_support.load(); } - std::atomic image_support{false}; - std::uint32_t max_context() const override { return 65536; } - std::vector session_states() const override { return sessions; } - std::vector sessions; - std::string model_path() const override { return served_path; } - std::string build_info() const override { return served_build; } - std::string served_path; - std::string served_build; - std::shared_ptr start_complete( - std::string_view prompt, std::size_t max_tokens, - const gufo::sampling::SamplingConfig& sampling, - const CancellationCheck& cancellation, bool stream, bool ignore_eos, - std::string_view client_id, - const std::vector& stop_sequences, bool return_progress, - bool cache_prompt) override { - last_ignore_eos = ignore_eos; - return std::make_shared( - TextGenerationBackend::start_complete( - prompt, max_tokens, sampling, cancellation, stream, false, - client_id, stop_sequences, return_progress, cache_prompt), - progress, admit.load()); - } - std::shared_ptr start_chat( - const gufo::server::ChatRequest& request, std::size_t max_tokens, - const gufo::sampling::SamplingConfig& sampling, - const CancellationCheck& cancellation, bool stream) override { - return std::make_shared( - TextGenerationBackend::start_chat(request, max_tokens, sampling, - cancellation, stream), - std::vector{}, admit.load()); - } - SamplingDefaults sampling_defaults() const override { return defaults; } - SamplingDefaults defaults; - gufo::ReasoningOptions reasoning_defaults() const override { - return reasoning; - } - InitialOutputState initial_output_state( - const gufo::server::ChatRequest& request) const override { - return initial_output_state_override.value_or( - TextGenerationBackend::initial_output_state(request)); - } - std::size_t count_tokens(std::string_view text) const override { - return text.size(); - } - std::optional count_chat_tokens( - const gufo::server::ChatRequest& request) const override { - counted_messages = request.messages.size(); - counted_tools = request.tools.size(); - return chat_token_count; - } - std::optional chat_token_count; - mutable std::size_t counted_messages{0}; - mutable std::size_t counted_tools{0}; - Result complete(std::string_view prompt, std::size_t limit, - const gufo::sampling::SamplingConfig& sampling, - const CancellationCheck& cancel, const TokenCallback& token, - std::string_view client_id = "anonymous", - const std::vector& stop_sequences = {}, - bool cache_prompt = true) override { - ++calls; - last_cache_prompt = cache_prompt; - if (failure == 1) - throw std::length_error("context exceeded"); - if (failure == 2) - throw std::invalid_argument("invalid prompt"); - // A failed probe: the scheduler reports device loss from then on. - if (failure == 3) { - lost = true; - throw gufo::server::TextGenerationError( - gufo::server::TextGenerationErrorCode::kDeviceLost, - gufo::server::kDeviceLostMessage); - } - // A failure on a device that still accepts work. - if (failure == 4) - throw std::runtime_error("unspecified launch failure"); - if (failure == 5) - throw std::runtime_error(""); - Result result; - if (wait_for_disconnect) { - entered.release(); - const auto deadline = - std::chrono::steady_clock::now() + std::chrono::seconds(2); - while (!(disconnected = cancel && cancel()) && - std::chrono::steady_clock::now() < deadline) - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - finished.release(); - result.cancelled = disconnected; - return result; - } - { - const std::lock_guard lock(mutex_); - last_ = {.prompt = std::string(prompt), - .max_tokens = limit, - .sampling = sampling, - .client_id = std::string(client_id), - .stop_sequences = stop_sequences, - .trace_request = gufo::server::Trace::CurrentRequest()}; - result.text = output_; - } - result.prompt_tokens = 10; - result.cached_prompt_tokens = 8; - result.cache_hit = true; - result.draft_accepted_tokens = 4; - result.draft_rounds = 2; - result.draft_tokens = 8; - result.prefill_tokens = 2; - result.prefill_ms = 4; - result.completion_tokens = 1; - result.decode_ms = 2; - result.finish_reason = - limit == 1 ? FinishReason::kLength : FinishReason::kStop; - if (!forced_stop_sequence.empty()) { - result.finish_reason = FinishReason::kStopSequence; - result.stop_sequence = forced_stop_sequence; - } - if (token && failure != 8) { - std::this_thread::sleep_for(token_delay); - (void)token(result.text); - } - if (failure == 6) - throw std::runtime_error("context exceeded"); - if (failure == 7) { - lost = true; - throw gufo::server::TextGenerationError( - gufo::server::TextGenerationErrorCode::kDeviceLost, - gufo::server::kDeviceLostMessage); - } - return result; - } - Result chat(const gufo::server::ChatRequest& request, std::size_t limit, - const gufo::sampling::SamplingConfig& sampling, - const CancellationCheck& cancel, - const TokenCallback& token) override { - auto result = complete("", limit, sampling, cancel, token, - request.client_id, request.stop_sequences); - { - const std::lock_guard lock(mutex_); - last_.chat = request; - } - return result; - } - std::atomic calls{0}; - std::atomic last_ignore_eos{false}; - std::atomic last_cache_prompt{true}; - std::vector progress; - // Report a scheduler admission before generation, as the HIP backend does. - std::atomic admit{false}; - std::chrono::milliseconds token_delay{0}; - std::atomic failure{0}; - std::atomic lost{false}; - std::string forced_stop_sequence; - gufo::ReasoningOptions reasoning; - std::optional initial_output_state_override; - bool wait_for_disconnect{false}; - std::atomic disconnected{false}; - std::binary_semaphore entered{0}; - std::binary_semaphore finished{0}; - -private: - std::mutex mutex_; - Call last_; - std::string output_{"ok"}; -}; - -class RunningServer { -public: - explicit RunningServer(gufo::server::HttpServerOptions options = {}, - bool handle_signals = false, - std::string host = "127.0.0.1") - : backend(std::make_shared()), - server(std::move(host), 0, backend, nullptr, nullptr, nullptr, - std::move(options)) { - server.add("GET", "/peer", [](const auto& request, auto&) { - return gufo::server::HttpResponse{.body = request.client_id}; - }); - server.add("POST", "/echo", [](const auto& request, auto&) { - return gufo::server::HttpResponse{.body = request.body}; - }); - server.add("POST", "/stream-error", [](const auto&, auto&) { - return gufo::server::HttpResponse{ - .streaming_body = [](const auto& write) { - (void)write("first chunk"); - throw std::runtime_error("injected stream failure"); - }}; - }); - server.add("POST", "/stream", [](const auto& request, auto&) { - auto log = std::make_shared(); - return gufo::server::HttpResponse{ - .streaming_body = - [log, fail = !request.body.empty(), - reported = request.body == "reported"](const auto& write) { - (void)write(std::string_view("a\0b", 3)); - (void)write(""); - (void)write("end"); - if (fail) - log->error_code = "injected"; - log->error_event_sent = reported; - }, - .stream_log = log, - }; - }); - server.add("POST", "/sse-idle", [](const auto&, auto&) { - return gufo::server::HttpResponse{ - .headers = {{"Content-Type", "text/event-stream"}}, - .streaming_body = - [](const auto& write) { - (void)write("data: first\n\n"); - std::this_thread::sleep_for(std::chrono::milliseconds(90)); - (void)write("data: last\n\n"); - (void)write("data: [DONE]\n\n"); - }, - }; - }); - server.add("POST", "/sse-finish", [](const auto& request, auto&) { - return gufo::server::HttpResponse{ - .headers = {{"Content-Type", "text/event-stream"}}, - .streaming_body = [fail = !request.body.empty()](const auto& write) { - (void)write("data: first\n\n"); - if (fail) - throw std::runtime_error("injected SSE failure"); - (void)write("data: [DONE]\n\n"); - }}; - }); - std::string error; - assert(server.start(&error)); - worker = std::jthread([this, handle_signals] { - server.run(handle_signals); - run_finished.release(); - }); - } - ~RunningServer() { - server.stop(); - worker.join(); - } - - int Connect(const std::string& peer = "127.0.0.1") { - sockaddr_storage address{}; - socklen_t length = 0; - auto* ipv4 = reinterpret_cast(&address); - auto* ipv6 = reinterpret_cast(&address); - if (::inet_pton(AF_INET, peer.c_str(), &ipv4->sin_addr) == 1) { - ipv4->sin_family = AF_INET; - ipv4->sin_port = htons(server.port()); - length = sizeof(*ipv4); - } else { - assert(::inet_pton(AF_INET6, peer.c_str(), &ipv6->sin6_addr) == 1); - ipv6->sin6_family = AF_INET6; - ipv6->sin6_port = htons(server.port()); - length = sizeof(*ipv6); - } - const int fd = ::socket(address.ss_family, SOCK_STREAM, 0); - assert(fd >= 0); - assert(net::SetSocketTimeouts(fd, 3'000)); - assert(::connect(fd, reinterpret_cast(&address), length) == - 0); - return fd; - } - std::string Send(std::string_view request, bool half_close = false, - const std::string& peer = "127.0.0.1") { - const int fd = Connect(peer); - while (!request.empty()) { - const auto count = net::SendNoSignal(fd, request.data(), request.size()); - assert(count > 0); - request.remove_prefix(static_cast(count)); - } - // A half-close allows malformed/truncated-body tests to complete without - // timing-dependent sleeps. - if (half_close) - ::shutdown(fd, SHUT_WR); - std::string response; - char buffer[4096]; - for (;;) { - const auto count = ::read(fd, buffer, sizeof(buffer)); - assert(count >= 0); - if (count == 0) { - break; - } - response.append(buffer, static_cast(count)); - } - net::CloseSocket(fd); - return response; - } - - std::string Post(std::string_view path, std::string_view body) { - return Send("POST " + std::string(path) + " HTTP/1.1\r\nContent-Length: " + - std::to_string(body.size()) + "\r\n\r\n" + std::string(body)); - } - - std::shared_ptr backend; - HttpServer server; - std::binary_semaphore run_finished{0}; - std::jthread worker; -}; - -void ExpectStatus(const std::string& response, int status) { - if (!response.starts_with("HTTP/1.1 " + std::to_string(status) + " ")) { - std::cerr << response << '\n'; - std::abort(); - } -} void TestAuthorization() { RunningServer secured({.api_key = "test-secret"}); @@ -546,9 +179,9 @@ std::string ChunkedBody(const std::string& response) { void TestContentTrace() { namespace fs = std::filesystem; using gufo::server::Trace; - const auto path = - fs::temp_directory_path() / - ("gufo-trace-test-" + std::to_string(::getpid()) + ".jsonl"); + const auto path = fs::temp_directory_path() / + ("gufo-trace-test-" + + std::to_string(gufo::platform::ProcessId()) + ".jsonl"); fs::remove(path); assert(!Trace::Open(path.string()).has_value()); std::string plain_id; @@ -775,17 +408,18 @@ void TestExpectContinue() { RunningServer server({.max_request_body_bytes = 8192}); for (const std::string expect : {"Expect: 100-continue", "expect: 100-Continue"}) { - const int fd = server.Connect(); + const net::Socket fd = server.Connect(); const std::string head = "POST /echo HTTP/1.1\r\n" + expect + "\r\nContent-Length: 5\r\n\r\n"; assert(net::SendNoSignal(fd, head.data(), head.size()) == - static_cast(head.size())); + static_cast(head.size())); // The interim response arrives before any body byte is sent. const std::string_view interim = "HTTP/1.1 100 Continue\r\n\r\n"; std::string response; char buffer[4096]; while (response.size() < interim.size()) { - const auto count = ::read(fd, buffer, interim.size() - response.size()); + const auto count = + net::Receive(fd, buffer, interim.size() - response.size()); assert(count > 0); response.append(buffer, static_cast(count)); } @@ -793,7 +427,7 @@ void TestExpectContinue() { assert(net::SendNoSignal(fd, "hello", 5) == 5); response.clear(); for (;;) { - const auto count = ::read(fd, buffer, sizeof(buffer)); + const auto count = net::Receive(fd, buffer, sizeof(buffer)); assert(count >= 0); if (count == 0) break; @@ -2444,18 +2078,19 @@ void TestDeviceLossWhileWriterBlocked() { }); const int fd = server.Connect(); const int receive_buffer = 1024; - assert(::setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &receive_buffer, + assert(::setsockopt(fd, SOL_SOCKET, SO_RCVBUF, + reinterpret_cast(&receive_buffer), sizeof(receive_buffer)) == 0); const std::string request = "POST /blocked HTTP/1.1\r\n\r\n"; assert(net::SendNoSignal(fd, request.data(), request.size()) == - static_cast(request.size())); + static_cast(request.size())); assert(writing.try_acquire_for(std::chrono::seconds(2))); // The peer never drains the response. No health request or completed write // is available to trigger shutdown; the listener must observe the loss. server.backend->lost = true; assert(hook_called.try_acquire_for(std::chrono::seconds(2))); assert(hook_calls == 1 && !write_finished); - ::shutdown(fd, SHUT_RDWR); + net::ShutdownSocket(fd, net::ShutdownHow::kReadWrite); net::CloseSocket(fd); } @@ -2474,8 +2109,8 @@ void TestInvalidBindSettings() { // Some CI containers have no IPv6 loopback; listener checks skip there. bool HasIpv6Loopback() { - const int fd = ::socket(AF_INET6, SOCK_STREAM, 0); - if (fd < 0) + const net::Socket fd = net::CreateStreamSocket(AF_INET6); + if (fd == net::kInvalidSocket) return false; sockaddr_in6 address{}; address.sin6_family = AF_INET6; @@ -2515,7 +2150,7 @@ void TestListenAddresses() { // Binds port 0 to find a free port, then releases it for the code under test. int FreePort(const std::string& host = "127.0.0.1") { sockaddr_storage address{}; - socklen_t length = 0; + net::SocklenT length = 0; auto* ipv4 = reinterpret_cast(&address); auto* ipv6 = reinterpret_cast(&address); if (::inet_pton(AF_INET, host.c_str(), &ipv4->sin_addr) == 1) { @@ -2526,8 +2161,8 @@ int FreePort(const std::string& host = "127.0.0.1") { ipv6->sin6_family = AF_INET6; length = sizeof(*ipv6); } - const int fd = ::socket(address.ss_family, SOCK_STREAM, 0); - assert(fd >= 0); + const net::Socket fd = net::CreateStreamSocket(address.ss_family); + assert(fd != net::kInvalidSocket); assert(::bind(fd, reinterpret_cast(&address), length) == 0); assert(::getsockname(fd, reinterpret_cast(&address), &length) == 0); @@ -2536,9 +2171,9 @@ int FreePort(const std::string& host = "127.0.0.1") { : ipv4->sin_port); } -int ConnectTo(const std::string& host, int port) { +net::Socket ConnectTo(const std::string& host, int port) { sockaddr_storage address{}; - socklen_t length = 0; + net::SocklenT length = 0; auto* ipv4 = reinterpret_cast(&address); auto* ipv6 = reinterpret_cast(&address); if (::inet_pton(AF_INET, host.c_str(), &ipv4->sin_addr) == 1) { @@ -2551,12 +2186,12 @@ int ConnectTo(const std::string& host, int port) { ipv6->sin6_port = htons(static_cast(port)); length = sizeof(*ipv6); } - const int fd = ::socket(address.ss_family, SOCK_STREAM, 0); - assert(fd >= 0); + const net::Socket fd = net::CreateStreamSocket(address.ss_family); + assert(fd != net::kInvalidSocket); assert(net::SetSocketTimeouts(fd, 3'000)); if (::connect(fd, reinterpret_cast(&address), length) != 0) { net::CloseSocket(fd); - return -1; + return net::kInvalidSocket; } return fd; } @@ -2564,18 +2199,18 @@ int ConnectTo(const std::string& host, int port) { // Sends each part separately, pausing between them, and reads to EOF. std::string SendParts(const std::string& host, int port, const std::vector& parts) { - const int fd = ConnectTo(host, port); - assert(fd >= 0); + const net::Socket fd = ConnectTo(host, port); + assert(fd != net::kInvalidSocket); for (std::size_t index = 0; index < parts.size(); ++index) { if (index > 0) std::this_thread::sleep_for(std::chrono::milliseconds(50)); assert(net::SendNoSignal(fd, parts[index].data(), parts[index].size()) == - static_cast(parts[index].size())); + static_cast(parts[index].size())); } std::string response; char buffer[4096]; for (;;) { - const auto count = ::read(fd, buffer, sizeof(buffer)); + const auto count = net::Receive(fd, buffer, sizeof(buffer)); assert(count >= 0); if (count == 0) break; @@ -2725,7 +2360,7 @@ void TestPeerDisconnect() { "Content-Type: application/json\r\nContent-Length: " + std::to_string(body.size()) + "\r\n\r\n" + body; assert(net::SendNoSignal(fd, request.data(), request.size()) == - static_cast(request.size())); + static_cast(request.size())); assert(server.backend->entered.try_acquire_for(std::chrono::seconds(2))); net::CloseSocket(fd); assert(server.backend->finished.try_acquire_for(std::chrono::seconds(2))); @@ -3054,48 +2689,6 @@ void TestSseHeartbeatShutdown() { } } -void TestSignalShutdown() { - // Process signals must never terminate the test runner itself. Prove that - // both idle listeners and active generation return through normal cleanup. - for (const int signal : {SIGINT, SIGTERM}) { - for (const bool active : {false, true}) { - const pid_t child = ::fork(); - assert(child >= 0); - if (child == 0) { - ::alarm(5); - { - RunningServer server({}, true); - // Accepting a request proves run() installed its handlers. - ExpectStatus(server.Post("/echo", "ready"), 200); - int fd = -1; - if (active) { - server.backend->wait_for_disconnect = true; - fd = server.Connect(); - const std::string body = R"({"prompt":"hello"})"; - const std::string request = - "POST /v1/completions HTTP/1.1\r\nContent-Length: " + - std::to_string(body.size()) + "\r\n\r\n" + body; - assert(net::SendNoSignal(fd, request.data(), request.size()) == - static_cast(request.size())); - assert(server.backend->entered.try_acquire_for( - std::chrono::seconds(2))); - } - assert(::kill(::getpid(), signal) == 0); - assert(server.run_finished.try_acquire_for(std::chrono::seconds(2))); - if (active) { - assert(server.backend->disconnected); - net::CloseSocket(fd); - } - } - ::_exit(0); - } - int status = 0; - assert(::waitpid(child, &status, 0) == child); - assert(WIFEXITED(status) && WEXITSTATUS(status) == 0); - } - } -} - } // namespace void TestRawCompletionCachePrompt() { @@ -3342,7 +2935,7 @@ int main() { // The log assertions below match "[LEVEL] [component]" text written to a // redirected stderr sink, so the real stderr's TTY state must not add ANSI // tint around the level tag. - ::setenv("NO_COLOR", "1", 1); + gufo::platform::SetEnv("NO_COLOR", "1"); TestRequestLogging(); TestContentTrace(); TestQuietTiersSuppressLifecycle(); @@ -3379,7 +2972,6 @@ int main() { TestSseHeartbeatShutdown(); TestDeferredStreamHeaders(); TestAdmittedStreamHeaders(); - TestSignalShutdown(); TestEmbeddingService(); std::cout << "HTTP transport checks passed.\n"; } diff --git a/tests/cli/http_server_test_support.hpp b/tests/cli/http_server_test_support.hpp new file mode 100644 index 00000000..8a70718f --- /dev/null +++ b/tests/cli/http_server_test_support.hpp @@ -0,0 +1,394 @@ +#ifndef GUFO_TESTS_CLI_HTTP_SERVER_TEST_SUPPORT_HPP_ +#define GUFO_TESTS_CLI_HTTP_SERVER_TEST_SUPPORT_HPP_ + +// The HTTP server test harness, shared by the main suite and the POSIX-only +// signal-shutdown suite. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "src/cli/serve/http_server.hpp" +#include "src/cli/serve/logging.hpp" +#include "src/cli/serve/trace.hpp" +#include "src/core/platform/net.hpp" + +namespace { +namespace net = gufo::net; + +using gufo::server::HttpServer; +using gufo::server::TextGenerationBackend; + +/// Reports an optional admission and fixed prompt progress before delegating +/// token generation. +class ProgressRequest final : public TextGenerationBackend::GenerationRequest { +public: + ProgressRequest(std::shared_ptr inner, + std::vector progress, + bool admit = false) + : inner_(std::move(inner)), + progress_(std::move(progress)), + admit_(admit) {} + + TextGenerationBackend::Result Wait( + const TextGenerationBackend::TokenCallback& on_token, + const TextGenerationBackend::ProgressCallback& on_progress, + const TextGenerationBackend::StartCallback& on_start) override { + if (admit_ && on_start && !on_start()) + inner_->Cancel(); + for (const auto& value : progress_) { + if (on_progress && !on_progress(value)) { + inner_->Cancel(); + break; + } + } + return inner_->Wait(on_token, on_progress, on_start); + } + void Cancel() noexcept override { inner_->Cancel(); } + +private: + std::shared_ptr inner_; + std::vector progress_; + bool admit_; +}; + +class FakeBackend final : public TextGenerationBackend { +public: + struct Call { + std::string prompt; + gufo::server::ChatRequest chat; + std::size_t max_tokens = 0; + gufo::sampling::SamplingConfig sampling; + std::string client_id; + std::vector stop_sequences; + std::string trace_request; + }; + Call LastCall() { + const std::lock_guard lock(mutex_); + return last_; + } + void SetOutput(std::string text) { + const std::lock_guard lock(mutex_); + output_ = std::move(text); + } + std::string model_id() const override { return "test"; } + std::vector model_ids() const override { return names; } + std::vector names{"test"}; + bool ready() const override { return true; } + bool device_lost() const override { return lost.load(); } + bool supports_images() const override { return image_support.load(); } + std::atomic image_support{false}; + std::uint32_t max_context() const override { return 65536; } + std::vector session_states() const override { return sessions; } + std::vector sessions; + std::string model_path() const override { return served_path; } + std::string build_info() const override { return served_build; } + std::string served_path; + std::string served_build; + std::shared_ptr start_complete( + std::string_view prompt, std::size_t max_tokens, + const gufo::sampling::SamplingConfig& sampling, + const CancellationCheck& cancellation, bool stream, bool ignore_eos, + std::string_view client_id, + const std::vector& stop_sequences, bool return_progress, + bool cache_prompt) override { + last_ignore_eos = ignore_eos; + return std::make_shared( + TextGenerationBackend::start_complete( + prompt, max_tokens, sampling, cancellation, stream, false, + client_id, stop_sequences, return_progress, cache_prompt), + progress, admit.load()); + } + std::shared_ptr start_chat( + const gufo::server::ChatRequest& request, std::size_t max_tokens, + const gufo::sampling::SamplingConfig& sampling, + const CancellationCheck& cancellation, bool stream) override { + return std::make_shared( + TextGenerationBackend::start_chat(request, max_tokens, sampling, + cancellation, stream), + std::vector{}, admit.load()); + } + SamplingDefaults sampling_defaults() const override { return defaults; } + SamplingDefaults defaults; + gufo::ReasoningOptions reasoning_defaults() const override { + return reasoning; + } + InitialOutputState initial_output_state( + const gufo::server::ChatRequest& request) const override { + return initial_output_state_override.value_or( + TextGenerationBackend::initial_output_state(request)); + } + std::size_t count_tokens(std::string_view text) const override { + return text.size(); + } + std::optional count_chat_tokens( + const gufo::server::ChatRequest& request) const override { + counted_messages = request.messages.size(); + counted_tools = request.tools.size(); + return chat_token_count; + } + std::optional chat_token_count; + mutable std::size_t counted_messages{0}; + mutable std::size_t counted_tools{0}; + Result complete(std::string_view prompt, std::size_t limit, + const gufo::sampling::SamplingConfig& sampling, + const CancellationCheck& cancel, const TokenCallback& token, + std::string_view client_id = "anonymous", + const std::vector& stop_sequences = {}, + bool cache_prompt = true) override { + ++calls; + last_cache_prompt = cache_prompt; + if (failure == 1) + throw std::length_error("context exceeded"); + if (failure == 2) + throw std::invalid_argument("invalid prompt"); + // A failed probe: the scheduler reports device loss from then on. + if (failure == 3) { + lost = true; + throw gufo::server::TextGenerationError( + gufo::server::TextGenerationErrorCode::kDeviceLost, + gufo::server::kDeviceLostMessage); + } + // A failure on a device that still accepts work. + if (failure == 4) + throw std::runtime_error("unspecified launch failure"); + if (failure == 5) + throw std::runtime_error(""); + Result result; + if (wait_for_disconnect) { + entered.release(); + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (!(disconnected = cancel && cancel()) && + std::chrono::steady_clock::now() < deadline) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + finished.release(); + result.cancelled = disconnected; + return result; + } + { + const std::lock_guard lock(mutex_); + last_ = {.prompt = std::string(prompt), + .max_tokens = limit, + .sampling = sampling, + .client_id = std::string(client_id), + .stop_sequences = stop_sequences, + .trace_request = gufo::server::Trace::CurrentRequest()}; + result.text = output_; + } + result.prompt_tokens = 10; + result.cached_prompt_tokens = 8; + result.cache_hit = true; + result.draft_accepted_tokens = 4; + result.draft_rounds = 2; + result.draft_tokens = 8; + result.prefill_tokens = 2; + result.prefill_ms = 4; + result.completion_tokens = 1; + result.decode_ms = 2; + result.finish_reason = + limit == 1 ? FinishReason::kLength : FinishReason::kStop; + if (!forced_stop_sequence.empty()) { + result.finish_reason = FinishReason::kStopSequence; + result.stop_sequence = forced_stop_sequence; + } + if (token && failure != 8) { + std::this_thread::sleep_for(token_delay); + (void)token(result.text); + } + if (failure == 6) + throw std::runtime_error("context exceeded"); + if (failure == 7) { + lost = true; + throw gufo::server::TextGenerationError( + gufo::server::TextGenerationErrorCode::kDeviceLost, + gufo::server::kDeviceLostMessage); + } + return result; + } + Result chat(const gufo::server::ChatRequest& request, std::size_t limit, + const gufo::sampling::SamplingConfig& sampling, + const CancellationCheck& cancel, + const TokenCallback& token) override { + auto result = complete("", limit, sampling, cancel, token, + request.client_id, request.stop_sequences); + { + const std::lock_guard lock(mutex_); + last_.chat = request; + } + return result; + } + std::atomic calls{0}; + std::atomic last_ignore_eos{false}; + std::atomic last_cache_prompt{true}; + std::vector progress; + // Report a scheduler admission before generation, as the HIP backend does. + std::atomic admit{false}; + std::chrono::milliseconds token_delay{0}; + std::atomic failure{0}; + std::atomic lost{false}; + std::string forced_stop_sequence; + gufo::ReasoningOptions reasoning; + std::optional initial_output_state_override; + bool wait_for_disconnect{false}; + std::atomic disconnected{false}; + std::binary_semaphore entered{0}; + std::binary_semaphore finished{0}; + +private: + std::mutex mutex_; + Call last_; + std::string output_{"ok"}; +}; + +class RunningServer { +public: + explicit RunningServer(gufo::server::HttpServerOptions options = {}, + bool handle_signals = false, + std::string host = "127.0.0.1") + : backend(std::make_shared()), + server(std::move(host), 0, backend, nullptr, nullptr, nullptr, + std::move(options)) { + server.add("GET", "/peer", [](const auto& request, auto&) { + return gufo::server::HttpResponse{.body = request.client_id}; + }); + server.add("POST", "/echo", [](const auto& request, auto&) { + return gufo::server::HttpResponse{.body = request.body}; + }); + server.add("POST", "/stream-error", [](const auto&, auto&) { + return gufo::server::HttpResponse{ + .streaming_body = [](const auto& write) { + (void)write("first chunk"); + throw std::runtime_error("injected stream failure"); + }}; + }); + server.add("POST", "/stream", [](const auto& request, auto&) { + auto log = std::make_shared(); + return gufo::server::HttpResponse{ + .streaming_body = + [log, fail = !request.body.empty(), + reported = request.body == "reported"](const auto& write) { + (void)write(std::string_view("a\0b", 3)); + (void)write(""); + (void)write("end"); + if (fail) + log->error_code = "injected"; + log->error_event_sent = reported; + }, + .stream_log = log, + }; + }); + server.add("POST", "/sse-idle", [](const auto&, auto&) { + return gufo::server::HttpResponse{ + .headers = {{"Content-Type", "text/event-stream"}}, + .streaming_body = + [](const auto& write) { + (void)write("data: first\n\n"); + std::this_thread::sleep_for(std::chrono::milliseconds(90)); + (void)write("data: last\n\n"); + (void)write("data: [DONE]\n\n"); + }, + }; + }); + server.add("POST", "/sse-finish", [](const auto& request, auto&) { + return gufo::server::HttpResponse{ + .headers = {{"Content-Type", "text/event-stream"}}, + .streaming_body = [fail = !request.body.empty()](const auto& write) { + (void)write("data: first\n\n"); + if (fail) + throw std::runtime_error("injected SSE failure"); + (void)write("data: [DONE]\n\n"); + }}; + }); + std::string error; + assert(server.start(&error)); + worker = std::jthread([this, handle_signals] { + server.run(handle_signals); + run_finished.release(); + }); + } + ~RunningServer() { + server.stop(); + worker.join(); + } + + net::Socket Connect(const std::string& peer = "127.0.0.1") { + sockaddr_storage address{}; + net::SocklenT length = 0; + auto* ipv4 = reinterpret_cast(&address); + auto* ipv6 = reinterpret_cast(&address); + if (::inet_pton(AF_INET, peer.c_str(), &ipv4->sin_addr) == 1) { + ipv4->sin_family = AF_INET; + ipv4->sin_port = htons(server.port()); + length = sizeof(*ipv4); + } else { + assert(::inet_pton(AF_INET6, peer.c_str(), &ipv6->sin6_addr) == 1); + ipv6->sin6_family = AF_INET6; + ipv6->sin6_port = htons(server.port()); + length = sizeof(*ipv6); + } + const net::Socket fd = net::CreateStreamSocket(address.ss_family); + assert(fd != net::kInvalidSocket); + assert(net::SetSocketTimeouts(fd, 3'000)); + assert(::connect(fd, reinterpret_cast(&address), length) == + 0); + return fd; + } + std::string Send(std::string_view request, bool half_close = false, + const std::string& peer = "127.0.0.1") { + const net::Socket fd = Connect(peer); + while (!request.empty()) { + const auto count = net::SendNoSignal(fd, request.data(), request.size()); + assert(count > 0); + request.remove_prefix(static_cast(count)); + } + // A half-close allows malformed/truncated-body tests to complete without + // timing-dependent sleeps. + if (half_close) + net::ShutdownSocket(fd, net::ShutdownHow::kWrite); + std::string response; + char buffer[4096]; + for (;;) { + const auto count = net::Receive(fd, buffer, sizeof(buffer)); + assert(count >= 0); + if (count == 0) { + break; + } + response.append(buffer, static_cast(count)); + } + net::CloseSocket(fd); + return response; + } + + std::string Post(std::string_view path, std::string_view body) { + return Send("POST " + std::string(path) + " HTTP/1.1\r\nContent-Length: " + + std::to_string(body.size()) + "\r\n\r\n" + std::string(body)); + } + + std::shared_ptr backend; + HttpServer server; + std::binary_semaphore run_finished{0}; + std::jthread worker; +}; + +void ExpectStatus(const std::string& response, int status) { + if (!response.starts_with("HTTP/1.1 " + std::to_string(status) + " ")) { + std::cerr << response << '\n'; + std::abort(); + } +} +} // namespace + +#endif // GUFO_TESTS_CLI_HTTP_SERVER_TEST_SUPPORT_HPP_ diff --git a/tests/cli/text_generation_scheduler_test.cpp b/tests/cli/text_generation_scheduler_test.cpp index 90e56f82..12efc6c6 100644 --- a/tests/cli/text_generation_scheduler_test.cpp +++ b/tests/cli/text_generation_scheduler_test.cpp @@ -30,6 +30,7 @@ #include "src/cli/serve/stop_sequences.hpp" #include "src/cli/serve/trace.hpp" #include "src/core/json.hpp" +#include "src/core/platform/env.hpp" namespace { @@ -2979,7 +2980,7 @@ int main() { TestPublicationWaitUsesRecentSubstantialPrefill(); // Log assertions in this binary match the plain "[LEVEL] [component]" text // captured from a redirected sink; a TTY stderr would tint the level tag. - ::setenv("NO_COLOR", "1", 1); + gufo::platform::SetEnv("NO_COLOR", "1"); TestProgressLoggingIsOptInAndBounded(); TestStructuredOutputIsPreparedBeforeServing(); TestAdmissionLoggingIsDebugTierOnly(); diff --git a/tests/cli/text_model_runner_test.cpp b/tests/cli/text_model_runner_test.cpp index ec8e2b7e..fbc66f20 100644 --- a/tests/cli/text_model_runner_test.cpp +++ b/tests/cli/text_model_runner_test.cpp @@ -1,7 +1,5 @@ #include "src/cli/serve/text_model_runner.hpp" -#include - #include #include #include @@ -23,8 +21,11 @@ #include #include "src/core/json.hpp" +#include "src/core/platform/env.hpp" +#include "src/core/platform/os_info.hpp" #include "src/models/qwen/control_tokens.hpp" #include "src/models/qwen/vision/prompt.hpp" +#include "tests/support/temporary_directory.hpp" using gufo::tokenization::kImEnd; using gufo::tokenization::kImStart; @@ -770,32 +771,6 @@ private: std::size_t max_context_; }; -class TemporaryDirectory { -public: - TemporaryDirectory() { - std::string pattern = (std::filesystem::temp_directory_path() / - "gufo-runner-disk-cache-XXXXXX") - .string(); - const char* created = ::mkdtemp(pattern.data()); - if (created == nullptr) { - throw std::runtime_error("failed to create runner cache directory"); - } - path_ = created; - } - - ~TemporaryDirectory() { - std::error_code error; - std::filesystem::remove_all(path_, error); - } - - [[nodiscard]] const std::filesystem::path& path() const noexcept { - return path_; - } - -private: - std::filesystem::path path_; -}; - void TestSnapshotForkAndUnsupportedCapabilities() { auto stats = std::make_shared(); SnapshotRunner runner(stats); @@ -970,7 +945,7 @@ void TestGeneratedFrontierForksBeforeMutation() { void TestGeneratedFrontierPersistsForForks() { // Persist every frontier; spacing is checked below and by the disk store. - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-runner-disk-cache"); const TextRunnerDiskCacheOptions disk{.directory = directory.path(), .capacity_bytes = 4096, .staging_capacity_bytes = 4096, @@ -998,7 +973,7 @@ void TestGeneratedFrontierPersistsForForks() { // With the default step, the generated frontier one token past the prompt // is not written; a restart resumes from the prompt and re-prefills it. - TemporaryDirectory spaced_directory; + TemporaryDirectory spaced_directory("gufo-runner-disk-cache"); const TextRunnerDiskCacheOptions spaced{.directory = spaced_directory.path(), .capacity_bytes = 4096, .staging_capacity_bytes = 4096}; @@ -1133,7 +1108,7 @@ void TestCancellationRetainsOnlyCompletedWork() { } void TestPersistentSnapshotRestoresAcrossPools() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-runner-disk-cache"); const TextRunnerDiskCacheOptions disk_cache{ .directory = directory.path(), .capacity_bytes = 4096, @@ -1212,7 +1187,7 @@ void TestPromptReuseCanBeDisabledPerRequest() { } void TestStableChatPrefixSurvivesInterruptedFraming() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-runner-disk-cache"); const TextRunnerDiskCacheOptions disk_cache{.directory = directory.path(), .capacity_bytes = 8192, .staging_capacity_bytes = 4096}; @@ -1532,7 +1507,7 @@ void TestChatFallbackSurvivesSnapshotBudgetPressure() { } void TestFullChatCheckpointRestoresWithoutSuffixPrefill() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-runner-disk-cache"); const TextRunnerDiskCacheOptions disk_cache{.directory = directory.path(), .capacity_bytes = 8192, .staging_capacity_bytes = 4096, @@ -1568,7 +1543,7 @@ void TestFullChatCheckpointRestoresWithoutSuffixPrefill() { "the earlier disk fallback survives exact full-prompt restoration"); changed.Invalidate(); - TemporaryDirectory legacy_directory; + TemporaryDirectory legacy_directory("gufo-runner-disk-cache"); const TextRunnerDiskCacheOptions legacy_disk{ .directory = legacy_directory.path(), .capacity_bytes = 8192, @@ -1640,7 +1615,7 @@ void TestNewImageGetsAStableCheckpoint() { } void TestSharedPrefixIsLearnedAndRestoredAcrossConversations() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-runner-disk-cache"); const TextRunnerDiskCacheOptions disk_cache{ .directory = directory.path(), .capacity_bytes = 8192, @@ -1800,7 +1775,7 @@ void TestLearnedBoundarySurvivesExactRepublication() { void TestCoincidentCacheBoundariesShareOneCopy() { for (const bool stable : {false, true}) { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-runner-disk-cache"); const TextRunnerDiskCacheOptions disk_cache{ .directory = directory.path(), .capacity_bytes = 1024 * 1024, @@ -1845,7 +1820,7 @@ void TestCoincidentCacheBoundariesShareOneCopy() { } void TestDiskOnlyCaptureReservesBudgetBeforeCommit() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-runner-disk-cache"); auto stats = std::make_shared(); auto runner = std::make_shared( stats, "artifact-A", sizeof(FakeSnapshot) - 1); @@ -1877,7 +1852,7 @@ void TestDiskOnlyCaptureReservesBudgetBeforeCommit() { } void TestDiskPreflightAvoidsUnusableCapture() { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-runner-disk-cache"); auto stats = std::make_shared(); auto runner = std::make_shared( stats, "artifact-A", sizeof(FakeSnapshot) - 1); @@ -2080,8 +2055,9 @@ void TestCgroupAvailableCountsReclaimableCache() { namespace fs = std::filesystem; using gufo::server::CgroupAvailableBytes; constexpr std::uint64_t gib = std::uint64_t{1} << 30; - const auto root = fs::temp_directory_path() / - ("gufo_cgroup_test_" + std::to_string(::getpid())); + const auto root = + fs::temp_directory_path() / + ("gufo_cgroup_test_" + std::to_string(gufo::platform::ProcessId())); fs::remove_all(root); const auto write = [&](const fs::path& dir, const char* name, const std::string& text) { @@ -2524,7 +2500,7 @@ int main() { TestPreparedConstraintVocabularyKeepsGrammarBehavior(); // The cache warning assertion in this binary matches the plain "[WARN] // [cache]" text captured from a redirected sink; a TTY stderr tints it. - ::setenv("NO_COLOR", "1", 1); + gufo::platform::SetEnv("NO_COLOR", "1"); TestInPassStableCheckpoint(); TestInPassFailureRetainsOnlyCompletedCheckpoints(); TestNewImageGetsAStableCheckpoint(); diff --git a/tests/cli/trace_test.cpp b/tests/cli/trace_test.cpp index e05cf67b..14685391 100644 --- a/tests/cli/trace_test.cpp +++ b/tests/cli/trace_test.cpp @@ -12,6 +12,7 @@ #include #include "src/core/json.hpp" +#include "src/core/platform/env.hpp" namespace { @@ -70,7 +71,7 @@ void TestPartialWriteIsRolledBack() { } // namespace int main() { - ::setenv("NO_COLOR", "1", 1); + gufo::platform::SetEnv("NO_COLOR", "1"); TestPartialWriteIsRolledBack(); std::cout << "Trace sink checks passed.\n"; } diff --git a/tests/cli/video_api_test.cpp b/tests/cli/video_api_test.cpp index 3c463e2a..f1f4d871 100644 --- a/tests/cli/video_api_test.cpp +++ b/tests/cli/video_api_test.cpp @@ -1,7 +1,5 @@ #include "src/cli/serve/video_api.hpp" -#include - #include #include #include diff --git a/tests/cli/video_jobs_test.cpp b/tests/cli/video_jobs_test.cpp index 89bf7cf4..858bca50 100644 --- a/tests/cli/video_jobs_test.cpp +++ b/tests/cli/video_jobs_test.cpp @@ -1,7 +1,5 @@ #include "src/cli/serve/video_jobs.hpp" -#include - #include #include #include diff --git a/tests/core/gguf_identity_test.cpp b/tests/core/gguf_identity_test.cpp index 467421b9..e3f181f9 100644 --- a/tests/core/gguf_identity_test.cpp +++ b/tests/core/gguf_identity_test.cpp @@ -17,6 +17,7 @@ #include "src/core/crypto/sha256.hpp" #include "src/core/gguf_reader.hpp" +#include "src/core/platform/env.hpp" namespace { @@ -148,7 +149,7 @@ void TestFileDigestCache() { const char* old_cache = std::getenv("XDG_CACHE_HOME"); const std::string saved_cache = old_cache ? old_cache : ""; const bool had_cache = old_cache != nullptr; - setenv("XDG_CACHE_HOME", temp, 1); + gufo::platform::SetEnv("XDG_CACHE_HOME", temp); auto bytes = BuildImage("model", "first", 64 * kWindow, 2 * kWindow); const auto path = root / "model.gguf"; auto write = [&] { @@ -186,9 +187,9 @@ void TestFileDigestCache() { } Expect(read() == Identity(bytes), "malformed cached digest is recomputed"); if (had_cache) - setenv("XDG_CACHE_HOME", saved_cache.c_str(), 1); + gufo::platform::SetEnv("XDG_CACHE_HOME", saved_cache); else - unsetenv("XDG_CACHE_HOME"); + gufo::platform::UnsetEnv("XDG_CACHE_HOME"); std::filesystem::remove_all(root); } diff --git a/tests/diagnostics/gpu_queues_test.cpp b/tests/diagnostics/gpu_queues_test.cpp index aa84de7e..13b667ca 100644 --- a/tests/diagnostics/gpu_queues_test.cpp +++ b/tests/diagnostics/gpu_queues_test.cpp @@ -1,7 +1,5 @@ #include "src/core/diagnostics/gpu_queues.h" -#include - #include #include #include @@ -10,6 +8,8 @@ #include #include +#include "src/core/platform/os_info.hpp" + namespace { void Expect(bool condition, std::string_view message) { @@ -25,7 +25,7 @@ class FakeKfd { public: FakeKfd() : root_(std::filesystem::temp_directory_path() / - ("gufo-queues-" + std::to_string(::getpid()))) { + ("gufo-queues-" + std::to_string(gufo::platform::ProcessId()))) { std::filesystem::remove_all(root_); std::filesystem::create_directories(proc_root()); } diff --git a/tests/functional/run.py b/tests/functional/run.py index 6a9336dc..007634b7 100644 --- a/tests/functional/run.py +++ b/tests/functional/run.py @@ -197,8 +197,11 @@ def server(command, log_path, startup_timeout, environment=None): process = None previous_handlers = {} try: + stop_signals = [signal.SIGTERM] + if hasattr(signal, "SIGHUP"): + stop_signals.append(signal.SIGHUP) previous_handlers = {signum: signal.signal(signum, interrupt) - for signum in (signal.SIGTERM, signal.SIGHUP)} + for signum in stop_signals} process = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT, env=environment) deadline = time.monotonic() + startup_timeout @@ -221,7 +224,13 @@ def server(command, log_path, startup_timeout, environment=None): yield process finally: if process is not None and process.poll() is None: - process.send_signal(signal.SIGINT) + try: + process.send_signal(signal.SIGINT) + except ValueError: + # Windows Popen cannot deliver SIGINT; TerminateProcess + # still reaps the server, but graceful shutdown is never + # exercised there. + process.terminate() try: process.wait(timeout=30) except subprocess.TimeoutExpired: diff --git a/tests/models/deepseek_v4_flash/load_test.cpp b/tests/models/deepseek_v4_flash/load_test.cpp index b723a0dd..00aaf703 100644 --- a/tests/models/deepseek_v4_flash/load_test.cpp +++ b/tests/models/deepseek_v4_flash/load_test.cpp @@ -9,6 +9,7 @@ #include #include +#include "src/core/platform/env.hpp" #include "src/models/deepseek_v4_flash/runtime/model.h" namespace { @@ -37,7 +38,7 @@ int main() { const auto root = std::filesystem::path(directory); const auto path = root / "invalid.gguf"; const auto lock_path = root / "instance.lock"; - assert(::setenv("DS4_LOCK_FILE", lock_path.c_str(), 1) == 0); + assert(gufo::platform::SetEnv("DS4_LOCK_FILE", lock_path.string())); const auto descriptors = OpenDescriptors(); for (int fixture = 0; fixture < 3; ++fixture) { { diff --git a/tests/models/embedding_gemma2/CMakeLists.txt b/tests/models/embedding_gemma2/CMakeLists.txt index c349a7e9..4f16aac6 100644 --- a/tests/models/embedding_gemma2/CMakeLists.txt +++ b/tests/models/embedding_gemma2/CMakeLists.txt @@ -62,7 +62,7 @@ endif() add_executable(embedding_gemma2_tokenizer_test tokenizer_test.cpp) target_link_libraries(embedding_gemma2_tokenizer_test PRIVATE - gufo_embedding_gemma2) + gufo_embedding_gemma2 gufo_platform) target_compile_definitions(embedding_gemma2_tokenizer_test PRIVATE GUFO_EMBEDDING_GEMMA2_TOKENIZER_GOLDEN="${PROJECT_SOURCE_DIR}/tests/fixtures/embedding_gemma2/tokenizer_golden.json") add_test(NAME embedding_gemma2_tokenizer_test diff --git a/tests/models/embedding_gemma2/tokenizer_test.cpp b/tests/models/embedding_gemma2/tokenizer_test.cpp index bd7cb4d0..c4a35129 100644 --- a/tests/models/embedding_gemma2/tokenizer_test.cpp +++ b/tests/models/embedding_gemma2/tokenizer_test.cpp @@ -1,7 +1,5 @@ #include "src/models/embedding_gemma2/tokenizer.hpp" -#include - #include #include #include @@ -16,6 +14,7 @@ #include #include "src/core/json.hpp" +#include "src/core/platform/os_info.hpp" namespace embedding_gemma2 = gufo::models::embedding_gemma2; @@ -74,7 +73,8 @@ std::string Format(const std::vector& ids) { void CheckMalformedInputs() { const std::filesystem::path dir = std::filesystem::temp_directory_path() / - ("embedding_gemma2_tokenizer_" + std::to_string(::getpid())); + ("embedding_gemma2_tokenizer_" + + std::to_string(gufo::platform::ProcessId())); std::filesystem::create_directories(dir); std::string error; Check(!embedding_gemma2::Tokenizer::Load((dir / "missing").string(), &error) diff --git a/tests/models/minimax_h3_media_test.cpp b/tests/models/minimax_h3_media_test.cpp index e43905a7..40b3e7d5 100644 --- a/tests/models/minimax_h3_media_test.cpp +++ b/tests/models/minimax_h3_media_test.cpp @@ -10,6 +10,7 @@ #include #include +#include "src/core/platform/env.hpp" #include "src/models/minimax_h3/media.hpp" namespace { @@ -186,14 +187,14 @@ int main() { const char* saved = std::getenv("GUFO_FFMPEG"); const std::string saved_value = saved == nullptr ? "" : saved; WriteSentinel(error_output); - setenv("GUFO_FFMPEG", "/bin/false", 1); + gufo::platform::SetEnv("GUFO_FFMPEG", "/bin/false"); const bool error_result = WriteSynchronizedMp4( error_output, frames, kFrames, kWidth, kHeight, 24, pcm, kSamples, 2, 32000, nullptr, &telemetry, &error); if (saved == nullptr) { - unsetenv("GUFO_FFMPEG"); + gufo::platform::UnsetEnv("GUFO_FFMPEG"); } else { - setenv("GUFO_FFMPEG", saved_value.c_str(), 1); + gufo::platform::SetEnv("GUFO_FFMPEG", saved_value); } if (error_result || !HasSentinel(error_output) || HasPartialFor(error_output)) { @@ -204,14 +205,14 @@ int main() { const std::string saved_probe_value = saved_probe == nullptr ? "" : saved_probe; WriteSentinel(probe_error_output); - setenv("GUFO_FFPROBE", "/bin/false", 1); + gufo::platform::SetEnv("GUFO_FFPROBE", "/bin/false"); const bool probe_error_result = WriteSynchronizedMp4( probe_error_output, frames, kFrames, kWidth, kHeight, 24, pcm, kSamples, 2, 32000, nullptr, &telemetry, &error); if (saved_probe == nullptr) { - unsetenv("GUFO_FFPROBE"); + gufo::platform::UnsetEnv("GUFO_FFPROBE"); } else { - setenv("GUFO_FFPROBE", saved_probe_value.c_str(), 1); + gufo::platform::SetEnv("GUFO_FFPROBE", saved_probe_value); } if (probe_error_result || !HasSentinel(probe_error_output) || HasPartialFor(probe_error_output)) { diff --git a/tests/models/minimax_h3_runtime_test.cpp b/tests/models/minimax_h3_runtime_test.cpp index e4c63a6e..97591ff8 100644 --- a/tests/models/minimax_h3_runtime_test.cpp +++ b/tests/models/minimax_h3_runtime_test.cpp @@ -1,7 +1,4 @@ -#include - #include -#include #include #include #include @@ -20,6 +17,7 @@ #include "src/models/minimax_h3/json.hpp" #include "src/models/minimax_h3/runtime.hpp" #include "src/models/minimax_h3/sha256.hpp" +#include "tests/support/temporary_directory.hpp" namespace { @@ -40,23 +38,6 @@ void Expect(bool condition, const std::string& message) { } } -struct TemporaryDirectory { - TemporaryDirectory() { - std::array pattern{}; - std::strcpy(pattern.data(), "/tmp/gufo-h3-runtime-XXXXXX"); - char* result = mkdtemp(pattern.data()); - if (result == nullptr) { - throw std::runtime_error("mkdtemp failed"); - } - path = result; - } - ~TemporaryDirectory() { - std::error_code error; - std::filesystem::remove_all(path, error); - } - std::filesystem::path path; -}; - struct TensorSpec { std::string name; std::string dtype; @@ -318,8 +299,8 @@ void TestStrictJson() { } void TestInspectionAndInventory() { - TemporaryDirectory temporary; - ModelFixture fixture(temporary.path); + TemporaryDirectory temporary("gufo-h3-runtime"); + ModelFixture fixture(temporary.path()); const ModelInventory inventory = InspectFixture(fixture); Expect(inventory.tensors().size() == 5, "all fixture tensors inspected"); Expect(inventory.shards().size() == 4, "all fixture shards inspected"); @@ -336,8 +317,8 @@ void TestInspectionAndInventory() { } void TestUnknownArchitectureRejected() { - TemporaryDirectory temporary; - ModelFixture fixture(temporary.path); + TemporaryDirectory temporary("gufo-h3-runtime"); + ModelFixture fixture(temporary.path()); fixture.WriteIndexes(true); std::string error; const auto inventory = ModelInventory::Inspect( @@ -349,8 +330,8 @@ void TestUnknownArchitectureRejected() { } void TestPinnedManifestRequiredByDefault() { - TemporaryDirectory temporary; - ModelFixture fixture(temporary.path); + TemporaryDirectory temporary("gufo-h3-runtime"); + ModelFixture fixture(temporary.path()); std::string error; const auto inventory = ModelInventory::Inspect( fixture.root, fixture.root / "source-manifest.json", &error); @@ -362,8 +343,8 @@ void TestPinnedManifestRequiredByDefault() { } void TestDeviceCopyAndMappedResidency() { - TemporaryDirectory temporary; - ModelFixture fixture(temporary.path); + TemporaryDirectory temporary("gufo-h3-runtime"); + ModelFixture fixture(temporary.path()); const ModelInventory inventory = InspectFixture(fixture); FakeBackend backend; { @@ -405,8 +386,8 @@ void TestDeviceCopyAndMappedResidency() { } void TestFailureInjectionIsTransactional() { - TemporaryDirectory temporary; - ModelFixture fixture(temporary.path); + TemporaryDirectory temporary("gufo-h3-runtime"); + ModelFixture fixture(temporary.path()); const ModelInventory inventory = InspectFixture(fixture); for (const ResidencyMode mode : {ResidencyMode::kMappedReadOnly, ResidencyMode::kDeviceCopy}) { @@ -447,8 +428,8 @@ void TestFailureInjectionIsTransactional() { } void TestCancellationAndPhasePeaks() { - TemporaryDirectory temporary; - ModelFixture fixture(temporary.path); + TemporaryDirectory temporary("gufo-h3-runtime"); + ModelFixture fixture(temporary.path()); const ModelInventory inventory = InspectFixture(fixture); FakeBackend backend; CancellationToken cancelled; diff --git a/tests/models/qwen/vision_serving_test.cpp b/tests/models/qwen/vision_serving_test.cpp index fca09cd2..7bf9c59b 100644 --- a/tests/models/qwen/vision_serving_test.cpp +++ b/tests/models/qwen/vision_serving_test.cpp @@ -1,5 +1,3 @@ -#include - #include #include #include @@ -14,6 +12,7 @@ #include "src/core/image.hpp" #include "src/models/qwen/hip/executor.hpp" #include "src/models/qwen38_flash_next/engine.hpp" +#include "tests/support/temporary_directory.hpp" namespace { using namespace gufo; @@ -546,19 +545,9 @@ int main(int argc, char** argv) { if (flash) CheckFlashIncrementalOracle(qfn, continuation, continued); - std::array pattern{}; - const std::string temporary = "/tmp/gufo-vision-cache-XXXXXX"; - std::copy(temporary.begin(), temporary.end(), pattern.begin()); - Require(::mkdtemp(pattern.data()) != nullptr, "cannot create test cache"); - struct Cleanup { - std::filesystem::path path; - ~Cleanup() { - std::error_code ignored; - std::filesystem::remove_all(path, ignored); - } - } cleanup{pattern.data()}; + TemporaryDirectory cache_root("gufo-vision-cache"); server::TextDiskCacheConfig cache{ - .directory = cleanup.path, + .directory = cache_root.path(), .capacity_bytes = std::size_t{2} << 30, .staging_capacity_bytes = std::size_t{512} << 20, .model_artifact_fingerprint = core::GgufIdentityHex(*reader)}; @@ -596,7 +585,7 @@ int main(int argc, char** argv) { for (const bool use_spec : {false, true}) { if (use_spec && draft.empty()) continue; - cache.directory = cleanup.path / (use_spec ? "spec" : "ar"); + cache.directory = cache_root.path() / (use_spec ? "spec" : "ar"); auto first = load(use_spec, cache); const auto red = first->chat(requests[0], 16, greedy); const auto sampled_red = first->chat(requests[0], 16, sampled); diff --git a/tests/models/qwen27b/serving_test.cpp b/tests/models/qwen27b/serving_test.cpp index ddf7d2d7..f7804b21 100644 --- a/tests/models/qwen27b/serving_test.cpp +++ b/tests/models/qwen27b/serving_test.cpp @@ -23,6 +23,7 @@ #include "src/models/qwen/generator.hpp" #include "src/models/qwen/hip/executor.hpp" #include "tests/models/qwen27b/sampling_cases.hpp" +#include "tests/support/temporary_directory.hpp" namespace { @@ -137,35 +138,6 @@ void CheckSamplingStrategies( "concurrent sampling did not exercise shared target execution"); } -class TemporaryDirectory { -public: - TemporaryDirectory() { - std::string pattern = - (std::filesystem::temp_directory_path() / "gufo-qwen-disk-cache-XXXXXX") - .string(); - const char* created = ::mkdtemp(pattern.data()); - if (created == nullptr) { - throw std::runtime_error("failed to create Qwen cache directory"); - } - path_ = created; - } - - ~TemporaryDirectory() { - std::error_code error; - std::filesystem::remove_all(path_, error); - } - - TemporaryDirectory(const TemporaryDirectory&) = delete; - TemporaryDirectory& operator=(const TemporaryDirectory&) = delete; - - [[nodiscard]] const std::filesystem::path& path() const noexcept { - return path_; - } - -private: - std::filesystem::path path_; -}; - void RunPromptReuseSmoke( const std::shared_ptr& model, gufo::server::InferenceBackend& backend) { @@ -288,7 +260,7 @@ void CheckGeneratedFrontierReuse( // The prompt checkpoint has a published but unconsumed token. Persist // both AR and DFlash2 checkpoints, then restore into fresh sessions. for (const bool speculative : {false, true}) { - TemporaryDirectory directory; + TemporaryDirectory directory("gufo-qwen-disk-cache"); const gufo::server::TextDiskCacheConfig disk{ .directory = directory.path(), .model_artifact_fingerprint = std::string(64, 'a'), @@ -952,7 +924,7 @@ int main(int argc, const char* const* argv) { const auto direct_fork = GenerateDirect(*direct, fork_prompt, 2); { - TemporaryDirectory cache_directory; + TemporaryDirectory cache_directory("gufo-qwen-disk-cache"); const gufo::server::TextDiskCacheConfig disk_cache{ .directory = cache_directory.path(), .capacity_bytes = 1024ULL * 1024ULL * 1024ULL, diff --git a/tests/support/temporary_directory.hpp b/tests/support/temporary_directory.hpp new file mode 100644 index 00000000..ae17ac9d --- /dev/null +++ b/tests/support/temporary_directory.hpp @@ -0,0 +1,60 @@ +#ifndef GUFO_TESTS_SUPPORT_TEMPORARY_DIRECTORY_HPP_ +#define GUFO_TESTS_SUPPORT_TEMPORARY_DIRECTORY_HPP_ + +// A unique directory under the temporary path, owner-only where the OS +// supports it, removed when the test ends. + +#include +#include +#include +#include +#include + +#include "src/core/platform/fs.hpp" +#include "src/core/platform/os_info.hpp" + +namespace { + +class TemporaryDirectory { +public: + explicit TemporaryDirectory(std::string name) { + const auto base = std::filesystem::temp_directory_path(); + // A random suffix keeps the name unpredictable in a shared temporary + // root, as mkdtemp's was. + std::random_device random; + for (int attempt = 0; attempt < 64; ++attempt) { + std::error_code error; + const auto candidate = + base / (name + "-" + std::to_string(gufo::platform::ProcessId()) + + "-" + std::to_string(random())); + if (std::filesystem::create_directory(candidate, error)) { + if (!gufo::platform::RestrictToOwner(candidate)) { + std::filesystem::remove(candidate, error); + throw std::runtime_error("failed to restrict " + name + " directory"); + } + path_ = candidate; + return; + } + } + throw std::runtime_error("failed to create " + name + " directory"); + } + + ~TemporaryDirectory() { + std::error_code error; + std::filesystem::remove_all(path_, error); + } + + TemporaryDirectory(const TemporaryDirectory&) = delete; + TemporaryDirectory& operator=(const TemporaryDirectory&) = delete; + + [[nodiscard]] const std::filesystem::path& path() const noexcept { + return path_; + } + +private: + std::filesystem::path path_; +}; + +} // namespace + +#endif // GUFO_TESTS_SUPPORT_TEMPORARY_DIRECTORY_HPP_