tests/testthat/test-3-readable.R

In ElianHugh/emitters: NodeJS-inspired events and io streams.

helper_file <- testthat::test_path("data", "helper_dummy.txt")

# ~~~~~~~ ReadableStream ~~~~~ #

test_that(
    "3.1.1 ReadableStreams can be created",
    code = {
        expect_error(ReadableStream$new(helper_file), NA)
        expect_error(ReadableStream$new(mtcars), NA)
        expect_error(ReadableStream$new(1:10), NA)
    }
)

test_that(
    "3.1.1 ReadableStream can be manually read from",
    code = {
        stream <- ReadableStream$new(helper_file)

        out <- c()
        stream$on("readable", function() {
            out <<- append(out, stream$read())
        })

        wait_for_change("out", c())
        testthat::expect_vector(out, ptype = character())

        out <- stream$read()
        testthat::expect_null(out)
        testthat::expect_length(out, 0)

        rm(stream)
    }
)

test_that(
    "3.1.2 ReadableStream can be read from via emit",
    code = {
        stream <- ReadableStream$new(helper_file)

        out <- c()
        stream$on("data", \(x) {
            out <<- append(out, x)
        })

        wait_for_change("out", c())
        testthat::expect_length(out, 1)

        rm(stream)
    }
)

test_that(
    "3.1.3 ReadableStream events are appropriately emitted",
    code = {
        stream <- ReadableStream$new(helper_file)

        data_emit <- FALSE
        pause_emit <- FALSE
        resume_emit <- FALSE
        close_emit <- FALSE

        expect_false(data_emit)
        expect_false(pause_emit)
        expect_false(resume_emit)
        expect_false(close_emit)

        stream$once("data", function(x) {
            data_emit <<- TRUE
        })

        stream$once("pause", function() {
            pause_emit <<- TRUE
        })

        stream$once("resume", function() {
            resume_emit <<- TRUE
        })

        stream$once("close", function() {
            close_emit <<- TRUE
        })

        wait_for_change("data_emit", FALSE)
        stream$pause()
        wait_for_change("pause_emit", FALSE)
        suppressWarnings(stream$resume())
        wait_for_change("resume_emit", FALSE)

        expect_true(data_emit)
        expect_true(pause_emit)
        expect_true(resume_emit)

        stream$destroy()
        wait_for_change("close_emit", FALSE)
        expect_true(close_emit)

        rm(stream)
    }
)