Testing async reactives

shiny::testServer() works with async reactives as it does with plain ones, with one difference: a task runs in a worker, so its result is not there the moment you set an input. The test has to let the worker finish and let Shiny see the result. This vignette shows the few lines that do that, using only the mock session’s public methods.

library(shiny)
library(shinyasync)
library(testthat)

Daemons

Start a small pool at the top of each test and shut it down when the test ends. Tests then run the same whether or not the app under test calls pool_async(), and never depend on a pool left behind by another test.

local_pool <- function(n = 2, env = parent.frame()) {
  mirai::daemons(n, dispatcher = TRUE)
  withr::defer(mirai::daemons(0), envir = env)
}

If a test does not start daemons, the first task starts the default pool with a message, held by the mock session, which is also fine for a one-off test.

Waiting for a result

Two things must happen before a test can read a result. later::run_now() runs the event loop, which delivers the worker’s result to the session. session$flushReact() then re-runs the reactives that depend on it, as the real session would. Repeat both until the condition you care about holds:

wait_for <- function(cond, session, timeout = 15) {
  start <- Sys.time()
  repeat {
    session$flushReact()
    if (isTRUE(shiny::isolate(cond()))) {
      # Let observers see a value that arrived during the last flush.
      session$flushReact()
      return(invisible(TRUE))
    }
    if (difftime(Sys.time(), start, units = "secs") > timeout) {
      stop(
        "Timed out after ", timeout, "s waiting for: ",
        paste(deparse(body(cond)), collapse = " "),
        call. = FALSE
      )
    }
    later::run_now(0.05)
  }
}

The second flush after the condition holds lets an observer that depends on the value run before the test goes on to check what it did.

The condition is any reactive read: the value, status_async() or has_value_async(). The read runs inside isolate(), so it does not register dependencies, but it still starts a task that has not run yet, the same way an output reading the reactive would.

Reading a value that is not there yet halts, as req() does, and a halt inside expect_equal() is a confusing failure. peek() reads a reactive and returns a marker instead:

peek <- function(r) {
  tryCatch(shiny::isolate(r()), shiny.silent.error = function(e) "<halted>")
}

Put the three helpers in tests/testthat/helper-async.R and every test file sees them.

A first test

The server below fits a model in a worker. Set an input, wait for the status to settle, then read the value exactly as you would in a plain Shiny test.

server <- function(input, output, session) {
  fit <- reactiveAsync({
    lm(mpg ~ wt, data = mtcars[seq_len(input$rows), ])
  })
  output$slope <- renderText(round(coef(fit())[["wt"]], 2))
}

test_that("the slope updates when the rows change", {
  local_pool()
  testServer(server, {
    session$setInputs(rows = 32)
    wait_for(function() status_async(fit) == "success", session)
    expect_equal(output$slope, "-5.34")

    session$setInputs(rows = 10)
    wait_for(function() output$slope != "-5.34", session)
    expect_equal(status_async(fit), "success")
  })
})
#> Test passed with 2 successes 🎊.

Reading fit() before the first result arrives halts silently, as req() does, so the test waits for the status rather than the value. After the first result, reading the value during a refresh returns the previous one; the second wait above uses the output changing as its condition for that reason.

Status, buttons and cancellation

Statuses follow the task. A bindEvent() task is "initial" until its event, "running" while the worker has it, and "success" or "error" after. cancel_async() sets it to "idle" while keeping the last value.

server <- function(input, output, session) {
  sim <- reactiveAsync({
    Sys.sleep(2)
    mean(rnorm(input$n))
  }) |> bindEvent(input$go)
  observe(cancel_async(sim)) |> bindEvent(input$stop)
}

test_that("a button task runs on click and stops on cancel", {
  local_pool()
  testServer(server, {
    session$setInputs(n = 100)
    expect_equal(status_async(sim), "initial")

    session$setInputs(go = 1)
    expect_equal(status_async(sim), "running")

    session$setInputs(stop = 1)
    expect_equal(status_async(sim), "idle")
    expect_false(has_value_async(sim))
  })
})
#> Test passed with 4 successes 🎊.

session$setInputs() flushes the session, so the status is "running" as soon as the click is set. That only holds because the task has work to do: a task that finishes within the flush that dispatched it settles in that flush, and a test that reads its status straight after setInputs() sees "success". The cancelled run had not produced a result, so there is no previous value to keep and has_value_async() is FALSE.

Errors

A failure in the worker surfaces when the reactive is read, as it would from reactive(). Wait for the status to become "error", then expect the error on the read.

server <- function(input, output, session) {
  parsed <- reactiveAsync(jsonlite::fromJSON(input$text))
}

test_that("a worker error is re-thrown on read", {
  local_pool()
  testServer(server, {
    session$setInputs(text = "{not json")
    wait_for(function() status_async(parsed) == "error", session)
    expect_error(parsed(), "lexical error")
  })
})
#> Test passed with 1 success 🎊.

Side effects

observeAsync() has no value to read. Its handle reports the status, and the side effect is what the test checks: here a line written to a log file.

log_path <- tempfile(fileext = ".log")

server <- function(input, output, session) {
  logged <- observeAsync(cat(input$spot, file = log_path, sep = "\n", append = TRUE))
}

test_that("each selected spot is logged", {
  local_pool()
  testServer(server, {
    session$setInputs(spot = "Muizenberg")
    wait_for(function() logged$status() == "success", session)
    expect_equal(readLines(log_path), "Muizenberg")
  })
})
#> Test passed with 1 success 🌈.

The observer starts on the next flush, like observe(), so there is nothing to read first. status_async(logged) reads the same status as logged$status().

Values fed by hand

reactiveValAsync() is set from an observer, usually with a mirai started by mirai_async(). Test it as any reactiveVal(), waiting for the task the setter was handed:

rank_spots <- function(spots) sort(spots)

server <- function(input, output, session) {
  ranking <- reactiveValAsync()
  observe(ranking(mirai_async(rank_spots(input$spots)))) |> bindEvent(input$rank)
}

test_that("the ranking arrives after the click", {
  local_pool()
  testServer(server, {
    session$setInputs(spots = c("Kommetjie", "Muizenberg", "Elands Bay"))
    expect_null(ranking())

    session$setInputs(rank = 1)
    wait_for(function() has_value_async(ranking), session)
    expect_equal(ranking()[[1]], "Elands Bay")
  })
})
#> Test passed with 2 successes 🥇.

Before the click the value is NULL, as it would be for a reactiveVal() created without one; only reactiveAsync() halts until its first result. mirai_async() ships rank_spots() with the task, as it would a helper from the app’s R/ directory.

Queued runs

With mode = "queue" every change runs, in order, where the default would cancel the superseded ones. An observer collecting the results into a reactiveVal() is the easiest way to see the order:

save_favourite <- function(spot) paste("saved", spot)

server <- function(input, output, session) {
  saved <- reactiveAsync({
    Sys.sleep(0.2)
    save_favourite(input$favourite)
  }, mode = "queue")
  seen <- reactiveVal(character())
  observe(seen(c(isolate(seen()), saved())))
}

test_that("every favourite is saved, in order", {
  local_pool()
  testServer(server, {
    session$setInputs(favourite = "Kommetjie")
    session$setInputs(favourite = "Muizenberg")
    session$setInputs(favourite = "Elands Bay")
    wait_for(function() identical(peek(saved), "saved Elands Bay"), session)
    expect_equal(seen(), c("saved Kommetjie", "saved Muizenberg", "saved Elands Bay"))
  })
})
#> Test passed with 1 success 🎊.

The three inputs are set before the first task finishes. A queued reactive starts, like any reactiveAsync(), only once something reads it, which the observer does on the first flush.

Chained reactives

An async reactive that reads another one keeps its previous value and reports "running" while its upstream is catching up, then runs once on the settled value. A test can see each step:

server <- function(input, output, session) {
  fit <- reactiveAsync({
    Sys.sleep(1)
    lm(mpg ~ wt, data = mtcars[seq_len(input$rows), ])
  })
  predicted <- reactiveAsync(predict(fit(), data.frame(wt = 3)))
}

test_that("a prediction waits for the model it depends on", {
  local_pool()
  testServer(server, {
    session$setInputs(rows = 32)
    wait_for(function() has_value_async(predicted), session)
    first <- predicted()

    session$setInputs(rows = 10)
    expect_equal(status_async(predicted), "running")
    expect_equal(peek(predicted), first)
    wait_for(function() !identical(peek(predicted), first), session)
    expect_equal(status_async(fit), "success")
  })
})
#> Test passed with 3 successes 🥳.

Waiting on the downstream reactive is enough: reading predicted starts fit, and the second wait ends only when a new model has been fitted and predicted from.

Modules

Nothing changes for a module: pass the module server to testServer() and the same helpers apply to the reactives it defines or returns.

summaryServer <- function(id) {
  moduleServer(id, function(input, output, session) {
    s <- reactiveAsync(summary(mtcars[[input$column]]))
    output$text <- renderPrint(s())
    s
  })
}

test_that("the module summarises the chosen column", {
  local_pool()
  testServer(summaryServer, {
    session$setInputs(column = "mpg")
    wait_for(function() has_value_async(s), session)
    expect_equal(s()[["Max."]], 33.9)
  })
})
#> Test passed with 1 success 😸.