File: test-python-output.R

package info (click to toggle)
r-cran-reticulate 1.41.0.1%2Bdfsg-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 3,088 kB
  • sloc: cpp: 5,154; python: 620; sh: 13; makefile: 2
file content (107 lines) | stat: -rw-r--r-- 3,189 bytes parent folder | download
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
context("output")

capture_test_output <- function(type) {
  sys <- import("sys")
  py_capture_output(type = type, {
    if ("stdout" %in% type)
      sys$stdout$write("out\n");
    if ("stderr" %in% type)
    sys$stderr$write("err\n");
  })
}

test_that("Python streams can be captured", {
  skip_if_no_python()
  expect_equal(capture_test_output(type = c("stdout", "stderr")) ,"out\nerr\n")
})

test_that("Python stdout stream can be captured", {
  skip_if_no_python()
  expect_equal(capture_test_output(type = "stdout") , "out\n")
})

test_that("Python stderr stream can be captured", {
  skip_if_no_python()
  expect_equal(capture_test_output(type = "stderr") , "err\n")
})

test_that("Python loggers work with py_capture_output", {

  skip_if(py_version() < "3.2")
  skip_on_os("windows")

  output <- py_capture_output({
    logging <- import("logging")
    l <- logging$getLogger("test.logger")
    l$addHandler(logging$StreamHandler())
    l$setLevel("INFO")
    l$info("info")
  })

  expect_equal(output, "info\n")

  l <- logging$getLogger("test.logger2")
  l$addHandler(logging$StreamHandler())
  l$setLevel("INFO")
  output <- py_capture_output(l$info("info"))

  expect_equal(output, "info\n")

})


test_that("nested py_capture_output() calls work", {

  # capture original py ids to check we restored
  # everything correctly at the end
  sys <- import("sys")
  og_sys.stdout_pd_id <- py_id(sys$stdout)
  og_sys.stderr_pd_id <- py_id(sys$stderr)
  og_sys.__stdout___pd_id <- py_id(sys$`__stdout__`)
  og_sys.__stderr___pd_id <- py_id(sys$`__stderr__`)

  # Outer level captures both stdout and stderr
  level_1 <- py_capture_output({

    py_run_string("print('Start outer')")

    # Middle level is configured to only capture stdout,
    # allowing stderr to propagate to the outer level
    level_2 <- py_capture_output(type = "stdout", {

      py_run_string("print('Start middle')")

      # Innermost level captures both stdout and stderr
      level_3 <- py_capture_output({
        py_run_string("print('Start inner')")
        py_run_string("import sys; print('Innermost error', file=sys.stderr)")
        py_run_string("print('End inner')")
      })

      # Middle level generating stderr, should be captured by level_1
      py_run_string("import sys; print('Middle level error', file=sys.stderr)")
      py_run_string("print('End middle')")

    })

    py_run_string("print('End outer')")
  })

  # level_1 captures both stdout and stderr, including the
  # stderr propagated from the middle level
  expect_equal(level_1, "Start outer\nMiddle level error\nEnd outer\n")

  # level_2 only captures stdout, so the stderr from the middle level is not here
  expect_equal(level_2, "Start middle\nEnd middle\n")

  # level 3 captures both stdout and stderr
  expect_equal(level_3, "Start inner\nInnermost error\nEnd inner\n")

  # Check the original streams were restored correctly
  sys <- import("sys")
  expect_identical(og_sys.stdout_pd_id, py_id(sys$stdout))
  expect_identical(og_sys.stderr_pd_id, py_id(sys$stderr))
  expect_identical(og_sys.__stdout___pd_id, py_id(sys$`__stdout__`))
  expect_identical(og_sys.__stderr___pd_id, py_id(sys$`__stderr__`))

})