# minitest - a minimal testing framework v0.0.2 --------------------------------
test_library <- function(package) library(package = package, character.only = TRUE)
test_true <- function(x) invisible(isTRUE(x) || {print(x); stop("the above was returned instead of TRUE")})
test_null <- function(x) invisible(is.null(x) || {print(x); stop("the above was returned instead of NULL")})
test_zero <- function(x) invisible(x == 0L || {print(x); stop("the above was returned instead of 0L")})
test_type <- function(type, x) invisible(typeof(x) == type || {stop("object of type '", typeof(x), "' was returned instead of '", type, "'")})
test_class <- function(class, x) invisible(inherits(x, class) || {stop("object of class '", paste(class(x), collapse = ", "), "' was returned instead of '", class, "'")})
test_equal <- function(a, b) invisible(a == b || {print(a); print(b); stop("the above expressions were not equal")})
test_identical <- function(a, b) invisible(identical(a, b) || {print(a); print(b); stop("the above expressions were not identical")})
test_print <- function(x) invisible(is.character(capture.output(print(x))) || stop("print output of expression cannot be captured as a character value"))
test_error <- function(x, containing = "") invisible(inherits(x <- tryCatch(x, error = identity), "error") && grepl(containing, x[["message"]], fixed = TRUE) || stop("Expected error message containing: ", containing, "\nActual error message: ", x[["message"]]))
# ------------------------------------------------------------------------------

test_library("mirai")
connection <- !is_error_value(collect_mirai(mirai(TRUE, .timeout = 2000L)))
# core tests
test_type("list", status())
test_zero(status()[["connections"]])
test_zero(status()[["daemons"]])
test_zero(daemons(0L))
test_error(mirai(), "missing expression, perhaps wrap in {}?")
test_error(mirai(a, 1), "all '...' arguments must be named")
test_error(mirai(a, .args = list(1)), "all items in '.args' must be named")
test_error(mirai_map(1:2, "a function"), "must be of type function, not character")
test_error(daemons(url = "URL"), "Invalid argument")
test_error(daemons(-1), "zero or greater")
test_error(daemons(raw(0L)), "must be numeric")
test_error(daemons(1, dispatcher = "p"), "must be either")
test_error(daemons(url = local_url(), dispatcher = "t"), "must be either")
test_error(dispatcher(client = "URL"), "must be 1 or greater")
test_error(daemon("URL"), "Invalid argument")
test_error(launch_local(1L), "daemons must be set")
test_type("character", host_url())
test_true(startsWith(host_url(tls = TRUE), "tls"))
test_true(grepl("5555", host_url(port = 5555), fixed = TRUE))
test_type("list", ssh_config("ssh://remotehost"))
test_type("list", ssh_config("ssh://remotehost", tunnel = TRUE, port = "5555"))
test_true(is_mirai_interrupt(r <- mirai:::mk_interrupt_error()))
test_print(r)
test_true(is_mirai_error(r <- `class<-`("Error in: testing\n", c("miraiError", "errorValue", "try-error"))))
test_print(r)
test_null(r$stack.trace)
test_equal(mirai:::.DollarNames.miraiError(r, "c"), "class")
test_true(mirai:::is.promising.mirai())
test_error(everywhere({}), "not found")
test_null(nextstream())
test_null(nextget("pid"))
for (i in 0:4)
  test_type("character", nextcode(i))
# mirai and daemons tests
connection && {
  Sys.sleep(1L)
  n <- function() m
  m <- mirai({
    Sys.sleep(0.1)
    q <- m + n() + 2L
    q / m
  }, m = 2L, .args = environment(), .timeout = 2000L)
  test_identical(call_mirai(m), m)
  if (!is_error_value(m$data)) test_equal(m$data, 3L)
  Sys.sleep(1L)
  `lang obj` <- quote(m + n + 2L)
  args <- c(m = 2L, n = 4L)
  m <- mirai(.expr = `lang obj`, .args = args, .timeout = 2000L)
  if (!is_error_value(call_mirai_(m)$data)) test_equal(m$data, 8L)
  test_true(!stop_mirai(m))
  Sys.sleep(1L)
  test_equal(1L, d <- daemons(1L, dispatcher = FALSE, asyncdial = FALSE, seed = 1546L))
  test_print(d)
  me <- mirai(mirai::mirai(), .timeout = 2000L)[]
  if (!is_mirai_error(me)) test_true(is_error_value(me))
  if (is_mirai_error(me)) test_type("list", me$stack.trace)
  if (is_mirai_error(me)) test_true(length(me$stack.trace) >= 2L)
  if (is_mirai_error(me)) test_true(all(as.logical(lapply(me$stack.trace, is.character))))
  test_true(!is_mirai_interrupt(me))
  test_class("errorValue", me)
  test_print(me)
  df <- data.frame(a = 1, b = 2)
  dm <- mirai(as.matrix(df), .args = list(df = df), .timeout = 2000L)
  test_true(is_mirai(call_mirai(dm)))
  test_true(!unresolved(dm))
  if (!is_error_value(dm$data)) test_class("matrix", dm$data)
  test_print(dm)
  test_type("integer", status()[["connections"]])
  test_type("character", status()[["daemons"]])
  test_type("character", mlc <- launch_remote())
  test_class("miraiLaunchCmd", mlc)
  test_print(mlc)
  test_error(launch_remote(1L, remote = remote_config(command = "echo", args = "invalid")), "must be an element")
  test_error(launch_remote(3L, remote = remote_config(command = "echo", args = list(c("test", "."), c("test", ".")))), "must equal the length")
  test_zero(daemons(0L))
  Sys.sleep(1L)
  test_equal(1L, daemons(1L, dispatcher = FALSE, maxtasks = 10L, walltime = 10000L, idletime = 20000L, cleanup = FALSE, output = TRUE, .compute = "new"))
  test_type("character", nextget("urls", .compute = "new"))
  test_type("integer", nextstream(.compute = "new"))
  Sys.sleep(1.5)
  test_type("list", everywhere({}, as.environment(df), .compute = "new"))
  mn <- mirai("test1", .compute = "new")
  mp <- mirai(b + 1, .compute = "new")
  Sys.sleep(1L)
  if (!unresolved(mn$data)) test_equal(mn$data, "test1")
  if (!unresolved(mp$data)) test_equal(mp$data, 3)
  Sys.sleep(1L)
  test_type("integer", status(.compute = "new")[["connections"]])
  test_zero(daemons(0L, .compute = "new"))
}
# additional daemons tests
connection && .Platform[["OS.type"]] != "windows" && {
  Sys.sleep(1L)
  test_zero(daemons(url = value <- local_url(), dispatcher = FALSE))
  test_identical(status()$daemons, value)
  test_identical(nextget("urls"), value)
  test_type("character", launch_remote(remote = remote_config(command = "echo", args = list(c("Test out:", ".", ">/dev/null")), rscript = "/usr/lib/R/bin/Rscript")))
  test_type("character", launch_remote(remote = ssh_config(remotes = c("ssh://remotehost", "ssh://remotenode"), tunnel = TRUE, port = "5555", command = "echo")))
  test_zero(daemons(0L))
  Sys.sleep(1L)
  test_zero(daemons(n = 2L, url = value <- "ws://:0", dispatcher = FALSE, remote = remote_config(quote = TRUE)))
  test_true(status()$daemons != value)
  test_zero(daemons(0L))
}
# mirai_map tests
connection && .Platform[["OS.type"]] != "windows" && {
  Sys.sleep(1L)
  m <- with(daemons(1, dispatcher = "none", .compute = "ml"), {
    if (is.null(tryCatch(mirai_map(list(1, "a", 2), sum, .compute = "ml")[.stop], error = function(e) NULL)))
      mirai_map(1:3, rnorm, .args = list(mean = 20, 2), .compute = "ml")[]
  })
  test_true(!is_mirai_map(m))
  test_type("list", m)
  test_equal(length(m), 3L)
  test_true(all(as.logical(lapply(m, is.numeric))))
  Sys.sleep(1L)
  test_print(suppressWarnings(mp <- mirai_map(list(x = "a"), function(...) do(...), do = function(x, y) sprintf("%s%s", x, y), .args = list("b"))))
  test_identical(collect_mirai(mp)[["x"]], "ab")
  test_identical(call_mirai(mp)[["x"]][["data"]], "ab")
  test_true(all(mirai_map(data.frame(1:3, 3:1), sum, .args = list(3L))[.flat] == 7L))
  test_true(all(mirai_map(list(c(a = 1, b = 1, c = 1), 3), sum)[.flat] == 3))
  test_zero(daemons(0L))
}
# parallel cluster tests
library(parallel)
test_null(tryCatch(mirai::register_cluster(), error = function(e) NULL))
connection && {
  Sys.sleep(1L)
  cluster <- make_cluster(1)
  test_class("miraiCluster", cluster)
  test_class("cluster", cluster)
  test_equal(length(cluster), 1L)
  test_class("miraiNode", cluster[[1]])
  test_print(cluster[[1]])
  test_type("list", cluster[1])
  test_type("character", launch_remote(cluster))
  test_type("character", launch_remote(cluster[[1L]]))
  test_type("list", status(cluster))
  clusterSetRNGStream(cluster, 123)
  j <- clusterEvalQ(cluster, expr = .GlobalEnv[[".Random.seed"]])
  a <- parSapply(cluster, 1:4, runif)
  setDefaultCluster(cluster)
  res <- parLapply(X = 1:10, fun = rnorm)
  test_type("list", res)
  test_equal(length(res), 10L)
  test_type("double", res[[1L]])
  test_equal(length(res[[1L]]), 1L)
  test_type("double", res[[10L]])
  test_equal(length(res[[10L]]), 10L)
  res <- parLapplyLB(X = 1:10, fun = rnorm)
  test_type("list", res)
  test_equal(length(res), 10L)
  test_type("double", res[[1L]])
  test_equal(length(res[[1L]]), 1L)
  test_type("double", res[[10L]])
  test_equal(length(res[[10L]]), 10L)
  test_identical(parSapply(NULL, 1:4, factorial), c(1, 2, 6, 24))
  test_identical(parSapplyLB(NULL, 1:8, factorial), c(1, 2, 6, 24, 120, 720, 5040, 40320))
  df <- data.frame(a = c(1, 2, 3), b = c(6, 7, 8))
  test_identical(parApply(cluster, df, 2, sum), `names<-`(c(6, 21), c("a", "b")))
  test_identical(parCapply(cluster, df, sum), `names<-`(c(6, 21), c("a", "b")))
  test_identical(parRapply(cluster, df, sum), `names<-`(c(7, 9, 11), c("1", "2", "3")))
  res <- clusterEvalQ(expr = .GlobalEnv[[".Random.seed"]][[1L]])
  test_type("integer", res[[1L]])
  test_error(clusterEvalQ(cluster, elephant()), "Error in elephant(): could not find function \"elephant\"")
  test_null(stop_cluster(cluster))
  Sys.sleep(1L)
  test_class("miraiCluster", cl <- make_cluster(1))
  test_true(attr(cl, "id") != attr(cluster, "id"))
  clusterSetRNGStream(cl, 123)
  k <- clusterEvalQ(cl, expr = .GlobalEnv[[".Random.seed"]])
  b <- parSapply(cl, 1:4, runif)
  test_identical(j, k)
  test_identical(a, b)
  test_identical(clusterApply(cl, 1:2, get("+"), 3), list(4, 5))
  xx <- 1
  clusterExport(cl, "xx", environment())
  test_identical(clusterCall(cl, function(y) xx + y, 2), list(3))
  test_identical(clusterMap(cl, function(x, y) seq_len(x) + y, c(a =  1, b = 2, c = 3), c(A = 10, B = 0, C = -10)),
            list(a = 11, b = c(1, 2), c = c(-9, -8, -7)))
  test_identical(parSapply(cl, 1:20, get("+"), 3), as.double(4:23))
  test_null(stopCluster(cl))
  test_error(parLapply(cluster, 1:10, runif), "cluster is no longer active")
  Sys.sleep(1L)
  test_print(cl <- make_cluster(url = local_url()))
  test_null(stopCluster(cl))
  Sys.sleep(1L)
  test_print(cl <- make_cluster(n = 1, url = local_url(), remote = remote_config()))
  test_null(stopCluster(cl))
}
# advanced daemons and dispatcher tests
connection && .Platform[["OS.type"]] != "windows" && Sys.getenv("NOT_CRAN") == "true" && {
  Sys.sleep(1L)
  test_zero(daemons(url = "ws://:0", correctype = 0L, token = TRUE))
  test_zero(daemons(0L))
  test_zero(with(daemons(url = "tcp://:0", correcttype = c(1, 0), token = TRUE), {8L - 9L + 1L}))
  test_zero(daemons(n = 2, "ws://:0"))
  test_type("integer", nextget("pid"))
  test_equal(length(nextget("urls")), 1L)
  Sys.sleep(1L)
  status <- daemons()
  test_type("list", status)
  test_zero(status[["connections"]])
  test_type("integer", status[["mirai"]])
  test_zero(status[["mirai"]][["awaiting"]])
  test_zero(status[["mirai"]][["executing"]])
  test_zero(status[["mirai"]][["completed"]])
  test_zero(daemons(0))
  test_equal(daemons(2, correcttype = NA), 2L)
  test_equal(daemons()[["connections"]], 2L)
  test_type("list", res <- mirai_map(c(1,1), rnorm)[.progress])
  test_true(res[[1L]] != res[[2L]])
  test_equal(2L, daemons()[["connections"]])
  test_zero(daemons(0L))
  Sys.sleep(1L)
  test_zero(daemons(url = "tls+tcp://127.0.0.1:0", dispatcher = TRUE))
  test_equal(launch_local(), 1L)
  Sys.sleep(1L)
  test_true(grepl("CERTIFICATE", launch_remote(), fixed = TRUE))
  q <- quote(list2env(list(b = 2), envir = .GlobalEnv))
  m <- mirai("Seattle", .timeout = 1000)
  if (!is_error_value(m[])) test_equal(m[], "Seattle")
  test_class("errorValue", mirai(q(), .timeout = 1000)[])
  test_zero(daemons(0))
}
# TLS tests
connection && Sys.getenv("NOT_CRAN") == "true" && {
  Sys.sleep(1L)
  cfg <- serial_config("custom", function(x) serialize(x, NULL), unserialize)
  test_zero(daemons(url = host_url(), pass = "test", serial = cfg))
  test_equal(launch_local(1L), 1L)
  Sys.sleep(1L)
  q <- quote({ list2env(list(b = 2), envir = .GlobalEnv); 0L})
  mm <- everywhere(q)
  test_type("list", mm)
  test_zero(collect_mirai(mm, ".flat"))
  m <- mirai(b, .timeout = 1000)
  if (!is_error_value(m[])) test_equal(m[], 2L)
  test_null(saisei(1))
  test_zero(daemons(0))
  test_tls <- function(cert) {
    file <- tempfile()
    on.exit(unlink(file))
    cat(cert[["server"]], file = file)
    daemons(url = "tls+tcp://127.0.0.1:0", tls = file) == 0L && daemons(0L) == 0L
  }
  test_true(test_tls(nanonext::write_cert(cn = "127.0.0.1")))
}
# promises tests
connection && requireNamespace("promises", quietly = TRUE) && Sys.getenv("NOT_CRAN") == "true" && {
  Sys.sleep(1L)
  test_equal(daemons(1, notused = "wrongtype"), 1L)
  test_true(grepl("://", launch_remote(1L), fixed = TRUE))
  test_true(promises::is.promise(p1 <- promises::as.promise(mirai("completed"))))
  test_true(promises::is.promise(p2 <- promises::`%...>%`(mirai("completed"), identity())))
  test_true(promises::is.promise(p3 <- promises::as.promise(call_mirai(mirai("completed")))))
  test_zero(mirai_map(0:1, function(x) x, .promise = identity)[][[1L]])
  test_true(is_mirai_map(mp <- mirai_map(matrix(1:4, nrow = 2L), function(x, y) x + y, .promise = list(identity))))
  test_true(all(mp[.flat, .stop] == c(4L, 6L)))
  test_null(names(mp[]))
  test_class("errorValue", mirai_map(1, function(x) stop(x), .promise = list(identity, identity))[][[1L]])
  Sys.sleep(1L)
  getNamespace("later")[["run_now"]]()
  test_zero(daemons(NULL))
}
# mirai daemon limits tests
connection && Sys.getenv("NOT_CRAN") == "true" && {
  Sys.sleep(1L)
  test_equal(daemons(1, cleanup = FALSE, maxtasks = 2L, id = 125L), 1L)
  test_equal(mirai(1)[], mirai(1)[])
  m <- mirai(0L)
  Sys.sleep(1L)
  res <- status()
  test_zero(res$connections)
  test_identical(res$events, c(125L, -125L))
  test_equal(res$mirai[["awaiting"]], 1L)
  test_equal(launch_local(1, idletime = 5000L, walltime = 500L, id = 129L), 1L)
  test_zero(m[])
  Sys.sleep(1L)
  res <- status()
  test_zero(res$connections)
  test_identical(res$events, c(129L, -129L))
  test_zero(daemons(0))
  test_equal(daemons(1, dispatcher = FALSE, maxtasks = 1L), 1L)
  test_zero(mirai(0L)[])
  Sys.sleep(0.5)
  test_zero(status()$connections)
  test_equal(launch_local(1, idletime = 200L, walltime = 1000L), 1L)
  test_zero(mirai(0)[])
  Sys.sleep(1L)
  test_zero(status()$connections)
  test_zero(daemons(0))
}
# mirai cancellation tests
connection && Sys.getenv("NOT_CRAN") == "true" && {
  Sys.sleep(1L)
  test_equal(daemons(1, dispatcher = TRUE, cleanup = FALSE), 1L)
  m1 <- mirai({ Sys.sleep(1); res <<- "m1 done" })
  m2 <- mirai({ Sys.sleep(1); res <<- "m2 done" })
  Sys.sleep(0.1)
  test_true(stop_mirai(m2))
  test_true(stop_mirai(m1))
  test_equal(m2$data, 20L)
  test_equal(m1$data, 20L)
  test_class("errorValue", mirai(res)[])
  m <- mirai_map(1:10, function(x) { Sys.sleep(2); y <<- TRUE })
  Sys.sleep(0.1)
  s <- stop_mirai(m)
  test_equal(sum(unlist(m[])), 200L)
  test_class("errorValue", mirai(y)[])
  test_identical(s, !logical(10L))
  test_equal(status()$connections, 1L)
  test_equal(length(nextget("urls")), 1L)
  test_class("miraiLaunchCmd", launch_remote(1))
  test_zero(daemons(0))
}
# additional stress testing
connection && Sys.getenv("NOT_CRAN") == "true" && {
  Sys.sleep(1L)
  q <- vector(mode = "list", length = 10000L)
  Sys.setenv(R_DEFAULT_PACKAGES = "stats,utils")
  test_equal(daemons(4), 4L)
  Sys.unsetenv("R_DEFAULT_PACKAGES")
  for (i in seq_len(10000L)) {q[[i]] <- mirai(1L); attr(q[[i]], "status") <- status()}
  test_equal(sum(unlist(collect_mirai(q))), 10000L)
  test_true(all(as.logical(lapply(lapply(q, attr, "status"), is.list))))
  for (i in seq_len(10000L)) {q[[i]] <- mirai({Sys.sleep(0.001); rnorm(1)}); attr(q[[i]], "status") <- status()}
  test_equal(length(unique(unlist(collect_mirai(q)))), 10000L)
  test_true(all(as.logical(lapply(lapply(q, attr, "status"), is.list))))
  test_equal(daemons()[["mirai"]][["completed"]], 20000L)
}
# legacy interface tests
connection && .Platform[["OS.type"]] != "windows" && Sys.getenv("NOT_CRAN") == "true" && {
  option <- 15L
  Sys.setenv(R_DEFAULT_PACKAGES = "stats,utils")
  test_equal(1L, daemons(1, dispatcher = "process", maxtasks = 10L, timerstart = 1L, walltime = 500L, idletime = 500L, seed = 1546, cleanup = option, autoexit = tools::SIGCONT))
  Sys.unsetenv("R_DEFAULT_PACKAGES")
  Sys.sleep(1L)
  mq <- mirai(runif(1L), .timeout = 1000)
  test_true(is.numeric(mq[]))
  mq <- mirai(Sys.sleep(0.7), .timeout = 500)
  test_class("matrix", status()[["daemons"]])
  test_null(saisei(i = 1L))
  Sys.sleep(1L)
  test_equal(daemons(url = "wss://127.0.0.1:0", dispatcher = "process", output = TRUE, token = TRUE, walltime = 500L, idletime = 505L), 1L)
  test_equal(nextget("n"), 1L)
  test_equal(length(nextget("urls")), 1L)
  test_class("matrix", status()$daemons)
  test_null(saisei(i = 0L))
  test_print(saisei(i = 1L))
  test_print(saisei(i = 1L, force = TRUE))
  Sys.sleep(0.1)
  test_zero(daemons(0))
  test_equal(daemons(n = 2L, url = "tls+tcp://127.0.0.1:0", dispatcher = "thread", token = TRUE, idletime = Inf), 2L)
  test_class("matrix", status()$daemons)
  Sys.sleep(0.1)
  test_zero(daemons(0))
}
test_zero(daemons(0))
Sys.sleep(1L)