Nothing
# ---------------------
# Pipeline construction
# ---------------------
describe(".empty_pipeline", {
it("returns an empty data.table", {
dt <- .empty_pipeline()
expect_true(data.table::is.data.table(dt))
expect_equal(nrow(dt), 0)
expect_equal(
names(dt),
c(
"step",
"fun",
"params",
"out",
"state",
"tags",
"locked",
"exec",
"time",
"depends",
"unbound",
"nodeId"
)
)
})
})
describe(".new_step", {
step <- .new_step(
step = "step2",
fun = function(x) x^2,
params = list(x = 1, y = ~step1),
depends = c(y = "step1"),
tags = c("t1", "t2"),
nodeId = 0
)
it("contains the expected elements", {
expect_equal(step$step, "step2")
expect_equal(step$fun[[1]](2), 4)
expect_equivalent(step$params[[1]], list(x = 1, y = ~step1))
expect_equal(step$depends, list(c(y = "step1")))
expect_equal(step$out, list(NULL))
expect_equal(step$tags, list(c("t1", "t2")))
expect_equal(step$state, "new")
expect_true(inherits(step$time, "POSIXct"))
expect_equal(step$locked, FALSE)
expect_equal(step$nodeId, 0)
expect_equal(step$unbound, list("x"))
})
it("aligns with the empty pipeline", {
expect_equal(names(step), names(.empty_pipeline()))
})
it("can be appended to the empty pipeline", {
dt <- data.table::rbindlist(list(.empty_pipeline(), step))
expect_equal(nrow(dt), 1)
})
})
# -------
# Wrapper
# -------
describe(".wrap_self", {
it("returns functions that don't refer to .self as they are", {
f <- \(x = 1) x + 1
expect_identical(.wrap_self(f, self = "p"), f)
# Indirect access is not detected
g <- \(x = 1) get(".self")
expect_identical(.wrap_self(g, self = "p"), g)
})
it("returns a copy that sees .self in a new environment", {
env <- new.env()
f <- local(\(x = 1) .self, envir = env)
attr(f, "note") <- "kept"
w <- .wrap_self(f, self = "p")
expect_equal(w(), "p")
expect_identical(parent.env(environment(w)), env)
expect_equal(attr(w, "note"), "kept")
expect_identical(environment(f), env)
expect_false(exists(".self", envir = env, inherits = FALSE))
})
it("detects .self in functions defined in the body", {
f <- \(x = 1) {
g <- function() .self
g()
}
expect_equal(.wrap_self(f, self = "p")(), "p")
})
})
# ------------------------------
# Parameter & dependency parsing
# ------------------------------
describe(".extract_fun_params", {
it("returns TRUE if function has no args", {
expect_equal(
.extract_fun_params(function() 1),
list()
)
})
it("returns default values for function arguments", {
expect_equal(
.extract_fun_params(function(x = 1) x),
list(x = 1)
)
expect_equal(
.extract_fun_params(function(x = 1, y = 2) x + y),
list(x = 1, y = 2)
)
})
it("signals parameters with no default values", {
expect_error(
.extract_fun_params(function(x, y = 1) x + y),
"'x' has no default value",
fixed = TRUE
)
expect_error(
.extract_fun_params(function(x, y) x + y),
"'x', 'y' have no default value",
fixed = TRUE
)
})
it("supports ...", {
expect_equal(
.extract_fun_params(function(x = 1, ...) x),
list(x = 1)
)
})
it("supports formula", {
expect_equal(
.extract_fun_params(function(x = 1, y = ~foo) x),
list(x = 1, y = ~foo)
)
})
it("works with default values declared outside the function", {
default_value <- 1
expect_equal(
.extract_fun_params(function(x = default_value) x),
list(x = 1)
)
})
})
describe(".extract_depends", {
steps <- c("s1", "s2", "s3")
it("returns an empty character vector if no params are defined", {
expect_equal(
.extract_depends(list(), steps),
character(0)
)
})
it("returns an empty character vector if no dependencies are defined", {
expect_equal(
.extract_depends(list(a = 1), steps),
character(0)
)
})
it("extracts all dependencies defined via step name", {
expect_equal(
.extract_depends(list(a = ~s1, b = 2), steps),
c(a = "s1")
)
expect_equal(
.extract_depends(list(a = ~s1, b = ~s2), steps),
c(a = "s1", b = "s2")
)
})
it("extracts all dependencies defined relative", {
expect_equal(
.extract_depends(list(x = ~ -1), steps = c("f1", "f2")),
c(x = "f1")
)
expect_equal(
.extract_depends(list(a = ~ -1, b = 2), steps),
c(a = "s2")
)
expect_equal(
.extract_depends(list(a = ~ -2, b = ~ -1), steps),
c(a = "s1", b = "s2")
)
})
it("signals bad steps input", {
expect_error(
.extract_depends(list(a = ~ -1), character(0)),
"toPos (0) must be at least 1",
fixed = TRUE
)
})
it("signals relative index out of bound", {
expect_error(
.extract_depends(list(a = ~ -1), "s1"),
"relative index a = ~-1 points outside pipeline",
fixed = TRUE
)
expect_error(
.extract_depends(list(a = ~ -4), steps),
"relative index a = ~-4 points outside pipeline"
)
})
it("extracts all dependencies if defined both ways", {
expect_equal(
.extract_depends(list(a = ~ -1, b = ~s1, c = 3), steps),
c(a = "s2", b = "s1")
)
})
it("signals toPos exceeding number of steps", {
expect_error(
.extract_depends(list(), steps, 4L),
"toPos (4) exceeds number of steps (3)",
fixed = TRUE
)
})
it("signals bad arg types", {
f <- .extract_depends
expect_error(
f("not a list", steps),
"params must be a list"
)
expect_error(
f(list(), steps = list("not a character")),
"steps must be a character vector"
)
})
})
# -------------
# State updates
# -------------
describe(".pip_restart", {
counter_env <- function(...) {
env <- new.env(parent = emptyenv())
vals <- list(...)
for (nm in names(vals)) {
env[[nm]] <- vals[[nm]]
}
env
}
it("signals invalid inputs", {
p <- pip_new()
expect_error(
p$restart(force = "yes"),
"force must be a single logical value"
)
expect_error(
p$restart(force = c(TRUE, FALSE)),
"force must be a single logical value"
)
expect_error(
p$restart(times = 0),
"times must be a single integer value >= 1"
)
expect_error(
p$restart(times = NA),
"times must be a single integer value >= 1"
)
expect_error(
p$restart(times = "a"),
"times must be a single integer value >= 1"
)
expect_error(
p$restart(times = c(1, 2)),
"times must be a single integer value >= 1"
)
})
it("restarts the pipeline when a step requests a restart", {
c <- counter_env(n = 0L)
p <- pip_new() |>
pip_add("s1", function(x = 1) {
c[["n"]] <- c[["n"]] + 1L
if (c[["n"]] == 1L) {
.self$restart()
}
c[["n"]]
})
pip_run(p, lgr = NULL)
expect_equal(c[["n"]], 2L)
expect_equal(get_run_state(p), "ready")
})
it("stops recursive restarts after the announced times", {
c <- counter_env(n = 0L)
p <- pip_new() |>
pip_add("s1", function(x = 1) {
c[["n"]] <- c[["n"]] + 1L
.self$restart(times = 2L)
c[["n"]]
})
pip_run(p, lgr = NULL)
expect_equal(c[["n"]], 3L)
expect_equal(p[["pipenv"]][[".restart_count"]], 0L)
})
it("re-runs all steps on restart when force = TRUE", {
c <- counter_env(a = 0L, b = 0L, cc = 0L)
p <- pip_new() |>
pip_add("a", function(x = 1) {
c[["a"]] <- c[["a"]] + 1L
x
}) |>
pip_add("b", function(x = ~a) {
c[["b"]] <- c[["b"]] + 1L
if (c[["b"]] == 1L) {
.self$restart(force = TRUE)
}
x + 1
}) |>
pip_add("cc", function(x = ~b) {
c[["cc"]] <- c[["cc"]] + 1L
x + 1
})
pip_run(p, lgr = NULL)
expect_equal(c[["a"]], 2L)
expect_equal(c[["b"]], 2L)
expect_equal(c[["cc"]], 1L)
expect_equal(unname(p[["out"]]), list(1, 2, 3))
})
it("skips already done steps on restart when force = FALSE", {
c <- counter_env(a = 0L, b = 0L, cc = 0L)
p <- pip_new() |>
pip_add("a", function(x = 1) {
c[["a"]] <- c[["a"]] + 1L
x
}) |>
pip_add("b", function(x = ~a) {
c[["b"]] <- c[["b"]] + 1L
if (c[["b"]] == 1L) {
.self$restart(force = FALSE)
}
x + 1
}) |>
pip_add("cc", function(x = ~b) {
c[["cc"]] <- c[["cc"]] + 1L
x + 1
})
pip_run(p, lgr = NULL)
expect_equal(c[["a"]], 1L)
expect_equal(c[["b"]], 1L)
expect_equal(c[["cc"]], 1L)
expect_equal(unname(p[["out"]]), list(1, 2, 3))
})
it("marks the pipeline as restarting when called before a run", {
p <- pip_new() |>
pip_add("s1", \(x = 1) x)
p$restart()
expect_equal(get_run_state(p), "restart")
expect_equal(p[["pipenv"]][[".restart_count"]], 1L)
logs <- character(0)
lgr <- function(level, msg) logs <<- c(logs, msg)
pip_run(p, lgr = lgr)
expect_true(any(grepl("Restarting run", logs)))
expect_equal(get_run_state(p), "ready")
})
it("restarts the underlying pipeline when called on a view", {
p <- pip_new() |>
pip_add("s1", \(x = 1) x)
v <- pip_view(p, step = "s1")
v$restart()
expect_equal(get_run_state(p), "restart")
expect_equal(p[["pipenv"]][[".restart_count"]], 1L)
expect_identical(v[["pipenv"]][["data"]], p[["pipenv"]][["data"]])
})
it("restarts a view run when a step requests a restart", {
c <- counter_env(n = 0L)
p <- pip_new("view-pipeline") |>
pip_add("s1", function(x = 1) {
c[["n"]] <- c[["n"]] + 1L
if (c[["n"]] == 1L) {
.self$restart()
}
x
}) |>
pip_add("s2", function(x = ~s1) x + 1)
v <- pip_view(p, step = "s2")
pip_run(v, lgr = NULL)
expect_equal(c[["n"]], 2L)
expect_equal(unname(p[["out"]]), list(1, 2))
expect_equal(get_run_state(p), "ready")
})
it("restarts without declaring .self in the step signature", {
c <- counter_env(n = 0L)
p <- pip_new() |>
pip_add("s1", function(x = 1) {
c[["n"]] <- c[["n"]] + 1L
if (c[["n"]] == 1L) {
.self$restart()
}
x
})
pip_run(p, lgr = NULL)
expect_equal(c[["n"]], 2L)
expect_equal(get_run_state(p), "ready")
})
it("re-runs done steps downstream of steps executed before restart", {
c <- counter_env(n = 0L)
p <- pip_new() |>
pip_add("s1", function(x = 1) {
c[["n"]] <- c[["n"]] + 1L
if (c[["n"]] == 1L) {
.self$restart(force = FALSE)
}
x
}) |>
pip_add("s2", function(x = ~s1) x + 1)
p[["s2", "state"]] <- "done"
pip_run(p, lgr = NULL)
expect_equal(c[["n"]], 1L)
expect_equal(unname(p[["out"]]), list(1, 2))
expect_equal(unname(p[["state"]]), c("done", "done"))
})
})
describe(".pip_halt", {
it("marks the pipeline as halted when called before a run", {
p <- pip_new() |>
pip_add("s1", \(x = 1) x)
p$halt()
expect_equal(get_run_state(p), "halted")
expect_equal(get_run_state(p), "halted")
})
it("aborts the run at the halting step and marks downstream outdated", {
p <- pip_new() |>
pip_add("s1", function(x = 1) x) |>
pip_add("s2", function(x = ~s1) {
.self$halt()
x + 1
}) |>
pip_add("s3", function(x = ~s2) x + 1)
p[["s3", "state"]] <- "done"
pip_run(p, lgr = NULL)
expect_equal(unname(p[["out"]]), list(1, 2, NULL))
expect_equal(
unname(p[["state"]]),
c("done", "done", "outdated")
)
expect_equal(get_run_state(p), "halted")
})
it("continues after the halt in the next run", {
p <- pip_new() |>
pip_add("s1", function(x = 1) x) |>
pip_add("s2", function(x = ~s1) {
.self$halt()
x + 1
}) |>
pip_add("s3", function(x = ~s2) x + 1)
pip_run(p, lgr = NULL)
expect_equal(get_run_state(p), "halted")
pip_run(p, lgr = NULL)
expect_equal(unname(p[["out"]]), list(1, 2, 3))
expect_equal(unname(p[["state"]]), rep("done", 3))
expect_equal(get_run_state(p), "ready")
})
it("keeps the halted run state if steps failed before the halt", {
p <- pip_new() |>
pip_add("s1", function(x = 1) stop("boom")) |>
pip_add("s2", function(x = 1) {
.self$halt()
x
}) |>
pip_add("s3", function(x = 1) x)
expect_warning(
pip_run(p, lgr = NULL, on_error = "continue"),
class = "pipeflow_run_failed"
)
expect_equal(get_run_state(p), "halted")
expect_equal(unname(p[["state"]]), c("failed", "done", "new"))
})
it("leaves new steps not reached due to the halt as new", {
p <- pip_new() |>
pip_add("s1", function(x = 1) x) |>
pip_add("s2", function(x = ~s1) {
.self$halt()
x + 1
}) |>
pip_add("s3", function(x = ~s2) x + 1)
pip_run(p, lgr = NULL)
expect_equal(unname(p[["state"]]), c("done", "done", "new"))
})
it("logs the manual halt message during the run", {
p <- pip_new() |>
pip_add("s1", function(x = 1) x) |>
pip_add("s2", function(x = ~s1) {
.self$halt()
x + 1
}) |>
pip_add("s3", function(x = ~s2) x + 1)
logs <- character(0)
lgr <- function(level, msg) logs <<- c(logs, msg)
pip_run(p, lgr = lgr)
expect_true(any(
grepl("Aborting pipeline execution on manual halt", logs)
))
expect_true(any(grepl("Step 1/3 s1", logs)))
expect_true(any(grepl("Step 2/3 s2", logs)))
expect_false(any(grepl("Step 3/3 s3", logs)))
})
it("does not execute steps after the halting step", {
ran <- character(0)
p <- pip_new() |>
pip_add("s1", function(x = 1) {
ran <<- c(ran, "s1")
x
}) |>
pip_add("s2", function(x = ~s1) {
ran <<- c(ran, "s2")
.self$halt()
x + 1
}) |>
pip_add("s3", function(x = ~s2) {
ran <<- c(ran, "s3")
x + 1
})
pip_run(p, lgr = NULL)
expect_equal(ran, c("s1", "s2"))
})
it("halts at the first step and marks all later steps outdated", {
p <- pip_new() |>
pip_add("s1", function(x = 1) {
.self$halt()
x
}) |>
pip_add("s2", function(x = ~s1) x + 1) |>
pip_add("s3", function(x = ~s2) x + 1)
p[c("s2", "s3"), "state"] <- "done"
pip_run(p, lgr = NULL)
expect_equal(unname(p[["out"]]), list(1, NULL, NULL))
expect_equal(
unname(p[["state"]]),
c("done", "outdated", "outdated")
)
})
it("halts without declaring .self in the step signature", {
p <- pip_new() |>
pip_add("s1", function(x = 1) x) |>
pip_add("s2", function(x = ~s1) {
.self$halt()
x + 1
}) |>
pip_add("s3", function(x = ~s2) x + 1)
pip_run(p, lgr = NULL)
expect_equal(unname(p[["out"]]), list(1, 2, NULL))
expect_equal(
unname(p[["state"]]),
c("done", "done", "new")
)
})
it("marks the underlying pipeline as halted when called on a view", {
p <- pip_new() |>
pip_add("s1", \(x = 1) x)
v <- pip_view(p, step = "s1")
v$halt()
expect_equal(get_run_state(p), "halted")
expect_identical(v[["pipenv"]][["data"]], p[["pipenv"]][["data"]])
})
it("aborts a view run at the halting step", {
p <- pip_new("view-pipeline") |>
pip_add("s1", function(x = 1) x) |>
pip_add("s2", function(x = ~s1) {
.self$halt()
x + 1
}) |>
pip_add("s3", function(x = ~s2) x + 1) |>
pip_add("s4", function(x = ~s3) x + 1)
p[c("s3", "s4"), "state"] <- "done"
v <- pip_view(p, step = "s4")
pip_run(v, lgr = NULL)
expect_equal(unname(p[["out"]]), list(1, 2, NULL, NULL))
expect_equal(
unname(p[["state"]]),
c("done", "done", "outdated", "outdated")
)
expect_equal(get_run_state(p), "halted")
})
})
describe(".pip_outdate_downstream", {
test_pip <- function() {
pip_new() |>
pip_add("a1", \(x = 1) x) |>
pip_add("a2", \(x = ~a1) x) |>
pip_add("b1", \(x = 1) x) |>
pip_add("a3", \(x = ~a1) x) |>
pip_add("b2", \(x = ~b1) x) |>
pip_run(lgr = NULL)
}
it("outdates states downstream of single node as expected", {
p <- test_pip()
expect_true(all(p[["state"]] == "done"))
.pip_outdate_downstream(p, "a1")
expect_equal(
unname(p[["state"]]),
c("outdated", "outdated", "done", "outdated", "done")
)
})
it("can outdate states downstream of multiple nodes", {
p <- test_pip()
.pip_outdate_downstream(p, steps = c("a1", "b1"))
expect_true(all(p[["state"]] == "outdated"))
p <- test_pip()
.pip_outdate_downstream(p, steps = c("a1", "b2"))
expect_equal(
unname(p[["state"]]),
c("outdated", "outdated", "done", "outdated", "outdated")
)
})
it("does not outdate the steps listed in 'keep'", {
p <- test_pip()
.pip_outdate_downstream(p, steps = "a1", keep = c("a1", "a3"))
expect_equal(
unname(p[["state"]]),
c("done", "outdated", "done", "done", "done")
)
})
it("leaves new steps as new", {
p <- test_pip()
pip_reset(p[c("a1", "a3")])
.pip_outdate_downstream(p, steps = "a1")
expect_equal(
unname(p[["state"]]),
c("new", "outdated", "done", "new", "done")
)
})
it("ignores unknown steps and handles empty input", {
p <- test_pip()
.pip_outdate_downstream(p, steps = c("nope", character()))
.pip_outdate_downstream(p, steps = character())
expect_true(all(p[["state"]] == "done"))
})
})
# ---------------------------
# Step lookup & DAG traversal
# ---------------------------
describe(".pip_steps_to_rows", {
test_pip <- function() {
pip_new() |>
pip_add("s1", \(x = 1) x) |>
pip_add("s2", \(x = ~s1) x) |>
pip_add("s3", \(x = ~s2) x)
}
it("maps step names to row positions for pipelines and views", {
p <- test_pip()
expect_equal(.pip_steps_to_rows(p, c("s3", "s1")), c(3L, 1L))
v <- pip_view(p, step = c("s2", "s3"))
expect_equal(.pip_steps_to_rows(v, c("s1", "s3")), c(1L, 3L))
})
it("signals invalid step selectors with default messages", {
p <- test_pip()
expect_error(
.pip_steps_to_rows(p, c("s1", "")),
"step names must be non-empty strings"
)
expect_error(
.pip_steps_to_rows(p, c("s1", NA_character_)),
"step names must not contain NA"
)
expect_error(
.pip_steps_to_rows(p, c("s1", "unknown")),
"Unknown step names: unknown"
)
})
it("uses simple error messages consistently", {
p <- test_pip()
expect_error(
.pip_steps_to_rows(p, c("s1", "")),
"step names must be non-empty strings"
)
expect_error(
.pip_steps_to_rows(p, c("s1", "unknown")),
"Unknown step names: unknown"
)
})
})
describe(".pip_view_rows", {
test_pip <- function() {
pip_new() |>
pip_add("s1", \(x = 1) x) |>
pip_add("s2", \(x = ~s1) x) |>
pip_add("s3", \(x = ~s2) x)
}
it("returns all rows for a pipeline", {
expect_equal(.pip_view_rows(test_pip()), 1:3)
expect_equal(.pip_view_rows(pip_new()), integer(0))
})
it("returns the rows of the steps covered by a view", {
p <- test_pip()
pipenv <- p[["pipenv"]]
view <- .pip_rows_to_view(pipenv, c(3L, 1L))
v <- .wrap_pipenv(pipenv, "pipe", view = view)
# The selector holds node ids, the rows come in pipeline order
expect_equal(view, unname(p[["nodeId"]][c(3L, 1L)]))
expect_equal(.pip_view_rows(v), c(1L, 3L))
})
it("follows the steps when rows shift and drops removed steps", {
p <- test_pip()
v <- pip_view(p, step = c("s2", "s3"))
pip_add(p, "s0", \(x = 0) x, after = 0)
expect_equal(.pip_view_rows(v), c(3L, 4L))
pip_remove(p, "s3")
expect_equal(.pip_view_rows(v), 3L)
pip_remove(p, "s2")
expect_equal(.pip_view_rows(v), integer(0))
expect_true(.is_pipeflow_view(v))
})
})
describe(".pip_dag", {
# Node ids a = 0, a2 = 4, b = 1, c = 2, d = 3; id 5 was removed with e
test_pip <- function() {
p <- pip_new("p") |>
pip_add("a", \(x = 1) x) |>
pip_add("b", \(x = ~a) x + 1) |>
pip_add("c", \(x = ~b, k = 2) x * k) |>
pip_add("d", \(x = ~a, y = ~c) x + y)
pip_add(p, "a2", \(x = ~a) x * 10, after = "a")
pip_add(p, "e", \(x = ~d) x)
pip_remove(p, "e")
}
it("returns the DAG of a pipeline as it is", {
pipenv <- test_pip()[["pipenv"]]
expect_identical(.pip_dag(pipenv), pipenv[[".dag"]])
})
it("rebuilds the DAG of a read pipeline with the same node ids", {
p <- test_pip()
pipenv <- save_and_read(p)[["pipenv"]]
expect_true(dag_is_null(pipenv[[".dag"]]))
d <- .pip_dag(pipenv)
expect_false(dag_is_null(d))
expect_identical(pipenv[[".dag"]], d)
expect_equal(pipenv[["data"]][["nodeId"]], c(0L, 4L, 1L, 2L, 3L))
expect_equal(dag_structure(d), dag_structure(p[["pipenv"]][[".dag"]]))
})
it("keeps the node ids of removed steps unused", {
p <- test_pip()
q <- save_and_read(p)
pip_add(p, "f", \(x = ~d) x)
pip_add(q, "f", \(x = ~d) x)
expect_equal(q[["nodeId"]][["f"]], 6L)
expect_equal(q[["nodeId"]], p[["nodeId"]])
})
it("rebuilds the DAG of an empty pipeline", {
pipenv <- save_and_read(pip_new())[["pipenv"]]
expect_equal(dag_size(.pip_dag(pipenv)), 0L)
})
it("signals a step to node map that does not match the step table", {
msg <- "the step to node map does not match the step table"
pipenv <- save_and_read(test_pip())[["pipenv"]]
rm("b", envir = pipenv[[".steps_to_nodes"]])
expect_error(.pip_dag(pipenv), msg)
pipenv <- save_and_read(test_pip())[["pipenv"]]
pipenv[[".steps_to_nodes"]][["b"]] <- 7L
expect_error(.pip_dag(pipenv), msg)
pipenv <- save_and_read(test_pip())[["pipenv"]]
pipenv[[".steps_to_nodes"]][["zz"]] <- 7L
expect_error(.pip_dag(pipenv), msg)
expect_true(dag_is_null(pipenv[[".dag"]]))
})
it("signals steps that depend on unknown steps", {
pipenv <- save_and_read(test_pip())[["pipenv"]]
data.table::set(
pipenv[["data"]],
i = 3L,
j = "depends",
value = list(list(c(x = "zz")))
)
expect_error(.pip_dag(pipenv), "steps depend on unknown steps")
})
})
# -----------------
# Pipeline addition
# -----------------
# ---------------------------
# Exported pipeline functions
# ---------------------------
describe("pip_new", {
it("creates a pipeflow pipeline with expected base structure", {
p <- pip_new()
expect_true(.is_pipeflow(p))
expect_true(is.list(p))
expect_equal(p[["name"]], "pipe")
expect_null(p[["view"]])
expect_true(is.environment(p[["pipenv"]]))
expect_true(data.table::is.data.table(p[["pipenv"]][["data"]]))
expect_equal(nrow(p[["pipenv"]][["data"]]), 0L)
expect_true(is.environment(p[["pipenv"]][[".steps_to_nodes"]]))
expect_equal(
ls(envir = p[["pipenv"]][[".steps_to_nodes"]]),
character(0)
)
expect_equal(length(dag_get_nodes_order(p[["pipenv"]][[".dag"]])), 0L)
})
it("supports custom pipeline names", {
p <- pip_new("custom")
expect_equal(p[["name"]], "custom")
})
it("signals invalid names", {
expect_error(pip_new(c("a", "b")), "name must be a single string")
expect_error(pip_new(1), "name must be a single string")
expect_error(pip_new(NA_character_), "name must not be NA")
})
})
describe("pipelines read with readRDS()", {
is_formula <- \(x) inherits(x, "formula")
# Node ids a = 0, a2 = 4, b = 1, c = 2, d = 3; id 5 was removed with e
test_pip <- function() {
p <- pip_new("p") |>
pip_add("a", \(x = 1) x) |>
pip_add("b", \(x = ~a) x + 1) |>
pip_add("c", \(x = ~b, k = 2) x * k) |>
pip_add("d", \(x = ~a, y = ~c) x + y)
pip_add(p, "a2", \(x = ~a) x * 10, after = "a")
pip_add(p, "e", \(x = ~d) x)
pip_remove(p, "e")
}
# Read pipeline `q` has the same steps and DAG as pipeline `p`. Params
# that refer to steps are compared via `depends`.
expect_same_pip <- function(q, p) {
cols <- c("step", "out", "state", "tags", "locked", "exec", "depends")
expect_equal(q[, cols], p[, cols])
unbound <- \(x) lapply(x[["params"]], Filter, f = Negate(is_formula))
expect_equal(unbound(q), unbound(p))
expect_equal(q[["nodeId"]], p[["nodeId"]])
expect_equal(
dag_structure(.pip_dag(q[["pipenv"]])),
dag_structure(.pip_dag(p[["pipenv"]]))
)
}
it("can get new steps with pip_add()", {
p <- test_pip()
q <- save_and_read(p)
for (x in list(p, q)) {
pip_add(x, "f", \(x = ~d) x - 1)
pip_add(x, "b2", \(x = ~b) x / 2, after = "b")
}
expect_equal(
q[["nodeId"]],
c(a = 0L, a2 = 4L, b = 1L, b2 = 7L, c = 2L, d = 3L, f = 6L)
)
expect_same_pip(q, p)
})
it("can be run with pip_run()", {
p <- test_pip()
q <- save_and_read(p)
pip_run(p, lgr = NULL)
expect_no_warning(pip_run(q, lgr = NULL))
expect_equal(q[["out"]], list(a = 1, a2 = 10, b = 2, c = 4, d = 5))
expect_same_pip(q, p)
})
it("can be run through a view", {
q <- save_and_read(test_pip())
pip_run(q["c"], lgr = NULL)
expect_equal(
q[["state"]],
c(a = "done", a2 = "new", b = "done", c = "done", d = "new")
)
})
it("outdates the downstream steps of changed params", {
p <- pip_run(test_pip(), lgr = NULL)
q <- save_and_read(p)
for (x in list(p, q)) {
pip_set_params(x, list(k = 3))
}
expect_equal(
q[["state"]],
c(
a = "done",
a2 = "done",
b = "done",
c = "outdated",
d = "outdated"
)
)
expect_same_pip(q, p)
})
it("can remove steps with pip_remove()", {
p <- test_pip()
q <- save_and_read(p)
for (x in list(p, q)) {
pip_remove(x, "a2")
}
expect_same_pip(q, p)
q <- save_and_read(p)
for (x in list(p, q)) {
expect_message(
pip_remove(x, "b", force = TRUE),
"downstream dependencies: 'c', 'd'"
)
}
expect_equal(unname(q[["step"]]), "a")
expect_same_pip(q, p)
})
it("can replace steps with pip_replace()", {
p <- pip_run(test_pip(), lgr = NULL)
q <- save_and_read(p)
for (x in list(p, q)) {
pip_replace(x, "c", \(x = ~a2) x * 3)
}
expect_equal(q[["depends"]][["c"]], c(x = "a2"))
expect_equal(
q[["state"]],
c(a = "done", a2 = "done", b = "done", c = "new", d = "outdated")
)
expect_same_pip(q, p)
})
it("can be cloned with pip_clone()", {
p <- test_pip()
q <- save_and_read(p)
cl <- pip_clone(q)
expect_same_pip(cl, p)
pip_add(cl, "f", \(x = ~d) x)
expect_equal(cl[["nodeId"]][["f"]], 6L)
expect_false("f" %in% q[["step"]])
})
it("can be shown with pip_graph()", {
p <- test_pip()
q <- save_and_read(p)
expect_equal(
pip_graph(q["c"], include_upstream = TRUE),
pip_graph(p["c"], include_upstream = TRUE)
)
})
it("can be changed before the DAG is rebuilt", {
q <- save_and_read(pip_run(test_pip(), lgr = NULL))
expect_no_warning({
pip_rename(q, from = "a2", to = "ten")
pip_tag(q["b"], "tag")
pip_lock(q["b"])
pip_reset(q)
})
expect_true(dag_is_null(q[["pipenv"]][[".dag"]]))
pip_run(q, lgr = NULL)
expect_equal(q[["out"]], list(a = 1, ten = 10, b = 2, c = 4, d = 5))
expect_equal(q[["nodeId"]], c(a = 0L, ten = 4L, b = 1L, c = 2L, d = 3L))
})
it("keeps views that were saved with the pipeline", {
p <- test_pip()
saved <- save_and_read(list(p = p, v = pip_view(p, step = c("b", "c"))))
q <- saved[["p"]]
v <- saved[["v"]]
expect_identical(v[["pipenv"]], q[["pipenv"]])
expect_equal(v[["view"]], c("b", "c"))
# Running the view rebuilds the DAG of the pipeline as well
pip_run(v, lgr = NULL)
expect_false(dag_is_null(q[["pipenv"]][[".dag"]]))
expect_equal(
q[["state"]],
c(a = "done", a2 = "new", b = "done", c = "done", d = "new")
)
pip_add(q, "b2", \(x = ~b) x, after = "b")
expect_message(pip_remove(v, "c", force = TRUE), "'d'")
expect_equal(v[["view"]], "b")
expect_equal(unname(q[["step"]]), c("a", "a2", "b", "b2"))
})
it("doesn't give new steps the node ids that views hold", {
p <- pip_new("p") |>
pip_add("a", \(x = 1) x) |>
pip_add("b", \(x = ~a) x + 1)
v <- pip_view(p, step = c("a", "b"))
pip_remove(p, "b")
expect_equal(v[["view"]], "a")
saved <- save_and_read(list(p = p, v = v))
pip_add(saved[["p"]], "c", \(x = ~a) x)
expect_equal(saved[["p"]][["nodeId"]], c(a = 0L, c = 2L))
expect_equal(saved[["v"]][["view"]], "a")
})
it("restores a view saved on its own with a copy of its pipeline", {
p <- test_pip()
w <- save_and_read(pip_view(p, step = "c"))
expect_equal(w[["view"]], "c")
pip_run(w, lgr = NULL)
expect_equal(w[["c", "out"]], 4)
expect_equal(unname(p[["state"]]), rep("new", 5))
})
it("works for a pipeline that changed itself in a restarted run", {
p <- pip_new("restart") |>
pip_add("init", \(xInit = 0) xInit) |>
pip_add("f1", \(x = ~init) x + 1) |>
pip_add("f2", \(x = ~f1) {
if (x > 10) {
.self |>
pip_add("f2a", \(x = ~f1) x + 21, after = "f1") |>
pip_add("f2b", \(x = ~f2a) x + 22, after = "f2a") |>
pip_replace("f3", \(x = ~f2b) x + 30) |>
pip_remove("f2")
.self$restart()
}
x + 2
}) |>
pip_add("f3", \(x = ~f2) x + 3)
pip_set_params(p, list(xInit = 11))
pip_run(p, lgr = NULL)
expect_equal(
unname(p[["step"]]),
c("init", "f1", "f2a", "f2b", "f3")
)
expect_equal(p[["f3", "out"]], 11 + 1 + 21 + 22 + 30)
q <- save_and_read(p)
expect_same_pip(q, p)
for (x in list(p, q)) {
pip_set_params(x, list(xInit = 1))
pip_add(x, "f4", \(x = ~f3) x * 2)
pip_run(x, lgr = NULL)
}
expect_equal(q[["f4", "out"]], (1 + 1 + 21 + 22 + 30) * 2)
expect_equal(q[["nodeId"]][["f4"]], 6L)
expect_same_pip(q, p)
})
})
describe("pip_add params overlapping the formals of fun", {
it("warns if params overlap the defaults of fun", {
p <- pip_new()
expect_warning(
pip_add(p, "s", \(x = 1, y = 2) x + y, params = list(x = 5, y = 6)),
"step 's': the defaults of fun take precedence over params: x, y",
fixed = TRUE
)
expect_equal(p[["params"]][["s"]][["x"]], 1)
expect_equal(p[["params"]][["s"]][["y"]], 2)
})
it("names only the overlapping params", {
p <- pip_new()
expect_warning(
pip_add(p, "s", \(x = 1) x, params = list(x = 5, z = 6)),
"over params: x$"
)
})
it("warns if the step is inserted via after", {
p <- pip_new() |> pip_add("a", \(x = 1) x)
expect_warning(
pip_add(p, "b", \(x = 1) x, params = list(x = 5), after = 0),
"step 'b': the defaults of fun take precedence over params: x",
fixed = TRUE
)
})
it("does not warn for params that are not formals of fun", {
p <- pip_new()
expect_no_warning(
pip_add(p, "s", \(x = 1) x, params = list(z = 5))
)
})
it("does not warn for params that only match '...'", {
p <- pip_new()
expect_no_warning(
pip_add(p, "s", \(x = 1, ...) x, params = list(... = 5))
)
})
it("does not warn without params", {
p <- pip_new()
expect_no_warning(pip_add(p, "s", \(x = 1) x))
})
})
describe("pip_add", {
it("signals if step is not a single non-empty string", {
p <- pip_new()
expect_error(pip_add(p, ""), "step must be a non-empty string")
expect_error(pip_add(p, c("a", "b")), "step must be a single string")
})
it("signals if fun is not a function", {
p <- pip_new()
expect_error(
pip_add(p, "s1", fun = "not a function"),
"fun must be a function"
)
})
it("signals if fun is a primitive function", {
p <- pip_new()
expect_error(
pip_add(p, "s1", fun = sum),
"fun must not be a primitive function; wrap it"
)
expect_equal(length(p), 0L)
})
it("signals duplicate step names", {
p <- pip_new()
pip_add(p, "s1", \(a = 0) a)
expect_error(pip_add(p, "s1", \(a = 0) a), "step 's1' already exists")
})
it("signals undefined dependencies", {
p <- pip_new()
expect_error(
pip_add(p, "s1", \(x = ~undefined) x),
"cannot reference unknown steps: 'undefined'"
)
})
it("step can refer to previous step by relative number", {
p <- pip_new()
pip_add(p, "s1", \(a = 5) a)
pip_add(p, "s2", \(x = ~ -1) 2 * x)
expect_equal(p[["pipenv"]][["data"]]$depends[[2]], c(x = "s1"))
})
it("a bad relative step referal is signalled", {
p <- pip_new()
pip_add(p, "s1", \(a = 5) a)
expect_error(pip_add(p, "s2", \(x = ~ -2) 2 * x))
expect_equal(length(p), 1L)
})
it("if add is aborted, pipeline remains unchanged", {
p <- pip_new()
pip_add(p, "s1", \(a = 5) a)
expect_error(
pip_add(p, "s2", \(x = ~ -2) 2 * x),
"relative index x = ~-2 points outside pipeline",
fixed = TRUE
)
})
it("can refer to the pipeline itself via the .self argument", {
p <- pip_new()
pip_add(p, "s1", \(x = 1) .self)
pip_run(p, lgr = NULL)
expect_true(identical(p[["out"]][[1]], p))
})
it("allows functions with wildcard arguments", {
p <- pip_new() |>
pip_add("s1", \(x = 1, ...) x)
pip_set_params(p, list(x = 2))
pip_run(p, lgr = NULL)
expect_equal(p[["out"]][[1]], 2)
})
test_pip <- function() {
pip_new("pipe") |>
pip_add("f1", \(x = 1) x) |>
pip_add("f2", \(x = ~f1) x + 1)
}
it("can insert after a step name", {
p <- test_pip()
pip_add(p, "f3", \(x = ~f1) x + 10, after = "f1")
expect_equal(unname(p[["step"]]), c("f1", "f3", "f2"))
expect_equal(p[["depends"]][[2]], c(x = "f1"))
expect_equal(p[["depends"]][[3]], c(x = "f1"))
})
it("appends a step referencing the same step twice without warning", {
p <- pip_new("pipe") |>
pip_add("a", \(x = 1) x)
expect_no_warning(
pip_add(p, "b", \(x = ~a, y = ~a) x + y)
)
pip_run(p, lgr = NULL)
expect_equal(p[["out"]][[2]], 2)
})
it("inserts a step referencing the same step twice without warning", {
p <- test_pip()
expect_no_warning(
pip_add(p, "f3", \(x = ~f1, y = ~f1) x + y, after = "f1")
)
pip_run(p, lgr = NULL)
expect_equal(p[["out"]][[2]], 2)
})
it("can insert by numeric index and supports insertion at beginning", {
p <- test_pip()
pip_add(p, "f0", \(x = 0) x, after = 0)
expect_equal(unname(p[["step"]]), c("f0", "f1", "f2"))
expect_equal(p[["depends"]][[2]], character(0))
expect_equal(p[["depends"]][[3]], c(x = "f1"))
})
it("uses default insertion position at end", {
p <- test_pip()
pip_add(p, "f3", \(x = ~f2) x + 1)
expect_equal(unname(p[["step"]]), c("f1", "f2", "f3"))
expect_equal(p[["depends"]][[3]], c(x = "f2"))
})
it("returns invisibly also for default append path", {
p <- test_pip()
expect_invisible(
pip_add(p, "f3", \(x = ~f2) x + 1)
)
})
it("can append a step after pipeline was run", {
p <- test_pip()
pip_run(p, lgr = NULL)
expect_invisible(
pip_add(p, "f3", \(x = ~f2) x + 1)
)
expect_equal(unname(p[["step"]]), c("f1", "f2", "f3"))
expect_equal(p[["depends"]][[3]], c(x = "f2"))
})
it("can insert a step after pipeline was run", {
p <- test_pip()
pip_run(p, lgr = NULL)
expect_invisible(
pip_add(p, "f3", \(x = ~f1) x + 10, after = "f1")
)
expect_equal(unname(p[["step"]]), c("f1", "f3", "f2"))
expect_equal(p[["depends"]][[2]], c(x = "f1"))
expect_equal(p[["depends"]][[3]], c(x = "f1"))
})
it("keeps existing step state and output for appended tail steps", {
p <- test_pip()
data.table::set(
p[["pipenv"]][["data"]],
i = 2,
j = "out",
value = list(42)
)
data.table::set(
p[["pipenv"]][["data"]],
i = 2,
j = "state",
value = "done"
)
pip_add(p, "f3", \(x = ~f1) x + 10, after = "f1")
i <- match("f2", p[["step"]])
expect_equal(p[["out"]][[i]], 42)
expect_equal(p[["state"]][[i]], "done")
})
it("signals invalid insertion position and unknown step reference", {
p <- test_pip()
expect_error(
pip_add(p, "f3", \(x = 1) x, after = "unknown"),
"step 'unknown' does not exist"
)
expect_error(
pip_add(p, "f3", \(x = 1) x, after = 3.1),
"after index must be a whole number"
)
expect_error(
pip_add(p, "f3", \(x = 1) x, after = -1),
"after index must be between 0 and 2"
)
expect_error(
pip_add(p, "f3", \(x = 1) x, after = 3),
"after index must be between 0 and 2"
)
expect_error(
pip_add(p, "f3", \(x = ~f2) x, after = "f1"),
"cannot reference unknown steps: 'f2'"
)
})
it("signals duplicate step names also when inserting at position", {
p <- test_pip()
expect_error(
pip_add(p, "f2", \(x = 1) x, after = "f1"),
"step 'f2' already exists in the pipeline"
)
})
it("merges params with function defaults, letting defaults win", {
p <- pip_new()
expect_warning(
pip_add(
p,
"s1",
function(x = 1, y = 2) x + y,
params = list(x = 10, y = 20, z = 99)
),
"defaults of fun take precedence over params: x, y"
)
pars <- p[["params"]][[1]]
expect_equal(pars[["x"]], 1) # fun default takes precedence
expect_equal(pars[["y"]], 2)
expect_equal(pars[["z"]], 99) # extra param carried through
})
it("uses extra params passed via ... at runtime", {
p <- pip_new()
pip_add(
p,
"s1",
function(x = 1, ...) c(x, ...),
params = list(size = 42)
)
pars <- p[["params"]][[1]]
expect_equal(pars[["x"]], 1)
expect_equal(pars[["size"]], 42)
pip_run(p, lgr = NULL)
expect_equal(unname(p[["out"]][[1]]), c(1, 42))
})
it("resolves step references given via params", {
p <- pip_new()
pip_add(p, "s1", function(x = 1) x)
pip_add(
p,
"s2",
function(...) list(...),
params = list(y = ~s1)
)
expect_equal(unname(p[["depends"]][[2]]), "s1")
expect_equal(names(p[["depends"]][[2]]), "y")
pip_run(p, lgr = NULL)
expect_equal(p[["out"]][[2]]$y, 1)
})
it("resolves relative step references given via params", {
p <- pip_new()
pip_add(p, "s1", function(x = 5) x)
pip_add(
p,
"s2",
function(...) list(...),
params = list(y = ~ -1)
)
expect_equal(unname(p[["depends"]][[2]]), "s1")
pip_run(p, lgr = NULL)
expect_equal(p[["out"]][[2]]$y, 5)
})
it("signals unknown steps referenced via params", {
p <- pip_new()
expect_error(
pip_add(
p,
"s1",
function(...) list(...),
params = list(x = ~undefined)
),
"cannot reference unknown steps: 'undefined'"
)
})
it("keeps referenced params out of the independent params", {
p <- pip_new()
pip_add(p, "s1", function(x = 1) x)
pip_add(
p,
"s2",
function(...) list(...),
params = list(y = ~s1, z = 2)
)
expect_equal(
names(p[["params"]][[2]]),
c("y", "z")
)
# y is a dependency, so it is not unbound
expect_equal(p[["unbound"]][[2]], "z")
expect_equal(names(pip_get_params(p)), c("x", "z"))
})
it("outdates the step state if one of the params is updated", {
p <- pip_new()
pip_add(
p,
"s1",
function(x = 1, y = 2, ...) x + y,
params = list(z = 99)
)
pip_run(p, lgr = NULL)
expect_equal(p[["state"]][[1]], "done")
pip_set_params(p, list(z = 100))
expect_equal(p[["state"]][[1]], "outdated")
})
it("signals parameter without default value", {
p <- pip_new()
expect_error(
pip_add(p, "s1", function(x) x),
"has no default value"
)
})
it("auto-provides .self to steps that do not declare it", {
p <- pip_new() |>
pip_add("s1", function(x = 1) .self)
pip_run(p, lgr = NULL)
expect_true(identical(p[["out"]][[1]], p))
expect_false(
".self" %in% names(p[["params"]][[1]])
)
expect_false(
".self" %in% names(formals(p[["fun"]][[1]]))
)
})
it("signals if .self is declared as a step parameter", {
p <- pip_new()
expect_error(
pip_add(p, "s1", function(x = 1, .self = NULL) x),
paste(
"'.self' is a reserved parameter name and must not be",
"declared in step 's1' - it is provided automatically"
),
fixed = TRUE
)
})
it("signals if .self is provided via params", {
p <- pip_new()
expect_error(
pip_add(p, "s1", function(x = 1) x, params = list(.self = p)),
"'.self' is a reserved parameter name"
)
})
it("refreshes auto-provided .self to the clone when run", {
p <- pip_new() |>
pip_add("s1", function(x = 1) .self)
pip_run(p, lgr = NULL)
p2 <- pip_clone(p)
pip_run(p2, lgr = NULL, force = TRUE)
expect_true(identical(p2[["out"]][[1]], p2))
expect_true(identical(p[["out"]][[1]], p))
})
it("stores the step function as given", {
f <- \(x = 1) .self[["name"]]
p <- pip_new() |> pip_add("s1", f)
expect_identical(p[["s1", "fun"]], f)
})
})
describe("pip_add exec modes", {
it("signals invalid exec modes", {
p <- pip_new()
expect_error(
pip_add(p, "s1", \(x = 1) x, exec = "invalid"),
"exec must be one of: auto, split, reduce, plain"
)
expect_error(
pip_add(p, "s1", \(x = 1) x, exec = NA_character_),
"exec must be a single string"
)
expect_error(
pip_add(p, "s1", \(x = 1) x, exec = c("auto", "split")),
"exec must be a single string"
)
})
it("accepts all valid exec modes", {
for (mode in c("auto", "split", "reduce", "plain")) {
p <- pip_new()
expect_no_error(pip_add(p, "s1", \(x = 1) x, exec = mode))
expect_equal(p[["exec"]][[1]], mode)
}
})
})
describe("pip_add try() references", {
it("resolves the referenced step and keeps the formula in the params", {
p <- pip_new() |>
pip_add("a", \(x = 1) x) |>
pip_add("a2", \(x = 2) x) |>
pip_add("a3", \(x = 3) x) |>
pip_add("b", \(x = ~a, y = ~ try(a2), z = ~ try(-1)) x)
expect_equal(p[["b", "depends"]], c(x = "a", y = "a2", z = "a3"))
expect_equal(p[["b", "params"]][["y"]], ~ try(a2), ignore_attr = TRUE)
})
it("resolves the referenced step when inserting a step", {
p <- pip_new() |>
pip_add("a", \(x = 1) x) |>
pip_add("c", \(x = ~a) x) |>
pip_add("b", \(x = ~ try(a)) x, after = "a")
expect_equal(unname(p[["step"]]), c("a", "b", "c"))
expect_equal(p[["b", "depends"]], c(x = "a"))
})
it("signals references to unknown steps", {
p <- pip_new() |>
pip_add("a", \(x = 1) x)
expect_error_fixed(
pip_add(p, "b", \(x = ~ try(nope)) x),
"cannot reference unknown steps: 'nope'"
)
expect_error_fixed(
pip_add(p, "b", \(x = ~ try(a, silent = TRUE)) x),
"cannot reference unknown steps: 'try(a, silent = TRUE)'"
)
expect_equal(unname(p[["step"]]), "a")
})
})
describe("pip_rename", {
test_pip <- function() {
pip_new("pipe") |>
pip_add("f1", \(a = 1) a) |>
pip_add("f2", \(b = ~f1) b) |>
pip_add("f3", \(a = ~f1, b = ~f2) a + b)
}
it("signals invalid inputs", {
p <- test_pip()
expect_error(pip_rename(1, "f1", "first"))
expect_error(pip_rename(p, c("f1", "f2"), "first"))
expect_error(pip_rename(p, NA_character_, "first"))
expect_error(pip_rename(p, "", "first"))
expect_error(pip_rename(p, "f1", c("a", "b")))
expect_error(pip_rename(p, "f1", NA_character_))
expect_error(pip_rename(p, "f1", ""))
})
it("signals missing old step and name clashes", {
p <- test_pip()
expect_error(
pip_rename(p, "unknown", "first"),
"step 'unknown' does not exist"
)
expect_error(
pip_rename(p, "f1", "f2"),
"step 'f2' already exists"
)
})
it("renames step and updates dependencies", {
p <- test_pip()
pip_rename(p, from = "f1", to = "first")
expect_equal(unname(p[["step"]]), c("first", "f2", "f3"))
expect_equal(
unname(p[["depends"]]),
list(
character(0),
c(b = "first"),
c(a = "first", b = "f2")
)
)
})
it("updates the formula references in params", {
p <- test_pip()
pip_rename(p, from = "f1", to = "first")
expect_equal(
lapply(unname(p[["params"]]), \(par) {
vapply(par, deparse1, character(1))
}),
list(
c(a = "1"),
c(b = "~first"),
c(a = "~first", b = "~f2")
)
)
})
it("keeps DAG and step mapping consistent", {
p <- test_pip()
pip_rename(p, from = "f1", to = "first")
expect_true(is.na(.pip_steps_to_nodes(p, "f1")[[1]]))
expect_true(!is.na(.pip_steps_to_nodes(p, "first")[[1]]))
nodes <- .pip_get_reachable_nodes(p, "first")
steps <- .pip_filter_nodes(p, nodes)[["step"]]
expect_setequal(steps, c("first", "f2", "f3"))
})
it("renames a step through a view", {
p <- test_pip()
v <- pip_view(p, step = c("f2", "f3"))
expect_error(
pip_rename(v, from = "f1", to = "first"),
"step 'f1' is not part of the view"
)
pip_rename(v, from = "f2", to = "second")
expect_equal(unname(p[["step"]]), c("f1", "second", "f3"))
expect_equal(
unname(p[["depends"]]),
list(
character(0),
c(b = "f1"),
c(a = "f1", b = "second")
)
)
})
it("returns the pipeline as-is if from and to are identical", {
p <- test_pip()
pip_rename(p, from = "f1", to = "f1")
expect_equal(unname(p[["step"]]), c("f1", "f2", "f3"))
expect_equal(
unname(p[["depends"]]),
list(
character(0),
c(b = "f1"),
c(a = "f1", b = "f2")
)
)
})
it("keeps the params of a single-step pipeline intact", {
p <- pip_new("pipe") |>
pip_add("s1", \(x = 5) x * 2)
pip_rename(p, from = "s1", to = "first")
expect_equal(p[["params"]][[1]], list(x = 5))
expect_equal(
p[["pipenv"]][["data"]][["params"]][[1]],
list(x = 5)
)
})
it("runs a bound pipeline whose steps were renamed", {
p <- pip_new("pipe") |>
pip_add("a", \(x = 1) x) |>
pip_add("b", \(x = ~a) x + 1)
a <- pip_new("a") |>
pip_add("a", \(x = 5) x * 2)
b <- pip_new("b") |>
pip_add("a", \(x = 1) x)
out <- rbind(p, a, b)
pip_run(out, lgr = NULL)
expect_equal(
pip_collect(out),
list(a = 1, b = 2, a2 = 10, a3 = 1)
)
})
})
describe("pip_remove", {
test_pip <- function() {
pip_new("pipe") |>
pip_add("f1", \(x = 1) x) |>
pip_add("f2", \(x = ~f1) x) |>
pip_add("f3", \(x = ~f2) x) |>
pip_add("f4", \(x = ~f1) x) |>
pip_add("g1", \(x = 9) x)
}
it("signals invalid inputs", {
p <- test_pip()
expect_error(pip_remove(1, "f1"), "x must be a pipeflow pip")
expect_error(
pip_remove(p, c("f1", "f2")),
"step must be a single string"
)
expect_error(pip_remove(p, NA_character_), "step must not be NA")
expect_error(pip_remove(p, "unknown"), "step 'unknown' does not exist")
expect_error(
pip_remove(p, "f1", force = NA),
"force must be a single logical value"
)
expect_error(
pip_remove(p, "f1", force = c(TRUE, FALSE)),
"force must be a single logical value"
)
})
it("removes a leaf step", {
p <- test_pip()
node <- as.integer(.pip_steps_to_nodes(p, "g1")[[1]])
beforeOrder <- dag_get_nodes_order(p[["pipenv"]][[".dag"]])
beforeReach <- .pip_filter_nodes(
p,
.pip_get_reachable_nodes(p, "f1")
)[["step"]]
expect_true(dag_has_node(p[["pipenv"]][[".dag"]], node))
expect_true(node %in% beforeOrder)
expect_setequal(beforeReach, c("f1", "f2", "f3", "f4"))
pip_remove(p, "g1")
afterOrder <- dag_get_nodes_order(p[["pipenv"]][[".dag"]])
afterReach <- .pip_filter_nodes(
p,
.pip_get_reachable_nodes(p, "f1")
)[["step"]]
expect_equal(
unname(p[["step"]]),
c("f1", "f2", "f3", "f4")
)
expect_true(is.na(.pip_steps_to_nodes(p, "g1")[[1]]))
expect_false(dag_has_node(p[["pipenv"]][[".dag"]], node))
expect_false(node %in% afterOrder)
expect_equal(length(afterOrder), length(beforeOrder) - 1L)
expect_setequal(afterReach, c("f1", "f2", "f3", "f4"))
})
it("errors when direct downstream dependencies exist", {
p <- test_pip()
expect_error(
pip_remove(p, "f1"),
paste(
"cannot remove step 'f1' because the following",
"steps depend on it: 'f2', 'f4'"
)
)
})
it("removes step and all downstream dependencies recursively", {
p <- test_pip()
steps <- c("f1", "f2", "f3", "f4", "g1")
nodeMap <- vapply(
steps,
FUN = \(s) as.integer(.pip_steps_to_nodes(p, s)[[1]]),
FUN.VALUE = integer(1)
)
beforeOrder <- dag_get_nodes_order(p[["pipenv"]][[".dag"]])
dag <- p[["pipenv"]][[".dag"]]
expect_true(all(vapply(
nodeMap,
FUN = \(nid) dag_has_node(dag, nid),
FUN.VALUE = logical(1)
)))
out <- utils::capture.output(
pip_remove(p, "f1", force = TRUE),
type = "message"
)
afterOrder <- dag_get_nodes_order(dag)
remainingNode <- as.integer(nodeMap[["g1"]])
expect_equal(unname(p[["step"]]), "g1")
expect_equal(unname(p[["nodeId"]]), remainingNode)
expect_equal(afterOrder, remainingNode)
expect_equal(length(afterOrder), length(beforeOrder) - 4L)
expect_true(dag_has_node(dag, remainingNode))
expect_false(dag_has_node(dag, as.integer(nodeMap[["f1"]])))
expect_false(dag_has_node(dag, as.integer(nodeMap[["f2"]])))
expect_false(dag_has_node(dag, as.integer(nodeMap[["f3"]])))
expect_false(dag_has_node(dag, as.integer(nodeMap[["f4"]])))
expect_equal(
.pip_filter_nodes(p, .pip_get_reachable_nodes(p, "g1"))[["step"]],
"g1"
)
expect_equal(
out,
paste(
"Removing step 'f1' and its downstream dependencies:",
"'f2', 'f3', 'f4'"
)
)
})
it("removes a step through a view, which stays valid", {
p <- test_pip()
v <- pip_view(p, step = c("f4", "g1"))
expect_equal(v[["view"]], c("f4", "g1"))
expect_error(
pip_remove(v, "f1"),
"step 'f1' is not part of the view"
)
pip_remove(v, "f4")
expect_equal(
unname(p[["step"]]),
c("f1", "f2", "f3", "g1")
)
expect_equal(v[["step"]], c(g1 = "g1"))
expect_equal(v[["view"]], "g1")
})
it("force-removes downstream steps through a view", {
p <- test_pip()
v <- pip_view(p, step = c("f1", "f2"))
suppressMessages(v <- pip_remove(v, "f1", force = TRUE))
expect_equal(unname(p[["step"]]), "g1")
expect_equal(length(v), 0L)
})
})
describe("pip_replace", {
test_pip <- function() {
pip_new("pipe") |>
pip_add("f1", \(x = 1) x) |>
pip_add("f2", \(x = 2) x) |>
pip_add("f3", \(x = ~f2) x + 1)
}
it("warns if params overlap the defaults of fun", {
p <- test_pip()
expect_warning(
pip_replace(p, "f2", \(x = 20) x, params = list(x = 5)),
"step 'f2': the defaults of fun take precedence over params: x",
fixed = TRUE
)
expect_equal(p[["params"]][["f2"]][["x"]], 20)
})
it("does not warn if params do not overlap the defaults of fun", {
p <- test_pip()
expect_no_warning(
pip_replace(p, "f2", \(x = 20) x, params = list(z = 5))
)
})
it("signals invalid inputs", {
p <- test_pip()
expect_error(
pip_replace(1, "f1", \(x = 1) x),
"x must be a pipeflow pip"
)
expect_error(
pip_replace(p, c("f1", "f2"), \(x = 1) x),
"step must be a single string"
)
expect_error(
pip_replace(p, NA_character_, \(x = 1) x),
"step must not be NA"
)
expect_error(
pip_replace(p, "", \(x = 1) x),
"step must be a non-empty string"
)
expect_error(
pip_replace(p, "unknown", \(x = 1) x),
"step 'unknown' does not exist"
)
expect_error(pip_replace(p, "f1", 1), "fun must be a function")
})
it("signals if fun is a primitive function", {
p <- test_pip()
expect_error(
pip_replace(p, "f1", sum),
"fun must not be a primitive function; wrap it"
)
expect_error(
p[["f1", "fun"]] <- sum,
"fun must not be a primitive function; wrap it"
)
expect_false(is.primitive(p[["f1", "fun"]]))
})
it("stores the new function as given", {
p <- test_pip()
f <- \(x = 3) .self[["name"]]
pip_replace(p, "f2", f)
expect_identical(p[["f2", "fun"]], f)
})
it("replaces a step in-place while keeping the original order", {
p <- test_pip()
pip_run(p, lgr = NULL)
expect_equal(p[["pipenv"]][["data"]][step == "f3", out][[1]], 3)
pip_replace(p, "f2", \(x = 4) x * 2)
expect_equal(unname(p[["step"]]), c("f1", "f2", "f3"))
pip_run(p, lgr = NULL)
expect_equal(p[["pipenv"]][["data"]][step == "f2", out][[1]], 8)
expect_equal(p[["pipenv"]][["data"]][step == "f3", out][[1]], 9)
})
it("verifies replacement dependencies against earlier steps only", {
p <- test_pip()
expect_error(
pip_replace(p, "f2", \(x = ~f3) x),
"cannot reference unknown steps: 'f3'"
)
expect_error(
pip_replace(p, "f2", \(x = ~foo) x),
"cannot reference unknown steps: 'foo'"
)
})
it("marks only downstream dependent steps as outdated", {
p <- pip_new("pipe") |>
pip_add("a1", \(x = 1) x) |>
pip_add("a2", \(x = ~a1) x + 1) |>
pip_add("b1", \(x = 10) x) |>
pip_add("a3", \(x = ~a2) x + 1)
pip_run(p, lgr = NULL)
pip_replace(p, "a2", \(x = ~a1) x + 2)
expect_equal(
unname(p[["state"]]),
c("done", "new", "done", "outdated")
)
})
it("leaves downstream steps that have never run as new", {
p <- pip_new("pipe") |>
pip_add("a1", \(x = 1) x) |>
pip_add("a2", \(x = ~a1) x + 1) |>
pip_add("a3", \(x = ~a2) x + 1)
pip_replace(p, "a2", \(x = ~a1) x + 2)
expect_equal(unname(p[["state"]]), c("new", "new", "new"))
})
it("updates tags on replaced step", {
p <- pip_new("pipe") |>
pip_add("a1", \(x = 1) x) |>
pip_add("a2", \(x = ~a1) x + 1)
pip_replace(
p,
"a2",
\(x = ~a1) x + 10,
tags = c("updated", "core")
)
i <- match("a2", p[["step"]])
expect_equal(
p[["tags"]][[i]],
c("updated", "core")
)
})
it("keeps tags and exec by default", {
p <- test_pip()
p[["f2", "tags"]] <- c("io", "raw")
p[["f2", "exec"]] <- "plain"
pip_replace(p, "f2", \(x = 4) x * 2)
expect_equal(p[["f2", "tags"]], c("io", "raw"))
expect_equal(p[["f2", "exec"]], "plain")
# the other steps are untouched
expect_equal(p[["f1", "tags"]], character(0))
expect_equal(p[["f1", "exec"]], "auto")
})
it("keeps empty tags and the default exec mode", {
p <- test_pip()
pip_replace(p, "f2", \(x = 4) x * 2)
expect_equal(p[["f2", "tags"]], character(0))
expect_equal(p[["f2", "exec"]], "auto")
})
it("treats tags = NULL and exec = NULL as keep", {
p <- test_pip()
p[["f2", "tags"]] <- "io"
p[["f2", "exec"]] <- "split"
pip_replace(p, "f2", \(x = 4) x, tags = NULL, exec = NULL)
expect_equal(p[["f2", "tags"]], "io")
expect_equal(p[["f2", "exec"]], "split")
})
it("overrides kept tags and exec with explicit values", {
p <- test_pip()
p[["f2", "tags"]] <- "io"
p[["f2", "exec"]] <- "plain"
pip_replace(
p,
"f2",
\(x = 4) x,
tags = c("new", "tags"),
exec = "split"
)
expect_equal(p[["f2", "tags"]], c("new", "tags"))
expect_equal(p[["f2", "exec"]], "split")
})
it("clears all tags with tags = character(0)", {
p <- test_pip()
p[["f2", "tags"]] <- c("io", "raw")
pip_replace(p, "f2", \(x = 4) x, tags = character(0))
expect_equal(p[["f2", "tags"]], character(0))
})
it("resets exec to the default with exec = \"auto\"", {
p <- test_pip()
p[["f2", "exec"]] <- "reduce"
pip_replace(p, "f2", \(x = 4) x, exec = "auto")
expect_equal(p[["f2", "exec"]], "auto")
})
it("changes tags and exec independently of each other", {
p <- test_pip()
p[["f2", "tags"]] <- "io"
p[["f2", "exec"]] <- "plain"
pip_replace(p, "f2", \(x = 4) x, exec = "auto")
expect_equal(p[["f2", "tags"]], "io")
expect_equal(p[["f2", "exec"]], "auto")
pip_replace(p, "f2", \(x = 4) x, tags = "other")
expect_equal(p[["f2", "tags"]], "other")
expect_equal(p[["f2", "exec"]], "auto")
})
it("validates an explicit exec mode but not a missing one", {
p <- test_pip()
expect_error(
pip_replace(p, "f2", \(x = 4) x, exec = "foo"),
"exec must be one of"
)
expect_error(
pip_replace(p, "f2", \(x = 4) x, exec = c("auto", "plain")),
"exec must be a single string"
)
expect_no_error(pip_replace(p, "f2", \(x = 4) x, exec = NULL))
})
it("keeps tags and exec of a step replaced through a view", {
p <- test_pip()
p[["f2", "tags"]] <- "io"
p[["f2", "exec"]] <- "plain"
v <- pip_view(p, step = c("f2", "f3"))
pip_replace(v, "f2", \(x = 4) x * 2)
expect_equal(p[["f2", "tags"]], "io")
expect_equal(p[["f2", "exec"]], "plain")
})
it("still resets params, output and state when keeping tags", {
p <- test_pip()
p[["f2", "tags"]] <- "io"
pip_run(p, lgr = NULL)
pip_lock(pip_view(p, step = "f2"))
pip_replace(p, "f2", \(x = 4) x * 2)
expect_equal(p[["f2", "tags"]], "io")
expect_equal(p[["f2", "state"]], "new")
expect_false(p[["f2", "locked"]])
expect_null(p[["f2", "out"]])
expect_equal(p[["f2", "params"]], list(x = 4))
})
it("replaces a step through a view", {
p <- test_pip()
v <- pip_view(p, step = c("f2", "f3"))
expect_error(
pip_replace(v, "f1", \(x = 1) x),
"step 'f1' is not part of the view"
)
pip_replace(v, "f2", \(x = 4) x * 2)
expect_equal(unname(p[["step"]]), c("f1", "f2", "f3"))
pip_run(p, lgr = NULL)
expect_equal(p[["pipenv"]][["data"]][step == "f2", out][[1]], 8)
expect_equal(p[["pipenv"]][["data"]][step == "f3", out][[1]], 9)
})
it("takes try() references from the new function", {
p <- pip_new() |>
pip_add("a", \(x = 1) stop("boom")) |>
pip_add("b", \(x = ~ try(a)) conditionMessage(x))
pip_replace(p, "b", \(y = ~a) y)
expect_equal(p[["b", "depends"]], c(y = "a"))
suppressWarnings(pip_run(p, lgr = NULL, on_error = "continue"))
expect_equal(p[["b", "state"]], "new")
pip_replace(p, "b", \(y = ~ try(a)) conditionMessage(y))
suppressWarnings(pip_run(p, lgr = NULL, on_error = "continue"))
expect_equal(p[["b", "out"]], "step 'a' failed: boom")
})
})
describe("pip_clone", {
test_pip <- function() {
pip_new("p1") |>
pip_add("s1", \(x = 1) .self[["name"]]) |>
pip_add("s2", \(x = ~s1) x)
}
it("signals invalid inputs", {
expect_error(pip_clone(1), "x must be a pipeflow pip")
expect_error(
pip_clone(pip_new(), name = NA_character_),
"name must be a single non-NA string"
)
expect_error(
pip_clone(pip_new(), name = c("a", "b")),
"name must be a single non-NA string"
)
})
it("returns a pipeflow pipeline copy with same content", {
p <- test_pip()
p2 <- pip_clone(p)
expect_true(.is_pipeflow(p2))
expect_false(identical(p2, p))
expect_equal(p2[["name"]], p[["name"]])
expect_equal(
p2[["step"]],
p[["step"]]
)
expect_equal(
p2[["depends"]],
p[["depends"]]
)
})
it("supports overriding the clone name", {
p <- test_pip()
p2 <- pip_clone(p, name = "copied")
expect_equal(p2[["name"]], "copied")
expect_equal(p[["name"]], "p1")
})
it("creates an independent copy", {
p <- test_pip()
p2 <- pip_clone(p)
env <- p2[["pipenv"]]
env[["data"]][["state"]][1] <- "done"
expect_equal(unname(p[["state"]][1]), "new")
pip_add(p2, "s3", \(x = ~s2) x)
expect_false("s3" %in% p[["step"]])
expect_true("s3" %in% p2[["step"]])
})
it("rebinds .self params to the cloned pipeline", {
p <- test_pip()
p2 <- pip_clone(p)
pip_run(p2, lgr = NULL)
expect_equal(p2[["out"]][[1]], "p1")
# The original pipeline still points at itself.
pip_run(p, lgr = NULL, force = TRUE)
expect_equal(p[["out"]][[1]], "p1")
})
it("does not change .self of the original when the clone runs", {
p <- test_pip()
p2 <- pip_clone(p, name = "p2")
pip_run(p2, lgr = NULL)
pip_run(p, lgr = NULL)
expect_equal(p2[["out"]][[1]], "p2")
expect_equal(p[["out"]][[1]], "p1")
})
it("keeps .self of a running step when a clone of it runs", {
p <- pip_new("self") |>
pip_add("s", \(x = 1) {
before <- .self[["name"]]
if (x == 1) {
cl <- pip_clone(.self, name = "inner-clone")
pip_set_params(cl, list(x = 2)) |> pip_run(lgr = NULL)
}
c(before = before, after = .self[["name"]])
})
pip_run(p, lgr = NULL)
expect_equal(p[["s", "out"]], c(before = "self", after = "self"))
})
it("clones an empty pipeline", {
p <- pip_new()
p2 <- pip_clone(p)
expect_equal(length(p2), 0L)
expect_equal(p2[["name"]], p[["name"]])
expect_false(identical(p2, p))
})
})
describe("pip_collect", {
it("returns empty list for empty pipeline", {
p <- pip_new()
expect_equal(pip_collect(p), list())
})
it("returns named flat list", {
p <- pip_new()
pip_add(p, "s1", \(x = 1) x, tags = "data")
pip_add(p, "s2", \(x = ~ -1) x + 1, tags = "model")
data.table::set(
p[["pipenv"]][["data"]],
j = "out",
value = list(10, 20)
)
out <- pip_collect(p)
expect_equal(names(out), c("s1", "s2"))
expect_equal(unname(out), list(10, 20))
})
it("works on pipeline views", {
p <- pip_new()
pip_add(p, "s1", \(x = 1) x, tags = "data")
pip_add(p, "s2", \(x = ~ -1) x + 1, tags = "model")
pip_add(p, "s3", \(x = ~ -1) x + 1, tags = "model")
data.table::set(
p[["pipenv"]][["data"]],
j = "out",
value = list("o1", "o2", "o3")
)
v <- pip_view(p, tags = "model")
out <- pip_collect(v)
expect_equal(names(out), c("s2", "s3"))
expect_equal(unname(out), list("o2", "o3"))
})
it("preserves NULL outputs", {
p <- pip_new("test") |>
pip_add("s1", \(x = NULL) x) |>
pip_add("s2", \(x = 3) x, tags = "g2") |>
pip_add("s3", \(x = ~s2, factor = 2) x * factor) |>
pip_add("s4", \(x = ~s2) x^2, tags = "g4")
out <- pip_collect(p)
expect_equal(out, list(s1 = NULL, s2 = NULL, s3 = NULL, s4 = NULL))
pip_run(p, lgr = NULL)
out <- pip_collect(p)
expect_equal(out, list(s1 = NULL, s2 = 3, s3 = 2 * 3, s4 = 3^2))
})
it("signals invalid arguments", {
p <- pip_new()
expect_error(pip_collect(1), "x must be a pipeflow pip or view")
})
test_group_pip <- function() {
p <- pip_new()
pip_add(p, "s1", \(x = 1) x, tags = c("io", "daily"))
pip_add(p, "s2", \(x = ~ -1) x + 1, tags = "io")
pip_add(p, "s3", \(x = ~ -1) x + 1, tags = "model")
pip_add(p, "s4", \(x = ~ -1) x + 1)
data.table::set(
p[["pipenv"]][["data"]],
j = "out",
value = list("o1", "o2", "o3", "o4")
)
p
}
it("groups outputs by tag", {
p <- test_group_pip()
out <- pip_collect(p, by = "tags")
expect_equal(
out,
list(
io = list(s1 = "o1", s2 = "o2"),
daily = list(s1 = "o1"),
model = list(s3 = "o3")
)
)
})
it("returns the grouped table with as.table = TRUE", {
p <- test_group_pip()
out <- pip_collect(p, by = "tags", as.table = TRUE)
expect_true(data.table::is.data.table(out))
expect_equal(colnames(out), c("tags", "out"))
expect_equal(out[["tags"]], c("io", "daily", "model"))
expect_equal(
out[["out"]],
list(
list(s1 = "o1", s2 = "o2"),
list(s1 = "o1"),
list(s3 = "o3")
)
)
})
it("returns the table results visibly", {
p <- test_group_pip()
expect_true(withVisible(
pip_collect(p, by = "tags", as.table = TRUE)
)[["visible"]])
expect_true(withVisible(
pip_collect(p, as.table = TRUE)
)[["visible"]])
})
it("returns a flat table with if groups are of size 1", {
p <- test_group_pip()
out <- pip_collect(p, as.table = TRUE)
expect_true(data.table::is.data.table(out))
expect_equal(colnames(out), c("step", "out"))
expect_equal(out[["step"]], c("s1", "s2", "s3", "s4"))
expect_equal(out[["out"]], list("o1", "o2", "o3", "o4"))
out <- pip_collect(p, by = "nodeId", as.table = TRUE)
expect_true(data.table::is.data.table(out))
expect_equal(colnames(out), c("nodeId", "out"))
expect_equal(out[["nodeId"]], 0:3)
expect_equal(out[["out"]], list("o1", "o2", "o3", "o4"))
})
it("returns a flat list with if groups are of size 1", {
p <- test_group_pip()
out <- pip_collect(p)
expect_equal(
out,
list(s1 = "o1", s2 = "o2", s3 = "o3", s4 = "o4")
)
out <- pip_collect(p, by = "nodeId")
expect_equal(
out,
list(`0` = "o1", `1` = "o2", `2` = "o3", `3` = "o4")
)
})
it("keeps single-step groups nested with simplify = FALSE", {
p <- test_group_pip()
v <- pip_view(p, step = "s3") # single tag: "model"
expect_equal(pip_collect(v, by = "tags"), list(model = "o3"))
expect_equal(
pip_collect(v, by = "tags", simplify = FALSE),
list(model = list(s3 = "o3"))
)
expect_error(
pip_collect(v, by = "tags", simplify = "yes"),
"simplify must be a single logical value"
)
})
it("groups by scalar columns", {
p <- test_group_pip()
out <- pip_collect(p, by = "state")
expect_equal(
out,
list(new = list(s1 = "o1", s2 = "o2", s3 = "o3", s4 = "o4"))
)
})
it("groups within a view", {
p <- test_group_pip()
v <- pip_view(p, tags = "io")
out <- pip_collect(v, by = "tags")
expect_equal(
out,
list(
io = list(s1 = "o1", s2 = "o2"),
daily = list(s1 = "o1")
)
)
})
it("returns an empty result for untagged or empty pipelines", {
p <- pip_new() |>
pip_add("s1", \(x = 1) x)
expect_equal(pip_collect(p, by = "tags"), list())
tbl <- pip_collect(p, by = "tags", as.table = TRUE)
expect_true(data.table::is.data.table(tbl))
expect_equal(colnames(tbl), c("tags", "out"))
expect_equal(nrow(tbl), 0L)
e <- pip_new()
expect_equal(pip_collect(e, by = "state"), list())
flat <- pip_collect(e, as.table = TRUE)
expect_true(data.table::is.data.table(flat))
expect_equal(colnames(flat), c("step", "out"))
expect_equal(nrow(flat), 0L)
})
it("signals invalid by and as.table arguments", {
p <- test_group_pip()
expect_error(
pip_collect(p, by = 1),
"by must be a single column name"
)
expect_error(
pip_collect(p, by = c("tags", "state")),
"by must be a single column name"
)
expect_error(pip_collect(p, by = "nope"), "unknown column: nope")
expect_error(
pip_collect(p, by = "out"),
"cannot be used as a grouping column"
)
expect_error(
pip_collect(p, by = "params"),
"cannot be used as a grouping column"
)
expect_error(
pip_collect(p, as.table = "yes"),
"as.table must be a single logical value"
)
})
it("keeps pip_collect_out as a deprecated alias", {
p <- pip_new() |>
pip_add("s1", \(x = 1) x, tags = "io") |>
pip_add("s2", \(x = ~s1) x + 1, tags = "model")
pip_run(p, lgr = NULL)
expect_warning(
out <- pip_collect_out(p),
"pip_collect_out"
)
expect_equal(out, pip_collect(p))
expect_equal(
suppressWarnings(
pip_collect_out(p, by = "tags", as.table = TRUE)
),
pip_collect(p, by = "tags", as.table = TRUE)
)
})
})
describe("pip_get_params", {
test_pip <- function() {
pip_new() |>
pip_add("s1", \(data = data.frame(a = 1:2)) data) |>
pip_add("s2", \(data = ~s1, x = 1) data[, 2] + x) |>
pip_add("s3", \(y = ~s2) y) |>
pip_add("s4", \(x = 9, y = ~s2) x + y)
}
it("returns independent params from a pipeline", {
p <- test_pip()
params <- pip_get_params(p)
expect_named(params, c("data", "x"))
expect_equal(params[["data"]], data.frame(a = 1:2))
expect_equal(params[["x"]], 1)
})
it("returns params from a view subset only", {
p <- pip_new()
data <- data.frame(a = 1:2, b = 3:4)
pip_add(p, "s1", \(data = data) data)
pip_add(p, "s2", \(data = ~s1, x = 1) data[, 2] + x)
pip_add(p, "s3", \(y = ~s2) y)
pip_add(p, "s4", \(x = 9, y = ~s2) x + y)
v <- pip_view(p, step = c("s3", "s4"))
params <- pip_get_params(v)
expect_named(params, "x")
expect_equal(params[["x"]], 9)
})
it("returns empty list for view without independent params", {
p <- pip_new()
pip_add(p, "s1", \(x = 1) x)
pip_add(p, "s2", \(y = ~s1) y)
v <- pip_view(p, step = "s2")
expect_equal(pip_get_params(v), list())
})
it("signals invalid input", {
expect_error(
pip_get_params(1),
"x must be a pipeflow pip or view"
)
})
})
describe("pip_graph", {
test_pip <- function() {
pip_new() |>
pip_add("s1", \(x = 1) x, tags = "io", exec = "split") |>
pip_add("s2", \(x = ~s1) x + 1, tags = "model", exec = "reduce") |>
pip_add("s3", \(x = ~s2) x + 2, tags = "model")
}
map_edge_labels <- function(graph) {
idToLabel <- stats::setNames(
graph[["nodes"]][["label"]],
as.character(graph[["nodes"]][["id"]])
)
paste0(
idToLabel[as.character(graph[["edges"]][["from"]])],
"->",
idToLabel[as.character(graph[["edges"]][["to"]])]
)
}
it("returns visNetwork-compatible nodes and edges for pipelines", {
p <- test_pip()
env <- p[["pipenv"]]
env[["data"]][["state"]] <- c("new", "done", "failed")
g <- pip_graph(p)
nodes <- g[["nodes"]]
edges <- g[["edges"]]
expect_named(g, c("nodes", "edges"))
expect_s3_class(nodes, "data.frame")
expect_s3_class(edges, "data.frame")
expect_equal(names(nodes), c("id", "label", "shape", "color"))
expect_equal(names(edges), c("from", "to", "arrows"))
nodeShape <- stats::setNames(nodes[["shape"]], nodes[["label"]])
expect_equal(nodeShape[["s1"]], "star")
expect_equal(nodeShape[["s2"]], "dot")
expect_equal(nodeShape[["s3"]], "hexagon")
expect_true(all(edges[["arrows"]] == "to"))
expectedColors <- vapply(
p[["state"]],
FUN = \(st) .step_states[[st]][["color"]],
FUN.VALUE = character(1)
)
expect_equal(nodes[["color"]], unname(expectedColors))
expect_setequal(map_edge_labels(g), c("s1->s2", "s2->s3"))
})
it("supports view-only graph or graph with upstream closure", {
p <- test_pip()
v <- pip_view(p, step = "s3")
gView <- pip_graph(v, include_upstream = FALSE)
expect_equal(gView[["nodes"]][["label"]], "s3")
expect_equal(nrow(gView[["edges"]]), 0L)
gUp <- pip_graph(v, include_upstream = TRUE)
expect_setequal(gUp[["nodes"]][["label"]], c("s1", "s2", "s3"))
expect_setequal(map_edge_labels(gUp), c("s1->s2", "s2->s3"))
})
it("signals invalid inputs", {
p <- test_pip()
expect_error(pip_graph(1), "x must be a pipeflow pip or view")
expect_error(
pip_graph(p, include_upstream = c(TRUE, FALSE)),
"include_upstream must be a single logical value"
)
})
it("returns empty nodes and edges for empty pipeline", {
g <- pip_graph(pip_new())
expect_equal(nrow(g[["nodes"]]), 0L)
expect_equal(nrow(g[["edges"]]), 0L)
expect_named(g, c("nodes", "edges"))
})
it("keeps pip_get_graph as a deprecated alias", {
p <- test_pip()
expect_warning(
g <- pip_get_graph(p),
"pip_get_graph"
)
expect_equal(pip_graph(p), g)
expect_error(
suppressWarnings(pip_get_graph(1)),
"x must be a pipeflow pip or view"
)
})
})
describe("pip_data", {
test_pip <- function() {
pip_new() |>
pip_add("s1", \(x = 1) x) |>
pip_add("s2", \(x = ~s1) x + 1)
}
it("returns the underlying step table", {
p <- test_pip()
d <- pip_data(p)
expect_true(data.table::is.data.table(d))
expect_equal(nrow(d), 2L)
expect_equal(d[["step"]], c("s1", "s2"))
})
it("returns the full table for views", {
p <- test_pip()
v <- pip_view(p, step = "s2")
expect_equal(nrow(pip_data(v)), 2L)
})
it("is exposed as a virtual method", {
p <- test_pip()
expect_true(data.table::is.data.table(p$data()))
expect_true(data.table::is.data.table(p[["data"]]()))
})
it("signals invalid inputs", {
expect_error(pip_data(1), "x must be a pipeflow pip or view")
})
})
describe("pip_run", {
test_pip <- function() {
pip_new() |>
pip_add("load_raw", \(x = 1) x, tags = "io") |>
pip_add("fit_model", \(x = ~ -1) x + 1, tags = "model") |>
pip_add("eval_model", \(x = ~fit_model) x, tags = "model") |>
pip_add("bla_bla", \(bla = "blabla") bla, tags = "bla")
}
describe("standard runs", {
it("runs all steps of the pipeline and marks them as done", {
p <- test_pip()
pip_run(p, lgr = NULL)
expect_equal(
unname(p[["out"]]),
list(1, 2, 2, "blabla")
)
expect_equal(unname(p[["state"]]), rep("done", 4))
})
it("marks downstream steps not reached due to abort as outdated", {
fail <- FALSE
p <- pip_new() |>
pip_add(
"load_raw",
\(x = 1) if (fail) stop("io error") else x,
tags = "io"
) |>
pip_add("fit_model", \(x = ~ -1) x + 1, tags = "model") |>
pip_add("eval_model", \(x = ~fit_model) x, tags = "model")
pip_run(p, lgr = NULL)
fail <- TRUE
expect_error(pip_run(p, lgr = NULL, force = TRUE), "io error")
expect_equal(
unname(p[["state"]]),
c("failed", "outdated", "outdated")
)
})
it("leaves new steps not reached due to abort as new", {
p <- pip_new() |>
pip_add("load_raw", \(x = 1) stop("io error"), tags = "io") |>
pip_add("fit_model", \(x = ~ -1) x + 1, tags = "model") |>
pip_add("eval_model", \(x = ~fit_model) x, tags = "model")
expect_error(pip_run(p, lgr = NULL), "io error")
expect_equal(
unname(p[["state"]]),
c("failed", "new", "new")
)
})
it("keeps the state of unreached steps whose inputs did not run", {
p <- pip_new() |>
pip_add("data", \(n = 3) n) |>
pip_add("a", \(d = ~data) d + 1) |>
pip_add("fails", \(x = 1) stop("boom")) |>
pip_add("b", \(d = ~data) d + 2)
p[c("data", "a", "b"), "state"] <- "done"
expect_error(pip_run(p, lgr = NULL), "boom")
expect_equal(
unname(p[["state"]]),
c("done", "done", "failed", "done")
)
})
it("marks the run state as failed when a step errors", {
p <- pip_new() |>
pip_add("s1", \(x = 1) x) |>
pip_add("s2", \(x = ~s1) stop("boom")) |>
pip_add("s3", \(x = ~s2) x + 1)
expect_error(pip_run(p, lgr = NULL), "boom")
expect_equal(get_run_state(p), "failed")
# A subsequent successful run resets the state to ready.
pip_replace(p, "s2", \(x = ~s1) x + 1)
pip_run(p, lgr = NULL)
expect_equal(get_run_state(p), "ready")
})
it("marks the run state as failed when a view run errors", {
p <- pip_new() |>
pip_add("a", \(x = 1) x) |>
pip_add("b", \(x = ~a) stop("view boom"))
v <- pip_view(p, step = "b")
expect_error(pip_run(v, lgr = NULL), "view boom")
expect_equal(get_run_state(p), "failed")
})
})
describe("signals step errors", {
catch_step_error <- function(x, ...) {
tryCatch(
pip_run(x, lgr = NULL, ...),
pipeflow_step_error = \(e) e
)
}
it("names the failed step and keeps the original condition", {
p <- pip_new() |>
pip_add("data", \(x = 1) x) |>
pip_add("a", \(x = ~data) stop("boom"))
expect_error(
pip_run(p, lgr = NULL),
"step 'a': boom",
fixed = TRUE,
class = "pipeflow_step_error"
)
e <- catch_step_error(p)
expect_s3_class(
e,
c("pipeflow_step_error", "error", "condition"),
exact = TRUE
)
expect_equal(conditionMessage(e), "step 'a': boom")
expect_equal(e[["step"]], "a")
expect_s3_class(e[["parent"]], "simpleError")
expect_equal(conditionMessage(e[["parent"]]), "boom")
expect_null(e[["call"]])
expect_equal(get_run_state(p), "failed")
})
it("keeps the class of the original condition", {
skip_if_not_installed("rlang")
p <- pip_new() |>
pip_add("a", \(x = 1) {
rlang::abort(
"boom",
class = "my_error",
body = c(i = "hint")
)
})
e <- catch_step_error(p)
expect_s3_class(e[["parent"]], "my_error")
expect_s3_class(e[["parent"]], "rlang_error")
expect_equal(
conditionMessage(e),
paste0("step 'a': ", conditionMessage(e[["parent"]]))
)
expect_match(conditionMessage(e), "hint", fixed = TRUE)
})
it("names the failed upstream step of a view run", {
p <- pip_new() |>
pip_add("a", \(x = 1) stop("upstream boom")) |>
pip_add("b", \(x = ~a) x)
e <- catch_step_error(pip_view(p, step = "b"))
expect_equal(e[["step"]], "a")
expect_equal(conditionMessage(e), "step 'a': upstream boom")
})
it("keeps the step error of a run after a restart", {
cnt <- new.env()
cnt[["n"]] <- 0L
p <- pip_new() |>
pip_add("s1", function(x = 1) {
cnt[["n"]] <- cnt[["n"]] + 1L
if (cnt[["n"]] == 1L) {
.self$restart()
}
x
}) |>
pip_add("s2", \(x = ~s1) stop("boom"))
e <- catch_step_error(p)
expect_equal(cnt[["n"]], 2L)
expect_equal(e[["step"]], "s2")
expect_equal(conditionMessage(e), "step 's2': boom")
expect_equal(get_run_state(p), "failed")
})
it("nests the step errors of inner pipelines", {
inner <- pip_new("inner") |>
pip_add("i", \(x = 1) stop("boom"))
p <- pip_new("outer") |>
pip_add("o", \(x = 1) pip_run(inner, lgr = NULL))
e <- catch_step_error(p)
expect_equal(conditionMessage(e), "step 'o': step 'i': boom")
expect_equal(e[["step"]], "o")
expect_s3_class(e[["parent"]], "pipeflow_step_error")
expect_equal(e[["parent"]][["step"]], "i")
})
it("keeps other errors of the run as they are", {
p <- pip_new() |>
pip_add("a", \(x = 1) x)
progress <- function(value, detail) stop("progress boom")
e <- tryCatch(
pip_run(p, lgr = NULL, progress = progress),
error = \(e) e
)
expect_false(inherits(e, "pipeflow_step_error"))
expect_equal(conditionMessage(e), "progress boom")
expect_null(e[["call"]])
expect_equal(get_run_state(p), "failed")
})
})
describe("running views", {
it("can run parts of the pipeline via views", {
p <- test_pip()
v <- pip_view(p, tags = "bla")
pip_run(v, lgr = NULL)
expect_equal(
unname(p[["out"]]),
list(NULL, NULL, NULL, "blabla")
)
})
it("runs all steps of the view plus upstream dependencies", {
p <- test_pip()
v <- pip_view(p, tags = "model")
pip_run(v, lgr = NULL)
expect_equal(unname(p[["out"]]), list(1, 2, 2, NULL))
p <- pip_new() |>
pip_add("f1", \(x = 1) x) |>
pip_add("f2", \(x = ~f1) x + 1) |>
pip_add("f3", \(x = 1) x + 1) |>
pip_add("f4", \(x = ~f2) x + 1) |>
pip_add("f5", \(x = ~f4) x + 1)
v <- pip_view(p, step = "f2")
pip_run(v, lgr = NULL)
expect_equal(
unname(p[["out"]]),
list(1, 2, NULL, NULL, NULL)
)
v <- pip_view(p, step = "f5")
pip_run(v, lgr = NULL)
expect_equal(
unname(p[["out"]]),
list(1, 2, NULL, 3, 4)
)
})
it("marks downstream steps outside the view as outdated", {
p <- test_pip()
pip_run(p, lgr = NULL)
v <- pip_view(p, tags = "io")
pip_run(v, lgr = NULL, force = TRUE)
expect_equal(
unname(p[["state"]]),
c("done", "outdated", "outdated", "done")
)
})
it("leaves new steps outside the view as new", {
p <- test_pip()
v <- pip_view(p, tags = "io")
pip_run(v, lgr = NULL)
expect_equal(
unname(p[["state"]]),
c("done", "new", "new", "new")
)
})
it("does not outdate steps when the view run executes nothing", {
p <- pip_new("p") |>
pip_add("data", \(n = 3) n) |>
pip_add("a", \(d = ~data) d + 1) |>
pip_add("b", \(d = ~data) d + 2)
pip_run(p, lgr = NULL)
pip_run(pip_view(p, step = "a"), lgr = NULL)
expect_equal(unname(p[["state"]]), c("done", "done", "done"))
})
it("outdates only steps downstream of executed view steps", {
p <- pip_new("p") |>
pip_add("data", \(n = 3) n) |>
pip_add("a", \(d = ~data) d + 1) |>
pip_add("a2", \(x = ~a) x) |>
pip_add("b", \(d = ~data) d + 2)
pip_run(p, lgr = NULL)
p[["a", "state"]] <- "outdated"
pip_run(pip_view(p, step = "a"), lgr = NULL)
expect_equal(
unname(p[["state"]]),
c("done", "done", "outdated", "done")
)
})
it("does not outdate locked steps", {
p <- pip_new("p") |>
pip_add("data", \(n = 3) n) |>
pip_add("a", \(d = ~data) d + 1) |>
pip_add("b", \(d = ~data) d + 2)
pip_run(p, lgr = NULL)
pip_lock(p["b"])
pip_run(pip_view(p, step = "a"), lgr = NULL, force = TRUE)
expect_equal(unname(p[["state"]]), c("done", "done", "done"))
})
it("adds [view]/[upstream] markers to view-run logs", {
p <- test_pip()
v <- pip_view(p, tags = "model")
logs <- character(0)
lgr <- function(level, msg) {
logs <<- c(logs, paste(level, msg))
}
pip_run(v, lgr = lgr)
expect_true(any(grepl("\\[upstream\\] load_raw", logs)))
expect_true(any(grepl("\\[view\\] fit_model", logs)))
expect_true(any(grepl("\\[view\\] eval_model", logs)))
})
})
describe("partition-aware execution", {
test_pip_partitioned <- function() {
pip_new() |>
pip_add(
"load",
\(
x = data.frame(
grp = c("a", "a", "b", "b"),
value = c(1, 3, 10, 20)
)
) {
x
}
) |>
pip_add(
"split",
\(x = ~load) split(x, f = x$grp),
exec = "split"
) |>
pip_add(
"mean_by_grp",
\(x = ~split) mean(x$value)
) |>
pip_add(
"plus_one",
\(x = ~mean_by_grp) x + 1
) |>
pip_add(
"overall",
\(x = ~plus_one) mean(unlist(x)),
exec = "reduce"
)
}
it("maps downstream calls over split output in auto mode", {
p <- test_pip_partitioned()
pip_run(p, lgr = NULL)
splitOut <- p[["pipenv"]][["data"]]$out[[2]]
meanOut <- p[["pipenv"]][["data"]]$out[[3]]
plusOneOut <- p[["pipenv"]][["data"]]$out[[4]]
expect_true(inherits(splitOut, "pipeflow_partitioned"))
expect_true(inherits(meanOut, "pipeflow_partitioned"))
expect_true(inherits(plusOneOut, "pipeflow_partitioned"))
expect_equal(meanOut[["a"]], 2)
expect_equal(meanOut[["b"]], 15)
expect_equal(plusOneOut[["a"]], 3)
expect_equal(plusOneOut[["b"]], 16)
expect_equal(p[["pipenv"]][["data"]]$out[[5]], 9.5)
})
it("errors when reduce mode receives only non-partitioned inputs", {
p <- pip_new() |>
pip_add("load", \(x = 1) x) |>
pip_add("sum", \(x = ~load) x + 1, exec = "reduce")
expect_error(
pip_run(p, lgr = NULL),
"reduce mode requires at least one partitioned input"
)
})
it("lets a reduce step receive a failure through ~try()", {
p <- pip_new() |>
pip_add(
"load",
\(x = list(a = 1, b = 2)) x,
exec = "split"
) |>
pip_add(
"per_key",
\(x = ~load) if (x == 2) stop("boom") else x
) |>
pip_add(
"total",
\(x = ~ try(per_key)) {
if (inherits(x, "pipeflow_failure")) {
"failed"
} else {
sum(unlist(x))
}
},
exec = "reduce"
)
expect_warning(
pip_run(p, on_error = "continue", lgr = NULL),
class = "pipeflow_run_failed"
)
expect_equal(p[["pipenv"]][["data"]]$out[[3]], "failed")
expect_equal(p[["pipenv"]][["data"]]$state[[3]], "done")
})
it("skips a reduce step without try() if its input failed", {
p <- pip_new() |>
pip_add(
"load",
\(x = list(a = 1, b = 2)) x,
exec = "split"
) |>
pip_add(
"per_key",
\(x = ~load) if (x == 2) stop("boom") else x
) |>
pip_add(
"total",
\(x = ~per_key) sum(unlist(x)),
exec = "reduce"
)
expect_warning(
pip_run(p, on_error = "continue", lgr = NULL),
"total"
)
expect_equal(p[["pipenv"]][["data"]]$state[[3]], "new")
})
it("errors when plain mode receives partitioned input", {
p <- pip_new() |>
pip_add(
"load",
\(
x = data.frame(
grp = c("a", "a", "b", "b"),
value = c(1, 3, 10, 20)
)
) {
x
}
) |>
pip_add(
"split",
\(x = ~load) split(x, f = x$grp),
exec = "split"
) |>
pip_add(
"strict",
\(x = ~split) mean(unlist(x)),
exec = "plain"
)
expect_error(
pip_run(p, lgr = NULL),
"plain mode does not accept partitioned inputs"
)
})
it("maps with two partitioned and one scalar input", {
p <- pip_new() |>
pip_add(
"left_raw",
\(
x = data.frame(
grp = c("a", "a", "b", "b"),
value = c(1, 3, 10, 20)
)
) {
x
}
) |>
pip_add(
"right_raw",
\(
x = data.frame(
grp = c("a", "a", "b", "b"),
value = c(2, 4, 6, 8)
)
) {
x
}
) |>
pip_add(
"left_split",
\(x = ~left_raw) split(x, f = x$grp),
exec = "split"
) |>
pip_add(
"right_split",
\(x = ~right_raw) split(x, f = x$grp),
exec = "split"
) |>
pip_add("offset", \(x = 100) x) |>
pip_add(
"combine",
\(a = ~left_split, b = ~right_split, c = ~offset) {
mean(a$value) + mean(b$value) + c
}
)
pip_run(p, lgr = NULL)
out <- p[["pipenv"]][["data"]]$out[[6]]
expect_true(inherits(out, "pipeflow_partitioned"))
expect_equal(out[["a"]], 105)
expect_equal(out[["b"]], 122)
})
it("includes failing partition key in mapped errors", {
p <- pip_new() |>
pip_add(
"load",
\(
x = data.frame(
grp = c("a", "a", "b", "b"),
value = c(1, 2, 3, 4)
)
) {
x
}
) |>
pip_add(
"split",
\(x = ~load) split(x, f = x$grp),
exec = "split"
) |>
pip_add(
"fragile",
\(x = ~split) {
key <- unique(x$grp)
if (identical(key, "b")) {
stop("boom")
}
sum(x$value)
}
)
expect_error(
pip_run(p, lgr = NULL),
"key 'b': boom"
)
e <- tryCatch(pip_run(p, lgr = NULL), pipeflow_step_error = \(e) e)
expect_equal(conditionMessage(e), "step 'fragile': key 'b': boom")
keyError <- e[["parent"]]
expect_s3_class(
keyError,
c("pipeflow_key_error", "error", "condition"),
exact = TRUE
)
expect_equal(keyError[["key"]], "b")
expect_equal(conditionMessage(keyError), "key 'b': boom")
expect_equal(conditionMessage(keyError[["parent"]]), "boom")
expect_null(keyError[["call"]])
})
})
describe("can be run with dynamic pipeline modifications", {
it(
paste(
"supports modifying the pipeline at runtime and",
"continues with the updated version"
),
{
pip <- pip_new("my-pipeline") |>
pip_add("init", function(xInit = 0) xInit) |>
pip_add("f1", function(x = ~init) x + 1) |>
pip_add(
"f2",
function(x = ~f1) {
if (x > 10) {
.self |>
pip_replace("f3", function(x = ~f1) x * 3)
return(x / 2)
}
x + 2
}
) |>
pip_add("f3", function(x = ~f2) x + 3)
pip_run(pip, lgr = NULL)
expect_equal(unname(pip[["out"]]), list(0, 1, 3, 6))
pip |> pip_set_params(list(xInit = 15)) |> pip_run(lgr = NULL)
expect_equal(unname(pip[["out"]]), list(15, 16, 16 / 2, 16 * 3))
expect_equal(
body(pip[["f3", "fun"]]),
body(function(x = ~f1) x * 3)
)
}
)
it(
paste(
"supports modifying the pipeline at runtime by",
"a step that was replaced"
),
{
pip <- pip_new("my-pipeline") |>
pip_add("init", function(xInit = 0) xInit) |>
pip_add("f1", function(x = ~init) x + 1) |>
pip_add("f2", function(x = ~f1) x + 2) |>
pip_add("f3", function(x = ~f2) x + 3)
pip |>
pip_replace(
"f2",
function(x = ~f1) {
if (x > 10) {
.self |>
pip_replace("f3", function(x = ~f1) x * 3)
return(x / 2)
}
x + 2
}
)
pip_run(pip, lgr = NULL)
expect_equal(unname(pip[["out"]]), list(0, 1, 3, 6))
pip |> pip_set_params(list(xInit = 15)) |> pip_run(lgr = NULL)
expect_equal(unname(pip[["out"]]), list(15, 16, 16 / 2, 16 * 3))
expect_equal(
body(pip[["f3", "fun"]]),
body(function(x = ~f1) x * 3)
)
}
)
it(
paste(
"keeps .self rebound correctly for downstream steps",
"that were copied during replacement"
),
{
pip <- pip_new("my-pipeline") |>
pip_add("init", function(xInit = 0) xInit) |>
pip_add("f1", function(x = ~init) x + 1) |>
pip_add("f2", function(x = ~f1) x + 2) |>
pip_add(
"f3",
function(x = ~f2) {
if (x > 10) {
.self |>
pip_replace("f4", function(x = ~f1) x * 4)
}
x + 3
}
) |>
pip_add("f4", function(x = ~f3) x + 4)
pip |> pip_replace("f2", function(x = ~f1) x + 10)
pip_run(pip, lgr = NULL)
expect_equal(unname(pip[["out"]]), list(0, 1, 11, 14, 4))
expect_equal(
body(pip[["f4", "fun"]]),
body(function(x = ~f1) x * 4)
)
}
)
it(
paste(
"persists runtime self-modifications across runs",
"without losing .self routing"
),
{
pip <- pip_new("my-pipeline") |>
pip_add("init", function(xInit = 0) xInit) |>
pip_add("f1", function(x = ~init) x + 1) |>
pip_add(
"f2",
function(x = ~f1) {
if (x > 10) {
.self |>
pip_replace("f3", function(x = ~f1) x * 3)
return(x / 2)
}
x + 2
}
) |>
pip_add("f3", function(x = ~f2) x + 3)
pip |> pip_set_params(list(xInit = 15)) |> pip_run(lgr = NULL)
expect_equal(unname(pip[["out"]]), list(15, 16, 8, 48))
expect_equal(
body(pip[["f3", "fun"]]),
body(function(x = ~f1) x * 3)
)
pip |> pip_set_params(list(xInit = 20)) |> pip_run(lgr = NULL)
expect_equal(unname(pip[["out"]]), list(20, 21, 10.5, 63))
expect_equal(
body(pip[["f3", "fun"]]),
body(function(x = ~f1) x * 3)
)
}
)
it(
paste(
"applies runtime replacements when running only a view",
"and updates steps outside the view"
),
{
pip <- pip_new("my-pipeline") |>
pip_add("init", function(xInit = 0) xInit) |>
pip_add("f1", function(x = ~init) x + 1) |>
pip_add(
"f2",
function(x = ~f1) {
if (x > 10) {
.self |>
pip_replace("f3", function(x = ~f1) x * 3)
return(x / 2)
}
x + 2
}
) |>
pip_add("f3", function(x = ~f2) x + 3) |>
pip_add("f4", function(x = ~f3) x + 1)
pip_run(pip, lgr = NULL)
expect_equal(unname(pip[["out"]]), list(0, 1, 3, 6, 7))
pip |> pip_set_params(list(xInit = 15))
v <- pip_view(pip, step = c("f1", "f2"))
pip_run(v, lgr = NULL, force = TRUE)
expect_equal(unname(pip[["out"]]), list(15, 16, 8, NULL, 7))
expect_equal(
body(pip[["f3", "fun"]]),
body(function(x = ~f1) x * 3)
)
expect_equal(
pip[["pipenv"]][["data"]][step == "f4", state][[1]],
"outdated"
)
pip_run(pip, lgr = NULL)
expect_equal(unname(pip[["out"]]), list(15, 16, 8, 48, 49))
}
)
})
it("inserts and removes steps at runtime as expected", {
pip <- pip_new("my-pipeline") |>
pip_add("init", function(xInit = 0) xInit) |>
pip_add("f1", function(x = ~init) x + 1) |>
pip_add(
"f2",
function(x = ~f1) {
if (x > 10) {
.self |>
pip_add(
"f2a",
function(x = ~f1) x + 21,
after = "f1",
) |>
pip_add(
"f2b",
function(x = ~f2a) x + 22,
after = "f2a"
) |>
pip_replace(
"f3",
function(x = ~f2b) x + 30
) |>
pip_remove("f2")
}
x + 2
}
) |>
pip_add("f3", function(x = ~f2) x + 3)
pip_set_params(pip, list(xInit = 11))
pip_run(pip, lgr = NULL)
expect_equal(unname(pip[["step"]]), c("init", "f1", "f2a", "f2b", "f3"))
expect_equal(
unname(pip[["state"]]),
c("done", "done", "new", "done", "new")
)
expect_equal(unname(pip[["out"]]), list(11, 12, NULL, NULL + 22, NULL))
pip_set_params(pip, list(xInit = 11))
pip_run(pip, lgr = NULL)
expect_equal(unname(pip[["out"]]), list(11, 12, 33, 55, 85))
})
it("can insert and remove steps at runtime", {
test_pip <- function() {
pip_new("my-pipeline") |>
pip_add("init", function(xInit = 0) xInit) |>
pip_add("f1", function(x = ~init) x + 1) |>
pip_add(
"f2",
function(x = ~f1) {
if (x > 10) {
.self |>
pip_add(
"f2a",
function(x = ~f1) x + 21,
after = "f1",
) |>
pip_add(
"f2b",
function(x = ~f2a) x + 22,
after = "f2a"
) |>
pip_replace(
"f3",
function(x = ~f2b) x + 30
) |>
pip_remove("f2")
.self$restart()
}
x + 2
}
) |>
pip_add("f3", function(x = ~f2) x + 3)
}
pip <- test_pip()
pip_set_params(pip, list(xInit = 11)) |> pip_run(lgr = NULL)
expect_equal(unname(pip[["step"]]), c("init", "f1", "f2a", "f2b", "f3"))
expect_equal(unname(pip[["state"]]), rep("done", 5))
expect_equal(unname(pip[["out"]]), list(11, 12, 33, 55, 85))
})
it("keeps .self bound to the same pipeline across a restart", {
captured <- list()
p <- pip_new("self-bound") |>
pip_add("init", function(xInit = 0) xInit) |>
pip_add("f1", function(x = ~init) x + 1) |>
pip_add("f2", function(x = ~f1) {
captured[[1]] <<- .self
.self$restart()
x + 2
})
pip_run(p, lgr = NULL)
expect_true(identical(captured[[1]], p))
})
it("limits restarts with times while the pipeline is self-modifying", {
count <- 0L
p <- pip_new("mod-times") |>
pip_add("init", function(xInit = 0) xInit) |>
pip_add("f1", function(x = ~init) x + 1) |>
pip_add(
"f2",
function(x = ~f1) {
count <<- count + 1L
if (x > 10 && count < 3L) {
.self |>
pip_replace("f3", function(x = ~f1) x * 3)
.self$restart(times = 2L)
}
x + 2
}
) |>
pip_add("f3", function(x = ~f2) x + 3)
p |> pip_set_params(list(xInit = 11)) |> pip_run(lgr = NULL)
expect_equal(count, 3L)
expect_equal(unname(p[["out"]]), list(11, 12, 14, 36))
expect_equal(unname(p[["state"]]), rep("done", 4))
expect_equal(
body(p[["fun"]][[4]]),
body(function(x = ~f1) x * 3)
)
})
it("force = TRUE re-executes all steps regardless of state", {
p <- test_pip()
pip_run(p, lgr = NULL)
env <- p[["pipenv"]]
env[["data"]][["out"]] <- list(99, 99, 99, 99)
pip_run(p, lgr = NULL, force = TRUE)
expect_equal(unname(p[["out"]]), list(1, 2, 2, "blabla"))
expect_equal(unname(p[["state"]]), rep("done", 4))
})
it("calls the progress callback before each step", {
p <- test_pip()
calls <- character(0)
progress <- function(value, detail) {
calls <<- c(calls, detail)
}
pip_run(p, lgr = NULL, progress = progress)
expect_equal(calls, c("load_raw", "fit_model", "eval_model", "bla_bla"))
})
it("skips locked steps during run", {
p <- test_pip()
data.table::set(
p[["pipenv"]][["data"]],
i = 2,
j = "locked",
value = TRUE
)
pip_run(p, lgr = NULL)
expect_equal(
unname(p[["state"]]),
c("done", "new", "done", "done")
)
expect_equal(
unname(p[["out"]]),
list(1, NULL, NULL, "blabla")
)
})
it("skips locked steps even with force = TRUE", {
p <- test_pip()
data.table::set(
p[["pipenv"]][["data"]],
i = 2,
j = "locked",
value = TRUE
)
data.table::set(
p[["pipenv"]][["data"]],
i = 2,
j = "out",
value = list(99)
)
pip_run(p, lgr = NULL, force = TRUE)
expect_equal(p[["out"]][[2]], 99)
})
it("forwards warnings and messages from steps to the logger", {
p <- pip_new() |>
pip_add("s1", function(x = 1) {
warning("careful now")
message("hey there")
x
})
logs <- character(0)
lgr <- function(level, msg) logs <<- c(logs, paste(level, msg))
suppressWarnings(suppressMessages(pip_run(p, lgr = lgr)))
expect_true(any(grepl("warn careful now", logs)))
expect_true(any(grepl("info hey there", logs)))
})
describe(".self in step functions", {
it("is provided only to step functions that refer to it", {
p <- pip_new("p") |>
pip_add("with", \(x = 1) .self[["name"]]) |>
pip_add("without", \(x = 1) exists(".self"))
pip_run(p, lgr = NULL)
expect_equal(p[["with", "out"]], "p")
expect_false(p[["without", "out"]])
})
it("is not written into the environment of the step function", {
env <- new.env()
f <- local(\(x = 1) .self[["name"]], envir = env)
p <- pip_new("p") |> pip_add("s", f)
pip_run(p, lgr = NULL)
expect_equal(p[["s", "out"]], "p")
expect_false(exists(".self", envir = env, inherits = FALSE))
})
it("doesn't keep the pipeline alive via outputs of other steps", {
# The step function is defined outside of this test (whose
# environment holds `p`), has no source references and forces
# `x`, so only `.self` could make the closure reach the pipeline.
adder <- utils::removeSource(\(x = 1) {
force(x)
function(y) x + y
})
environment(adder) <- globalenv()
p <- pip_new() |>
pip_add("big", \(n = 1e5) rnorm(n)) |>
pip_add("adder", adder)
pip_run(p, lgr = NULL)
size <- length(serialize(p[["adder", "out"]], connection = NULL))
expect_lt(size, 1e4)
})
})
describe("continuing after errors", {
# 'b' and 'd' depend on the failing step 'a', 'c' does not
test_pip <- function() {
pip_new("p") |>
pip_add("data", \(n = 3) n) |>
pip_add("a", \(x = ~data) stop("boom")) |>
pip_add("b", \(x = ~a) x + 1) |>
pip_add("c", \(x = ~data) x * 2) |>
pip_add("d", \(x = ~b) x + 1)
}
run_continue <- function(x, ...) {
suppressWarnings(
pip_run(x, lgr = NULL, on_error = "continue", ...),
classes = "pipeflow_run_failed"
)
}
it("signals invalid on_error values", {
expect_error(
pip_run(test_pip(), lgr = NULL, on_error = "nope"),
"should be one of"
)
})
it("runs all steps that do not depend on a failed step", {
p <- test_pip()
run_continue(p)
expect_equal(
unname(p[["state"]]),
c("done", "failed", "new", "done", "new")
)
expect_equal(p[["c", "out"]], 6)
expect_null(p[["b", "out"]])
})
it("still aborts at the first failed step by default", {
p <- test_pip()
expect_error(pip_run(p, lgr = NULL), "boom")
expect_equal(
unname(p[["state"]]),
c("done", "failed", "new", "new", "new")
)
expect_null(p[["a", "out"]])
expect_equal(get_run_state(p), "failed")
})
it("keeps the original condition as output of the failed step", {
p <- pip_new() |>
pip_add("a", \(x = 1) {
stop(structure(
class = c("my_error", "error", "condition"),
list(message = "custom", call = NULL)
))
})
run_continue(p)
out <- p[["a", "out"]]
expect_s3_class(out, "my_error")
expect_false(inherits(out, "pipeflow_step_error"))
expect_equal(conditionMessage(out), "custom")
})
it("does not repeat the step name in failures and the warning", {
skip_if_not_installed("rlang")
p <- pip_new("p") |>
pip_add("a", \(x = 1) {
rlang::abort("boom", class = "my_error")
}) |>
pip_add("doc", \(x = ~ try(a)) x)
w <- tryCatch(
pip_run(p, lgr = NULL, on_error = "continue"),
pipeflow_run_failed = \(w) w
)
expect_s3_class(p[["a", "out"]], "my_error")
expect_false(inherits(p[["a", "out"]], "pipeflow_step_error"))
expect_equal(
conditionMessage(p[["doc", "out"]]),
"step 'a' failed: boom"
)
expect_s3_class(w[["failed"]][["a"]], "my_error")
expect_match(conditionMessage(w), "'a': boom", fixed = TRUE)
expect_no_match(conditionMessage(w), "step 'a': ", fixed = TRUE)
})
it("sets the run state to 'continued' and signals a warning", {
p <- test_pip()
w <- tryCatch(
pip_run(p, lgr = NULL, on_error = "continue"),
pipeflow_run_failed = \(w) w
)
expect_s3_class(
w,
c("pipeflow_run_failed", "warning", "condition"),
exact = TRUE
)
expect_equal(names(w[["failed"]]), "a")
expect_equal(conditionMessage(w[["failed"]][["a"]]), "boom")
expect_equal(w[["skipped"]], c("b", "d"))
expect_match(
conditionMessage(w),
"Run of pipeline 'p' continued after 1 failed step:",
fixed = TRUE
)
expect_match(conditionMessage(w), "'a': boom", fixed = TRUE)
expect_match(
conditionMessage(w),
"Steps not run because an input failed: 'b', 'd'",
fixed = TRUE
)
expect_null(w[["call"]])
expect_equal(get_run_state(p), "continued")
})
it("signals the warning after all states and outputs are stored", {
p <- test_pip()
seen <- NULL
withCallingHandlers(
pip_run(p, lgr = NULL, on_error = "continue"),
pipeflow_run_failed = function(w) {
seen <<- list(state = p[["state"]], c = p[["c", "out"]])
invokeRestart("muffleWarning")
}
)
expect_equal(seen[["state"]], p[["state"]])
expect_equal(seen[["c"]], 6)
})
it("neither warns nor changes the run state if nothing fails", {
p <- pip_new() |>
pip_add("a", \(x = 1) x)
expect_no_warning(pip_run(p, lgr = NULL, on_error = "continue"))
expect_equal(get_run_state(p), "ready")
})
it("passes failure objects to arguments with try() references", {
p <- test_pip() |>
pip_add(
"doc",
\(c = ~c, a = ~ try(a), d = ~ try(d)) {
list(c = c, a = a, d = d)
}
)
run_continue(p)
expect_equal(p[["doc", "state"]], "done")
out <- p[["doc", "out"]]
expect_equal(out[["c"]], 6)
a <- out[["a"]]
expect_s3_class(
a,
c("pipeflow_failure", "error", "condition"),
exact = TRUE
)
expect_equal(conditionMessage(a), "step 'a' failed: boom")
expect_equal(a[["step"]], "a")
expect_equal(a[["failed_step"]], "a")
expect_equal(conditionMessage(a[["parent"]]), "boom")
expect_null(a[["call"]])
d <- out[["d"]]
expect_s3_class(d, "pipeflow_failure")
expect_equal(
conditionMessage(d),
"step 'd' was not run: step 'a' failed: boom"
)
expect_equal(d[["step"]], "d")
expect_equal(d[["failed_step"]], "a")
expect_identical(d[["parent"]], a[["parent"]])
})
it("does not run a step if any of its failed inputs is not allowed", {
p <- test_pip() |>
pip_add("doc", \(a = ~ try(a), d = ~d) list(a = a, d = d)) |>
pip_add("final", \(x = ~doc) x)
w <- tryCatch(
pip_run(p, lgr = NULL, on_error = "continue"),
pipeflow_run_failed = \(w) w
)
expect_equal(
unname(p[["state"]]),
c("done", "failed", "new", "done", "new", "new", "new")
)
expect_equal(w[["skipped"]], c("b", "d", "doc", "final"))
})
it("marks skipped done steps outdated and keeps new ones new", {
fail <- FALSE
p <- pip_new() |>
pip_add("a", \(x = 1) if (fail) stop("boom") else x) |>
pip_add("b", \(x = ~a) x + 1) |>
pip_add("c", \(x = 1) x) |>
pip_add("d", \(x = ~b, y = ~c) x + y)
pip_run(p, lgr = NULL)
pip_add(p, "e", \(x = ~b) x)
fail <- TRUE
p[["a", "state"]] <- "outdated"
run_continue(p)
expect_equal(
unname(p[["state"]]),
c("failed", "outdated", "done", "outdated", "new")
)
expect_equal(p[["b", "out"]], 2)
})
it("retries failed steps and re-runs the steps given failures", {
fail <- TRUE
p <- pip_new() |>
pip_add("a", \(x = 1) if (fail) stop("boom") else x) |>
pip_add("b", \(x = ~a) x + 1) |>
pip_add(
"doc",
\(x = ~ try(b)) {
if (inherits(x, "pipeflow_failure")) NA else x
}
) |>
pip_add("final", \(x = ~doc) x * 10)
run_continue(p)
expect_equal(
unname(p[["state"]]),
c("failed", "new", "done", "done")
)
expect_equal(p[["final", "out"]], NA_real_)
fail <- FALSE
run_continue(p)
expect_equal(unname(p[["state"]]), rep("done", 4))
expect_equal(p[["final", "out"]], 20)
expect_equal(get_run_state(p), "ready")
})
it("re-runs steps given failures if the step fails again", {
msg <- "first"
p <- pip_new() |>
pip_add("a", \(x = 1) stop(msg)) |>
pip_add("doc", \(x = ~ try(a)) conditionMessage(x))
run_continue(p)
expect_equal(p[["doc", "out"]], "step 'a' failed: first")
msg <- "second"
run_continue(p)
expect_equal(p[["doc", "out"]], "step 'a' failed: second")
})
it("keeps the key prefix of failed partitions", {
calc <- function(x = ~parts) if (x > 1) stop("too big") else x
p <- pip_new() |>
pip_add("parts", \(x = 1) list(a = 1, b = 2), exec = "split") |>
pip_add("calc", calc) |>
pip_add("doc", \(x = ~ try(calc)) x)
run_continue(p)
f <- p[["doc", "out"]]
expect_s3_class(f, "pipeflow_failure")
expect_equal(
conditionMessage(f),
"step 'calc' failed: key 'b': too big"
)
expect_equal(conditionMessage(f[["parent"]]), "key 'b': too big")
out <- p[["calc", "out"]]
expect_s3_class(out, "pipeflow_key_error")
expect_equal(out[["key"]], "b")
expect_equal(conditionMessage(out[["parent"]]), "too big")
expect_identical(f[["parent"]], out)
})
it("does not treat locked steps with failed inputs as failed", {
fail <- FALSE
p <- pip_new() |>
pip_add("a", \(x = 1) if (fail) stop("boom") else x) |>
pip_add("locked", \(x = ~a) x + 1) |>
pip_add("after", \(x = ~locked) x * 10)
pip_run(p, lgr = NULL)
pip_lock(p["locked"])
fail <- TRUE
run_continue(p, force = TRUE)
expect_equal(
unname(p[["state"]]),
c("failed", "done", "done")
)
expect_equal(p[["after", "out"]], 20)
})
it("continues in view runs", {
p <- test_pip() |>
pip_add("doc", \(c = ~c, b = ~ try(b)) b)
run_continue(pip_view(p, step = "doc"))
expect_equal(
unname(p[["state"]]),
c("done", "failed", "new", "done", "new", "done")
)
expect_s3_class(p[["doc", "out"]], "pipeflow_failure")
})
it("keeps continuing after a restart", {
cnt <- new.env()
cnt[["n"]] <- 0L
p <- pip_new() |>
pip_add("s1", function(x = 1) {
cnt[["n"]] <- cnt[["n"]] + 1L
if (cnt[["n"]] == 1L) {
.self$restart()
}
x
}) |>
pip_add("s2", \(x = ~s1) stop("boom")) |>
pip_add("s3", \(x = 1) x)
expect_warning(
pip_run(p, lgr = NULL, on_error = "continue"),
class = "pipeflow_run_failed"
)
expect_equal(cnt[["n"]], 2L)
expect_equal(unname(p[["state"]]), c("done", "failed", "done"))
expect_equal(get_run_state(p), "continued")
})
it("logs skipped steps and the summary of the failures", {
p <- test_pip()
logs <- character(0)
lgr <- function(level, msg) logs <<- c(logs, paste(level, msg))
suppressWarnings(pip_run(p, lgr = lgr, on_error = "continue"))
expect_true(any(grepl("error boom", logs)))
expect_true(any(grepl(
"Step 3/5 b - skipping step, input 'x' failed",
logs,
fixed = TRUE
)))
expect_true(any(grepl(
"warn Run of pipeline 'p' continued after 1 failed step",
logs,
fixed = TRUE
)))
})
})
})
describe("pip_reset", {
it("resets states and outputs of all steps", {
p <- pip_new() |>
pip_add("a", \(x = 1) x) |>
pip_add("b", \(x = ~a) x + 1)
pip_run(p, lgr = NULL)
expect_equal(unname(p[["state"]]), c("done", "done"))
pip_reset(p)
expect_equal(unname(p[["state"]]), c("new", "new"))
expect_true(all(vapply(
p[["out"]],
is.null,
logical(1)
)))
expect_equal(get_run_state(p), "ready")
})
it("keeps params and tags but clears outputs", {
p <- pip_new() |>
pip_add("a", \(x = 1, n = 5) x, tags = "io")
pip_run(p, lgr = NULL)
pip_set_params(p, list(n = 10))
pip_reset(p)
expect_equal(p[["params"]][[1]][["n"]], 10)
expect_equal(p[["tags"]][[1]], "io")
})
it("resets only the steps covered by a view", {
p <- pip_new() |>
pip_add("a", \(x = 1) x) |>
pip_add("b", \(x = ~a) x + 1)
pip_run(p, lgr = NULL)
v <- pip_view(p, step = "b")
pip_reset(v)
expect_equal(unname(p[["state"]]), c("done", "new"))
})
it("allows re-running the pipeline from scratch after reset", {
p <- pip_new() |>
pip_add("a", \(x = 1) x) |>
pip_add("b", \(x = ~a) x + 1)
pip_run(p, lgr = NULL)
pip_reset(p)
pip_run(p, lgr = NULL)
expect_equal(unname(p[["out"]]), list(1, 2))
expect_equal(get_run_state(p), "ready")
})
it("skips locked steps, keeping their state and output", {
p <- pip_new() |>
pip_add("a", \(x = 1) x) |>
pip_add("b", \(x = ~a) x + 1)
pip_run(p, lgr = NULL)
pip_lock(pip_view(p, step = "b"))
pip_reset(p)
expect_equal(unname(p[["state"]]), c("new", "done"))
expect_null(p[["out"]][[1]])
expect_equal(p[["out"]][[2]], 2)
expect_equal(unname(p[["locked"]]), c(FALSE, TRUE))
})
it("warns when all selected steps are locked", {
p <- pip_new() |>
pip_add("a", \(x = 1) x)
pip_run(p, lgr = NULL)
pip_lock(p)
expect_message(pip_reset(p), "all selected steps are locked")
expect_equal(unname(p[["state"]]), "done")
expect_equal(p[["out"]][[1]], 1)
})
})
describe("pip_run_state", {
it("has the run states defined in the package as levels", {
p <- pip_new()
state <- .pip_pipenv(p)[[".run_state"]]
expect_equal(levels(state), names(.run_states))
expect_equal(
unname(vapply(.run_states, `[[`, character(1), "name")),
names(.run_states)
)
expect_true(all(vapply(
.run_states,
\(st) grepl("^#[0-9a-f]{8}$", st[["color"]]),
logical(1)
)))
})
it("is 'ready' for a new pipeline and after a successful run", {
p <- pip_new() |>
pip_add("a", \(x = 1) x)
expect_equal(pip_run_state(p), "ready")
pip_run(p, lgr = NULL)
expect_equal(pip_run_state(p), "ready")
})
it("is 'running' while a step is executed", {
p <- pip_new() |>
pip_add("a", \(x = 1) pip_run_state(.self))
pip_run(p, lgr = NULL)
expect_equal(p[["a", "out"]], "running")
})
it("is 'halted' after a step halted the run", {
p <- pip_new() |>
pip_add("a", function(x = 1) {
.self$halt()
x
}) |>
pip_add("b", \(x = ~a) x)
pip_run(p, lgr = NULL)
expect_equal(pip_run_state(p), "halted")
})
it("is 'failed' after a run stopped on an error", {
p <- pip_new() |>
pip_add("a", \(x = 1) stop("boom"))
expect_error(pip_run(p, lgr = NULL), "boom")
expect_equal(pip_run_state(p), "failed")
})
it("is 'continued' after a run went on after failed steps", {
p <- pip_new() |>
pip_add("a", \(x = 1) stop("boom")) |>
pip_add("b", \(x = 1) x)
expect_warning(
pip_run(p, lgr = NULL, on_error = "continue"),
class = "pipeflow_run_failed"
)
expect_equal(pip_run_state(p), "continued")
})
it("is 'ready' again after a reset", {
p <- pip_new() |>
pip_add("a", \(x = 1) stop("boom"))
expect_error(pip_run(p, lgr = NULL), "boom")
pip_reset(p)
expect_equal(pip_run_state(p), "ready")
})
it("returns the run state of the pipeline for a view", {
p <- pip_new() |>
pip_add("a", \(x = 1) x) |>
pip_add("b", \(x = 1) stop("boom"))
v <- pip_view(p, step = "a")
expect_error(pip_run(p, lgr = NULL), "boom")
expect_equal(pip_run_state(v), "failed")
})
it("signals an error if x is not a pipeflow", {
expect_error(pip_run_state(list()), "must be a pipeflow")
})
})
describe("pip_set_params", {
test_pip <- function() {
pip_new() |>
pip_add("s1", \(x = 1, data = data.frame(a = 1:2)) x) |>
pip_add("s2", \(x = ~s1, y = 2) x + y) |>
pip_add("s3", \(x = 1, z = 3) x + z) |>
pip_add("s4", \(a = ~s1, b = ~s3) a + b)
}
it("sets independent parameters in a pipeline", {
p <- test_pip()
params <- list(x = 11, z = 33, data = data.frame(b = 3:4))
pip_set_params(p, params = params)
after <- p[["params"]]
expect_equal(after[[1]][["x"]], 11)
expect_equal(after[[1]][["data"]], data.frame(b = 3:4))
expect_equal(after[[2]][["y"]], 2)
expect_equal(after[[3]][["x"]], 11)
expect_equal(after[[3]][["z"]], 33)
})
it("can be used with pip or view", {
p <- test_pip()
res <- pip_set_params(p, params = list(x = 11, y = 22))
expect_true(inherits(res, "pipeflow"))
v <- pip_view(p, step = "s2")
res <- pip_set_params(v, params = list(y = 22))
expect_true(.is_pipeflow_view(res))
})
it("ignores locked steps", {
p <- test_pip()
env <- p[["pipenv"]]
env[["data"]][["locked"]][[1]] <- TRUE
pip_set_params(p, params = list(x = 99))
after <- p[["params"]]
expect_equal(after[[1]][["x"]], 1)
})
it("does not warn for parameters defined only in locked steps", {
p <- pip_new() |>
pip_add("s1", \(x = 1) x) |>
pip_add("s2", \(y = 2) y)
pip_lock(p["s1"])
expect_no_warning(pip_set_params(p, params = list(x = 99, y = 22)))
after <- p[["params"]]
expect_equal(after[[1]][["x"]], 1)
expect_equal(after[[2]][["y"]], 22)
})
it("warns for unused parameters", {
p <- pip_new()
pip_add(p, "s1", \(x = 1, ...) x)
expect_warning(
pip_set_params(p, params = list(foo = 1)),
"Trying to set parameters not defined in the target: foo"
)
})
it("ignores unused parameters silently if unknown = 'ignore'", {
p <- test_pip()
expect_no_warning(
pip_set_params(p, params = list(x = 5, foo = 1), unknown = "ignore")
)
expect_equal(p[["params"]][[1]][["x"]], 5)
})
it("fails for unused parameters if unknown = 'error'", {
p <- test_pip()
expect_error(
pip_set_params(p, params = list(foo = 1), unknown = "error"),
"Trying to set parameters not defined in the target: foo"
)
})
it("changes nothing if unknown = 'error' fails", {
p <- test_pip() |> pip_run(lgr = NULL)
params <- p[["params"]]
expect_error(
pip_set_params(p, params = list(x = 5, foo = 1), unknown = "error"),
"foo"
)
expect_equal(p[["params"]], params)
expect_equal(unname(p[["state"]]), rep("done", 4))
})
it("checks unknown parameters against the selected view", {
p <- test_pip()
v <- pip_view(p, step = "s3")
expect_error(
pip_set_params(v, params = list(y = 22), unknown = "error"),
"Trying to set parameters not defined in the target: y"
)
})
it("fails for unused parameters if all steps are locked", {
p <- test_pip() |> pip_lock()
expect_no_message(
expect_error(
pip_set_params(p, params = list(foo = 1), unknown = "error"),
"not defined in the target: foo"
)
)
})
it("does not fail if unknown = 'error' and all params are known", {
p <- test_pip()
expect_no_error(
pip_set_params(p, params = list(x = 5, y = 3), unknown = "error")
)
expect_equal(p[["params"]][[2]][["y"]], 3)
})
it("fails for invalid values of unknown", {
p <- test_pip()
expect_error(
pip_set_params(p, params = list(x = 5), unknown = "foo"),
"'arg' should be one of"
)
})
it("warns by default for unused parameters via the $ method", {
p <- test_pip()
expect_warning(
p$set_params(params = list(foo = 1)),
"not defined in the target: foo"
)
expect_error(
p$set_params(params = list(foo = 1), unknown = "error"),
"not defined in the target: foo"
)
})
it("sets parameters only within the selected view", {
p <- test_pip()
v <- pip_view(p, step = "s3")
expect_warning(
pip_set_params(v, params = list(z = 33, x = 11, y = 22)),
"Trying to set parameters not defined in the target: y"
)
after <- p[["params"]]
expect_equal(after[[1]][["x"]], 1)
expect_equal(after[[2]][["y"]], 2)
expect_equal(after[[3]][["x"]], 11)
expect_equal(after[[3]][["z"]], 33)
})
it("marks changed and dependent downstream steps as 'outdated'", {
p <- test_pip() |> pip_run(lgr = NULL)
pip_set_params(p, params = list(x = 5))
expect_equal(
unname(p[["state"]]),
c("outdated", "outdated", "outdated", "outdated")
)
p <- test_pip() |> pip_run(lgr = NULL)
pip_set_params(p, params = list(y = 5))
expect_equal(
unname(p[["state"]]),
c("done", "outdated", "done", "done")
)
p <- test_pip() |> pip_run(lgr = NULL)
pip_set_params(p, params = list(z = 5))
expect_equal(
unname(p[["state"]]),
c("done", "done", "outdated", "outdated")
)
})
it("leaves steps that have never run as 'new'", {
p <- test_pip()
pip_set_params(p, params = list(x = 5))
expect_equal(unname(p[["state"]]), rep("new", 4))
p <- test_pip() |> pip_run(lgr = NULL)
pip_reset(p[3:4])
pip_set_params(p, params = list(x = 5))
expect_equal(
unname(p[["state"]]),
c("outdated", "outdated", "new", "new")
)
})
it("does not outdate steps if the values are identical", {
p <- test_pip() |> pip_run(lgr = NULL)
pip_set_params(p, params = list(x = 1, data = data.frame(a = 1:2)))
expect_equal(unname(p[["state"]]), rep("done", 4))
})
it("outdates steps with identical values if force = TRUE", {
p <- test_pip() |> pip_run(lgr = NULL)
pip_set_params(p, params = list(y = 2), force = TRUE)
expect_equal(
unname(p[["state"]]),
c("done", "outdated", "done", "done")
)
pip_set_params(p, params = list(x = 1), force = TRUE)
expect_equal(unname(p[["state"]]), rep("outdated", 4))
})
it("refreshes steps with environments modified in place if forced", {
e <- new.env()
p <- pip_new() |> pip_add("s", \(e = NULL) e)
pip_set_params(p, params = list(e = e))
pip_run(p, lgr = NULL)
e$a <- 1
pip_set_params(p, params = list(e = e))
expect_equal(p[["state"]][[1]], "done")
pip_set_params(p, params = list(e = e), force = TRUE)
expect_equal(p[["state"]][[1]], "outdated")
})
it("keeps steps that have never run 'new' if force = TRUE", {
p <- test_pip()
pip_set_params(p, params = list(x = 1), force = TRUE)
expect_equal(unname(p[["state"]]), rep("new", 4))
})
it("fails if force is not TRUE or FALSE", {
p <- test_pip()
expect_error(
pip_set_params(p, params = list(x = 1), force = NA),
"force must be TRUE or FALSE"
)
expect_error(
pip_set_params(p, params = list(x = 1), force = "yes"),
"force must be TRUE or FALSE"
)
})
it("outdates only the steps whose values change", {
p <- test_pip() |> pip_run(lgr = NULL)
pip_set_params(p, params = list(x = 1, y = 2, z = 5))
expect_equal(
unname(p[["state"]]),
c("done", "done", "outdated", "outdated")
)
expect_equal(p[["params"]][[3]][["z"]], 5)
})
it("outdates a step if one of its values changes", {
p <- test_pip() |> pip_run(lgr = NULL)
pip_set_params(p, params = list(x = 1, data = data.frame(a = 3:4)))
expect_equal(
unname(p[["state"]]),
c("outdated", "outdated", "done", "outdated")
)
expect_equal(p[["params"]][[1]][["data"]], data.frame(a = 3:4))
})
it("compares the values of each step separately", {
p <- test_pip()
pip_set_params(pip_view(p, step = "s3"), params = list(x = 5))
pip_run(p, lgr = NULL)
pip_set_params(p, params = list(x = 1))
expect_equal(
unname(p[["state"]]),
c("done", "done", "outdated", "outdated")
)
expect_equal(p[["params"]][[3]][["x"]], 1)
})
it("compares values with identical()", {
p <- test_pip() |> pip_run(lgr = NULL)
pip_set_params(p, params = list(y = 2L))
expect_equal(
unname(p[["state"]]),
c("done", "outdated", "done", "done")
)
})
it("no-ops on empty params list", {
p <- test_pip()
pip_set_params(p, params = list())
expect_equal(unname(p[["state"]]), rep("new", 4))
})
it("no-ops if all considered steps are locked, with a warning", {
p <- test_pip() |> pip_lock()
params <- pip_get_params(p)
expect_message(
pip_set_params(p, params = list(x = 5)),
"all selected steps are locked"
)
expect_equal(pip_get_params(p), params) # verify that nothing changed
})
it("warns for unused parameters if all considered steps are locked", {
p <- test_pip() |> pip_lock()
expect_message(
expect_warning(
pip_set_params(p, params = list(x = 5, foo = 1)),
"Trying to set parameters not defined in the target: foo"
),
"all selected steps are locked"
)
expect_equal(p[["params"]][[1]][["x"]], 1)
})
it("signals unnamed params", {
p <- test_pip()
expect_error(
pip_set_params(p, params = list(1, 2)),
"All parameters must be named"
)
})
it("signals non-list params", {
p <- test_pip()
expect_error(
pip_set_params(p, params = c(a = 1, b = 2)),
"params must be a list"
)
})
})
tag_lock_test_pip <- function() {
pip_new("pipe") |>
pip_add("s1", \(x = 1) x, tags = c("init", "daily")) |>
pip_add("s2", \(x = ~s1) x + 1, tags = c("daily", "model")) |>
pip_add("s3", \(x = ~s2) x + 1, tags = "report")
}
describe("pip_tag", {
it("signals invalid inputs", {
p <- tag_lock_test_pip()
expect_error(pip_tag(1), "x must be a pipeflow pip or view")
expect_error(pip_tag(p, tags = 1), "tags must be a character vector")
})
it("adds tags to all selected steps while preserving existing tags", {
p <- tag_lock_test_pip()
pip_tag(p, tags = c("daily", "core"))
expect_equal(
p[["tags"]][[1]],
c("init", "daily", "core")
)
expect_equal(
p[["tags"]][[2]],
c("daily", "model", "core")
)
expect_equal(
p[["tags"]][[3]],
c("report", "daily", "core")
)
})
it("updates only rows in a view and skips locked steps", {
p <- tag_lock_test_pip()
data.table::set(
p[["pipenv"]][["data"]],
i = 2,
j = "locked",
value = TRUE
)
data.table::set(
p[["pipenv"]][["data"]],
i = 2,
j = "tags",
value = list("keep")
)
v <- pip_view(p, step = c("s2", "s3"))
pip_tag(v, tags = "view")
expect_equal(p[["tags"]][[1]], c("init", "daily"))
expect_equal(p[["tags"]][[2]], "keep")
expect_equal(
p[["tags"]][[3]],
c("report", "view")
)
})
})
describe("pip_untag", {
it("signals invalid inputs", {
p <- tag_lock_test_pip()
expect_error(pip_untag(1), "x must be a pipeflow pip or view")
expect_error(pip_untag(p, tags = 1), "tags must be a character vector")
})
it("removes tags from all selected steps", {
p <- tag_lock_test_pip()
pip_untag(p, tags = c("daily", "report"))
expect_equal(p[["tags"]][[1]], "init")
expect_equal(p[["tags"]][[2]], "model")
expect_equal(p[["tags"]][[3]], character(0))
})
it("updates only rows in a view and skips locked steps", {
p <- tag_lock_test_pip()
data.table::set(
p[["pipenv"]][["data"]],
i = 2,
j = "locked",
value = TRUE
)
data.table::set(
p[["pipenv"]][["data"]],
i = 2,
j = "tags",
value = list(c("daily", "model"))
)
v <- pip_view(p, step = c("s2", "s3"))
pip_untag(v, tags = "daily")
expect_equal(p[["tags"]][[1]], c("init", "daily"))
expect_equal(
p[["tags"]][[2]],
c("daily", "model")
)
expect_equal(p[["tags"]][[3]], "report")
})
})
describe("pip_lock", {
it("signals invalid input", {
expect_error(pip_lock(1), "x must be a pipeflow pip or view")
})
it("locks selected steps", {
p <- tag_lock_test_pip()
pip_lock(p)
expect_true(all(p[["locked"]]))
})
it("locks only rows covered by a view", {
p <- tag_lock_test_pip()
v <- pip_view(p, step = c("s2", "s3"))
pip_lock(v)
expect_false(p[["locked"]][[1]])
expect_true(p[["locked"]][[2]])
expect_true(p[["locked"]][[3]])
})
})
describe("pip_unlock", {
it("signals invalid input", {
expect_error(pip_unlock(1), "x must be a pipeflow pip or view")
})
it("unlocks selected steps", {
p <- tag_lock_test_pip()
env <- p[["pipenv"]]
env[["data"]][["locked"]] <- rep(TRUE, nrow(env[["data"]]))
pip_unlock(p)
expect_false(any(env[["data"]][["locked"]]))
})
it("unlocks only rows covered by a view", {
p <- tag_lock_test_pip()
data.table::set(
p[["pipenv"]][["data"]],
j = "locked",
value = rep(TRUE, nrow(p[["pipenv"]][["data"]]))
)
v <- pip_view(p, step = c("s2", "s3"))
pip_unlock(v)
expect_true(p[["locked"]][[1]])
expect_false(p[["locked"]][[2]])
expect_false(p[["locked"]][[3]])
})
})
describe("pip_view", {
it("returns a view object with the expected structure", {
p <- pip_new("test_pipeline")
pip_add(p, "load", \(x = 1) x, tags = "io")
v <- pip_view(p)
expect_true(.is_pipeflow_view(v))
expect_true("view" %in% names(v))
expect_identical(v[["pipenv"]][["data"]], p[["pipenv"]][["data"]])
expect_identical(v[["name"]], "test_pipeline")
})
it("can filter by multiple properties with fixed matching", {
p <- pip_new()
pip_add(p, "load", \(x = 1) x, tags = "io")
pip_add(p, "fit", \(x = ~ -1) x + 1, tags = "model")
pip_add(p, "eval", \(x = ~fit) x, tags = "model")
p[["pipenv"]][["data"]][2, state := "done"]
v <- pip_view(p, tags = "model", state = "done")
expect_equal(v[["view"]], "fit")
})
it("can filter by depends column with multiple entries", {
p <- pip_new()
pip_add(p, "load", \(x = 1) x, tags = "io")
pip_add(p, "fit", \(x = ~ -1) x + 1, tags = "model")
pip_add(p, "eval", \(x = ~load, y = ~fit) x, tags = "model")
v <- pip_view(p, depends = "fit", tags = "model")
v
expect_equal(v[["view"]], "eval")
})
it("can filter by tags", {
p <- pip_new()
pip_add(p, "s1", \(x = 1) x, tags = c("core", "daily"))
pip_add(p, "s2", \(x = ~ -1) x + 1, tags = "model")
pip_add(p, "s3", \(x = ~ -1) x, tags = c("daily", "report"))
v <- pip_view(p, tags = "daily")
expect_equal(v[["view"]], c("s1", "s3"))
})
it("can filter by step names", {
p <- pip_new()
pip_add(p, "a1", \(x = 1) x, tags = "g1")
pip_add(p, "a2", \(x = ~ -1) x, tags = "g2")
pip_add(p, "a3", \(x = ~ -1) x, tags = "g2")
v <- pip_view(p, step = c("a1", "a3"))
expect_equal(v[["view"]], c("a1", "a3"))
v <- pip_view(p, step = c("a1", "a2"), tags = "g2")
expect_equal(v[["view"]], "a2")
})
it("can filter by parameter names", {
p <- pip_new()
pip_add(p, "s1", \(x = 1, n = 5) x, tags = "g1")
pip_add(p, "s2", \(y = ~ -1, z = 2) y, tags = "g2")
v <- pip_view(p, params = "n")
expect_equal(v[["view"]], "s1")
v <- pip_view(p, params = c("x", "z"))
expect_equal(v[["view"]], c("s1", "s2"))
})
it("matches dependency parameters via the params filter", {
p <- pip_new()
pip_add(p, "s1", \(x = 1) x)
pip_add(p, "s2", \(y = ~s1) y)
v <- pip_view(p, params = "y")
expect_equal(v[["view"]], "s2")
})
it("can filter by exec mode", {
p <- pip_new()
pip_add(p, "s1", \(x = 1) x)
pip_add(p, "s2", \(x = 2) x, exec = "split")
v <- pip_view(p, exec = "split")
expect_equal(v[["view"]], "s2")
})
it("can filter by regex when fixed is FALSE", {
p <- pip_new()
pip_add(p, "data", \(x = 1) x)
pip_add(p, "fit_model", \(x = ~ -1) x + 1)
pip_add(p, "eval_model", \(x = ~data, y = ~fit_model) x)
v <- pip_view(p, step = "_model$", fixed = FALSE)
expect_equal(v[["view"]], c("fit_model", "eval_model"))
})
it("intersects multiple filters", {
p <- pip_new()
pip_add(p, "a1", \(x = 1) x, tags = "g1")
pip_add(p, "a2", \(x = ~ -1) x, tags = "g2")
pip_add(p, "a3", \(x = ~ -1) x, tags = "g2")
v <- pip_view(p, step = c("a1", "a2"), tags = "g2")
expect_equal(v[["view"]], "a2")
})
it("unions multiple filters with join = union", {
p <- pip_new()
pip_add(p, "a1", \(x = 1) x, tags = "g1")
pip_add(p, "a2", \(x = ~ -1) x, tags = "g2")
pip_add(p, "a3", \(x = ~ -1) x, tags = "g2")
v <- pip_view(p, step = "a1", tags = "g2", join = "union")
expect_equal(v[["view"]], c("a1", "a2", "a3"))
})
it("defaults to intersect and allows partial join names", {
p <- pip_new()
pip_add(p, "a1", \(x = 1) x, tags = "g1")
pip_add(p, "a2", \(x = ~ -1) x, tags = "g2")
expect_equal(
pip_view(p, step = "a1", tags = "g2")[["view"]],
character(0)
)
expect_equal(
pip_view(p, step = "a1", tags = "g2", join = "uni")[["view"]],
c("a1", "a2")
)
})
it("signals invalid join values", {
p <- pip_new()
pip_add(p, "a1", \(x = 1) x)
expect_error(
pip_view(p, step = "a1", join = "bogus"),
"should be one of"
)
})
it("can be called on views to further subset rows", {
p <- pip_new("test_pipeline")
pip_add(p, "s1", \(x = 1) x, tags = c("core", "daily"))
pip_add(p, "s2", \(x = ~ -1) x + 1, tags = "model")
pip_add(p, "s3", \(x = ~ -1) x, tags = c("daily", "report"))
v <- pip_view(p, tags = "daily")
expect_equal(v[["view"]], c("s1", "s3"))
expect_equal(v[["name"]], "test_pipeline")
v2 <- pip_view(v, tags = "report")
expect_equal(v2[["view"]], "s3")
expect_equal(v2[["name"]], "test_pipeline")
})
it("signals invalid filter names", {
p <- pip_new()
pip_add(p, "s1", \(x = 1) x)
expect_error(
pip_view(p, not_a_column = "x"),
"Invalid filter name"
)
})
it("signals non-character filter values", {
p <- pip_new()
pip_add(p, "s1", \(x = 1) x)
expect_error(
pip_view(p, step = 2L),
"filter 'step' must be a character vector, not integer"
)
expect_error(
pip_view(p, tags = 1L),
"filter 'tags' must be a character vector, not integer"
)
expect_error(
pip_view(p, params = TRUE),
"filter 'params' must be a character vector, not logical"
)
})
describe("when the pipeline changes", {
view_pip <- function() {
pip_new() |>
pip_add("a", \(x = 1) x, tags = "t") |>
pip_add("b", \(x = ~a) x + 1) |>
pip_add("c", \(x = ~b) x + 1, tags = "t")
}
it("keeps covering its steps when steps are inserted", {
p <- view_pip()
v <- pip_view(p, step = c("b", "c"))
pip_add(p, "first", \(x = 0) x, after = 0)
pip_add(p, "mid", \(x = ~b) x, after = "b")
expect_equal(v[["view"]], c("b", "c"))
expect_equal(nrow(v), 2L)
})
it("does not cover new steps that match its filters", {
p <- view_pip()
v <- pip_view(p, tags = "t")
pip_add(p, "d", \(x = ~c) x, tags = "t")
expect_equal(v[["view"]], c("a", "c"))
})
it("drops steps that were removed through another handle", {
p <- view_pip()
v <- pip_view(p, step = c("a", "c"))
other <- p[c("b", "c")]
pip_remove(other, "c")
expect_equal(v[["view"]], "a")
expect_equal(other[["view"]], "b")
})
it("keeps covering steps that were renamed through any handle", {
p <- view_pip()
v <- pip_view(p, step = c("b", "c"))
pip_rename(p, "c", "c2")
pip_rename(v, "b", "b2") # v is not reassigned
expect_equal(v[["view"]], c("b2", "c2"))
})
it("keeps covering replaced steps", {
p <- view_pip()
v <- pip_view(p, step = "b")
pip_replace(p, "b", \(x = ~a) x * 10)
pip_run(v, lgr = NULL)
expect_equal(v[["view"]], "b")
expect_equal(v[["b", "out"]], 10)
})
it("runs its steps after steps were inserted before them", {
p <- view_pip()
v <- pip_view(p, step = "c")
pip_add(p, "first", \(x = 0) x, after = 0)
pip_run(v, lgr = NULL)
expect_equal(v[["c", "out"]], 3)
expect_equal(p[["first", "state"]], "new")
})
})
})
describe("pip_param_value", {
# S3 methods are looked up from the package namespace, so test methods
# must be registered there.
register_param_value <- function(cls, fn) {
registerS3method(
"pip_param_value",
class = cls,
method = fn,
envir = asNamespace("pipeflow")
)
}
it("returns x unchanged by default", {
expect_identical(pip_param_value(1:3), 1:3)
expect_identical(pip_param_value("a"), "a")
expect_null(pip_param_value(NULL))
})
it("unwraps unbound arguments before the step function sees them", {
register_param_value("pv_unit", function(x, ...) unclass(x)$value)
val <- structure(list(value = 5, unit = "m"), class = "pv_unit")
p <- pip_new() |>
pip_add("s", \(x = 1, y = 2) x + y, params = list())
pip_set_params(p, list(x = val))
pip_run(p, lgr = NULL)
expect_equal(p[["pipenv"]][["data"]]$out[[1]], 7)
})
it("is called once per unbound argument, including defaults", {
calls <- new.env()
calls$n <- 0L
register_param_value("pv_count", function(x, ...) {
calls$n <- calls$n + 1L
unclass(x)[[1]]
})
a <- structure(list(1), class = "pv_count")
b <- structure(list(2), class = "pv_count")
p <- pip_new() |>
pip_add("s", \(a = 0, b = 0, c = 3) a + b + c)
pip_set_params(p, list(a = a, b = b))
pip_run(p, lgr = NULL)
expect_equal(calls$n, 2L)
expect_equal(p[["pipenv"]][["data"]]$out[[1]], 6)
})
it("does not change the stored params", {
register_param_value("pv_store", function(x, ...) unclass(x)[[1]])
val <- structure(list(1), class = "pv_store")
p <- pip_new() |> pip_add("s", \(x = 0) x)
pip_set_params(p, list(x = val))
pip_run(p, lgr = NULL)
expect_identical(pip_get_params(p)$x, val)
})
it("is never called for ~step inputs", {
register_param_value("pv_never", function(x, ...) {
stop("pip_param_value must not be called")
})
mk <- function() structure(list(1), class = "pv_never")
p <- pip_new() |>
pip_add("a", \(z = 1) mk()) |>
pip_add("b", \(x = ~a) class(x))
expect_no_error(pip_run(p, lgr = NULL))
expect_equal(p[["pipenv"]][["data"]]$out[[2]], "pv_never")
})
it("is applied in every exec mode", {
register_param_value("pv_modes", function(x, ...) unclass(x)[[1]])
k <- structure(list(10), class = "pv_modes")
p <- pip_new() |>
pip_add(
"load",
\(x = list(a = 1, b = 2), k = 0) lapply(x, `+`, k),
exec = "split"
) |>
pip_add("map", \(x = ~load, k = 0) x + k) |>
pip_add(
"total",
\(x = ~map, k = 0) sum(unlist(x)) + k,
exec = "reduce"
)
pip_set_params(p, list(k = k))
pip_run(p, lgr = NULL)
# load: (1 + 10) + (2 + 10); map adds 10 per key; total adds 10
expect_equal(p[["pipenv"]][["data"]]$out[[3]], 11 + 10 + 12 + 10 + 10)
p <- pip_new() |>
pip_add("one", \(x = 1) x) |>
pip_add("plain", \(x = ~one, k = 0) x + k, exec = "plain")
pip_set_params(p, list(k = k))
pip_run(p, lgr = NULL)
expect_equal(p[["pipenv"]][["data"]]$out[[2]], 11)
})
it("turns an error in a method into a step error", {
register_param_value("pv_error", function(x, ...) stop("cannot unwrap"))
bad <- structure(list(1), class = "pv_error")
p <- pip_new() |> pip_add("s", \(x = 0) x)
pip_set_params(p, list(x = bad))
expect_error(pip_run(p, lgr = NULL), "step 's': cannot unwrap")
expect_equal(p[["pipenv"]][["data"]]$state[[1]], "failed")
})
it("lets on_error = 'continue' go on after a failing method", {
register_param_value("pv_cont", function(x, ...) stop("cannot unwrap"))
bad <- structure(list(1), class = "pv_cont")
p <- pip_new() |>
pip_add("s", \(x = 0) x) |>
pip_add("t", \(y = 1) y)
pip_set_params(p, list(x = bad))
expect_warning(
pip_run(p, on_error = "continue", lgr = NULL),
class = "pipeflow_run_failed"
)
expect_equal(p[["pipenv"]][["data"]]$state, c("failed", "done"))
})
})
describe("benchmarking", {
skip("benchmarking tests are skipped by default")
v <- c("hello", "world")
w <- c("hello", "this", "is", "my", "world")
grepv(pattern = v, x = w)
`%chin%` <- data.table::`%chin%`
N <- 1e3
u <- as.character(as.hexmode(1:10000))
y <- sample(u, N, replace = TRUE)
x <- sample(u, 100)
system.time(x %in% y)
system.time(x %chin% y)
microbenchmark(
y %in% x,
y %chin% x,
times = 1000
)
# Please type 'example(chmatch)' to run this and see timings on your machine
microbenchmark(x %in% y, x %chin% y, times = 100)
system.time(a <- x %in% y) # 4.5s
system.time(b <- x %chin% y) # 1.7s
identical(a, b)
# Different example with more unique strings ...
u <- as.character(as.hexmode(1:(N / 10)))
y <- sample(u, N, replace = TRUE)
x <- sample(u, N, replace = TRUE)
system.time(a <- match(x, y)) # 46s
system.time(b <- chmatch(x, y)) # 16s
identical(a, b)
})
Any scripts or data that you put into this service are public.
Add the following code to your website.
For more information on customizing the embed code, read Embedding Snippets.