Add tests to increase test coverage.

The following files now have 100% test coverage:

 * R/as.cluster.R
 * R/availableConnections.R
 * R/isConnectionValid.R
 * R/isForkedNode.R
 * R/serializedSize.R
 * R/stealth_sample.R

Increased coverage for:

 * R/makeClusterSequential.R [95.74%]

Author: HenrikBengtsson

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: parallelly
-Version: 1.45.1-9012
+Version: 1.45.1-9013
 Title: Enhancing the 'parallel' Package
 Imports:
     parallel,

R/makeClusterSequential.R

@@ -73,7 +73,6 @@ sendData.sequential_node <- function(node, data) {
     data <- data[["data"]]  ## sic!
     fun <- data[["fun"]]
     args <- data[["args"]]
-    ret <- data[["return"]]
 
     ## Don't evaluate in the global environment, which is the default
     if (identical(args[["envir"]], globalenv())) {

inst/testme/_prologue/090.context.R

@@ -8,5 +8,3 @@ on_solaris <- grepl("^solaris", R.version[["os"]])
 
 covr_testing <- ("covr" %in% loadedNamespaces())
 on_githubactions <- isTRUE(as.logical(Sys.getenv("GITHUB_ACTIONS")))
-
-useXDR <- isTRUE(parallelly:::getOption2("parallelly.makeNodePSOCK.useXDR"))

inst/testme/test-as.cluster.R

@@ -1,6 +1,7 @@
 library(parallelly)
 
 stopCluster <- parallel::stopCluster
+useXDR <- isTRUE(parallelly:::getOption2("parallelly.makeNodePSOCK.useXDR"))
 
 message("*** cluster operations ...")
 
@@ -69,3 +70,25 @@ if (length(remotes) > 0) {
 message("*** cluster operations - makeClusterPSOCK(remotes) ... DONE")
 
 message("*** cluster operations ... DONE")
+
+message("*** cluster operations - as.cluster(<non-cluster>) ...")
+res <- tryCatch({
+  as.cluster(1)
+}, error = function(ex) ex)
+stopifnot(inherits(res, "error"))
+message("*** cluster operations - as.cluster(<non-cluster>) ... DONE")
+
+message("*** cluster operations - c(...) with duplicated nodes ...")
+cl_dup1 <- makeClusterPSOCK(1L)
+on.exit(stopCluster(cl_dup1), add = TRUE)
+cl_combined <- c(cl_dup1, cl_dup1)
+stopifnot(inherits(cl_combined, "cluster"), length(cl_combined) == 2L)
+message("*** cluster operations - c(...) with duplicated nodes ... DONE")
+
+message("*** cluster operations - as.cluster(SOCKnode) ...")
+cl_base <- parallel::makeCluster(1L, type = "PSOCK")
+node_base <- cl_base[[1]]
+res_base <- as.cluster(node_base)
+stopifnot(inherits(res_base, "cluster"), inherits(res_base[[1]], "SOCKnode"))
+parallel::stopCluster(cl_base)
+message("*** cluster operations - as.cluster(SOCKnode) ... DONE")

inst/testme/test-availableConnections.R

@@ -0,0 +1,52 @@
+library(parallelly)
+
+message("availableConnections() ...")
+
+## Test with overriding option
+options(parallelly.availableConnections = 200L)
+n <- availableConnections()
+stopifnot(n == 200L)
+options(parallelly.availableConnections = NULL)
+
+## Reset memoized value
+assign("max", NULL, envir = environment(availableConnections))
+
+## Test with invalid option
+res <- tryCatch({
+  options(parallelly.availableConnections = "abc")
+  availableConnections()
+}, error = function(e) e)
+stopifnot(inherits(res, "simpleError"))
+options(parallelly.availableConnections = NULL)
+
+## Reset memoized value
+assign("max", NULL, envir = environment(availableConnections))
+
+## Test with a small number of tries to trigger +Inf
+options(parallelly.availableConnections.tries = 10L)
+n <- availableConnections()
+stopifnot(is.infinite(n))
+options(parallelly.availableConnections.tries = NULL)
+
+## Reset memoized value
+assign("max", NULL, envir = environment(availableConnections))
+
+## Test with invalid tries option
+res <- tryCatch({
+  options(parallelly.availableConnections.tries = -1L)
+  availableConnections()
+}, error = function(e) e)
+stopifnot(inherits(res, "simpleError"))
+options(parallelly.availableConnections.tries = NULL)
+
+## Reset memoized value
+assign("max", NULL, envir = environment(availableConnections))
+
+## Basic functionality
+n <- availableConnections()
+stopifnot(is.integer(n) || is.infinite(n), n >= 3L)
+
+f <- freeConnections()
+stopifnot(is.integer(f) || is.infinite(f), f <= n)
+
+message("availableConnections() ... done")

inst/testme/test-isConnectionValid.R

@@ -1,52 +1,68 @@
 library(parallelly)
 
-stopCluster <- parallel::stopCluster
-
-message("*** Connections ...")
-
-con <- stdin()
-idx <- as.integer(con)
-id <- connectionId(con)
-valid <- isConnectionValid(con)
-stopifnot(inherits(con, "connection"), idx == 0L, id == 0L, isTRUE(valid))
-
-con <- stdout()
-idx <- as.integer(con)
-id <- connectionId(con)
-valid <- isConnectionValid(con)
-stopifnot(inherits(con, "connection"), idx == 1L, id == 1L, isTRUE(valid))
-
-con <- stderr()
-idx <- as.integer(con)
-id <- connectionId(con)
-valid <- isConnectionValid(con)
-stopifnot(inherits(con, "connection"), idx == 2L, id == 2L, isTRUE(valid))
-
-message("- Connections cannot be serialized")
-
-con <- file(tempfile(), open = "w")
-x <- list(value = 42, stderr = stderr(), con = con)
-y <- unserialize(serialize(x, connection = NULL))
-print(connectionId(x$stderr))
-print(connectionId(x$con))
-print(isConnectionValid(x$stderr))
-print(isConnectionValid(x$con))
-print(connectionId(y$stderr))
-print(connectionId(y$con))
-print(isConnectionValid(y$stderr))
-print(isConnectionValid(y$con))
-
-stopifnot(
-  identical(y$value, x$value),
-  connectionId(x$stderr) == 2L,
-  isConnectionValid(x$stderr),
-  isConnectionValid(y$stderr),
-  identical(connectionId(y$stderr), connectionId(x$stderr)),
-  connectionId(x$con) >= 3L,
-  isConnectionValid(x$con),
-  connectionId(y$con) == -1L,
-  !isConnectionValid(y$con)
-)
-close(con)
-
-message("*** Connections ... DONE")
+message("isConnectionValid() ...")
+
+## Test connectionId()
+
+## Test with stdin, stdout, stderr
+stopifnot(connectionId(stdin()) == 0L)
+stopifnot(connectionId(stdout()) == 1L)
+stopifnot(connectionId(stderr()) == 2L)
+
+## Test with a connection that has no conn_id attribute
+con_no_id <- file(tempfile(), open = "w+")
+class(con_no_id) <- c("test_connection", class(con_no_id)) ## Inherit from connection
+attr(con_no_id, "conn_id") <- NULL ## Ensure no conn_id
+res <- connectionId(con_no_id)
+stopifnot(is.na(res))
+close(con_no_id)
+
+## Test with a serialized connection (should return -1L)
+con_ser <- file(tempfile(), open = "w")
+x_ser <- list(con = con_ser)
+y_ser <- unserialize(serialize(x_ser, connection = NULL))
+stopifnot(connectionId(y_ser$con) == -1L)
+close(con_ser)
+
+## Test with a valid file connection
+con_valid <- file(tempfile(), open = "w+")
+id_valid <- connectionId(con_valid)
+stopifnot(is.integer(id_valid), id_valid >= 3L)
+close(con_valid)
+
+
+## Test isConnectionValid()
+
+## Test with stdin, stdout, stderr
+stopifnot(isConnectionValid(stdin()))
+stopifnot(isConnectionValid(stdout()))
+stopifnot(isConnectionValid(stderr()))
+
+## Test with a serialized connection (should be FALSE)
+con_ser_valid <- file(tempfile(), open = "w")
+x_ser_valid <- list(con = con_ser_valid)
+y_ser_valid <- unserialize(serialize(x_ser_valid, connection = NULL))
+res_ser_valid <- isConnectionValid(y_ser_valid$con)
+stopifnot(!res_ser_valid, inherits(attr(res_ser_valid, "reason"), "character"))
+close(con_ser_valid)
+
+## Test with a valid connection
+con_real_valid <- file(tempfile(), open = "w+")
+stopifnot(isConnectionValid(con_real_valid))
+close(con_real_valid)
+
+
+## Test with a connection where index is not found in getAllConnections()
+# 1. Create a connection
+con_temp_file <- file(tempfile(), open = "w+")
+
+# 2. Close it, so its index is no longer in getAllConnections()
+close(con_temp_file)
+
+# 3. Now `con_temp_file` is a "zombie" object - its index is preserved
+#    but `is.element(as.integer(con_temp_file), getAllConnections())` is FALSE.
+res_non_existent <- isConnectionValid(con_temp_file)
+stopifnot(!res_non_existent, inherits(attr(res_non_existent, "reason"), "character"))
+
+
+message("isConnectionValid() ... done")

inst/testme/test-isForkedNode.R

@@ -0,0 +1,38 @@
+isForkedNode <- parallelly:::isForkedNode
+
+message("isForkedNode() ...")
+
+# 1. Test with a default object
+x <- 123
+res <- isForkedNode(x)
+stopifnot(is.na(res))
+
+x <- list(a = 1, b = 2)
+res <- isForkedNode(x)
+stopifnot(is.na(res))
+
+
+# 2. Test with a forknode object
+## Create a dummy forknode object
+## It's enough to set its class attribute
+node <- structure(list(), class = "forknode")
+res <- isForkedNode(node)
+stopifnot(res)
+
+
+# 3. Test with a cluster object
+## Create dummy cluster objects
+node1 <- structure(list(), class = "forknode")
+node2 <- structure(list(), class = "notforknode") ## A dummy node that is not a forknode
+node3 <- structure(list(), class = "forknode")
+
+cluster_obj <- list(node1, node2, node3)
+class(cluster_obj) <- "cluster"
+
+res <- isForkedNode(cluster_obj)
+stopifnot(is.logical(res), length(res) == 3)
+stopifnot(res[1])
+stopifnot(is.na(res[2]))
+stopifnot(res[3])
+
+message("isForkedNode() ... done")

inst/testme/test-makeClusterPSOCK.R

@@ -5,6 +5,8 @@ is_ip_number <- parallelly:::is_ip_number
 is_localhost <- parallelly:::is_localhost
 find_rshcmd <- parallelly:::find_rshcmd
 
+useXDR <- isTRUE(parallelly:::getOption2("parallelly.makeNodePSOCK.useXDR"))
+
 message("*** makeClusterPSOCK() ...")
 
 message("- makeClusterPSOCK() - internal utility functions")

inst/testme/test-makeClusterSequential.R

@@ -1,31 +1,98 @@
-if (getRversion() >= "4.4.0") {
+message("makeClusterSequential() ...")
+
+## Requires R (>= 4.4.0)
+if (getRversion() < "4.4.0") {
+  message("Skipping because R version is less than 4.4.0")
+  message("makeClusterSequential() ... skipped")
+} else {
   library(parallelly)
+
+  ## Temporarily override parallel::makeCluster and parallel::stopCluster
+  ## to use the sequential versions
+  old_makeCluster <- parallel::makeCluster
+  old_stopCluster <- parallel::stopCluster
+  on.exit({
+    unlockBinding("makeCluster", getNamespace("parallel"))
+    assign("makeCluster", old_makeCluster, envir = getNamespace("parallel"))
+    unlockBinding("stopCluster", getNamespace("parallel"))
+    assign("stopCluster", old_stopCluster, envir = getNamespace("parallel"))
+  }, add = TRUE)
+
+  unlockBinding("makeCluster", getNamespace("parallel"))
+  assign("makeCluster", makeClusterSequential, envir = getNamespace("parallel"))
+  unlockBinding("stopCluster", getNamespace("parallel"))
+  assign("stopCluster", function(cl) {
+    ## Do nothing for sequential_cluster, as it's not a real parallel cluster
+    ## Except for invalidating the node state
+    for (node in cl) {
+      if (inherits(node, "sequential_node")) {
+        node$envir[["...parallelly.valid..."]] <- FALSE
+      }
+    }
+  }, envir = getNamespace("parallel"))
+
   library(parallel)
-  
+
   cl <- makeClusterSequential()
-  print(cl)
+  stopifnot(inherits(cl, "sequential_cluster"), length(cl) == 1L)
 
-  y_truth <- lapply(X = 1:3, FUN = sqrt)
+  ## Test print.sequential_cluster
+  capture.output(print(cl))
+
+  ## Test print.sequential_node
+  capture.output(print(cl[[1]]))
+
+  ## Test basic parLapply functionality
   y <- parLapply(cl, X = 1:3, fun = sqrt)
-  str(y)
-  stopifnot(identical(y, y_truth))
-  
-  pid <- Sys.getpid()
-  print(pid)
-  y <- clusterEvalQ(cl, Sys.getpid())
-  str(y)
-  stopifnot(identical(y[[1]], pid))
-  
+  stopifnot(identical(y, lapply(1:3, sqrt)))
+
+  ## Test clusterEvalQ with side effects
+  abc <- 3.14
   y <- clusterEvalQ(cl, { abc <- 42; abc })
-  str(y)
-  stopifnot(!exists("abc", inherits = FALSE))
+  stopifnot(identical(y, list(42)), abc == 3.14)
 
+  ## Test stopping the cluster
   stopCluster(cl)
-  print(cl)
+
+  ## Test accessing an invalid node (after stopCluster)
+  res <- tryCatch({
+    parallel::sendData(cl[[1]], list(type = "EXEC", data = list(fun = identity, args = list(1))))
+  }, error = function(e) e)
+  stopifnot(inherits(res, "simpleError"), grepl("node is no longer valid", res$message))
+
+
+  ## Re-create cluster for other tests
+  cl <- makeClusterSequential()
+  node <- cl[[1]]
+
+  ## Test sendData with "DONE" type
+  parallel::sendData(node, list(type = "DONE"))
+  stopifnot(!isTRUE(node$envir[["...parallelly.valid..."]]))
+
+  ## Test sendData with unknown type
+  ## Test sendData with unknown type
+  ## Re-create cluster for this test, as the previous one was invalidated
+  cl_unknown <- makeClusterSequential()
+  node_unknown <- cl_unknown[[1]]
 
   res <- tryCatch({
-    y <- clusterEvalQ(cl, { 42 })
-  }, error = identity)
-  print(res)
-  stopifnot(inherits(res, "error"))
-} ## if (getRversion() >= "4.4.0")
+    parallel::sendData(node_unknown, list(type = "UNKNOWN"))
+  }, error = function(e) e)
+  message("Error message for UNKNOWN type: ", res$message)
+  stopifnot(inherits(res, "simpleError"), grepl("type = ['‘]UNKNOWN['’] not yet implemented", res$message))
+
+  ## Test recvData internal error
+  ## Re-create cluster for this test
+  cl_internal_error <- makeClusterSequential()
+  node_internal_error <- cl_internal_error[[1]]
+
+  ## Manipulate internal state to trigger error
+  node_internal_error$envir[["value"]] <- list(type = "BAD_TYPE")
+
+  res <- tryCatch({
+    parallel::recvData(node_internal_error)
+  }, error = function(e) e)
+  stopifnot(inherits(res, "simpleError"), grepl("INTERNAL ERROR", res$message))
+
+  message("makeClusterSequential() ... done")
+}