old_contents: """Tests for the logging bridge."""
  
  from __future__ import absolute_import
  
  from unittest import TestCase
  import threading
  
  from twisted.python import threadable
  
  from .._eventloop import ThreadLogObserver
  
  
  class ThreadLogObserverTest(TestCase):
      """
      Tests for ThreadLogObserver.
      """
      def test_stop(self):
          """
          ThreadLogObserver.stop() stops the thread started in __init__.
          """
          threadLog = ThreadLogObserver(None)
          self.assertTrue(threadLog._thread.is_alive())
          threadLog.stop()
          threadLog._thread.join()
          self.assertFalse(threadLog._thread.is_alive())
  
      def test_emit(self):
          """
          ThreadLogObserver.emit runs the wrapped observer's in its thread, with
          the given message.
          """
          messages = []
          def observer(msg):
              messages.append((threading.current_thread().ident, msg))
  
          threadLog = ThreadLogObserver(observer)
          ident = threadLog._thread.ident
          msg1 = {}
          msg2 = {"a": "b"}
          threadLog(msg1)
          threadLog(msg2)
          threadLog.stop()
          # Wait for writing to finish:
          threadLog._thread.join()
          self.assertEqual(messages, [(ident, msg1), (ident, msg2)])
  
  
      def test_ioThreadUnchanged(self):
          """
          ThreadLogObserver does not change the Twisted I/O thread (which is
          supposed to match the thread the main reactor is running in.)
          """
          threadLog = ThreadLogObserver(None)
          threadLog.stop()
          threadLog._thread.join()
          self.assertIn(threadable.ioThread,
                        # Either reactor was never run, or run in thread running
                        # the tests:
                        (None, threading.current_thread().ident))
  
new_contents: """Tests for the logging bridge."""
  
  from __future__ import absolute_import
  
  from twisted.trial.unittest import SynchronousTestCase
  import threading
  
  from twisted.python import threadable
  
  from .._eventloop import ThreadLogObserver
  
  
  class ThreadLogObserverTest(SynchronousTestCase):
      """
      Tests for ThreadLogObserver.
  
      We use Twisted's SyncTestCase to ensure that unhandled logged errors get
      reported as errors, in particular for test_error.
      """
      def test_stop(self):
          """
          ThreadLogObserver.stop() stops the thread started in __init__.
          """
          threadLog = ThreadLogObserver(None)
          self.assertTrue(threadLog._thread.is_alive())
          threadLog.stop()
          threadLog._thread.join()
          self.assertFalse(threadLog._thread.is_alive())
  
      def test_emit(self):
          """
          ThreadLogObserver.emit runs the wrapped observer's in its thread, with
          the given message.
          """
          messages = []
          def observer(msg):
              messages.append((threading.current_thread().ident, msg))
  
          threadLog = ThreadLogObserver(observer)
          ident = threadLog._thread.ident
          msg1 = {}
          msg2 = {"a": "b"}
          threadLog(msg1)
          threadLog(msg2)
          threadLog.stop()
          # Wait for writing to finish:
          threadLog._thread.join()
          self.assertEqual(messages, [(ident, msg1), (ident, msg2)])
  
      def test_errors(self):
          """
          ThreadLogObserver.emit catches and silently drops exceptions from its
          observer.
          """
          messages = []
          counter = []
          def observer(msg):
              counter.append(1)
              if len(counter) == 2:
                  raise RuntimeError("ono a bug")
              messages.append(msg)
  
          threadLog = ThreadLogObserver(observer)
          msg1 = {"m": "1"}
          msg2 = {"m": "2"}
          msg3 = {"m": "3"}
          threadLog(msg1)
          threadLog(msg2)
          threadLog(msg3)
          threadLog.stop()
          # Wait for writing to finish:
          threadLog._thread.join()
          self.assertEqual(messages, [msg1, msg3])
  
      def test_ioThreadUnchanged(self):
          """
          ThreadLogObserver does not change the Twisted I/O thread (which is
          supposed to match the thread the main reactor is running in.)
          """
          threadLog = ThreadLogObserver(None)
          threadLog.stop()
          threadLog._thread.join()
          self.assertIn(threadable.ioThread,
                        # Either reactor was never run, or run in thread running
                        # the tests:
                        (None, threading.current_thread().ident))
  
subject: Test for unhandled errors coming out of destination logging infrastructure.