Belle II Software  release-05-01-25
pylogging.py
1 #!/usr/bin/env python3
2 
3 import basf2
4 import unittest
5 from b2test_utils import run_in_subprocess, clean_working_directory
6 from contextlib import redirect_stdout
7 import io
8 import json
9 
10 # @cond internal_test
11 
12 
13 class PythonLogInterface(unittest.TestCase):
14  def assertDeath(self, function, *args, **kwargs):
15  """Run function in child process and check if it died. Only way to check for B2FATAL"""
16  exitcode = run_in_subprocess(target=function, *args, **kwargs)
17  self.assertNotEqual(exitcode, 0)
18 
19  def check_logs(self, target):
20  """Check the output of the log messages"""
21  with open("logmessages.txt") as f:
22  self.assertEqual(target, f.read())
23 
24  def setUp(self):
25  # disable error summary
26  basf2.logging.enable_summary(False)
27  # log to a file. We assume this is run in a temporary directory so
28  # fixed filename is fine
29  basf2.reset_log()
30  basf2.log_to_console(True)
31  basf2.log_to_file("logmessages.txt")
32  # Reset log level
33  basf2.logging.log_level = basf2.LogLevel.INFO
34  basf2.logging.debug_level = 100
35  # modify logging to remove the useless information
36  for level in basf2.LogLevel.values.values():
37  basf2.logging.set_info(level, basf2.LogInfo.LEVEL | basf2.LogInfo.MESSAGE)
38 
39  def test_debug(self):
40  # check argument handling
41  with self.assertRaises(TypeError):
42  basf2.B2DEBUG()
43  with self.assertRaises(TypeError):
44  basf2.B2DEBUG(1)
45  with self.assertRaises(TypeError):
46  basf2.B2DEBUG("foo", "bar")
47  with self.assertRaises(TypeError):
48  basf2.B2DEBUG(3.14, "bar")
49  with self.assertRaises(TypeError):
50  basf2.B2DEBUG([34, 324], "bar")
51 
52  # check visibility
53  basf2.B2DEBUG(101, "should not show")
54  basf2.B2DEBUG(100, "this neither")
55  basf2.logging.log_level = basf2.LogLevel.DEBUG
56  basf2.B2DEBUG(101, "should still not show")
57  basf2.B2DEBUG(100, "this should")
58  basf2.B2DEBUG(99, "and with variables", foo="bar", bar="foo", int=42, float=3.14)
59  self.check_logs("[DEBUG:100] this should\n"
60  "[DEBUG:99] and with variables\n"
61  "\tbar = foo\n"
62  "\tfloat = 3.14\n"
63  "\tfoo = bar\n"
64  "\tint = 42\n")
65 
66  def test_fatal(self):
67  # check argument handling
68  with self.assertRaises(TypeError):
69  basf2.B2FATAL()
70 
71  # check that fatal actually kills the process
72  def checkfatal():
73  basf2.B2FATAL("exit")
74  basf2.B2ERROR("should not show")
75  self.assertDeath(checkfatal)
76 
77  self.check_logs("[FATAL] exit\n")
78 
79  def test_fatal_pythonlogging(self):
80  # but use sys.stdout for logging
81  basf2.logging.enable_python_logging = True
82 
83  # check that fatal actually kills the process
84  def checkfatal():
85  try:
86  basf2.B2FATAL("exit")
87  except Exception as e:
88  basf2.B2ERROR("raised exception: ", e)
89  basf2.B2ERROR("should not show")
90  self.assertDeath(checkfatal)
91 
92  self.check_logs("[FATAL] exit\n")
93 
94  def test_others(self):
95  # check argument handling
96  for i, f in enumerate([basf2.B2INFO, basf2.B2WARNING, basf2.B2ERROR]):
97  with self.assertRaises(TypeError):
98  f()
99 
100  f("Show me", " a message with index ", i, index=i, dictvar={"i": i})
101 
102  # and visibility
103  basf2.logging.log_level = basf2.LogLevel.ERROR
104  basf2.B2INFO("No output here")
105  basf2.B2WARNING("Nor here")
106  basf2.B2ERROR("But here")
107 
108  self.check_logs("[INFO] Show me a message with index 0\n"
109  "\tdictvar = {'i': 0}\n"
110  "\tindex = 0\n"
111  "[WARNING] Show me a message with index 1\n"
112  "\tdictvar = {'i': 1}\n"
113  "\tindex = 1\n"
114  "[ERROR] Show me a message with index 2\n"
115  "\tdictvar = {'i': 2}\n"
116  "\tindex = 2\n"
117  "[ERROR] But here\n")
118 
119  def test_inspect(self):
120  # no change log info to show everything except time
121  li = basf2.LogInfo
122  basf2.logging.set_info(basf2.LogLevel.INFO, li.MESSAGE | li.LEVEL | li.PACKAGE | li.FUNCTION | li.FILE | li.LINE)
123  # sometimes the path to the file is absolute so make sure the filename
124  # is the same as reported for this frame. Also the line number changes
125  # every time we touch this file so determine it automatically
126  import inspect
127  filename = inspect.currentframe().f_code.co_filename
128  lineno = inspect.currentframe().f_lineno + 2
129  # and print a message
130  basf2.B2INFO("show current frame info", why="because we can")
131  self.check_logs(
132  "[INFO] show current frame info\n"
133  "\twhy = because we can { package: steering function: test_inspect @%s:%d }\n" %
134  (filename, lineno))
135 
136 
137 class PythonLogJSON(unittest.TestCase):
138  def setUp(self):
139  # disable error summary
140  basf2.logging.enable_summary(False)
141  # but use sys.stdout for logging
142  basf2.logging.enable_python_logging = True
143  # log to a file. We assume this is run in a temporary directory so
144  # fixed filename is fine
145  basf2.reset_log()
146  basf2.logging.add_json()
147  # Reset log level
148  basf2.logging.log_level = basf2.LogLevel.INFO
149  basf2.logging.debug_level = 100
150  # modify logging to remove the useless information
151  for level in basf2.LogLevel.values.values():
152  basf2.logging.set_info(level, basf2.LogInfo.LEVEL | basf2.LogInfo.MESSAGE)
153 
154  def getLogMessage(self, message, **argk):
155  out = io.StringIO()
156  with redirect_stdout(out):
157  basf2.B2INFO(message, **argk)
158  raw = out.getvalue()
159  # make sure string ends in new line
160  self.assertEqual(raw[-1], "\n")
161  # make sure there are no newlines in the raw string
162  self.assertTrue(raw[:-1].find("\n") < 0)
163  # make sure we can parse json
164  logobject = json.loads(raw)
165  return logobject
166 
167  def assertLogMessage(self, message, **argk):
168  # and that the message and level are there
169  logobject = self.getLogMessage(message, **argk)
170  if len(argk):
171  stringified = {str(key): str(val) for key, val in argk.items()}
172  self.assertDictEqual(logobject, {"level": "INFO", "message": message, "variables": stringified})
173  else:
174  self.assertDictEqual(logobject, {"level": "INFO", "message": message})
175 
176  def test_info(self):
177  self.assertLogMessage("message")
178 
179  def test_multiline(self):
180  self.assertLogMessage("message\ncontaining\nnewlines")
181 
182  def test_vars(self):
183  self.assertLogMessage("message", var1="foo", var2="bar", int=3)
184 
185  def test_vars_newline(self):
186  self.assertLogMessage("message", var1="foo\nbar", var2="bar\nboo")
187 
188  def test_utf8(self):
189  zalgo = "h͌̉e̳̞̞͆ͨ̏͋̕ ͍͚̱̰̀͡c͟o͛҉̟̰̫͔̟̪̠m̴̀ͯ̿͌ͨ̃͆e̡̦̦͖̳͉̗ͨͬ̑͌̃ͅt̰̝͈͚͍̳͇͌h̭̜̙̦̣̓̌̃̓̀̉͜!̱̞̻̈̿̒̀͢!̋̽̍̈͐ͫ͏̠̹̺̜̬͍ͅ"
190  self.assertLogMessage(zalgo, **{zalgo: zalgo})
191 
192  def test_loginfo(self):
193  """Make sure all loginfo settings are honored in json output"""
194  cum_info = 0
195  # level is always added
196  cum_keys = {"level"}
197  # go through all loginfo values and check if the correct field is in
198  # the json, once separately and also cumulative
199  for val, key in basf2.LogInfo.values.items():
200  cum_info |= val
201  key = key.name.lower()
202  cum_keys.add(key)
203  for info, keys in (val, {"level", key}), (cum_info, cum_keys):
204  basf2.logging.set_info(basf2.LogLevel.INFO, info)
205  logobject = self.getLogMessage("simple")
206  self.assertEqual(set(logobject.keys()), keys)
207  varkeys = keys.copy()
208  # variables are parts of messages ... but they only show up if
209  # message is shown
210  if "message" in keys:
211  varkeys.add("variables")
212  logobject = self.getLogMessage("with var", var="val")
213  self.assertEqual(set(logobject.keys()), varkeys)
214 
215  def test_complete(self):
216  """Test that we always get all fields when setting output to complete"""
217  basf2.reset_log()
218  basf2.logging.add_json(True)
219  logobject = self.getLogMessage("message")
220  self.assertSetEqual(set(logobject.keys()), {"level", "message", "variables", "module", "package",
221  "function", "file", "line", "timestamp", "proc"})
222 
223 
224 if __name__ == "__main__":
225  with clean_working_directory():
226  unittest.main(verbosity=0)
227 
228 # @endcond