skip_other_platforms <- function(platform) { if (os_type() != platform) skip(paste("only run it on", platform)) } skip_if_no_tool <- function(tool) { if (Sys.which(tool) == "") skip(paste0("`", tool, "` is not available")) } skip_extra_tests <- function() { if (Sys.getenv("PROCESSX_EXTRA_TESTS") == "") skip("no extra tests") } skip_if_no_ps <- function() { if (!requireNamespace("ps", quietly = TRUE)) skip("ps package needed") if (!ps::ps_is_supported()) skip("ps does not support this platform") } try_silently <- function(expr) { tryCatch( expr, error = function(x) "error", warning = function(x) "warning", message = function(x) "message" ) } get_pid_by_name <- function(name) { if (os_type() == "windows") { get_pid_by_name_windows(name) } else if (is_linux()) { get_pid_by_name_linux(name) } else { get_pid_by_name_unix(name) } } get_pid_by_name_windows <- function(name) { ## TODO } ## Linux does not exclude the ancestors of the pgrep process ## from the list, so we have to do that manually. We remove every ## process that contains 'pgrep' in its command line, which is ## not the proper solution, but for testing it will do. ## ## Unfortunately Ubuntu 12.04 pgrep does not have a -a switch, ## so we cannot just output the full command line and then filter ## it in R. So we first run pgrep to get all matching process ids ## (without their command lines), and then use ps to list the processes ## again. At this time the first pgrep process is not running any ## more, but another process might have its id, so we filter again the ## result for 'name' get_pid_by_name_linux <- function(name) { ## TODO } skip_in_covr <- function() { if (Sys.getenv("R_COVR", "") == "true") skip("in covr") } httpbin <- webfakes::new_app_process( webfakes::httpbin_app(), opts = webfakes::server_opts(num_threads = 6) ) interrupt_me <- function(expr, after = 1) { tryCatch({ p <- callr::r_bg(function(pid, after) { Sys.sleep(after) ps::ps_interrupt(ps::ps_handle(pid)) }, list(pid = Sys.getpid(), after = after)) expr p$kill() }, interrupt = function(e) e) } expect_error <- function(..., class = "error") { testthat::expect_error(..., class = class) } local_temp_dir <- function(pattern = "file", tmpdir = tempdir(), fileext = "", envir = parent.frame()) { path <- tempfile(pattern = pattern, tmpdir = tmpdir, fileext = fileext) dir.create(path) withr::local_dir(path, .local_envir = envir) withr::defer(unlink(path, recursive = TRUE), envir = envir) invisible(path) } has_locale <- function(l) { has <- TRUE tryCatch( withr::with_locale(c(LC_CTYPE = l), "foobar"), warning = function(w) has <<- FALSE, error = function(e) has <<- FALSE ) has } run_script <- function(expr, ..., quoted = NULL, encoding = "") { dir.create(dir <- tempfile()) sf <- file.path(dir, "script.R") sf2 <- file.path(dir, "script2.R") so <- paste0(sf, "out") se <- paste0(sf, "err") on.exit(unlink(c(dir), recursive = TRUE), add = TRUE) if (is.null(quoted)) quoted <- substitute(expr) writeLines(deparse(quoted), con = sf) writeLines( deparse(substitute({ options(keep.source = TRUE) source(sf) }, list(sf = basename(sf)))), con = sf2 ) out <- callr::rscript( basename(sf2), stdout = so, stderr = se, fail_on_status = FALSE, show = FALSE, wd = dirname(sf) ) enc <- function(x) iconv(list(x), encoding, "UTF-8") list( script = readLines(sf), stdout = enc(readBin(so, "raw", file.size(so))), stderr = enc(readBin(se, "raw", file.size(se))), status = out$status ) } scrub_px <- function(x) { sub("'px.exe'", "'px'", x, fixed = TRUE) } scrub_srcref <- function(x) { x <- sub(" at cnd-abort.R:[0-9]+:[0-9]+", "", x) x <- sub(" at standalone-errors.R:[0-9]+:[0-9]+", "", x) x <- sub(" at run.R:[0-9]+:[0-9]+", "", x) x <- sub("\033[90m\033[39m", "", x, fixed = TRUE) x } err$register_testthat_print()