tests/testthat/test-epinow.R

In EpiNow2: Estimate and Forecast Real-Time Infection Dynamics

skip_on_cran()

# set example reporting delay
reporting_delay <- LogNormal(
  meanlog = Normal(0.6, 0.06),
  sdlog = Normal(0.5, 0.1),
  max = 10
)

reported_cases <- EpiNow2::example_confirmed[1:30]

futile.logger::flog.threshold("FATAL")

df_non_zero <- function(df) {
  expect_true(nrow(df) > 0)
}
expected_out <- c("fit", "args", "observations", "timing")

# Integration tests (MCMC-based) ------------------------------------------
# These tests run actual MCMC sampling and are slow. Tests are divided into:
# - Core tests: Essential tests that always run to catch critical failures
# - Variant tests: Configuration variations that only run weekly (gated by EPINOW2_SKIP_INTEGRATION)

# Variant test: epinow is tested via estimate_infections underneath.
# This test verifies wrapper-specific functionality (plots, CrIs).
test_that("epinow produces expected output when run with default settings", {
  skip_integration()
  outputs <- capture.output(suppressMessages(suppressWarnings(
    out <- epinow(
      data = reported_cases,
      generation_time = gt_opts(example_generation_time),
      delays = delay_opts(example_incubation_period + reporting_delay),
      stan = stan_opts(
        samples = 25, warmup = 25,
        cores = 1, chains = 2,
        control = list(adapt_delta = 0.8)
      ),
      CrIs = c(0.95),
      logs = NULL, verbose = FALSE
    )
  )))

  expect_equal(names(out), expected_out)
  # Test new accessor methods work correctly
  df_non_zero(get_samples(out))
  df_non_zero(summary(out, type = "parameters"))
  df_non_zero(estimates_by_report_date(out)$summarised)
  expect_true(!is.null(summary(out)))
  expect_equal(
    names(plot(out, type = "all")),
    c("summary", "infections", "reports", "R", "growth_rate")
  )

  # Verify CrIs are present in output
  expect_true(length(extract_CrIs(summary(out, type = "parameters"))) > 0)
  expect_true(length(extract_CrIs(estimates_by_report_date(out)$summarised)) > 0)
})

test_that("epinow produces expected output with cmdstanr backend", {
  skip_integration()
  skip_on_os("windows")
  output <- capture.output(suppressMessages(suppressWarnings(
    out <- epinow(
      data = reported_cases,
      generation_time = gt_opts(example_generation_time),
      delays = delay_opts(example_incubation_period + reporting_delay),
      stan = stan_opts(backend = "cmdstanr"),
      logs = NULL, verbose = FALSE
    )
  )))

  expect_equal(names(out), expected_out)
  # Test new accessor methods work correctly
  df_non_zero(get_samples(out))
  df_non_zero(summary(out, type = "parameters"))
  df_non_zero(estimates_by_report_date(out)$summarised)
  expect_true(!is.null(summary(out)))
  expect_equal(
    names(plot(out, type = "all")),
    c("summary", "infections", "reports", "R", "growth_rate")
  )
})

test_that("epinow produces expected output with laplace algorithm", {
  skip_integration()
  skip_on_os("windows")
  output <- capture.output(suppressMessages(suppressWarnings(
    out <- epinow(
      data = reported_cases,
      generation_time = gt_opts(example_generation_time),
      delays = delay_opts(example_incubation_period + reporting_delay),
      stan = stan_opts(method = "laplace", backend = "cmdstanr"),
      logs = NULL, verbose = FALSE
    )
  )))
  expect_equal(names(out), expected_out)
  # Test new accessor methods work correctly
  df_non_zero(get_samples(out))
  df_non_zero(summary(out, type = "parameters"))
  df_non_zero(estimates_by_report_date(out)$summarised)
  expect_true(!is.null(summary(out)))
  expect_equal(
    names(plot(out, type = "all")),
    c("summary", "infections", "reports", "R", "growth_rate")
  )
})

test_that("epinow produces expected output with pathfinder algorithm", {
  skip_integration()
  skip_on_os("windows")
  output <- capture.output(suppressMessages(suppressWarnings(
    out <- epinow(
      data = reported_cases,
      generation_time = gt_opts(example_generation_time),
      delays = delay_opts(example_incubation_period + reporting_delay),
      stan = stan_opts(method = "pathfinder", backend = "cmdstanr"),
      logs = NULL, verbose = FALSE
    )
  )))
  expect_equal(names(out), expected_out)
  # Test new accessor methods work correctly
  df_non_zero(get_samples(out))
  df_non_zero(summary(out, type = "parameters"))
  df_non_zero(estimates_by_report_date(out)$summarised)
  expect_true(!is.null(summary(out)))
  expect_equal(
    names(plot(out, type = "all")),
    c("summary", "infections", "reports", "R", "growth_rate")
  )
})

test_that("epinow runs without error when saving to disk", {
  skip_integration()
  output <- capture.output(suppressMessages(suppressWarnings(
    out <- epinow(
      data = reported_cases,
      generation_time = gt_opts(example_generation_time),
      delays = delay_opts(example_incubation_period + reporting_delay),
      stan = stan_opts(
        samples = 25, warmup = 25, cores = 1, chains = 2,
        control = list(adapt_delta = 0.8)
      ),
      target_folder = tempdir(check = TRUE),
      logs = NULL, verbose = FALSE
    )
  )))
  expect_null(out)
})

test_that("epinow can produce partial output as specified", {
  skip_integration()
  output <- capture.output(suppressMessages(suppressWarnings(
    out <- epinow(
      data = reported_cases,
      generation_time = gt_opts(
        example_generation_time,
        weight_prior = FALSE
      ),
      delays = delay_opts(example_incubation_period + reporting_delay),
      stan = stan_opts(
        samples = 25, warmup = 25,
        cores = 1, chains = 2,
        control = list(adapt_delta = 0.8)
      ),
      output = NULL,
      logs = NULL, verbose = FALSE
    )
  )))
  expect_equal(names(out), c("fit", "args", "observations"))
  # Test new accessor methods work correctly
  df_non_zero(get_samples(out))
  df_non_zero(summary(out, type = "parameters"))
  df_non_zero(estimates_by_report_date(out)$summarised)
  expect_true(!is.null(summary(out)))
})

test_that("epinow propagates target_date into the forecast horizon", {
  skip_integration()
  max_date <- max(reported_cases$date)
  extra_days <- 3
  target_date <- as.character(max_date + extra_days)
  base_horizon <- 7
  expected_horizon <- base_horizon + extra_days

  output <- capture.output(suppressMessages(suppressWarnings(
    out <- epinow(
      data = reported_cases,
      generation_time = gt_opts(example_generation_time),
      delays = delay_opts(example_incubation_period + reporting_delay),
      forecast = forecast_opts(horizon = base_horizon),
      stan = stan_opts(
        samples = 25, warmup = 25,
        cores = 1, chains = 1,
        control = list(adapt_delta = 0.8)
      ),
      target_date = target_date,
      logs = NULL, verbose = FALSE
    )
  )))

  expect_equal(out$args$horizon, expected_horizon)
  reported <- estimates_by_report_date(out)$summarised
  expect_equal(max(reported$date), as.Date(target_date) + base_horizon)
})

test_that("epinow fails as expected when given a short timeout", {
  skip_integration()
  expect_error(suppressWarnings(x <- epinow(
    data = reported_cases,
    generation_time = gt_opts(example_generation_time),
    delays = delay_opts(example_incubation_period + reporting_delay),
    stan = stan_opts(
      samples = 100, warmup = 100,
      cores = 1, chains = 2,
      control = list(adapt_delta = 0.8),
      max_execution_time = 1
    ),
    logs = NULL, verbose = FALSE
  )))
})

# Argument validation tests (fast - no MCMC) ------------------------------


test_that("epinow fails if given NUTs arguments when using variational inference", {
  expect_error(capture.output(suppressMessages(suppressWarnings(
    epinow(
      data = reported_cases,
      generation_time = gt_opts(example_generation_time),
      delays = delay_opts(example_incubation_period + reporting_delay),
      stan = stan_opts(
        samples = 100, warmup = 100,
        cores = 1, chains = 2,
        method = "vb"
      ),
      logs = NULL, verbose = FALSE
    )
  ))))
})


test_that("epinow fails if given variational inference arguments when using NUTs", {
  expect_error(capture.output(suppressMessages(suppressWarnings(
    epinow(
      data = reported_cases,
      generation_time = gt_opts(example_generation_time),
      delays = delay_opts(example_incubation_period + reporting_delay),
      stan = stan_opts(method = "sampling", tol_rel_obj = 1),
      logs = NULL, verbose = FALSE
    )
  ))))
})