def stop(self):
     """Stop this Tracer."""
     if hasattr(sys, "gettrace") and self.warn:
         if sys.gettrace() != self._trace:
             msg = "Trace function changed, measurement is likely wrong: %r"
             self.warn(msg % sys.gettrace())
     sys.settrace(None)

        def wrapper(*args, **kwargs):
            result = []

            def test_func_thread(test_func, *args, **kwargs):
                try:
                    test_func(*args, **kwargs)
                except Exception as e:
                    result.append(e)
                finally:
                    result.append("done")

            test_timer = threading.Timer(interval, result.append, args=("interrupted",))
            test_thread = threading.Thread(target=test_func_thread,
                                           args=(test_func,) + args,
                                           kwargs=kwargs)
            test_thread.setDaemon(not sys.gettrace())  # in debug mode make thread blocking main
            test_thread.start()
            if not sys.gettrace():  # disable this in debug mode
                test_timer.start()
            while True:
                if test_thread.is_alive() and test_timer.is_alive():
                    time.sleep(0.1)
                else:
                    break

            if "interrupted" in result:
                raise Exception("%s did not finish in %s seconds" % (test_func.__name__,
                                                                     interval))
            else:
                test_timer.cancel()
                for test_result in result:
                    if test_result not in ["done", "interrupted"]:
                        raise test_result

def test_trace_merge():
    with hunter.trace(function="a"):
        with hunter.trace(function="b"):
            with hunter.trace(function="c"):
                assert sys.gettrace().handler == When(Q(function="c"), CodePrinter)
            assert sys.gettrace().handler == When(Q(function="b"), CodePrinter)
        assert sys.gettrace().handler == When(Q(function="a"), CodePrinter)

    def test_checking(self):

        def fake_trace(*args):
            return
        old_trace = sys.gettrace()
        sys.settrace(fake_trace)

        with self.assertRaises(AssertionError):
            # We know f1's behavior but we run f2, raise attention on it
            with checking(name='test_check', trace_all=True):
                f2('Traced code')

        # It has created a summary file
        self.assertTrue(os.path.exists('test_check-last-change.txt'))
        with open('test_check-last-change.txt') as f:
            change = f.read()
        change_lines = change.splitlines()
        self.assertEqual(len(change_lines), 1)
        self.assertIn('test_tools:f', change)
        self.assertEqual(change[0], '!')

        # It has put our trace function back
        after_trace = sys.gettrace()
        sys.settrace(old_trace)
        self.assertEqual(after_trace, fake_trace)

def test_call_trace():
    from syn.base_utils import call_trace, reset_trace, capture, CallTrace

    def foo():
        bar()

    def bar():
        pass

    foo()
    tr = sys.gettrace()
    with capture() as (out, err):
        with reset_trace():
            call_trace()
            foo()
    assert sys.gettrace() is tr

    assert out.getvalue() == 'foo\n bar\n__exit__\n reset_trace\n'
    assert err.getvalue() == ''

    t = CallTrace()
    with capture() as (out, err):
        t(sys._getframe(), 'call', None)
    
    assert out.getvalue() == 'test_call_trace\n'
    assert err.getvalue() == ''
    assert t.indent == 1

    t(sys._getframe(), 'return', None)
    assert t.indent == 0

 def test_start_and_stop(self):
     assert None is sys.gettrace()
     self.tm.start()
     self.assertIn(self.tm, tracerlib._active_managers)
     self.assertEqual(tracerlib._global_tracer, sys.gettrace())
     self.tm.stop()
     assert None is sys.gettrace()

def check_with_no_trace():
    if False:
        print('break on check_with_trace')
    frame = sys._getframe()
    if frame.f_trace is not None:
        raise AssertionError('Expected %s to be None' % (frame.f_trace,))

    if sys.gettrace() is not pydevd_frame_tracing.dummy_tracing_holder.dummy_trace_func:
        raise AssertionError('Expected %s to be dummy_trace_func' % (sys.gettrace(),))

 def stop(self):
     self.stopped = True
     if self.thread != threading.currentThread():
         return
     if hasattr(sys, 'gettrace') and self.warn:
         if sys.gettrace() != self._trace:
             msg = 'Trace function changed, measurement is likely wrong: %r'
             self.warn(msg % (sys.gettrace(),))
     sys.settrace(None)

 def stop(self):
     """
     Stop this Tracer.
     """
     if hasattr(sys, "gettrace") and self.log:
         if sys.gettrace() != self._trace:
             msg = "Trace function changed, measurement is likely wrong: %r"
             print >> sys.stdout, msg % sys.gettrace()
     sys.settrace(None)

 def stop(self):
     """Stop this Tracer."""
     if hasattr(sys, "gettrace") and self.warn:
         if sys.gettrace() != self._trace:
             msg = "Trace function changed, measurement is likely wrong: %r"
             self.warn(msg % (sys.gettrace(),))
             #--debug
             #from coverage.misc import short_stack
             #self.warn(msg % (sys.gettrace()))#, short_stack()))
     sys.settrace(None)

 def __call__(self, enable=True):
     if enable:
         try:
             self.thread_local.trace_backup.append(sys.gettrace())
         except AttributeError:
             self.thread_local.trace_backup = []
             self.thread_local.trace_backup.append(sys.gettrace())
         sys.settrace(self._traceit)
     else:
         sys.settrace(self.thread_local.trace_backup.pop())

    def test_gettrace(self):
        '''TODO: revisit'''
        self.assertEqual(sys.gettrace(), None)

        def temp_func(*args, **kwargs):
            pass

        sys.settrace(temp_func)
        self.assertEqual(sys.gettrace(), temp_func)
        sys.settrace(None)
        self.assertEqual(sys.gettrace(), None)

def test_gettrace():
    '''TODO: revisit'''
    AreEqual(sys.gettrace(), None)
    
    def temp_func(*args, **kwargs):
        pass
        
    sys.settrace(temp_func)
    AreEqual(sys.gettrace(), temp_func)
    sys.settrace(None)
    AreEqual(sys.gettrace(), None)

 def run_thread():           # pragma: nested
     """Check that coverage is stopping properly in threads."""
     deadline = time.time() + 5
     ident = threading.currentThread().ident
     if sys.gettrace() is not None:
         has_started_coverage.append(ident)
     while sys.gettrace() is not None:
         # Wait for coverage to stop
         time.sleep(0.01)
         if time.time() > deadline:
             return
     has_stopped_coverage.append(ident)

    def test_error_when_set_multiple(self):
        self.monitor._profile.replace(self.bar)
        self.assertIs(sys.gettrace(), self.bar)
        self.monitor._profile.replace(self.bar)
        self.assertIs(sys.gettrace(), self.bar)
        self.monitor._profile.recover()

        self.monitor._profile.replace(self.bar)
        self.assertIs(sys.gettrace(), self.bar)
        with self.assertRaises(RuntimeError):
            self.monitor._profile.replace(None)
            self.assertIs(sys.gettrace(), self.bar)
        self.monitor._profile.recover()

 def test_multiple_managers(self):
     tm = self.tm
     tm2 = tracerlib.TracerManager()
     with tm:
         with tm2:
             self.assertEqual(2, len(tracerlib._active_managers))
             self.assertIn(tm, tracerlib._active_managers)
             self.assertIn(tm2, tracerlib._active_managers)
             self.assertEqual(tracerlib._global_tracer, sys.gettrace())
         self.assertEqual(1, len(tracerlib._active_managers))
         self.assertEqual(tracerlib._global_tracer, sys.gettrace())
     self.assertEqual(0, len(tracerlib._active_managers))
     self.assertEqual(None, sys.gettrace())

def _begin():
    """Start tracking program state.

    If begin() has previously been called, any labels that occur during this
    execution will also be made visible to previous begin calls.
    """
    sys.settrace(None)
    assert sys.gettrace() is None
    global prev_state
    prev_state = 0
    push_array()
    sys.settrace(tracer)
    assert sys.gettrace() is tracer

    def prepareTestCase(self, test):
        # TODO Test this for module-level tests
        log.info('Tracing calls from %s', test)

        # Let's assume that this Python interpreter *does* implement
        # sys.gettrace/settrace
        try:
            sys.gettrace()
        except RuntimeError:
            raise SystemError('NoseDive requires sys.settrace')

        tracer = TestTracer(inspect.getmodule(test.test).__file__)
        sys.settrace(tracer)

 def wrap(f, self, request, *args, **kwargs):
     #from api.handlers import log
     c_type = request.content_type
     try:
         #data = data_loaders[c_type](request.raw_post_data)[field_name]
         data = decode_from_string(c_type, request.raw_post_data)[field_name]
     except KeyError as e:
         if find_in_root:
             data = decode_from_string(c_type, request.raw_post_data)
             #data = data_loaders[c_type](request.raw_post_data)
         else:
             raise e, sys.gettrace()
     setattr(request, operation, data)
     return f(self, request, *args, **kwargs)

def test_inject_cleans_up_tracefn_on_except():
    import sys
    old_trace = sys.gettrace()
    @inject(a=5)
    def foo():
        raise Exception()

    try:
        foo()
        assert False, "should have raised an exception"
    except:
        pass
    assert sys.gettrace() == old_trace
    sys.settrace(old_trace)