tests/testthat/helpers.R

In ps: List, Query, Manipulate System Processes

format_regexp <- function() {
  "<ps::ps_handle> PID=[0-9]+, NAME=.*, AT="
}

parse_ps <- function(args) {
  out <- processx::run("ps", args)$stdout
  sub(" *$", "", strsplit(out, "\n")[[1]][[2]])
}

parse_time <- function(x) {
  x <- utils::tail(c(0, 0, 0, as.numeric(strsplit(x, ":")[[1]])), 3)
  x[1] * 60 * 60 + x[2] * 60 + x[3]
}

wait_for_status <- function(ps, status, timeout = 5) {
  limit <- Sys.time() + timeout
  while (ps_status(ps) != status && Sys.time() < limit) {
    Sys.sleep(0.05)
  }
}

px <- function() get_tool("px")

skip_in_rstudio <- function() {
  if (Sys.getenv("RSTUDIO") != "") skip("Cannot test in RStudio")
}

has_processx <- function() {
  requireNamespace("processx", quietly = TRUE) &&
    package_version(getNamespaceVersion("processx")) >= "3.1.0.9005"
}

skip_if_no_processx <- function() {
  if (!has_processx()) skip("Needs processx >= 3.1.0.9005 to run")
}

skip_without_program <- function(prog) {
  if (Sys.which(prog) == "") skip(paste(prog, "is not available"))
}

have_ipv6_support <- function() {
  ps_os_type()[["WINDOWS"]] ||
    !is.null(ps_env$constants$address_families$AF_INET6)
}

skip_without_ipv6 <- function() {
  if (!have_ipv6_support()) skip("Needs IPv6")
}

ipv6_url <- function() {
  paste0("https://", ipv6_host())
}

ipv6_host <- function() {
  "ipv6.test-ipv6.com"
}

have_ipv6_connection <- local({
  ok <- NULL
  myurl <- NULL
  function(url = ipv6_url()) {
    if (is.null(ok) || myurl != url) {
      myurl <<- url
      opt <- options(warn = 2)
      on.exit(options(opt), add = TRUE)
      tryCatch(
        {
          cx <- curl::curl(url)
          open(cx)
          ok <<- TRUE
        },
        error = function(x) ok <<- FALSE,
        finally = close(cx)
      )
    }
    ok
  }
})

skip_without_ipv6_connection <- function() {
  if (!have_ipv6_connection()) skip("Needs working IPv6 connection")
}

wait_for_string <- function(proc, string, timeout) {
  deadline <- Sys.time() + as.difftime(timeout / 1000, units = "secs")
  str <- ""
  repeat {
    left <- max(as.double(deadline - Sys.time(), units = "secs"), 0)
    pr <- processx::poll(list(proc), as.integer(left * 1000))
    str <- paste(str, proc$read_error())
    if (grepl(string, str)) {
      return()
    }
    if (proc$has_output_connection()) {
      read_output()
    }
    if (deadline < Sys.time()) {
      stop("Cannot start proces")
    }
    if (!proc$is_alive()) stop("Cannot start process")
  }
}

## This is not perfect, e.g. we don't check that the numbers are <255,
## but will do for our purposes

is_ipv4_address <- function(x) {
  grepl("^[0-9]+\\.[0-9]+\\.[0-9]+\\.[0-9]+$", x)
}

cleanup_process <- function(p) {
  tryCatch(close(p$get_input_connection()), error = function(x) x)
  tryCatch(close(p$get_output_connection()), error = function(x) x)
  tryCatch(close(p$get_error_connection()), error = function(x) x)
  tryCatch(close(p$get_poll_connection()), error = function(x) x)
  tryCatch(p$kill(), error = function(x) x)
}

httpbin <- webfakes::new_app_process(
  webfakes::httpbin_app(),
  opts = webfakes::server_opts(num_threads = 6)
)

capture_success_failure <- function(expr) {
  cnd <- NULL
  n_success <- 0
  n_failure <- 0
  failures <- list()
  withCallingHandlers(
    expr,
    expectation_failure = function(cnd) {
      failures[[length(failures) + 1]] <<- cnd
      n_failure <<- n_failure + 1
      invokeRestart("continue_test")
    },
    expectation_success = function(cnd) {
      n_success <<- n_success + 1
      invokeRestart("continue_test")
    }
  )
  list(
    n_success = n_success,
    n_failure = n_failure,
    failures = failures
  )
}