Mrc 5114 Add a submit function to the queue object

R/queue.R

@@ -34,6 +34,20 @@ Queue <- R6::R6Class("Queue", #nolint
                                    controller = self$controller)
     },
 
+    submit = function(reportname, parameters = NULL,
+                      branch = "master", ref = "HEAD") {
+      run_args <- list(
+        self$root,
+        reportname,
+        parameters,
+        branch,
+        ref
+      )
+      rrq::rrq_task_create_call(runner_run, run_args,
+                                separate_process = TRUE,
+                                controller = self$controller)
+    },
+
     # Just until we add queue status for testing
     number_of_workers = function() {
       rrq::rrq_worker_len(self$controller)

R/runner.R

@@ -0,0 +1,59 @@
+runner_run <- function(orderly_root, reportname, parameters, branch, ref, ...) {
+  # Helper functions
+  # ==================
+  point_head_to_ref <- function(worker_path, branch, ref) {
+    gert::git_fetch(repo = worker_path)
+    gert::git_branch_checkout(branch, repo = worker_path)
+    gert::git_reset_hard(ref, repo = worker_path)
+  }
+
+  add_dir_parent_if_empty <- function(files_to_delete, path) {
+    contained_files <- list.files(path, full.names = TRUE)
+    if (length(setdiff(contained_files, files_to_delete)) > 0) {
+      return(files_to_delete)
+    }
+    add_dir_parent_if_empty(c(files_to_delete, path), dirname(path))
+  }
+
+  get_empty_dirs <- function(worker_path) {
+    dirs <- fs::dir_ls(worker_path, recurse = TRUE, type = "directory")
+    Reduce(add_dir_parent_if_empty, c(list(character()), dirs))
+  }
+
+  git_clean <- function(worker_path) {
+    # gert does not have git clean but this should achieve the same thing
+    res <- tryCatch(
+      gert::git_stash_save(
+        include_untracked = TRUE,
+        include_ignored = TRUE,
+        repo = worker_path
+      ),
+      error = function(e) NULL
+    )
+    if (!is.null(res)) {
+      gert::git_stash_drop(repo = worker_path)
+    }
+    # however git ignores all directories, only cares about files, so we may
+    # have empty directories left
+    unlink(get_empty_dirs(worker_path), recursive = TRUE)
+  }
+  # ==================
+
+
+  # Actual runner code
+  # ==================
+  # Setup
+  worker_id <- Sys.getenv("RRQ_WORKER_ID")
+  worker_path <- file.path(orderly_root, ".packit", "workers", worker_id)
+  point_head_to_ref(worker_path, branch, ref)
+
+  # Run
+  withr::with_envvar(
+    c(ORDERLY_SRC_ROOT = file.path(worker_path, "src", reportname)),
+    orderly2::orderly_run(reportname, parameters = parameters,
+                          root = orderly_root, ...)
+  )
+
+  # Cleanup
+  git_clean(worker_path)
+}

tests/testthat/helper-orderly-runner.R

@@ -79,6 +79,53 @@ helper_add_git <- function(path) {
   list(path = path, user = user, branch = branch, sha = sha, url = url)
 }
 
+new_queue_quietly <- function(root, ...) {
+  suppressMessages(Queue$new(root, ...))
+}
+
+make_worker_dirs <- function(orderly_root, ids) {
+  packit_path <- file.path(orderly_root, ".packit")
+  dir.create(packit_path)
+  workers <- file.path(packit_path, "workers")
+  dir.create(workers)
+  lapply(ids, function(id) {
+    worker_path <- file.path(workers, id)
+    dir.create(worker_path)
+    gert::git_clone(orderly_root, path = worker_path)
+  })
+
+}
+
+start_queue_workers_quietly <- function(n_workers,
+                                        controller, env = parent.frame()) {
+  worker_manager <- suppressMessages(
+    rrq::rrq_worker_spawn2(n_workers, controller = controller)
+  )
+  withr::defer(rrq::rrq_worker_stop(controller = controller), env = env)
+  worker_manager
+}
+
+skip_if_no_redis <- function() {
+  available <- redux::redis_available()
+  if (!available) {
+    testthat::skip("Skipping test as redis is not available")
+  }
+  invisible(available)
+}
+
+expect_worker_task_complete <- function(task_id, controller, n_tries) {
+  is_completed <- FALSE
+  for (i in seq_len(n_tries)) {
+    is_completed <- rrq::rrq_task_status(
+      task_id, controller = controller
+    ) == "COMPLETE"
+    if (is_completed == TRUE) {
+      break
+    }
+    Sys.sleep(1)
+  }
+  expect_equal(is_completed, TRUE)
+}
 
 initialise_git_repo <- function() {
   t <- tempfile()

tests/testthat/test-queue.R

@@ -58,3 +58,76 @@ test_that("Generated namespaced id if no ids exist", {
   q <- new_queue_quietly(root)
   expect_match(q$controller$queue_id, "orderly.runner")
 })
+
+
+test_that("Can submit task", {
+  skip_if_no_redis()
+
+  root <- test_prepare_orderly_example("data")
+  gert::git_init(root)
+  gert::git_add(c("src", "orderly_config.yml"), repo = root)
+  gert::git_commit("first commit", repo = root)
+
+  q <- new_queue_quietly(root)
+  worker_manager <- start_queue_workers_quietly(1, q$controller)
+  make_worker_dirs(root, worker_manager$id)
+
+  task_id <- q$submit("data")
+  expect_worker_task_complete(task_id, q$controller, 10)
+})
+
+
+test_that("Can submit 2 tasks on different branches", {
+  skip_if_no_redis()
+
+  root <- test_prepare_orderly_example("data")
+  gert::git_init(root)
+  gert::git_add(c("src", "orderly_config.yml"), repo = root)
+  gert::git_commit("first commit", repo = root)
+
+  gert::git_branch_create("branch1", repo = root)
+  gert::git_branch_checkout("branch1", repo = root)
+  write.table("test", file = file.path(root, "test.txt"))
+  gert::git_add("test.txt", repo = root)
+  gert::git_commit("branch1 commit", repo = root)
+
+  q <- new_queue_quietly(root)
+  worker_manager <- start_queue_workers_quietly(2, q$controller)
+  make_worker_dirs(root, worker_manager$id)
+
+  task_id1 <- q$submit("data", branch = "master")
+  task_id2 <- q$submit("data", branch = "branch1")
+  expect_worker_task_complete(task_id1, q$controller, 10)
+  expect_worker_task_complete(task_id2, q$controller, 10)
+
+  worker_id2 <- rrq::rrq_task_info(task_id2, controller = q$controller)$worker
+  worker2_txt <- file.path(root, ".packit", "workers", worker_id2, "test.txt")
+  expect_equal(file.exists(worker2_txt), TRUE)
+})
+
+
+test_that("Can submit 2 tasks on different commit hashes", {
+  skip_if_no_redis()
+
+  root <- test_prepare_orderly_example("data")
+  gert::git_init(root)
+  gert::git_add(c("src", "orderly_config.yml"), repo = root)
+  sha1 <- gert::git_commit("first commit", repo = root)
+
+  write.table("test", file = file.path(root, "test.txt"))
+  gert::git_add("test.txt", repo = root)
+  sha2 <- gert::git_commit("second commit", repo = root)
+
+  q <- new_queue_quietly(root)
+  worker_manager <- start_queue_workers_quietly(2, q$controller)
+  make_worker_dirs(root, worker_manager$id)
+
+  task_id1 <- q$submit("data", ref = sha1)
+  task_id2 <- q$submit("data", ref = sha2)
+  expect_worker_task_complete(task_id1, q$controller, 10)
+  expect_worker_task_complete(task_id2, q$controller, 10)
+
+  worker_id2 <- rrq::rrq_task_info(task_id2, controller = q$controller)$worker
+  worker2_txt <- file.path(root, ".packit", "workers", worker_id2, "test.txt")
+  expect_equal(file.exists(worker2_txt), TRUE)
+})

tests/testthat/test-runner.R

@@ -0,0 +1,45 @@
+test_that("runner runs as expected", {
+  orderly_root <- test_prepare_orderly_example("data")
+  gert::git_init(orderly_root)
+  gert::git_add(c("src", "orderly_config.yml"), repo = orderly_root)
+  gert::git_commit("first commit", repo = orderly_root)
+
+  worker_id <- "worker1"
+  make_worker_dirs(orderly_root, worker_id)
+  worker_root <- file.path(orderly_root, ".packit", "workers", worker_id)
+
+  suppressMessages(withr::with_envvar(
+    c(RRQ_WORKER_ID = worker_id),
+    runner_run(orderly_root, "data", NULL, "master", "HEAD", echo = FALSE)
+  ))
+
+  # report has been run with data in archive
+  expect_equal(length(list.files(file.path(orderly_root, "archive"))), 1)
+  # cleanup has deleted draft folder
+  expect_equal(file.exists(file.path(worker_root, "draft")), FALSE)
+})
+
+test_that("runner runs as expected with parameters", {
+  orderly_root <- test_prepare_orderly_example("parameters")
+  gert::git_init(orderly_root)
+  gert::git_add(c("src", "orderly_config.yml"), repo = orderly_root)
+  gert::git_commit("first commit", repo = orderly_root)
+
+  worker_id <- "worker1"
+  make_worker_dirs(orderly_root, worker_id)
+  worker_root <- file.path(orderly_root, ".packit", "workers", worker_id)
+
+  parameters <- list(a = -1, b = -2, c = -3)
+  suppressMessages(withr::with_envvar(
+    c(RRQ_WORKER_ID = worker_id),
+    runner_run(orderly_root, "parameters", parameters,
+               "master", "HEAD", echo = FALSE)
+  ))
+
+  report_archive <- file.path(orderly_root, "archive", "parameters")
+  rds_path <- file.path(report_archive, list.files(report_archive), "data.rds")
+  output <- readRDS(rds_path)
+
+  expect_equal(output, parameters)
+  expect_equal(file.exists(file.path(worker_root, "draft")), FALSE)
+})