""" Tests for L{eliot._output}. """ from sys import stdout from unittest import TestCase, skipUnless # Make sure to use StringIO that only accepts unicode: from io import BytesIO, StringIO import json as pyjson from tempfile import mktemp from time import time from uuid import UUID from threading import Thread try: import numpy as np except ImportError: np = None from zope.interface.verify import verifyClass from .._output import ( MemoryLogger, ILogger, Destinations, Logger, to_file, FileDestination, _safe_unicode_dictionary, ) from .._action import start_action from .._validation import ValidationError, Field, _MessageSerializer from .._traceback import write_traceback from ..testing import assertContainsFields from .common import CustomObject, CustomJSONEncoder class MemoryLoggerTests(TestCase): """ Tests for L{MemoryLogger}. """ def test_interface(self): """ L{MemoryLogger} implements L{ILogger}. """ verifyClass(ILogger, MemoryLogger) def test_write(self): """ Dictionaries written with L{MemoryLogger.write} are stored on a list. """ logger = MemoryLogger() logger.write({"a": "b"}) logger.write({"c": 1}) self.assertEqual(logger.messages, [{"a": "b"}, {"c": 1}]) logger.validate() def test_notStringFieldKeys(self): """ Field keys must be unicode or bytes; if not L{MemoryLogger.validate} raises a C{TypeError}. """ logger = MemoryLogger() logger.write({123: "b"}) self.assertRaises(TypeError, logger.validate) def test_bytesMustBeUTF8(self): """ Field keys can be bytes, but only if they're UTF-8 encoded Unicode. """ logger = MemoryLogger() logger.write({"\u1234".encode("utf-16"): "b"}) self.assertRaises(UnicodeDecodeError, logger.validate) def test_serializer(self): """ L{MemoryLogger.validate} calls the given serializer's C{validate()} method with the message, as does L{MemoryLogger.write}. """ class FakeValidator(list): def validate(self, message): self.append(message) def serialize(self, obj): return obj validator = FakeValidator() logger = MemoryLogger() message = {"message_type": "mymessage", "X": 1} logger.write(message, validator) self.assertEqual(validator, [message]) logger.validate() self.assertEqual(validator, [message, message]) def test_failedValidation(self): """ L{MemoryLogger.validate} will allow exceptions raised by the serializer to pass through. """ serializer = _MessageSerializer( [Field.forValue("message_type", "mymessage", "The type")] ) logger = MemoryLogger() logger.write({"message_type": "wrongtype"}, serializer) self.assertRaises(ValidationError, logger.validate) def test_JSON(self): """ L{MemoryLogger.validate} will encode the output of serialization to JSON. """ serializer = _MessageSerializer( [ Field.forValue("message_type", "type", "The type"), Field("foo", lambda value: object(), "The type"), ] ) logger = MemoryLogger() logger.write( {"message_type": "type", "foo": "will become object()"}, serializer ) self.assertRaises(TypeError, logger.validate) @skipUnless(np, "NumPy is not installed.") def test_EliotJSONEncoder(self): """ L{MemoryLogger.validate} uses the EliotJSONEncoder by default to do encoding testing. """ logger = MemoryLogger() logger.write({"message_type": "type", "foo": np.uint64(12)}, None) logger.validate() def test_JSON_custom_encoder(self): """ L{MemoryLogger.validate} will use a custom JSON encoder if one was given. """ logger = MemoryLogger(encoder=CustomJSONEncoder) logger.write( {"message_type": "type", "custom": CustomObject()}, None, ) logger.validate() def test_serialize(self): """ L{MemoryLogger.serialize} returns a list of serialized versions of the logged messages. """ serializer = _MessageSerializer( [ Field.forValue("message_type", "mymessage", "The type"), Field("length", len, "The length"), ] ) messages = [ {"message_type": "mymessage", "length": "abc"}, {"message_type": "mymessage", "length": "abcd"}, ] logger = MemoryLogger() for message in messages: logger.write(message, serializer) self.assertEqual( logger.serialize(), [ {"message_type": "mymessage", "length": 3}, {"message_type": "mymessage", "length": 4}, ], ) def test_serializeCopies(self): """ L{MemoryLogger.serialize} does not mutate the original logged messages. """ serializer = _MessageSerializer( [ Field.forValue("message_type", "mymessage", "The type"), Field("length", len, "The length"), ] ) message = {"message_type": "mymessage", "length": "abc"} logger = MemoryLogger() logger.write(message, serializer) logger.serialize() self.assertEqual(logger.messages[0]["length"], "abc") def write_traceback(self, logger, exception): """ Write an exception as a traceback to the logger. """ try: raise exception except: write_traceback(logger) def test_tracebacksCauseTestFailure(self): """ Logging a traceback to L{MemoryLogger} will add its exception to L{MemoryLogger.tracebackMessages}. """ logger = MemoryLogger() exception = Exception() self.write_traceback(logger, exception) self.assertEqual(logger.tracebackMessages[0]["reason"], exception) def test_flushTracebacksNoTestFailure(self): """ Any tracebacks cleared by L{MemoryLogger.flushTracebacks} (as specified by exception type) are removed from L{MemoryLogger.tracebackMessages}. """ logger = MemoryLogger() exception = RuntimeError() self.write_traceback(logger, exception) logger.flushTracebacks(RuntimeError) self.assertEqual(logger.tracebackMessages, []) def test_flushTracebacksReturnsExceptions(self): """ L{MemoryLogger.flushTracebacks} returns the traceback messages. """ exceptions = [ZeroDivisionError(), ZeroDivisionError()] logger = MemoryLogger() logger.write({"x": 1}) for exc in exceptions: self.write_traceback(logger, exc) logger.write({"x": 1}) flushed = logger.flushTracebacks(ZeroDivisionError) self.assertEqual(flushed, logger.messages[1:3]) def test_flushTracebacksUnflushedTestFailure(self): """ Any tracebacks uncleared by L{MemoryLogger.flushTracebacks} (because they are of a different type) are still listed in L{MemoryLogger.tracebackMessages}. """ logger = MemoryLogger() exception = RuntimeError() self.write_traceback(logger, exception) logger.flushTracebacks(KeyError) self.assertEqual(logger.tracebackMessages[0]["reason"], exception) def test_flushTracebacksUnflushedUnreturned(self): """ Any tracebacks uncleared by L{MemoryLogger.flushTracebacks} (because they are of a different type) are not returned. """ logger = MemoryLogger() exception = RuntimeError() self.write_traceback(logger, exception) self.assertEqual(logger.flushTracebacks(KeyError), []) def test_reset(self): """ L{MemoryLogger.reset} clears all logged messages and tracebacks. """ logger = MemoryLogger() logger.write({"key": "value"}, None) logger.reset() self.assertEqual( (logger.messages, logger.serializers, logger.tracebackMessages), ([], [], []), ) def test_threadSafeWrite(self): """ L{MemoryLogger.write} can be called from multiple threads concurrently. """ # Some threads will log some messages thread_count = 10 # A lot of messages. This will keep the threads running long enough # to give them a chance to (try to) interfere with each other. write_count = 10000 # They'll all use the same MemoryLogger instance. logger = MemoryLogger() # Each thread will have its own message and serializer that it writes # to the log over and over again. def write(msg, serializer): for i in range(write_count): logger.write(msg, serializer) # Generate a single distinct message for each thread to log. msgs = list({"i": i} for i in range(thread_count)) # Generate a single distinct serializer for each thread to log. serializers = list(object() for i in range(thread_count)) # Pair them all up. This gives us a simple invariant we can check # later on. write_args = zip(msgs, serializers) # Create the threads. threads = list(Thread(target=write, args=args) for args in write_args) # Run them all. Note threads early in this list will start writing to # the log before later threads in the list even get a chance to start. # That's part of why we have each thread write so many messages. for t in threads: t.start() # Wait for them all to finish. for t in threads: t.join() # Check that we got the correct number of messages in the log. expected_count = thread_count * write_count self.assertEqual(len(logger.messages), expected_count) self.assertEqual(len(logger.serializers), expected_count) # Check the simple invariant we created above. Every logged message # must be paired with the correct serializer, where "correct" is # defined by ``write_args`` above. for position, (msg, serializer) in enumerate( zip(logger.messages, logger.serializers) ): # The indexes must match because the objects are paired using # zip() above. msg_index = msgs.index(msg) serializer_index = serializers.index(serializer) self.assertEqual( msg_index, serializer_index, "Found message #{} with serializer #{} at position {}".format( msg_index, serializer_index, position ), ) class MyException(Exception): """ Custom exception. """ class BadDestination(list): """ A destination that throws an exception the first time it is called. """ called = 0 def __call__(self, msg): if not self.called: self.called = True raise MyException("ono") self.append(msg) class DestinationsTests(TestCase): """ Tests for L{Destinations}. """ def test_send(self): """ L{Destinations.send} calls all destinations added with L{Destinations.add} with the given dictionary. """ destinations = Destinations() message = {"hoorj": "blargh"} dest = [] dest2 = [] dest3 = [] destinations.add(dest.append, dest2.append) destinations.add(dest3.append) destinations.send(message) self.assertEqual(dest, [message]) self.assertEqual(dest2, [message]) self.assertEqual(dest3, [message]) def test_destination_exception_multiple_destinations(self): """ If one destination throws an exception, other destinations still get the message. """ destinations = Destinations() dest = [] dest2 = BadDestination() dest3 = [] destinations.add(dest.append) destinations.add(dest2) destinations.add(dest3.append) message = {"hello": 123} destinations.send(message) self.assertIn(message, dest) self.assertIn(message, dest3) def test_destination_exception_continue(self): """ If a destination throws an exception, future messages are still sent to it. """ destinations = Destinations() dest = BadDestination() destinations.add(dest) msg1 = {"hello": 123} msg2 = {"world": 456} destinations.send(msg1) self.assertNotIn(msg1, dest) destinations.send(msg2) self.assertIn(msg2, dest) def test_remove(self): """ A destination removed with L{Destinations.remove} will no longer receive messages from L{Destionations.add} calls. """ destinations = Destinations() message = {"hello": 123} dest = [] destinations.add(dest.append) destinations.remove(dest.append) destinations.send(message) self.assertEqual(dest, []) def test_removeNonExistent(self): """ Removing a destination that has not previously been added with result in a C{ValueError} being thrown. """ destinations = Destinations() self.assertRaises(ValueError, destinations.remove, [].append) def test_addGlobalFields(self): """ L{Destinations.addGlobalFields} adds the given fields and values to the messages being passed in. """ destinations = Destinations() dest = [] destinations.add(dest.append) destinations.addGlobalFields(x=123, y="hello") destinations.send({"z": 456}) self.assertEqual(dest, [{"x": 123, "y": "hello", "z": 456}]) def test_addGlobalFieldsCumulative(self): """ L{Destinations.addGlobalFields} adds the given fields to those set by previous calls. """ destinations = Destinations() dest = [] destinations.add(dest.append) destinations.addGlobalFields(x=123, y="hello") destinations.addGlobalFields(x=456, z=456) destinations.send({"msg": "X"}) self.assertEqual(dest, [{"x": 456, "y": "hello", "z": 456, "msg": "X"}]) def test_buffering(self): """ Before any destinations are set up to 1000 messages are buffered, and then delivered to the first registered destinations. """ destinations = Destinations() messages = [{"k": i} for i in range(1050)] for m in messages: destinations.send(m) dest, dest2 = [], [] destinations.add(dest.append, dest2.append) self.assertEqual((dest, dest2), (messages[-1000:], messages[-1000:])) def test_buffering_second_batch(self): """ The second batch of added destination don't get the buffered messages. """ destinations = Destinations() message = {"m": 1} message2 = {"m": 2} destinations.send(message) dest = [] dest2 = [] destinations.add(dest.append) destinations.add(dest2.append) destinations.send(message2) self.assertEqual((dest, dest2), ([message, message2], [message2])) def test_global_fields_buffering(self): """ Global fields are added to buffered messages, when possible. """ destinations = Destinations() message = {"m": 1} destinations.send(message) destinations.addGlobalFields(k=123) dest = [] destinations.add(dest.append) self.assertEqual(dest, [{"m": 1, "k": 123}]) def makeLogger(): """ Return a tuple (L{Logger} instance, C{list} of written messages). """ logger = Logger() logger._destinations = Destinations() written = [] logger._destinations.add(written.append) return logger, written class LoggerTests(TestCase): """ Tests for L{Logger}. """ def test_interface(self): """ L{Logger} implements L{ILogger}. """ verifyClass(ILogger, Logger) def test_global(self): """ A global L{Destinations} is used by the L{Logger} class. """ self.assertIsInstance(Logger._destinations, Destinations) def test_write(self): """ L{Logger.write} sends the given dictionary L{Destinations} object. """ logger, written = makeLogger() d = {"hello": 1} logger.write(d) self.assertEqual(written, [d]) def test_serializer(self): """ If a L{_MessageSerializer} is passed to L{Logger.write}, it is used to serialize the message before it is passed to the destination. """ logger, written = makeLogger() serializer = _MessageSerializer( [ Field.forValue("message_type", "mymessage", "The type"), Field("length", len, "The length of a thing"), ] ) logger.write({"message_type": "mymessage", "length": "thething"}, serializer) self.assertEqual(written, [{"message_type": "mymessage", "length": 8}]) def test_passedInDictionaryUnmodified(self): """ The dictionary passed in to L{Logger.write} is not modified. """ logger, written = makeLogger() serializer = _MessageSerializer( [ Field.forValue("message_type", "mymessage", "The type"), Field("length", len, "The length of a thing"), ] ) d = {"message_type": "mymessage", "length": "thething"} original = d.copy() logger.write(d, serializer) self.assertEqual(d, original) def test_safe_unicode_dictionary(self): """ L{_safe_unicode_dictionary} converts the given dictionary's values and keys to unicode using C{safeunicode}. """ class badobject(object): def __repr__(self): raise TypeError() dictionary = {badobject(): 123, 123: badobject()} badMessage = "eliot: unknown, str() raised exception" self.assertEqual( eval(_safe_unicode_dictionary(dictionary)), {badMessage: "123", "123": badMessage}, ) def test_safe_unicode_dictionary_fallback(self): """ If converting the dictionary failed for some reason, L{_safe_unicode_dictionary} runs C{repr} on the object. """ self.assertEqual(_safe_unicode_dictionary(None), "None") def test_safe_unicode_dictionary_fallback_failure(self): """ If all else fails, L{_safe_unicode_dictionary} just gives up. """ class badobject(object): def __repr__(self): raise TypeError() self.assertEqual( _safe_unicode_dictionary(badobject()), "eliot: unknown, str() raised exception", ) def test_serializationErrorTraceback(self): """ If serialization fails in L{Logger.write}, a traceback is logged, along with a C{eliot:serialization_failure} message for debugging purposes. """ logger, written = makeLogger() def raiser(i): raise RuntimeError("oops") serializer = _MessageSerializer( [ Field.forValue("message_type", "mymessage", "The type"), Field("fail", raiser, "Serialization fail"), ] ) message = {"message_type": "mymessage", "fail": "will"} logger.write(message, serializer) self.assertEqual(len(written), 2) tracebackMessage = written[0] assertContainsFields( self, tracebackMessage, { "exception": "%s.RuntimeError" % (RuntimeError.__module__,), "message_type": "eliot:traceback", }, ) self.assertIn("RuntimeError: oops", tracebackMessage["traceback"]) # Calling _safe_unicode_dictionary multiple times leads to # inconsistent results due to hash ordering, so compare contents: assertContainsFields( self, written[1], {"message_type": "eliot:serialization_failure"} ) self.assertEqual( eval(written[1]["message"]), dict((repr(key), repr(value)) for (key, value) in message.items()), ) def test_destination_exception_caught(self): """ If a destination throws an exception, an appropriate error is logged. """ logger = Logger() logger._destinations = Destinations() dest = BadDestination() logger._destinations.add(dest) message = {"hello": 123} logger.write({"hello": 123}) assertContainsFields( self, dest[0], { "message_type": "eliot:destination_failure", "message": _safe_unicode_dictionary(message), "reason": "ono", "exception": "eliot.tests.test_output.MyException", }, ) def test_destination_multiple_exceptions_caught(self): """ If multiple destinations throw an exception, an appropriate error is logged for each. """ logger = Logger() logger._destinations = Destinations() logger._destinations.add(BadDestination()) logger._destinations.add(lambda msg: 1 / 0) messages = [] logger._destinations.add(messages.append) try: 1 / 0 except ZeroDivisionError as e: zero_divide = str(e) zero_type = ZeroDivisionError.__module__ + ".ZeroDivisionError" message = {"hello": 123} logger.write({"hello": 123}) def remove(key): return [message.pop(key) for message in messages[1:]] # Make sure we have task_level & task_uuid in exception messages. task_levels = remove("task_level") task_uuids = remove("task_uuid") timestamps = remove("timestamp") self.assertEqual( ( abs(timestamps[0] + timestamps[1] - 2 * time()) < 1, task_levels == [[1], [1]], len([UUID(uuid) for uuid in task_uuids]) == 2, messages, ), ( True, True, True, [ message, { "message_type": "eliot:destination_failure", "message": _safe_unicode_dictionary(message), "reason": "ono", "exception": "eliot.tests.test_output.MyException", }, { "message_type": "eliot:destination_failure", "message": _safe_unicode_dictionary(message), "reason": zero_divide, "exception": zero_type, }, ], ), ) def test_destination_exception_caught_twice(self): """ If a destination throws an exception, and the logged error about it also causes an exception, then just drop that exception on the floor, since there's nothing we can do with it. """ logger = Logger() logger._destinations = Destinations() def always_raise(message): raise ZeroDivisionError() logger._destinations.add(always_raise) # Just a message. No exception raised; since everything is dropped no # other assertions to be made. logger.write({"hello": 123}) # With an action. No exception raised; since everything is dropped no # other assertions to be made. with start_action(logger, "sys:do"): logger.write({"hello": 123}) class PEP8Tests(TestCase): """ Tests for PEP 8 method compatibility. """ def test_flush_tracebacks(self): """ L{MemoryLogger.flush_tracebacks} is the same as L{MemoryLogger.flushTracebacks} """ self.assertEqual(MemoryLogger.flush_tracebacks, MemoryLogger.flushTracebacks) class ToFileTests(TestCase): """ Tests for L{to_file}. """ def test_to_file_adds_destination(self): """ L{to_file} adds a L{FileDestination} destination with the given file. """ f = BytesIO() to_file(f) expected = FileDestination(file=f) self.addCleanup(Logger._destinations.remove, expected) self.assertIn(expected, Logger._destinations._destinations) def test_to_file_custom_encoder(self): """ L{to_file} accepts a custom encoder, and sets it on the resulting L{FileDestination}. """ from json import JSONEncoder f = stdout class MyEncoder(JSONEncoder): def default(self, o): return 17 to_file(f, encoder=MyEncoder) added = Logger._destinations._destinations[-1] self.addCleanup(Logger._destinations.remove, added) self.assertEqual(added._json_default(object()), 17) def test_to_file_custom_json_default(self): """ L{to_file} accepts a custom JSON default object, and sets it on the resulting L{FileDestination}. """ f = BytesIO() def default(o): return 23 to_file(f, json_default=default) added = Logger._destinations._destinations[-1] self.addCleanup(Logger._destinations.remove, added) self.assertEqual(added._json_default(object()), 23) @skipUnless(np, "NumPy is not installed.") def test_default_encoder_supports_numpy(self): """The default encoder can encode NumPy objects.""" message = {"x": np.int64(3)} f = StringIO() destination = FileDestination(file=f) destination(message) self.assertEqual( [pyjson.loads(line) for line in f.getvalue().splitlines()], [{"x": 3}] ) def test_filedestination_writes_json_bytes(self): """ L{FileDestination} writes JSON-encoded messages to a file that accepts bytes. """ message1 = {"x": 123} message2 = {"y": None, "x": "abc"} bytes_f = BytesIO() destination = FileDestination(file=bytes_f) destination(message1) destination(message2) self.assertEqual( [pyjson.loads(line) for line in bytes_f.getvalue().splitlines()], [message1, message2], ) def test_filedestination_custom_encoder(self): """ L{FileDestionation} can use a custom encoder. """ custom = object() class CustomEncoder(pyjson.JSONEncoder): def default(self, o): if o is custom: return "CUSTOM!" else: return pyjson.JSONEncoder.default(self, o) message = {"x": 123, "z": custom} f = BytesIO() destination = FileDestination(file=f, encoder=CustomEncoder) destination(message) self.assertEqual( pyjson.loads(f.getvalue().splitlines()[0]), {"x": 123, "z": "CUSTOM!"} ) def test_filedestination_custom_json_default(self): """ L{FileDestination} accepts a custom JSON default callable. """ custom = object() def default(o): if o is custom: return "CUSTOM!" else: raise TypeError message = {"x": 123, "z": custom} f = BytesIO() destination = FileDestination(file=f, json_default=default) destination(message) self.assertEqual( pyjson.loads(f.getvalue().splitlines()[0]), {"x": 123, "z": "CUSTOM!"} ) def test_filedestination_flushes(self): """ L{FileDestination} flushes after every write, to ensure logs get written out even if the local buffer hasn't filled up. """ path = mktemp() # File with large buffer: f = open(path, "wb", 1024 * 1024 * 10) # and a small message that won't fill the buffer: message1 = {"x": 123} destination = FileDestination(file=f) destination(message1) # Message got written even though buffer wasn't filled: self.assertEqual( [ pyjson.loads(line.decode("utf-8")) for line in open(path, "rb").read().splitlines() ], [message1], ) def test_filedestination_writes_json_unicode(self): """ L{FileDestination} writes JSON-encoded messages to file that only accepts Unicode. """ message = {"x": "\u1234"} unicode_f = StringIO() destination = FileDestination(file=unicode_f) destination(message) self.assertEqual(pyjson.loads(unicode_f.getvalue()), message) def test_filedestination_unwriteable_file(self): """ L{FileDestination} raises a runtime error if the given file isn't writeable. """ path = mktemp() open(path, "w").close() f = open(path, "r") with self.assertRaises(RuntimeError): FileDestination(f)