5 from b2test_utils
import run_in_subprocess, clean_working_directory
6 from contextlib
import redirect_stdout
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)
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())
26 basf2.logging.enable_summary(
False)
30 basf2.log_to_console(
True)
31 basf2.log_to_file(
"logmessages.txt")
33 basf2.logging.log_level = basf2.LogLevel.INFO
34 basf2.logging.debug_level = 100
36 for level
in basf2.LogLevel.values.values():
37 basf2.logging.set_info(level, basf2.LogInfo.LEVEL | basf2.LogInfo.MESSAGE)
41 with self.assertRaises(TypeError):
43 with self.assertRaises(TypeError):
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")
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"
68 with self.assertRaises(TypeError):
74 basf2.B2ERROR(
"should not show")
75 self.assertDeath(checkfatal)
77 self.check_logs(
"[FATAL] exit\n")
79 def test_fatal_pythonlogging(self):
81 basf2.logging.enable_python_logging =
True
87 except Exception
as e:
88 basf2.B2ERROR(
"raised exception: ", e)
89 basf2.B2ERROR(
"should not show")
90 self.assertDeath(checkfatal)
92 self.check_logs(
"[FATAL] exit\n")
94 def test_others(self):
96 for i, f
in enumerate([basf2.B2INFO, basf2.B2WARNING, basf2.B2ERROR]):
97 with self.assertRaises(TypeError):
100 f(
"Show me",
" a message with index ", i, index=i, dictvar={
"i": i})
103 basf2.logging.log_level = basf2.LogLevel.ERROR
104 basf2.B2INFO(
"No output here")
105 basf2.B2WARNING(
"Nor here")
106 basf2.B2ERROR(
"But here")
108 self.check_logs(
"[INFO] Show me a message with index 0\n"
109 "\tdictvar = {'i': 0}\n"
111 "[WARNING] Show me a message with index 1\n"
112 "\tdictvar = {'i': 1}\n"
114 "[ERROR] Show me a message with index 2\n"
115 "\tdictvar = {'i': 2}\n"
117 "[ERROR] But here\n")
119 def test_inspect(self):
122 basf2.logging.set_info(basf2.LogLevel.INFO, li.MESSAGE | li.LEVEL | li.PACKAGE | li.FUNCTION | li.FILE | li.LINE)
127 filename = inspect.currentframe().f_code.co_filename
128 lineno = inspect.currentframe().f_lineno + 2
130 basf2.B2INFO(
"show current frame info", why=
"because we can")
132 "[INFO] show current frame info\n"
133 "\twhy = because we can { package: steering function: test_inspect @%s:%d }\n" %
137 class PythonLogJSON(unittest.TestCase):
140 basf2.logging.enable_summary(
False)
142 basf2.logging.enable_python_logging =
True
146 basf2.logging.add_json()
148 basf2.logging.log_level = basf2.LogLevel.INFO
149 basf2.logging.debug_level = 100
151 for level
in basf2.LogLevel.values.values():
152 basf2.logging.set_info(level, basf2.LogInfo.LEVEL | basf2.LogInfo.MESSAGE)
154 def getLogMessage(self, message, **argk):
156 with redirect_stdout(out):
157 basf2.B2INFO(message, **argk)
160 self.assertEqual(raw[-1],
"\n")
162 self.assertTrue(raw[:-1].find(
"\n") < 0)
164 logobject = json.loads(raw)
167 def assertLogMessage(self, message, **argk):
169 logobject = self.getLogMessage(message, **argk)
171 stringified = {str(key): str(val)
for key, val
in argk.items()}
172 self.assertDictEqual(logobject, {
"level":
"INFO",
"message": message,
"variables": stringified})
174 self.assertDictEqual(logobject, {
"level":
"INFO",
"message": message})
177 self.assertLogMessage(
"message")
179 def test_multiline(self):
180 self.assertLogMessage(
"message\ncontaining\nnewlines")
183 self.assertLogMessage(
"message", var1=
"foo", var2=
"bar", int=3)
185 def test_vars_newline(self):
186 self.assertLogMessage(
"message", var1=
"foo\nbar", var2=
"bar\nboo")
189 zalgo =
"h͌̉e̳̞̞͆ͨ̏͋̕ ͍͚̱̰̀͡c͟o͛҉̟̰̫͔̟̪̠m̴̀ͯ̿͌ͨ̃͆e̡̦̦͖̳͉̗ͨͬ̑͌̃ͅt̰̝͈͚͍̳͇͌h̭̜̙̦̣̓̌̃̓̀̉͜!̱̞̻̈̿̒̀͢!̋̽̍̈͐ͫ͏̠̹̺̜̬͍ͅ"
190 self.assertLogMessage(zalgo, **{zalgo: zalgo})
192 def test_loginfo(self):
193 """Make sure all loginfo settings are honored in json output"""
199 for val, key
in basf2.LogInfo.values.items():
201 key = key.name.lower()
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()
210 if "message" in keys:
211 varkeys.add(
"variables")
212 logobject = self.getLogMessage(
"with var", var=
"val")
213 self.assertEqual(set(logobject.keys()), varkeys)
215 def test_complete(self):
216 """Test that we always get all fields when setting output to complete"""
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"})
224 if __name__ ==
"__main__":
225 with clean_working_directory():
226 unittest.main(verbosity=0)