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