eventlet/tests/debug_test.py

   1 import sys
   2 from unittest import TestCase
   3
   4 from eventlet import debug
   5 from eventlet.support import capture_stderr, six
   6 from tests import LimitedTestCase, main
   7 import eventlet
   8
   9
  10 class TestSpew(TestCase):
  11     def setUp(self):
  12         self.orig_trace = sys.settrace
  13         sys.settrace = self._settrace
  14         self.tracer = None
  15
  16     def tearDown(self):
  17         sys.settrace = self.orig_trace
  18         sys.stdout = sys.__stdout__
  19
  20     def _settrace(self, cb):
  21         self.tracer = cb
  22
  23     def test_spew(self):
  24         debug.spew()
  25         assert isinstance(self.tracer, debug.Spew)
  26
  27     def test_unspew(self):
  28         debug.spew()
  29         debug.unspew()
  30         assert self.tracer is None
  31
  32     def test_line(self):
  33         sys.stdout = six.StringIO()
  34         s = debug.Spew()
  35         f = sys._getframe()
  36         s(f, "line", None)
  37         lineno = f.f_lineno - 1  # -1 here since we called with frame f in the line above
  38         output = sys.stdout.getvalue()
  39         assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
  40         assert "f=<frame object at" in output
  41
  42     def test_line_nofile(self):
  43         sys.stdout = six.StringIO()
  44         s = debug.Spew()
  45         g = globals().copy()
  46         del g['__file__']
  47         f = eval("sys._getframe()", g)
  48         lineno = f.f_lineno
  49         s(f, "line", None)
  50         output = sys.stdout.getvalue()
  51         assert "[unknown]:%i" % lineno in output, "Didn't find [unknown]:%i in %s" % (lineno, output)
  52         assert "VM instruction #" in output, output
  53
  54     def test_line_global(self):
  55         global GLOBAL_VAR
  56         sys.stdout = six.StringIO()
  57         GLOBAL_VAR = debug.Spew()
  58         f = sys._getframe()
  59         GLOBAL_VAR(f, "line", None)
  60         lineno = f.f_lineno - 1  # -1 here since we called with frame f in the line above
  61         output = sys.stdout.getvalue()
  62         assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
  63         assert "f=<frame object at" in output
  64         assert "GLOBAL_VAR" in f.f_globals
  65         assert "GLOBAL_VAR=<eventlet.debug.Spew object at" in output
  66         del GLOBAL_VAR
  67
  68     def test_line_novalue(self):
  69         sys.stdout = six.StringIO()
  70         s = debug.Spew(show_values=False)
  71         f = sys._getframe()
  72         s(f, "line", None)
  73         lineno = f.f_lineno - 1  # -1 here since we called with frame f in the line above
  74         output = sys.stdout.getvalue()
  75         assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
  76         assert "f=<frame object at" not in output
  77
  78     def test_line_nooutput(self):
  79         sys.stdout = six.StringIO()
  80         s = debug.Spew(trace_names=['foo'])
  81         f = sys._getframe()
  82         s(f, "line", None)
  83         output = sys.stdout.getvalue()
  84         assert output == ""
  85
  86
  87 class TestDebug(LimitedTestCase):
  88     def test_everything(self):
  89         debug.hub_exceptions(True)
  90         debug.hub_exceptions(False)
  91         debug.tpool_exceptions(True)
  92         debug.tpool_exceptions(False)
  93         debug.hub_listener_stacks(True)
  94         debug.hub_listener_stacks(False)
  95         debug.hub_timer_stacks(True)
  96         debug.hub_timer_stacks(False)
  97         debug.format_hub_listeners()
  98         debug.format_hub_timers()
  99
 100     def test_hub_exceptions(self):
 101         debug.hub_exceptions(True)
 102         server = eventlet.listen(('0.0.0.0', 0))
 103         client = eventlet.connect(('127.0.0.1', server.getsockname()[1]))
 104         client_2, addr = server.accept()
 105
 106         def hurl(s):
 107             s.recv(1)
 108             {}[1]  # keyerror
 109
 110         with capture_stderr() as fake:
 111             gt = eventlet.spawn(hurl, client_2)
 112             eventlet.sleep(0)
 113             client.send(b' ')
 114             eventlet.sleep(0)
 115             # allow the "hurl" greenlet to trigger the KeyError
 116             # not sure why the extra context switch is needed
 117             eventlet.sleep(0)
 118         self.assertRaises(KeyError, gt.wait)
 119         debug.hub_exceptions(False)
 120         # look for the KeyError exception in the traceback
 121         assert 'KeyError: 1' in fake.getvalue(), "Traceback not in:\n" + fake.getvalue()
 122
 123 if __name__ == "__main__":
 124     main()