Improve snapshotting cleanup in parallel tests (#1800)

We cannot forward all work of the snapshot reporter to the main process, because `expect_snapshot_helper()` uses the return value of the `take_snapshot()` method. So this must be called in the subprocess.

This is also good for performance, because the snapshot comparison code still runs in the subprocess, and we don't copy potentially large objects between processes.

The disadvantage is that we need slightly more complicated reporters, both in the subprocess and in the main process, to make sure that both processes are doing the right type of snapshot cleanup.

Closes #1797.

Simplifies and improves #1788 and #1793.

R/parallel.R

@@ -45,9 +45,6 @@ test_files_parallel <- function(
                        load_package = c("none", "installed", "source")
                        ) {
 
-
-  reporters <- test_files_reporter(reporter, parallel = TRUE)
-
   # TODO: support timeouts. 20-30s for each file by default?
 
   num_workers <- min(default_num_cpus(), length(test_paths))
@@ -71,6 +68,7 @@ test_files_parallel <- function(
   )
 
   withr::with_dir(test_dir, {
+    reporters <- test_files_reporter_parallel(reporter)
     with_reporter(reporters$multi, {
       parallel_updates <- reporter$capabilities$parallel_updates
       if (parallel_updates) {
@@ -79,11 +77,29 @@ test_files_parallel <- function(
         parallel_event_loop_chunky(queue, reporters, ".")
       }
     })
+
+    test_files_check(reporters$list$get_results(),
+      stop_on_failure = stop_on_failure,
+      stop_on_warning = stop_on_warning
+    )
   })
+}
 
-  test_files_check(reporters$list$get_results(),
-    stop_on_failure = stop_on_failure,
-    stop_on_warning = stop_on_warning
+test_files_reporter_parallel <- function(reporter, .env = parent.frame()) {
+  lister <- ListReporter$new()
+  snapshotter <- MainprocessSnapshotReporter$new("_snaps", fail_on_new = FALSE)
+  reporters <- list(
+    find_reporter(reporter),
+    lister, # track data
+    snapshotter
+  )
+  withr::local_options(
+    "testthat.snapshotter" = snapshotter,
+    .local_envir = .env
+  )
+  list(
+    multi = MultiReporter$new(reporters = compact(reporters)),
+    list = lister
   )
 }
 
@@ -133,21 +149,21 @@ parallel_event_loop_smooth <- function(queue, reporters, test_dir) {
       }
 
       if (m$cmd != "DONE") {
-        # Set working directory so expect_location() generates correct links
-        withr::with_dir(test_dir, {
-          reporters$multi$start_file(m$filename)
+        reporters$multi$start_file(m$filename)
+        reporters$multi$start_test(m$context, m$test)
+        if (m$type == "snapshotter") {
+          snapshotter <- getOption("testthat.snapshotter")
+          do.call(snapshotter[[m$cmd]], m$args)
+        } else {
           do.call(reporters$multi[[m$cmd]], m$args)
           updated <- TRUE
-        })
+        }
       }
     }
 
     # We need to spin, even if there were no events
     if (!updated) {
-      # Set working directory so expect_location() generates correct links
-      withr::with_dir(test_dir, {
-        reporters$multi$update()
-      })
+      reporters$multi$update()
     }
   }
 }
@@ -172,20 +188,22 @@ parallel_event_loop_chunky <- function(queue, reporters, test_dir) {
       if (m$cmd != "DONE") {
         files[[m$filename]] <- append(files[[m$filename]], list(m))
       } else {
-        # Set working directory so expect_location() generates correct links
-        withr::with_dir(test_dir, {
-          replay_events(reporters$multi, files[[m$filename]])
-          reporters$multi$end_context_if_started()
-        })
+        replay_events(reporters$multi, files[[m$filename]])
+        reporters$multi$end_context_if_started()
         files[[m$filename]] <- NULL
       }
     }
   }
 }
 
 replay_events <- function(reporter, events) {
-  for (event in events) {
-    do.call(reporter[[event$cmd]], event$args)
+  snapshotter <- getOption("testthat.snapshotter")
+  for (m in events) {
+    if (m$type == "snapshotter") {
+      do.call(snapshotter[[m$cmd]], m$args)
+    } else {
+      do.call(reporter[[m$cmd]], m$args)
+    }
   }
 }
 
@@ -254,8 +272,17 @@ queue_task <- function(path) {
   env <- .GlobalEnv$.test_env
 
   withr::local_envvar("TESTTHAT_IS_PARALLEL" = "true")
-  reporters <- test_files_reporter(SubprocessReporter$new())
-  with_reporter(reporters$multi, test_one_file(path, env = env))
+  snapshotter <- SubprocessSnapshotReporter$new(
+    snap_dir = "_snaps",
+    fail_on_new = FALSE
+  )
+  withr::local_options(testthat.snapshotter = snapshotter)
+  reporters <- list(
+    SubprocessReporter$new(),
+    snapshotter
+  )
+  multi <- MultiReporter$new(reporters = reporters)
+  with_reporter(multi, test_one_file(path, env = env))
   NULL
 }
 
@@ -326,9 +353,12 @@ SubprocessReporter <- R6::R6Class("SubprocessReporter",
       private$event("start_file", filename)
     },
     start_test = function(context, test) {
+      private$context <- context
+      private$test <- test
       private$event("start_test", context, test)
     },
     start_context = function(context) {
+      private$context <- context
       private$event("start_context", context)
     },
     add_result = function(context, test, result) {
@@ -355,11 +385,16 @@ SubprocessReporter <- R6::R6Class("SubprocessReporter",
 
   private = list(
     filename = NULL,
+    context = NULL,
+    test = NULL,
     event = function(cmd, ...) {
       msg <- list(
         code = PROCESS_MSG,
+        type = "reporter",
         cmd = cmd,
         filename = private$filename,
+        context = private$context,
+        test = private$test,
         time = proc.time()[[3]],
         args = list(...)
       )

R/snapshot-reporter-parallel.R

@@ -0,0 +1,70 @@
+
+MainprocessSnapshotReporter <- R6::R6Class("MainprocessSnapshotReporter",
+  inherit = SnapshotReporter,
+  public = list(
+    end_file = function() {
+      # No thing to do, this is done in the subprocess
+    }
+  )
+)
+
+SubprocessSnapshotReporter <- R6::R6Class("SubprocessSnapshotReporter",
+  inherit = SnapshotReporter,
+  public = list(
+    start_file = function(path, test = NULL) {
+      private$filename <- path
+      private$test <- test
+      super$start_file(path, test)
+    },
+
+    end_file = function() {
+      private$filename <- NULL
+      super$end_file()
+    },
+
+    end_context = function(context) {
+      private$context <- NULL
+      super$end_context()
+    },
+
+    end_test = function(context, test) {
+      private$context <- NULL
+      private$test <- NULL
+      super$end_test(context, test)
+    },
+
+    start_test = function(context, test) {
+      private$context <- context
+      private$test <- test
+      super$start_test(context, test)
+    },
+
+    announce_file_snapshot = function(name) {
+      private$event("announce_file_snapshot", name)
+      super$announce_file_snapshot(name)
+    },
+
+    end_reporter = function() {
+      # do not delete unused snapshots, that's up to the main process
+    }
+  ),
+  private = list(
+    filename = NULL,
+    context = NULL,
+    test = NULL,
+    event = function(cmd, ...) {
+      msg <- list(
+        code = PROCESS_MSG,
+        type = "snapshotter",
+        cmd = cmd,
+        filename = private$filename,
+        context = context,
+        test = test,
+        time = NULL,
+        args = list(...)
+      )
+      class(msg) <- c("testthat_message", "callr_message", "condition")
+      signalCondition(msg)
+    }
+  )
+)

R/snapshot-reporter.R

@@ -9,16 +9,14 @@ SnapshotReporter <- R6::R6Class("SnapshotReporter",
     snap_file_seen = character(),
     variants_changed = FALSE,
     fail_on_new = FALSE,
-    parallel = FALSE,
 
     old_snaps = NULL,
     cur_snaps = NULL,
     new_snaps = NULL,
 
-    initialize = function(snap_dir = "_snaps", fail_on_new = FALSE, parallel = FALSE) {
+    initialize = function(snap_dir = "_snaps", fail_on_new = FALSE) {
       self$snap_dir <- normalizePath(snap_dir, mustWork = FALSE)
       self$fail_on_new <- fail_on_new
-      self$parallel <- parallel
     },
 
     start_file = function(path, test = NULL) {
@@ -129,11 +127,6 @@ SnapshotReporter <- R6::R6Class("SnapshotReporter",
     },
 
     end_file = function() {
-      if (self$parallel) {
-        # This is only needs to be done in the child threads.
-        return()
-      }
-
       dir.create(self$snap_dir, showWarnings = FALSE)
 
       self$cur_snaps$write()
@@ -200,13 +193,9 @@ get_snapshotter <- function() {
 #'
 #' @export
 #' @keywords internal
-local_snapshotter <- function(snap_dir = NULL, cleanup = FALSE, fail_on_new = FALSE, parallel = FALSE, .env = parent.frame()) {
+local_snapshotter <- function(snap_dir = NULL, cleanup = FALSE, fail_on_new = FALSE, .env = parent.frame()) {
   snap_dir <- snap_dir %||% withr::local_tempdir(.local_envir = .env)
-  reporter <- SnapshotReporter$new(
-    snap_dir = snap_dir,
-    fail_on_new = fail_on_new,
-    parallel = parallel
-  )
+  reporter <- SnapshotReporter$new(snap_dir = snap_dir, fail_on_new = fail_on_new)
   if (!identical(cleanup, FALSE)) {
     warn("`cleanup` is deprecated")
   }

R/test-files.R

@@ -277,17 +277,12 @@ test_files_setup_state <- function(
   withr::defer(source_test_teardown(".", env), frame)      # old school
 }
 
-test_files_reporter <- function(reporter, .env = parent.frame(), parallel = FALSE) {
+test_files_reporter <- function(reporter, .env = parent.frame()) {
   lister <- ListReporter$new()
   reporters <- list(
     find_reporter(reporter),
     lister, # track data
-    local_snapshotter(
-      "_snaps",
-      fail_on_new = FALSE,
-      parallel = parallel,
-      .env = .env
-    )
+    local_snapshotter("_snaps", fail_on_new = FALSE, .env = .env)
   )
   list(
     multi = MultiReporter$new(reporters = compact(reporters)),

tests/testthat/test-parallel.R

@@ -34,3 +34,86 @@ test_that("ok", {
   expect_equal(tdf$failed, c(0,1,0))
   expect_equal(tdf$skipped, c(FALSE, FALSE, TRUE))
 })
+
+test_that("snapshots", {
+  withr::local_envvar(c(TESTTHAT_PARALLEL = "TRUE"))
+  on.exit(unlink(tmp, recursive = TRUE), add = TRUE)
+  dir.create(tmp <- tempfile("testthat-snap-"))
+  file.copy(test_path("test-parallel", "snap"), tmp, recursive = TRUE)
+  suppressMessages(ret <- test_local(
+    file.path(tmp, "snap"),
+    reporter = "silent",
+    stop_on_failure = FALSE
+  ))
+  tdf <- as.data.frame(ret)
+  tdf <- tdf[order(tdf$file), ]
+  expect_equal(tdf$failed, c(0,0,1))
+  snaps <- file.path(tmp, "snap", "tests", "testthat", "_snaps")
+  expect_true(file.exists(file.path(snaps, "snap-1.md")))
+  expect_true(file.exists(file.path(snaps, "snap-2.md")))
+  expect_true(file.exists(file.path(snaps, "snap-3.md")))
+})
+
+test_that("new snapshots are added", {
+  withr::local_envvar(c(TESTTHAT_PARALLEL = "TRUE"))
+  on.exit(unlink(tmp, recursive = TRUE), add = TRUE)
+  dir.create(tmp <- tempfile("testthat-snap-"))
+  file.copy(test_path("test-parallel", "snap"), tmp, recursive = TRUE)
+  unlink(file.path(tmp, "snap", "tests", "testthat", "_snaps", "snap-2.md"))
+  suppressMessages(ret <- test_local(
+    file.path(tmp, "snap"),
+    reporter = "silent",
+    stop_on_failure = FALSE
+  ))
+  tdf <- as.data.frame(ret)
+  tdf <- tdf[order(tdf$file), ]
+  expect_equal(tdf$failed, c(0,0,1))
+  snaps <- file.path(tmp, "snap", "tests", "testthat", "_snaps")
+  expect_true(file.exists(file.path(snaps, "snap-1.md")))
+  expect_true(file.exists(file.path(snaps, "snap-2.md")))
+  expect_true(file.exists(file.path(snaps, "snap-3.md")))
+})
+
+test_that("snapshots are removed if test file has no snapshots", {
+  withr::local_envvar(c(TESTTHAT_PARALLEL = "TRUE"))
+  on.exit(unlink(tmp, recursive = TRUE), add = TRUE)
+  dir.create(tmp <- tempfile("testthat-snap-"))
+  file.copy(test_path("test-parallel", "snap"), tmp, recursive = TRUE)
+  writeLines(
+    "test_that(\"2\", { expect_true(TRUE) })",
+    file.path(tmp, "snap", "tests", "testthat", "test-snap-2.R")
+  )
+  suppressMessages(ret <- test_local(
+    file.path(tmp, "snap"),
+    reporter = "silent",
+    stop_on_failure = FALSE
+  ))
+  tdf <- as.data.frame(ret)
+  tdf <- tdf[order(tdf$file), ]
+  expect_equal(tdf$failed, c(0,0,1))
+  snaps <- file.path(tmp, "snap", "tests", "testthat", "_snaps")
+  expect_true(file.exists(file.path(snaps, "snap-1.md")))
+  expect_false(file.exists(file.path(snaps, "snap-2.md")))
+  expect_true(file.exists(file.path(snaps, "snap-3.md")))
+})
+
+test_that("snapshots are removed if test file is removed", {
+  withr::local_envvar(c(TESTTHAT_PARALLEL = "TRUE"))
+  on.exit(unlink(tmp, recursive = TRUE), add = TRUE)
+  dir.create(tmp <- tempfile("testthat-snap-"))
+  file.copy(test_path("test-parallel", "snap"), tmp, recursive = TRUE)
+  unlink(file.path(tmp, "snap", "tests", "testthat", "test-snap-2.R"))
+  withr::local_envvar(CI = NA_character_)
+  suppressMessages(ret <- test_local(
+    file.path(tmp, "snap"),
+    reporter = "silent",
+    stop_on_failure = FALSE
+  ))
+  tdf <- as.data.frame(ret)
+  tdf <- tdf[order(tdf$file), ]
+  expect_equal(tdf$failed, c(0,1))
+  snaps <- file.path(tmp, "snap", "tests", "testthat", "_snaps")
+  expect_true(file.exists(file.path(snaps, "snap-1.md")))
+  expect_false(file.exists(file.path(snaps, "snap-2.md")))
+  expect_true(file.exists(file.path(snaps, "snap-3.md")))
+})