import sys

from eventlet import debug
import six
import tests
import eventlet


class TestSpew(tests.LimitedTestCase):
    def setUp(self):
        self.orig_trace = sys.settrace
        sys.settrace = self._settrace
        self.tracer = None

    def tearDown(self):
        sys.settrace = self.orig_trace
        sys.stdout = sys.__stdout__

    def _settrace(self, cb):
        self.tracer = cb

    def test_spew(self):
        debug.spew()
        assert isinstance(self.tracer, debug.Spew)

    def test_unspew(self):
        debug.spew()
        debug.unspew()
        assert self.tracer is None

    def test_line(self):
        if sys.version_info >= (3, 7):
            frame_str = "f=<frame at"
        else:
            frame_str = "f=<frame object at"

        sys.stdout = six.StringIO()
        s = debug.Spew()
        f = sys._getframe()
        s(f, "line", None)
        lineno = f.f_lineno - 1  # -1 here since we called with frame f in the line above
        output = sys.stdout.getvalue()
        assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
        assert frame_str in output

    def test_line_nofile(self):
        sys.stdout = six.StringIO()
        s = debug.Spew()
        g = globals().copy()
        del g['__file__']
        f = eval("sys._getframe()", g)
        lineno = f.f_lineno
        s(f, "line", None)
        output = sys.stdout.getvalue()
        assert "[unknown]:%i" % lineno in output, "Didn't find [unknown]:%i in %s" % (lineno, output)
        assert "VM instruction #" in output, output

    def test_line_global(self):
        if sys.version_info >= (3, 7):
            frame_str = "f=<frame at"
        else:
            frame_str = "f=<frame object at"

        global GLOBAL_VAR
        sys.stdout = six.StringIO()
        GLOBAL_VAR = debug.Spew()
        f = sys._getframe()
        GLOBAL_VAR(f, "line", None)
        lineno = f.f_lineno - 1  # -1 here since we called with frame f in the line above
        output = sys.stdout.getvalue()
        assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
        assert frame_str in output
        assert "GLOBAL_VAR" in f.f_globals
        assert "GLOBAL_VAR=<eventlet.debug.Spew object at" in output
        del GLOBAL_VAR

    def test_line_novalue(self):
        sys.stdout = six.StringIO()
        s = debug.Spew(show_values=False)
        f = sys._getframe()
        s(f, "line", None)
        lineno = f.f_lineno - 1  # -1 here since we called with frame f in the line above
        output = sys.stdout.getvalue()
        assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
        assert "f=<frame object at" not in output

    def test_line_nooutput(self):
        sys.stdout = six.StringIO()
        s = debug.Spew(trace_names=['foo'])
        f = sys._getframe()
        s(f, "line", None)
        output = sys.stdout.getvalue()
        assert output == ""


class TestDebug(tests.LimitedTestCase):
    def test_everything(self):
        debug.hub_exceptions(True)
        debug.hub_exceptions(False)
        debug.tpool_exceptions(True)
        debug.tpool_exceptions(False)
        debug.hub_listener_stacks(True)
        debug.hub_listener_stacks(False)
        debug.hub_timer_stacks(True)
        debug.hub_timer_stacks(False)
        debug.format_hub_listeners()
        debug.format_hub_timers()

    def test_hub_exceptions(self):
        debug.hub_exceptions(True)
        server = eventlet.listen(('0.0.0.0', 0))
        client = eventlet.connect(('127.0.0.1', server.getsockname()[1]))
        client_2, addr = server.accept()

        def hurl(s):
            s.recv(1)
            {}[1]  # keyerror

        with tests.capture_stderr() as fake:
            gt = eventlet.spawn(hurl, client_2)
            eventlet.sleep(0)
            client.send(b' ')
            eventlet.sleep(0)
            # allow the "hurl" greenlet to trigger the KeyError
            # not sure why the extra context switch is needed
            eventlet.sleep(0)
        self.assertRaises(KeyError, gt.wait)
        debug.hub_exceptions(False)
        # look for the KeyError exception in the traceback
        assert 'KeyError: 1' in fake.getvalue(), "Traceback not in:\n" + fake.getvalue()