test_logging.py 5.35 KB
Newer Older
André Anjos's avatar
André Anjos committed
1
2
3
4
5
6
7
8
#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
# Andre Anjos <andre.anjos@idiap.ch>
# Tue 09 Jul 2013 13:24:49 CEST

"""Tests for the logging subsystem
"""

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
import bob.core

def test_from_python():
  logger = bob.core.log.setup("bob.core")
  bob.core.log.set_verbosity_level(logger, 2)

  # send a log message
  logger.debug("This is a test debug message")
  logger.info("This is a test info message")
  bob.core.log.set_verbosity_level(logger, 0)
  logger.warn("This is a test warn message")
  logger.error("This is a test error message")


def test_from_python_output():
  import logging, sys

  if sys.version_info[0] < 3:
    from StringIO import StringIO
  else:
    from io import StringIO
  # create an artificial logger using the logging module
  logger = logging.getLogger("XXX.YYY")

  # add handlers
  out, err = StringIO(), StringIO()
  _warn_err = logging.StreamHandler(err)
  _warn_err.setLevel(logging.WARNING)
  logger.addHandler(_warn_err)

  _debug_info = logging.StreamHandler(out)
  _debug_info.setLevel(logging.DEBUG)
  _debug_info.addFilter(bob.core.log._InfoFilter())
  logger.addHandler(_debug_info)

  # now, set up the logger
  logger = bob.core.log.setup("XXX.YYY")

  # send log messages
  bob.core.log.set_verbosity_level(logger, 2)
  logger.debug("This is a test debug message")
  logger.info("This is a test info message")

  bob.core.log.set_verbosity_level(logger, 0)
  logger.warn("This is a test warn message")
  logger.error("This is a test error message")

  out = out.getvalue().rstrip()
  err = err.getvalue().rstrip()

  assert out.startswith("XXX.YYY")
  assert "INFO" in out
  assert out.endswith("This is a test info message")

  assert err.startswith("XXX.YYY")
  assert "ERROR" in err
  assert err.endswith("This is a test error message")

André Anjos's avatar
André Anjos committed
67
68

def test_from_cxx():
69
70
  from bob.core._test import _test_log_message
  _test_log_message(1, 'error', 'this is a test message')
André Anjos's avatar
André Anjos committed
71
72

def test_from_cxx_multithreaded():
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
  from bob.core._test import _test_log_message_mt
  _test_log_message_mt(2, 1, 'error', 'this is a test message')



def test_from_cxx_disable():
  from bob.core._test import _test_output_disable
  import sys, os
  import threading

  class OutputGrabber(object):
      """
      Class used to grab standard output or another stream.
      This is a slight modification of what was proposed here:
      http://stackoverflow.com/a/29834357/3301902
      """
      escape_char = "\b"

      def __init__(self, stream=None, threaded=False):
          self.origstream = stream
          self.threaded = threaded
          if self.origstream is None:
              self.origstream = sys.stdout
          self.origstreamfd = self.origstream.fileno()
          self.capturedtext = ""
          # Create a pipe so the stream can be captured:
          self.pipe_out, self.pipe_in = os.pipe()

      def start(self):
          """
          Start capturing the stream data.
          """
          self.capturedtext = ""
          # Save a copy of the stream:
          self.streamfd = os.dup(self.origstreamfd)
          # Replace the Original stream with our write pipe
          os.dup2(self.pipe_in, self.origstreamfd)
          if self.threaded:
              # Start thread that will read the stream:
              self.workerThread = threading.Thread(target=self.readOutput)
              self.workerThread.start()
              # Make sure that the thread is running and os.read is executed:
              time.sleep(0.01)

      def stop(self):
          """
          Stop capturing the stream data and save the text in `capturedtext`.
          """
          # Flush the stream to make sure all our data goes in before
          # the escape character.
          self.origstream.flush()
          # Print the escape character to make the readOutput method stop:
          self.origstream.write(self.escape_char)
          self.origstream.flush()
          if self.threaded:
              # wait until the thread finishes so we are sure that
              # we have until the last character:
              self.workerThread.join()
          else:
              self.readOutput()
          # Close the pipe:
          os.close(self.pipe_out)
          # Restore the original stream:
          os.dup2(self.streamfd, self.origstreamfd)

      def readOutput(self):
          """
          Read the stream data (one byte at a time)
          and save the text in `capturedtext`.
          """
          while True:
              data = os.read(self.pipe_out, 1)  # Read One Byte Only
              if self.escape_char in data:
                  break
              if not data:
                  break
              self.capturedtext += data


  try:
    # redirect output and error streams
    out = OutputGrabber(sys.stdout)
    err = OutputGrabber(sys.stderr)
    out.start()
    err.start()

    # run our code
    _test_output_disable()

  finally:
    # Clean up the pipe and restore the original stdout
    out.stop()
    err.stop()

  # output should contain two lines
  outs = out.capturedtext.rstrip().split("\n")
  assert len(outs) == 2
  assert outs[0] == "This is a debug message"
  assert outs[1] == "This is an info message"

  # error should contain three lines; error message is printed twice
  errs = err.capturedtext.rstrip().split("\n")
  assert len(errs) == 3
  assert errs[0] == "This is a warning message"
  assert errs[1] == "This is an error message"
  assert errs[2] == "This is an error message"