tests/testthat/test-limits.R

In zukomp: Portable Byte Compression with a Pluggable Codec Registry

test_that("max_output is enforced by the driver", {
  expect_codec_error(
    zu_test_stream(new_payload("zeros", 4096L), "identity", "decode",
                   max_output = 100),
    "zukomp_output_limit"
  )
})

test_that("max_output is exact, not approximate", {
  # Exactly at the cap must pass; one byte over must fail. An off-by-one here
  # would either reject legitimate payloads or let a bomb through.
  x <- new_payload("zeros", 1000L)
  expect_silent(zu_test_stream(x, "identity", "decode", max_output = 1000))
  expect_codec_error(
    zu_test_stream(x, "identity", "decode", max_output = 999),
    "zukomp_output_limit"
  )
})

test_that("max_output is enforced whatever the chunk sizes", {
  # The cap shrinks the window handed to the codec, so it must hold when the
  # caller's own output buffer is smaller than the cap, and when it is larger.
  x <- new_payload("zeros", 4096L)
  for (cout in c(1L, 7L, 64L, 8192L)) {
    expect_codec_error(
      zu_test_stream(x, "identity", "decode", out_chunk = cout, max_output = 100),
      "zukomp_output_limit"
    )
  }
})

test_that("max_output = 0 means unlimited", {
  x <- new_payload("ascii", 20000L)
  expect_identical(zu_test_stream(x, "identity", "decode", max_output = 0), x)
})

test_that("a decompression bomb is stopped without allocating its output", {
  # The point of the limit is that proving it works must itself be cheap:
  # never allocate the bomb to show the cap holds.
  bomb <- new_payload("zeros", 10L * 1024L * 1024L)
  expect_codec_error(
    zu_test_stream(bomb, "identity", "decode", max_output = 1024),
    "zukomp_output_limit"
  )
})

test_that("the output-limit condition carries design 7's metadata", {
  err <- expect_error(
    zu_test_stream(new_payload("zeros", 4096L), "identity", "decode",
                   max_output = 10),
    class = "zukomp_output_limit"
  )
  expect_identical(err$codec, "identity")
  expect_identical(err$input_bytes, 4096L)
  expect_false(is.na(err$native_status))
})

test_that("growth arithmetic refuses to overflow", {
  expect_codec_error(zu_test_grow(near_size_max = TRUE), "zukomp_memory_error")
})

test_that("ordinary growth still succeeds", {
  # A positive control: the overflow test above must fail for the right
  # reason, not because zu_test_grow() always errors.
  expect_true(zu_test_grow(near_size_max = FALSE))
})

test_that("max_ratio admits a codec that does not expand", {
  # identity's output always equals its input, so ratio 1 is the tightest
  # limit it can satisfy. The interesting case -- a ratio that actually
  # trips -- needs a codec that expands, so it lands in Stage 6 with
  # deflate. This is the positive control for the arithmetic in the meantime.
  x <- new_payload("ascii", 4096L)
  expect_identical(zu_test_stream(x, "identity", "decode", max_ratio = 1), x)
  expect_identical(zu_test_stream(x, "identity", "decode", max_ratio = 100), x)
})

test_that("limits are independent of each other", {
  x <- new_payload("zeros", 2048L)
  expect_identical(
    zu_test_stream(x, "identity", "decode", max_output = 2048, max_ratio = 1),
    x
  )
  expect_codec_error(
    zu_test_stream(x, "identity", "decode", max_output = 8, max_ratio = 100),
    "zukomp_output_limit"
  )
})

test_that("max_ratio stops an expanding stream", {
  # Now reachable: zeros compress ~250:1, so decoding them trips any modest
  # ratio cap. This is the path identity could never exercise.
  z <- zu_test_stream(new_payload("zeros", 8192L), "zlib", "encode", level = 9L)
  expect_lt(length(z), 100L)
  expect_codec_error(
    zu_test_stream(z, "zlib", "decode", max_ratio = 10),
    "zukomp_ratio_limit"
  )
})

test_that("max_ratio admits a stream inside the cap", {
  z <- zu_test_stream(new_payload("zeros", 8192L), "zlib", "encode", level = 9L)
  x <- zu_test_stream(z, "zlib", "decode", max_ratio = 100000)
  expect_length(x, 8192L)
})

test_that("max_output and max_ratio report distinct conditions", {
  z <- zu_test_stream(new_payload("zeros", 8192L), "zlib", "encode", level = 9L)
  expect_codec_error(zu_test_stream(z, "zlib", "decode", max_output = 100),
                     "zukomp_output_limit")
  expect_codec_error(zu_test_stream(z, "zlib", "decode", max_ratio = 2),
                     "zukomp_ratio_limit")
})

test_that("an empty payload decodes into a zero-capacity sink", {
  # The codec used to answer ZU_NEED_OUTPUT whenever avail_out was 0, without
  # asking whether output was actually needed, so a stream with nothing left
  # to produce could never reach its trailer: a consumer sizing its sink from
  # a known-zero Content-Length got an output-limit error instead of 0 bytes.
  for (codec in c("identity", "deflate-raw", "zlib", "gzip")) {
    z <- komp_compress(raw(0), if (codec == "identity") "identity" else codec)
    expect_identical(zu_test_decompress_one(z, codec, 0), raw(0),
                     info = paste("codec =", codec))
  }
})

test_that("a non-empty payload into a zero-capacity sink is an output limit", {
  x <- charToRaw("not empty")
  for (codec in c("identity", "deflate-raw", "zlib", "gzip")) {
    z <- komp_compress(x, codec)
    expect_error(zu_test_decompress_one(z, codec, 0),
                 class = "zukomp_output_limit")
  }
})

test_that("the one-shot ABI fills a sink sized exactly right", {
  withr::local_seed(20260908)
  x <- new_payload("ascii", 5000)
  for (codec in c("identity", "deflate-raw", "zlib", "gzip")) {
    z <- komp_compress(x, codec)
    expect_identical(zu_test_decompress_one(z, codec, length(x)), x,
                     info = paste("codec =", codec))
    expect_error(zu_test_decompress_one(z, codec, length(x) - 1L),
                 class = "zukomp_output_limit")
  }
})