From f70c24a21cb4a0222387f03d28a8e3bcfe9fd076 Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 4 Oct 2026 13:47:01 +0000 Subject: [PATCH 1/2] test(r): skip mock IPC tests on CRAN and fix server races --- r-pkg/tests/testthat/helper-mock-server.R | 106 +++++++++++-------- r-pkg/tests/testthat/test-mock-server.R | 120 ++++++++++++++++++++++ 2 files changed, 185 insertions(+), 41 deletions(-) create mode 100644 r-pkg/tests/testthat/test-mock-server.R diff --git a/r-pkg/tests/testthat/helper-mock-server.R b/r-pkg/tests/testthat/helper-mock-server.R index 3ffee5c..0928493 100644 --- a/r-pkg/tests/testthat/helper-mock-server.R +++ b/r-pkg/tests/testthat/helper-mock-server.R @@ -11,25 +11,60 @@ # 3. Responds to metrics_request messages with approximate values # 4. Collects all received JSON messages # 5. Returns collected messages when the device sends "close" +# +# These subprocess/socket integration tests run in CI, but are skipped on CRAN. + +# Shared by both transports: wait() does not terminate a process on timeout. +collect_mock_server = function(bg, timeout, name = "Mock server") { + bg$wait(timeout) + if (bg$is_alive()) { + bg$kill() + stop(name, " timed out after ", timeout, + " ms waiting for completion; background process terminated") + } + + status = bg$get_exit_status() + read_error = function() { + # kill() can close stderr; preserve the exit-status error in that case. + tryCatch(bg$read_error(), error = function(e) "stderr unavailable") + } + if (is.null(status) || is.na(status)) { + stop(name, " exited with unknown exit status: ", read_error()) + } + if (status != 0) { + stop(name, " exited with error (status ", status, "): ", read_error()) + } + bg$get_result() +} -start_mock_server_local = function(send_welcome = FALSE, transport = "unix") { +start_mock_server_local = function( + send_welcome = FALSE, + transport = "unix", + envir = parent.frame() +) { + skip_on_cran() skip_if_not_installed("callr") skip_if_not_installed("processx") skip_if_not_installed("jsonlite") is_windows = (.Platform$OS.type == "windows") + # Keep paths until the caller finishes, rather than until this helper returns. + ready_file = withr::local_tempfile( + pattern = "jgd-test-ready-", fileext = ".txt", .local_envir = envir + ) if (is_windows) { # Windows: named pipe via processx pipe_name = sprintf("jgd-test-%d-%s", Sys.getpid(), - basename(tempfile())) + basename(ready_file)) # processx expects \\?\pipe\NAME (extended-length path prefix); # the C client uses \\.\pipe\NAME (device namespace) — both resolve # to the same kernel pipe object win_path = paste0("\\\\?\\pipe\\", pipe_name) - ready_file = tempfile(pattern = "jgd-test-ready-", fileext = ".txt") } else { socket_path = tempfile(pattern = "jgd-test-", fileext = ".sock") + # local_tempfile() uses recursive unlink, which does not remove Unix sockets. + withr::defer(unlink(socket_path), envir = envir) } bg = callr::r_bg( @@ -37,11 +72,9 @@ start_mock_server_local = function(send_welcome = FALSE, transport = "unix") { `%||%` = function(x, y) if (is.null(x)) y else x server = processx::conn_create_unix_socket(conn_path) - # Signal readiness: on Windows write a ready file (pipe has no filesystem - # presence); on Unix the socket file itself signals readiness - if (!is.null(ready_file)) { - writeLines("ready", ready_file) - } + # Signal readiness only after socket creation (including listen) returns. + # The Unix socket path can appear before the server is listening. + writeLines("ready", ready_file) # Wait for client connection (30s timeout) # poll() returns "connect" (not "ready") for new connections on a @@ -124,34 +157,27 @@ start_mock_server_local = function(send_welcome = FALSE, transport = "unix") { }, args = list( conn_path = if (is_windows) win_path else socket_path, - ready_file = if (is_windows) ready_file else NULL, + ready_file = ready_file, send_welcome = send_welcome, transport = transport ), supervise = TRUE ) + # Both Unix sockets and Windows named pipes use an explicit readiness signal. + for (i in seq_len(30)) { + if (file.exists(ready_file)) break + Sys.sleep(0.1) + } + if (!file.exists(ready_file)) { + bg$kill() + unlink(ready_file) + if (!is_windows) unlink(socket_path) + skip("Mock server not ready in time") + } if (is_windows) { - # Wait for the ready file to appear (pipe has no filesystem presence) - for (i in seq_len(30)) { - if (file.exists(ready_file)) break - Sys.sleep(0.1) - } - if (!file.exists(ready_file)) { - bg$kill() - skip("Mock server pipe not ready in time") - } client_uri = paste0("npipe:////./pipe/", pipe_name) } else { - # Wait for the socket file to appear - for (i in seq_len(30)) { - if (file.exists(socket_path)) break - Sys.sleep(0.1) - } - if (!file.exists(socket_path)) { - bg$kill() - skip("Mock server socket not created in time") - } client_uri = socket_path } @@ -159,11 +185,7 @@ start_mock_server_local = function(send_welcome = FALSE, transport = "unix") { bg = bg, socket_path = client_uri, collect = function(timeout = 10000) { - bg$wait(timeout) - if (bg$get_exit_status() != 0) { - stop("Mock server exited with error: ", bg$read_error()) - } - bg$get_result() + collect_mock_server(bg, timeout) }, cleanup = function() { if (bg$is_alive()) { @@ -171,19 +193,25 @@ start_mock_server_local = function(send_welcome = FALSE, transport = "unix") { } if (!is_windows) { unlink(socket_path) - } else { - unlink(ready_file) } + unlink(ready_file) } ) } # TCP mock server using base R sockets (works on all platforms including Windows) -start_mock_server_tcp = function(send_welcome = FALSE, transport = "tcp") { +start_mock_server_tcp = function( + send_welcome = FALSE, + transport = "tcp", + envir = parent.frame() +) { + skip_on_cran() skip_if_not_installed("callr") skip_if_not_installed("jsonlite") - port_file = tempfile(pattern = "jgd-tcp-port-", fileext = ".txt") + port_file = withr::local_tempfile( + pattern = "jgd-tcp-port-", fileext = ".txt", .local_envir = envir + ) bg = callr::r_bg( function(port_file, send_welcome, transport) { @@ -295,11 +323,7 @@ start_mock_server_tcp = function(send_welcome = FALSE, transport = "tcp") { port = port, socket_url = sprintf("tcp://127.0.0.1:%d", port), collect = function(timeout = 10000) { - bg$wait(timeout) - if (bg$get_exit_status() != 0) { - stop("Mock TCP server exited with error: ", bg$read_error()) - } - bg$get_result() + collect_mock_server(bg, timeout, "Mock TCP server") }, cleanup = function() { if (bg$is_alive()) { diff --git a/r-pkg/tests/testthat/test-mock-server.R b/r-pkg/tests/testthat/test-mock-server.R new file mode 100644 index 0000000..c8f93ea --- /dev/null +++ b/r-pkg/tests/testthat/test-mock-server.R @@ -0,0 +1,120 @@ +test_that("mock server entry points skip before starting subprocesses on CRAN", { + withr::local_envvar(NOT_CRAN = "false") + + expect_condition(start_mock_server_local(), "On CRAN", class = "skip") + expect_condition(start_mock_server_tcp(), "On CRAN", class = "skip") + expect_condition( + with_mock_jgd(stop("must not run")), + "On CRAN", + class = "skip" + ) + expect_condition( + with_mock_jgd(stop("must not run"), transport = "tcp"), + "On CRAN", + class = "skip" + ) +}) + +test_that("collection reports unknown and nonzero exit statuses", { + # Unknown statuses are rare in real processx processes, so model them here. + for (status in list(NULL, NA_integer_, 1L)) { + bg = list( + wait = function(timeout) NULL, + is_alive = function() FALSE, + get_exit_status = function() status, + read_error = function() "server diagnostic", + get_result = function() stop("must not read a failed result") + ) + message = if (is.null(status) || is.na(status)) { + "unknown exit status: server diagnostic" + } else { + "error \\(status 1\\): server diagnostic" + } + expect_error(collect_mock_server(bg, 1), message) + } +}) + +test_that("local collection timeout terminates the waiting mock server", { + server = start_mock_server_local() + withr::defer(server$cleanup()) + + expect_error(server$collect(timeout = 1), "Mock server timed out after 1 ms") + expect_false(server$bg$is_alive()) +}) + +test_that("TCP collection timeout terminates the waiting mock server", { + server = start_mock_server_tcp() + withr::defer(server$cleanup()) + + expect_error( + server$collect(timeout = 1), + "Mock TCP server timed out after 1 ms" + ) + expect_false(server$bg$is_alive()) +}) + +test_that("collection reports a terminated subprocess for both transports", { + lapply(list(start_mock_server_local, start_mock_server_tcp), function(start) { + server = start() + withr::defer(server$cleanup()) + server$bg$kill() + + expect_error(server$collect(), "exited with error \\(status") + }) +}) + +test_that("local readiness waits for socket creation to return", { + skip_on_cran() + skip_if_not_installed("callr") + skip_if_not_installed("processx") + skip_if_not_installed("jsonlite") + + returned_file = withr::local_tempfile() + files = new.env(parent = emptyenv()) + r_bg = callr::r_bg + local_mocked_bindings( + r_bg = function(func, args, ...) { + files$ready = args$ready_file + r_bg( + function(func, args, returned_file) { + create_socket = processx::conn_create_unix_socket + # Delay the constructor after the socket/pipe exists. The parent must + # still wait until the constructor returns and publishes readiness. + assignInNamespace( + "conn_create_unix_socket", + function(...) { + server = create_socket(...) + Sys.sleep(0.5) + writeLines("returned", returned_file) + server + }, + ns = "processx" + ) + do.call(func, args) + }, + args = list(func = func, args = args, returned_file = returned_file), + ... + ) + }, + .package = "callr" + ) + + run_server = function() { + server = start_mock_server_local() + withr::defer(if (server$bg$is_alive()) server$bg$kill()) + expect_true(file.exists(returned_file)) + expect_true(file.exists(files$ready)) + + jgd(socket = server$socket_path) + dev.off() + expect_identical(tail(server$collect(), 1)[[1]]$type, "close") + server$socket_path + } + socket_path = run_server() + # withr removes the paths when the caller finishes, even without + # an explicit server$cleanup() call. + expect_false(file.exists(files$ready)) + if (.Platform$OS.type != "windows") { + expect_false(file.exists(socket_path)) + } +}) From 2f4ce7bc679d6998a66d1b1ffa5c2fcdff5406bb Mon Sep 17 00:00:00 2001 From: eitsupi Date: Sun, 4 Oct 2026 14:00:14 +0000 Subject: [PATCH 2/2] test(r): arm named pipe connection before signaling readiness --- r-pkg/tests/testthat/helper-mock-server.R | 4 ++++ r-pkg/tests/testthat/test-mock-server.R | 24 ++++++++++++++++++++--- 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/r-pkg/tests/testthat/helper-mock-server.R b/r-pkg/tests/testthat/helper-mock-server.R index 0928493..de7c6fc 100644 --- a/r-pkg/tests/testthat/helper-mock-server.R +++ b/r-pkg/tests/testthat/helper-mock-server.R @@ -72,6 +72,10 @@ start_mock_server_local = function( `%||%` = function(x, y) if (is.null(x)) y else x server = processx::conn_create_unix_socket(conn_path) + # On Windows the first poll starts ConnectNamedPipe. Arm it before a + # fast client can connect and close, leaving an unaccepted closed pipe. + processx::poll(list(server), 0) + # Signal readiness only after socket creation (including listen) returns. # The Unix socket path can appear before the server is listening. writeLines("ready", ready_file) diff --git a/r-pkg/tests/testthat/test-mock-server.R b/r-pkg/tests/testthat/test-mock-server.R index c8f93ea..e3ee279 100644 --- a/r-pkg/tests/testthat/test-mock-server.R +++ b/r-pkg/tests/testthat/test-mock-server.R @@ -63,20 +63,21 @@ test_that("collection reports a terminated subprocess for both transports", { }) }) -test_that("local readiness waits for socket creation to return", { +test_that("local readiness waits for socket creation and connection polling", { skip_on_cran() skip_if_not_installed("callr") skip_if_not_installed("processx") skip_if_not_installed("jsonlite") returned_file = withr::local_tempfile() + polled_file = withr::local_tempfile() files = new.env(parent = emptyenv()) r_bg = callr::r_bg local_mocked_bindings( r_bg = function(func, args, ...) { files$ready = args$ready_file r_bg( - function(func, args, returned_file) { + function(func, args, returned_file, polled_file) { create_socket = processx::conn_create_unix_socket # Delay the constructor after the socket/pipe exists. The parent must # still wait until the constructor returns and publishes readiness. @@ -90,9 +91,25 @@ test_that("local readiness waits for socket creation to return", { }, ns = "processx" ) + poll_connections = processx::poll + assignInNamespace( + "poll", + function(processes, ms) { + # Let the client connect and close before the blocking poll. + # Windows must already have a pending ConnectNamedPipe request. + if (ms == 30000) Sys.sleep(0.5) + result = poll_connections(processes, ms) + if (ms == 0) writeLines("polled", polled_file) + result + }, + ns = "processx" + ) do.call(func, args) }, - args = list(func = func, args = args, returned_file = returned_file), + args = list( + func = func, args = args, + returned_file = returned_file, polled_file = polled_file + ), ... ) }, @@ -103,6 +120,7 @@ test_that("local readiness waits for socket creation to return", { server = start_mock_server_local() withr::defer(if (server$bg$is_alive()) server$bg$kill()) expect_true(file.exists(returned_file)) + expect_true(file.exists(polled_file)) expect_true(file.exists(files$ready)) jgd(socket = server$socket_path)