1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
|
test_that("fails in windows", {
skip_other_platforms("windows")
expect_error(process$new("R", pty = TRUE), "only implemented on Unix",
class = "error")
})
test_that("pty works", {
skip_other_platforms("unix")
skip_on_os("solaris")
skip_on_cran()
p <- process$new("cat", pty = TRUE)
on.exit(p$kill(), add = TRUE)
expect_true(p$is_alive())
if (!p$is_alive()) stop("process not running")
pr <- p$poll_io(0)
expect_equal(pr[["output"]], "timeout")
p$write_input("foobar\n")
pr <- p$poll_io(300)
expect_equal(pr[["output"]], "ready")
if (pr[["output"]] != "ready") stop("no output")
expect_equal(p$read_output(), "foobar\r\n")
})
test_that("pty echo", {
skip_other_platforms("unix")
skip_on_os("solaris")
skip_on_cran()
p <- process$new("cat", pty = TRUE, pty_options = list(echo = TRUE))
on.exit(p$kill(), add = TRUE)
expect_true(p$is_alive())
if (!p$is_alive()) stop("process not running")
pr <- p$poll_io(0)
expect_equal(pr[["output"]], "timeout")
p$write_input("foo")
pr <- p$poll_io(300)
expect_equal(pr[["output"]], "ready")
if (pr[["output"]] != "ready") stop("no output")
expect_equal(p$read_output(), "foo")
p$write_input("bar\n")
pr <- p$poll_io(300)
expect_equal(pr[["output"]], "ready")
if (pr[["output"]] != "ready") stop("no output")
expect_equal(p$read_output(), "bar\r\nfoobar\r\n")
})
test_that("read_output_lines() fails for pty", {
skip_other_platforms("unix")
skip_on_os("solaris")
skip_on_cran()
p <- process$new("cat", pty = TRUE)
p$write_input("foobar\n")
expect_error(p$read_output_lines(), "Cannot read lines from a pty")
pr <- p$poll_io(300)
expect_equal(pr[["output"]], "ready")
if (pr[["output"]] != "ready") stop("no output")
expect_equal(p$read_output(), "foobar\r\n")
})
|
