1############################################################################### 2# 3# The MIT License (MIT) 4# 5# Copyright (c) Crossbar.io Technologies GmbH 6# 7# Permission is hereby granted, free of charge, to any person obtaining a copy 8# of this software and associated documentation files (the "Software"), to deal 9# in the Software without restriction, including without limitation the rights 10# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 11# copies of the Software, and to permit persons to whom the Software is 12# furnished to do so, subject to the following conditions: 13# 14# The above copyright notice and this permission notice shall be included in 15# all copies or substantial portions of the Software. 16# 17# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 18# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 19# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 20# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 21# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 22# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 23# THE SOFTWARE. 24# 25############################################################################### 26 27from __future__ import print_function 28 29from collections import namedtuple 30from io import BytesIO, StringIO 31 32import os 33import six 34import pytest 35import txaio 36 37Log = namedtuple('Log', ['args']) 38 39 40class LoggingHandler(BytesIO): 41 42 @property 43 def messages(self): 44 # Because we print the \n after, there will always be an empty 45 # 'message', so just don't include it. 46 return self.getvalue().split(os.linesep.encode('ascii'))[:-1] 47 48 49_handler = LoggingHandler() 50 51 52@pytest.fixture 53def log_started(framework): 54 """ 55 Sets up the logging, which we can only do once per run. 56 """ 57 early_log = txaio.make_logger() 58 early_log.info("early log") 59 60 txaio.start_logging(out=_handler, level='debug') 61 62 63@pytest.fixture(scope='function') 64def handler(log_started): 65 """ 66 Resets the global TestHandler instance for each test. 67 """ 68 global _handler 69 _handler.truncate(0) 70 _handler.seek(0) 71 return _handler 72 73 74def test_categories(handler, framework): 75 """ 76 Calling ``txaio.add_log_categories`` with a dict mapping category keys to 77 category log formats will allow log messages which have the 78 ``log_category`` key take the associated format. 79 """ 80 logger = txaio.make_logger() 81 82 txaio.add_log_categories({"TX100": u"{adjective} {nouns[2]}"}) 83 84 # do something a little fancy, with attribute access etc. 85 logger.critical( 86 "you won't see me", 87 log_category="TX100", 88 adjective='hilarious', 89 nouns=['skunk', 'elephant', 'wombat'], 90 ) 91 92 assert len(handler.messages) == 1 93 assert handler.messages[0].endswith(b"hilarious wombat") 94 95 96def test_categories_subsequent(handler, framework): 97 """ 98 Later calls to add_log_categories update the list of log categories and 99 take precedence. 100 """ 101 logger = txaio.make_logger() 102 103 txaio.add_log_categories({"TX100": u"{adjective} {nouns[2]}"}) 104 txaio.add_log_categories({"TX100": u"{adjective} {nouns[1]}"}) 105 106 # do something a little fancy, with attribute access etc. 107 logger.critical( 108 log_category="TX100", 109 adjective='hilarious', 110 nouns=['skunk', 'elephant', 'wombat'], 111 ) 112 113 assert len(handler.messages) == 1 114 assert handler.messages[0].endswith(b"hilarious elephant") 115 116 117def test_critical(handler, framework): 118 logger = txaio.make_logger() 119 120 # do something a little fancy, with attribute access etc. 121 logger.critical( 122 "{adjective} {nouns[2]}", 123 adjective='hilarious', 124 nouns=['skunk', 'elephant', 'wombat'], 125 ) 126 127 assert len(handler.messages) == 1 128 assert handler.messages[0].endswith(b"hilarious wombat") 129 130 131def test_info(handler, framework): 132 logger = txaio.make_logger() 133 134 # do something a little fancy, with attribute access etc. 135 logger.info( 136 "{adjective} {nouns[1]}", 137 adjective='hilarious', 138 nouns=['skunk', 'elephant', 'wombat'], 139 ) 140 141 assert len(handler.messages) == 1 142 assert handler.messages[0].endswith(b"hilarious elephant") 143 144 145def test_legacy_error_with_traceback(handler, framework): 146 if framework.using_twisted: 147 return pytest.skip('test only for asyncio users') 148 149 import logging 150 151 try: 152 raise RuntimeError("the bad stuff") 153 except RuntimeError: 154 logging.error("bad stuff", exc_info=True) 155 156 assert 'RuntimeError: the bad stuff' in str(handler.messages) 157 158 159def test_trace(handler, framework): 160 logger = txaio.make_logger() 161 old_log = txaio.get_global_log_level() 162 txaio.set_global_log_level("trace") 163 164 # the txaio_trace variable should be in it 165 logger.trace( 166 "trace {txaio_trace}", 167 ) 168 169 txaio.set_global_log_level(old_log) 170 171 assert len(handler.messages) == 1 172 assert handler.messages[0].endswith(b"trace True") 173 174 175def test_emit_noop(handler, framework): 176 """ 177 emit() with a too-low level is an no-op. 178 """ 179 logger = txaio.make_logger() 180 181 old_log = txaio.get_global_log_level() 182 txaio.set_global_log_level("info") 183 184 logger.emit("debug", "foobar") 185 186 txaio.set_global_log_level(old_log) 187 188 assert len(handler.messages) == 0 189 190 191def test_emit_ok(handler, framework): 192 """ 193 emit() with an OK level emits the message. 194 """ 195 logger = txaio.make_logger() 196 197 old_log = txaio.get_global_log_level() 198 txaio.set_global_log_level("trace") 199 200 logger.emit("trace", "foobar") 201 logger.emit("info", "barbaz") 202 203 txaio.set_global_log_level(old_log) 204 205 assert len(handler.messages) == 2 206 assert handler.messages[0].endswith(b"foobar") 207 assert handler.messages[1].endswith(b"barbaz") 208 209 210def test_bad_failures(handler, framework): 211 # just ensuring this doesn't explode 212 txaio.failure_format_traceback("not a failure") 213 txaio.failure_message("not a failure") 214 215 216def test_debug_with_object(handler, framework): 217 logger = txaio.make_logger() 218 219 class Shape(object): 220 sides = 4 221 name = "bamboozle" 222 config = dict(foo='bar') 223 224 logger.info( 225 "{what.config[foo]} {what.sides} {what.name}", 226 what=Shape(), 227 ) 228 229 assert len(handler.messages) == 1 230 assert handler.messages[0].endswith(b"bar 4 bamboozle") 231 232 233def test_log_noop_trace(handler, framework): 234 # trace should be a no-op, because we set the level to 'debug' in 235 # the fixture 236 logger = txaio.make_logger() 237 238 logger.trace("a trace message") 239 240 assert len(handler.messages) == 0 241 242 243def test_double_start(handler, framework): 244 try: 245 txaio.start_logging() 246 except RuntimeError: 247 assert False, "shouldn't get exception" 248 249 250def test_invalid_level(framework): 251 try: 252 txaio.start_logging(level='foo') 253 assert False, "should get exception" 254 except RuntimeError as e: 255 assert 'Invalid log level' in str(e) 256 257 258def test_class_descriptor(handler, framework): 259 class Something(object): 260 log = txaio.make_logger() 261 262 def do_a_thing(self): 263 self.log.info("doing a thing") 264 265 s = Something() 266 s.do_a_thing() 267 268 assert len(handler.messages) == 1 269 assert handler.messages[0].endswith(b"doing a thing") 270 271 272def test_class_attribute(handler, framework): 273 class Something(object): 274 def __init__(self): 275 self.log = txaio.make_logger() 276 277 def do_a_thing(self): 278 self.log.info("doing a thing") 279 280 s = Something() 281 s.do_a_thing() 282 283 assert len(handler.messages) == 1 284 assert handler.messages[0].endswith(b"doing a thing") 285 286 287def test_log_converter(handler, framework): 288 pytest.importorskip("twisted.logger") 289 # this checks that we can convert a plain Twisted Logger calling 290 # failure() into a traceback on our observers. 291 from twisted.logger import Logger 292 from txaio.tx import _LogObserver 293 294 out = six.StringIO() 295 observer = _LogObserver(out) 296 logger = Logger(observer=observer) 297 298 try: 299 raise RuntimeError("failed on purpose") 300 except RuntimeError: 301 logger.failure(None) 302 303 output = out.getvalue() 304 assert "failed on purpose" in output 305 assert "Traceback" in output 306 307 308def test_txlog_write_binary(handler, framework): 309 """ 310 Writing to a binary stream is supported. 311 """ 312 pytest.importorskip("twisted.logger") 313 from txaio.tx import _LogObserver 314 315 out_file = BytesIO() 316 observer = _LogObserver(out_file) 317 318 observer({ 319 "log_format": "hi: {testentry}", 320 "testentry": "hello", 321 "log_level": observer.to_tx["info"], 322 "log_time": 1442890018.002233 323 }) 324 325 output = out_file.getvalue() 326 assert b"hi: hello" in output 327 328 329def test_txlog_write_text(handler, framework_tx): 330 """ 331 Writing to a text stream is supported. 332 """ 333 pytest.importorskip("twisted.logger") 334 from txaio.tx import _LogObserver 335 336 out_file = StringIO() 337 observer = _LogObserver(out_file) 338 339 observer({ 340 "log_format": "hi: {testentry}", 341 "testentry": "hello", 342 "log_level": observer.to_tx["info"], 343 "log_time": 1442890018.002233 344 }) 345 346 output = out_file.getvalue() 347 assert u"hi: hello" in output 348 349 350def test_aiolog_write_binary(handler, framework_aio): 351 """ 352 Writing to a binary stream is supported. 353 """ 354 pytest.importorskip("txaio.aio") 355 from txaio.aio import _TxaioFileHandler 356 357 out_file = BytesIO() 358 observer = _TxaioFileHandler(out_file) 359 360 observer.emit(Log(args={ 361 "log_message": "hi: {testentry}", 362 "testentry": "hello", 363 "log_time": 1442890018.002233 364 })) 365 366 output = out_file.getvalue() 367 assert b"hi: hello" in output 368 369 370def test_aiolog_write_text(handler, framework_aio): 371 """ 372 Writing to a text stream is supported. 373 """ 374 pytest.importorskip("txaio.aio") 375 from txaio.aio import _TxaioFileHandler 376 377 out_file = StringIO() 378 observer = _TxaioFileHandler(out_file) 379 380 observer.emit(Log(args={ 381 "log_message": "hi: {testentry}", 382 "testentry": "hello", 383 "log_time": 1442890018.002233 384 })) 385 386 output = out_file.getvalue() 387 assert u"hi: hello" in output 388