test_functionality.py

import os
import sys
import time
import threading
import unittest
import yappi
import _yappi
import utils
import multiprocessing
import subprocess

_counter = 0


class BasicUsage(utils.YappiUnitTestCase):

    def test_callback_function_int_return_overflow(self):
        # this test is just here to check if any errors are generated, as the err
        # is printed in C side, I did not include it here. THere are ways to test
        # this deterministically, I did not bother
        import ctypes

        def _unsigned_overflow_margin():
            return 2**(ctypes.sizeof(ctypes.c_void_p) * 8) - 1

        def foo():
            pass

        #with utils.captured_output() as (out, err):
        yappi.set_context_id_callback(_unsigned_overflow_margin)
        yappi.set_tag_callback(_unsigned_overflow_margin)
        yappi.start()
        foo()

    def test_issue60(self):

        def foo():
            buf = bytearray()
            buf += b't' * 200
            view = memoryview(buf)[10:]
            view = view.tobytes()
            del buf[:10]  # this throws exception
            return view

        yappi.start(builtins=True)
        foo()
        self.assertTrue(
            len(
                yappi.get_func_stats(
                    filter_callback=lambda x: yappi.
                    func_matches(x, [memoryview.tobytes])
                )
            ) > 0
        )
        yappi.stop()

    def test_issue54(self):

        def _tag_cbk():
            global _counter
            _counter += 1
            return _counter

        def a():
            pass

        def b():
            pass

        yappi.set_tag_callback(_tag_cbk)
        yappi.start()
        a()
        a()
        a()
        yappi.stop()
        stats = yappi.get_func_stats()
        self.assertEqual(stats.pop().ncall, 3)  # aggregated if no tag is given
        stats = yappi.get_func_stats(tag=1)

        for i in range(1, 3):
            stats = yappi.get_func_stats(tag=i)
            stats = yappi.get_func_stats(
                tag=i, filter_callback=lambda x: yappi.func_matches(x, [a])
            )

            stat = stats.pop()
            self.assertEqual(stat.ncall, 1)

        yappi.set_tag_callback(None)
        yappi.clear_stats()
        yappi.start()
        b()
        b()
        stats = yappi.get_func_stats()
        self.assertEqual(len(stats), 1)
        stat = stats.pop()
        self.assertEqual(stat.ncall, 2)

    def test_filter(self):

        def a():
            pass

        def b():
            a()

        def c():
            b()

        _TCOUNT = 5

        ts = []
        yappi.start()
        for i in range(_TCOUNT):
            t = threading.Thread(target=c)
            t.start()
            ts.append(t)

        for t in ts:
            t.join()

        yappi.stop()

        ctx_ids = []
        for tstat in yappi.get_thread_stats():
            if tstat.name == '_MainThread':
                main_ctx_id = tstat.id
            else:
                ctx_ids.append(tstat.id)

        fstats = yappi.get_func_stats(filter={"ctx_id": 9})
        self.assertTrue(fstats.empty())
        fstats = yappi.get_func_stats(
            filter={
                "ctx_id": main_ctx_id,
                "name": "c"
            }
        )  # main thread
        self.assertTrue(fstats.empty())

        for i in ctx_ids:
            fstats = yappi.get_func_stats(
                filter={
                    "ctx_id": i,
                    "name": "a",
                    "ncall": 1
                }
            )
            self.assertEqual(fstats.pop().ncall, 1)
            fstats = yappi.get_func_stats(filter={"ctx_id": i, "name": "b"})
            self.assertEqual(fstats.pop().ncall, 1)
            fstats = yappi.get_func_stats(filter={"ctx_id": i, "name": "c"})
            self.assertEqual(fstats.pop().ncall, 1)

        yappi.clear_stats()
        yappi.start(builtins=True)
        time.sleep(0.1)
        yappi.stop()
        fstats = yappi.get_func_stats(filter={"module": "time"})
        self.assertEqual(len(fstats), 1)

        # invalid filters`
        self.assertRaises(
            Exception, yappi.get_func_stats, filter={'tag': "sss"}
        )
        self.assertRaises(
            Exception, yappi.get_func_stats, filter={'ctx_id': "None"}
        )

    def test_filter_callback(self):

        def a():
            time.sleep(0.1)

        def b():
            a()

        def c():
            pass

        def d():
            pass

        yappi.set_clock_type("wall")
        yappi.start(builtins=True)
        a()
        b()
        c()
        d()
        stats = yappi.get_func_stats(
            filter_callback=lambda x: yappi.func_matches(x, [a, b])
        )
        #stats.print_all()
        r1 = '''
        tests/test_functionality.py:98 a      2      0.000000  0.200350  0.100175
        tests/test_functionality.py:101 b     1      0.000000  0.120000  0.100197
        '''
        self.assert_traces_almost_equal(r1, stats)
        self.assertEqual(len(stats), 2)
        stats = yappi.get_func_stats(
            filter_callback=lambda x: yappi.
            module_matches(x, [sys.modules[__name__]])
        )
        r1 = '''
        tests/test_functionality.py:98 a      2      0.000000  0.230130  0.115065
        tests/test_functionality.py:101 b     1      0.000000  0.120000  0.109011
        tests/test_functionality.py:104 c     1      0.000000  0.000002  0.000002
        tests/test_functionality.py:107 d     1      0.000000  0.000001  0.000001
        '''
        self.assert_traces_almost_equal(r1, stats)
        self.assertEqual(len(stats), 4)

        stats = yappi.get_func_stats(
            filter_callback=lambda x: yappi.func_matches(x, [time.sleep])
        )
        self.assertEqual(len(stats), 1)
        r1 = '''
        time.sleep                            2      0.206804  0.220000  0.103402
        '''
        self.assert_traces_almost_equal(r1, stats)

    def test_print_formatting(self):

        def a():
            pass

        def b():
            a()

        func_cols = {
            1: ("name", 48),
            0: ("ncall", 5),
            2: ("tsub", 8),
        }
        thread_cols = {
            1: ("name", 48),
            0: ("ttot", 8),
        }

        yappi.start()
        a()
        b()
        yappi.stop()
        fs = yappi.get_func_stats()
        cs = fs[1].children
        ts = yappi.get_thread_stats()
        #fs.print_all(out=sys.stderr, columns={1:("name", 70), })
        #cs.print_all(out=sys.stderr, columns=func_cols)
        #ts.print_all(out=sys.stderr, columns=thread_cols)
        #cs.print_all(out=sys.stderr, columns={})

        self.assertRaises(
            yappi.YappiError, fs.print_all, columns={1: ("namee", 9)}
        )
        self.assertRaises(
            yappi.YappiError, cs.print_all, columns={1: ("dd", 0)}
        )
        self.assertRaises(
            yappi.YappiError, ts.print_all, columns={1: ("tidd", 0)}
        )

    def test_get_clock(self):
        yappi.set_clock_type('cpu')
        self.assertEqual('cpu', yappi.get_clock_type())
        clock_info = yappi.get_clock_info()
        self.assertTrue('api' in clock_info)
        self.assertTrue('resolution' in clock_info)

        yappi.set_clock_type('wall')
        self.assertEqual('wall', yappi.get_clock_type())

        t0 = yappi.get_clock_time()
        time.sleep(0.1)
        duration = yappi.get_clock_time() - t0
        self.assertTrue(0.05 < duration < 0.3)

    def test_profile_decorator(self):

        def aggregate(func, stats):
            fname = f"tests/{func.__name__}.profile"
            try:
                stats.add(fname)
            except OSError:
                pass
            stats.save(fname)
            raise Exception("messing around")

        @yappi.profile(return_callback=aggregate)
        def a(x, y):
            if x + y == 25:
                raise Exception("")
            return x + y

        def b():
            pass

        try:
            os.remove(
                "tests/a.profile"
            )  # remove the one from prev test, if available
        except:
            pass

        # global profile is on to mess things up
        yappi.start()
        b()

        # assert functionality and call function at same time
        try:
            self.assertEqual(a(1, 2), 3)
        except:
            pass
        try:
            self.assertEqual(a(2, 5), 7)
        except:
            pass
        try:
            a(4, 21)
        except:
            pass
        stats = yappi.get_func_stats().add("tests/a.profile")
        fsa = utils.find_stat_by_name(stats, 'a')
        self.assertEqual(fsa.ncall, 3)
        self.assertEqual(len(stats), 1)  # b() should be cleared out.

        @yappi.profile(return_callback=aggregate)
        def count_down_rec(n):
            if n == 0:
                return
            count_down_rec(n - 1)

        try:
            os.remove(
                "tests/count_down_rec.profile"
            )  # remove the one from prev test, if available
        except:
            pass

        try:
            count_down_rec(4)
        except:
            pass
        try:
            count_down_rec(3)
        except:
            pass

        stats = yappi.YFuncStats("tests/count_down_rec.profile")
        fsrec = utils.find_stat_by_name(stats, 'count_down_rec')
        self.assertEqual(fsrec.ncall, 9)
        self.assertEqual(fsrec.nactualcall, 2)

    def test_strip_dirs(self):

        def a():
            pass

        stats = utils.run_and_get_func_stats(a, )
        stats.strip_dirs()
        fsa = utils.find_stat_by_name(stats, "a")
        self.assertEqual(fsa.module, os.path.basename(fsa.module))

    @unittest.skipIf(os.name == "nt", "do not run on Windows")
    def test_run_as_script(self):
        import re
        p = subprocess.Popen(
            ['yappi', os.path.join('./tests', 'run_as_script.py')],
            stdout=subprocess.PIPE
        )
        out, err = p.communicate()
        self.assertEqual(p.returncode, 0)
        func_stats, thread_stats = re.split(
            b'name\\s+id\\s+tid\\s+ttot\\s+scnt\\s*\n', out
        )
        self.assertTrue(b'FancyThread' in thread_stats)

    def test_yappi_overhead(self):
        LOOP_COUNT = 100000

        def a():
            pass

        def b():
            for i in range(LOOP_COUNT):
                a()

        t0 = time.time()
        yappi.start()
        b()
        yappi.stop()
        time_with_yappi = time.time() - t0
        t0 = time.time()
        b()
        time_without_yappi = time.time() - t0
        if time_without_yappi == 0:
            time_without_yappi = 0.000001

        # in latest v0.82, I calculated this as close to "7.0" in my machine.
        # however, %83 of this overhead is coming from tickcount(). The other %17
        # seems to have been evenly distributed to the internal bookkeeping
        # structures/algorithms which seems acceptable. Note that our test only
        # tests one function being profiled at-a-time in a short interval.
        # profiling high number of functions in a small time
        # is a different beast, (which is pretty unlikely in most applications)
        # So as a conclusion: I cannot see any optimization window for Yappi that
        # is worth implementing as we will only optimize %17 of the time.
        sys.stderr.write("\r\nYappi puts %0.1f times overhead to the profiled application in average.\r\n" % \
            (time_with_yappi / time_without_yappi))

    def test_clear_stats_while_running(self):

        def a():
            pass

        yappi.start()
        a()
        yappi.clear_stats()
        a()
        stats = yappi.get_func_stats()
        fsa = utils.find_stat_by_name(stats, 'a')
        self.assertEqual(fsa.ncall, 1)

    def test_generator(self):

        def _gen(n):
            while (n > 0):
                yield n
                n -= 1

        yappi.start()
        for x in _gen(5):
            pass
        self.assertTrue(
            yappi.convert2pstats(yappi.get_func_stats()) is not None
        )

    def test_slice_child_stats_and_strip_dirs(self):

        def b():
            for i in range(10000000):
                pass

        def a():
            b()

        yappi.start(builtins=True)
        a()
        stats = yappi.get_func_stats()
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        self.assertTrue(fsa.children[0:1] is not None)
        prev_afullname = fsa.full_name
        prev_bchildfullname = fsa.children[fsb].full_name
        stats.strip_dirs()
        self.assertTrue(len(prev_afullname) > len(fsa.full_name))
        self.assertTrue(
            len(prev_bchildfullname) > len(fsa.children[fsb].full_name)
        )

    def test_children_stat_functions(self):
        _timings = {"a_1": 5, "b_1": 3, "c_1": 1}
        _yappi._set_test_timings(_timings)

        def b():
            pass

        def c():
            pass

        def a():
            b()
            c()

        yappi.start()
        a()
        b()  # non-child call
        c()  # non-child call
        stats = yappi.get_func_stats()
        fsa = utils.find_stat_by_name(stats, 'a')
        childs_of_a = fsa.children.get().sort("tavg", "desc")
        prev_item = None
        for item in childs_of_a:
            if prev_item:
                self.assertTrue(prev_item.tavg > item.tavg)
            prev_item = item
        childs_of_a.sort("name", "desc")
        prev_item = None
        for item in childs_of_a:
            if prev_item:
                self.assertTrue(prev_item.name > item.name)
            prev_item = item
        childs_of_a.clear()
        self.assertTrue(childs_of_a.empty())

    def test_no_stats_different_clock_type_load(self):

        def a():
            pass

        yappi.start()
        a()
        yappi.stop()
        yappi.get_func_stats().save("tests/ystats1.ys")
        yappi.clear_stats()
        yappi.set_clock_type("WALL")
        yappi.start()
        yappi.stop()
        stats = yappi.get_func_stats().add("tests/ystats1.ys")
        fsa = utils.find_stat_by_name(stats, 'a')
        self.assertTrue(fsa is not None)

    def test_subsequent_profile(self):
        _timings = {"a_1": 1, "b_1": 1}
        _yappi._set_test_timings(_timings)

        def a():
            pass

        def b():
            pass

        yappi.start()
        a()
        yappi.stop()
        yappi.start()
        b()
        yappi.stop()
        stats = yappi.get_func_stats()
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        self.assertTrue(fsa is not None)
        self.assertTrue(fsb is not None)
        self.assertEqual(fsa.ttot, 1)
        self.assertEqual(fsb.ttot, 1)

    def test_lambda(self):
        f = lambda: time.sleep(0.3)
        yappi.set_clock_type("wall")
        yappi.start()
        f()
        stats = yappi.get_func_stats()
        fsa = utils.find_stat_by_name(stats, '<lambda>')
        self.assertTrue(fsa.ttot > 0.1)

    def test_module_stress(self):
        self.assertEqual(yappi.is_running(), False)

        yappi.start()
        yappi.clear_stats()
        self.assertRaises(_yappi.error, yappi.set_clock_type, "wall")

        yappi.stop()
        yappi.clear_stats()
        yappi.set_clock_type("cpu")
        self.assertRaises(yappi.YappiError, yappi.set_clock_type, "dummy")
        self.assertEqual(yappi.is_running(), False)
        yappi.clear_stats()
        yappi.clear_stats()

    def test_stat_sorting(self):
        _timings = {"a_1": 13, "b_1": 10, "a_2": 6, "b_2": 1}
        _yappi._set_test_timings(_timings)

        self._ncall = 1

        def a():
            b()

        def b():
            if self._ncall == 2:
                return
            self._ncall += 1
            a()

        stats = utils.run_and_get_func_stats(a)
        stats = stats.sort("totaltime", "desc")
        prev_stat = None
        for stat in stats:
            if prev_stat:
                self.assertTrue(prev_stat.ttot >= stat.ttot)
            prev_stat = stat
        stats = stats.sort("totaltime", "asc")
        prev_stat = None
        for stat in stats:
            if prev_stat:
                self.assertTrue(prev_stat.ttot <= stat.ttot)
            prev_stat = stat
        stats = stats.sort("avgtime", "asc")
        prev_stat = None
        for stat in stats:
            if prev_stat:
                self.assertTrue(prev_stat.tavg <= stat.tavg)
            prev_stat = stat
        stats = stats.sort("name", "asc")
        prev_stat = None
        for stat in stats:
            if prev_stat:
                self.assertTrue(prev_stat.name <= stat.name)
            prev_stat = stat
        stats = stats.sort("subtime", "asc")
        prev_stat = None
        for stat in stats:
            if prev_stat:
                self.assertTrue(prev_stat.tsub <= stat.tsub)
            prev_stat = stat

        self.assertRaises(
            yappi.YappiError, stats.sort, "invalid_func_sorttype_arg"
        )
        self.assertRaises(
            yappi.YappiError, stats.sort, "totaltime",
            "invalid_func_sortorder_arg"
        )

    def test_start_flags(self):
        self.assertEqual(_yappi._get_start_flags(), None)
        yappi.start()

        def a():
            pass

        a()
        self.assertEqual(_yappi._get_start_flags()["profile_builtins"], 0)
        self.assertEqual(_yappi._get_start_flags()["profile_multicontext"], 1)
        self.assertEqual(len(yappi.get_thread_stats()), 1)

    def test_builtin_profiling(self):

        def a():
            time.sleep(0.4)  # is a builtin function

        yappi.set_clock_type('wall')

        yappi.start(builtins=True)
        a()
        stats = yappi.get_func_stats()
        fsa = utils.find_stat_by_name(stats, 'sleep')
        self.assertTrue(fsa is not None)
        self.assertTrue(fsa.ttot > 0.3)
        yappi.stop()
        yappi.clear_stats()

        def a():
            pass

        yappi.start()
        t = threading.Thread(target=a)
        t.start()
        t.join()
        stats = yappi.get_func_stats()

    def test_singlethread_profiling(self):
        yappi.set_clock_type('wall')

        def a():
            time.sleep(0.2)

        class Worker1(threading.Thread):

            def a(self):
                time.sleep(0.3)

            def run(self):
                self.a()

        yappi.start(profile_threads=False)

        c = Worker1()
        c.start()
        c.join()
        a()
        stats = yappi.get_func_stats()
        fsa1 = utils.find_stat_by_name(stats, 'Worker1.a')
        fsa2 = utils.find_stat_by_name(stats, 'a')
        self.assertTrue(fsa1 is None)
        self.assertTrue(fsa2 is not None)
        self.assertTrue(fsa2.ttot > 0.1)

    def test_run(self):

        def profiled():
            pass

        yappi.clear_stats()
        try:
            with yappi.run():
                profiled()
            stats = yappi.get_func_stats()
        finally:
            yappi.clear_stats()

        self.assertIsNotNone(utils.find_stat_by_name(stats, 'profiled'))

    def test_run_recursive(self):

        def profiled():
            pass

        def not_profiled():
            pass

        yappi.clear_stats()
        try:
            with yappi.run():
                with yappi.run():
                    profiled()
                # Profiling stopped here
                not_profiled()
            stats = yappi.get_func_stats()
        finally:
            yappi.clear_stats()

        self.assertIsNotNone(utils.find_stat_by_name(stats, 'profiled'))
        self.assertIsNone(utils.find_stat_by_name(stats, 'not_profiled'))


class StatSaveScenarios(utils.YappiUnitTestCase):

    def test_pstats_conversion(self):

        def pstat_id(fs):
            return (fs.module, fs.lineno, fs.name)

        def a():
            d()

        def b():
            d()

        def c():
            pass

        def d():
            pass

        _timings = {"a_1": 12, "b_1": 7, "c_1": 5, "d_1": 2}
        _yappi._set_test_timings(_timings)
        stats = utils.run_and_get_func_stats(a, )
        stats.strip_dirs()
        stats.save("tests/a1.pstats", type="pstat")
        fsa_pid = pstat_id(utils.find_stat_by_name(stats, "a"))
        fsd_pid = pstat_id(utils.find_stat_by_name(stats, "d"))
        yappi.clear_stats()
        _yappi._set_test_timings(_timings)
        stats = utils.run_and_get_func_stats(a, )
        stats.strip_dirs()
        stats.save("tests/a2.pstats", type="pstat")
        yappi.clear_stats()
        _yappi._set_test_timings(_timings)
        stats = utils.run_and_get_func_stats(b, )
        stats.strip_dirs()
        stats.save("tests/b1.pstats", type="pstat")
        fsb_pid = pstat_id(utils.find_stat_by_name(stats, "b"))
        yappi.clear_stats()
        _yappi._set_test_timings(_timings)
        stats = utils.run_and_get_func_stats(c, )
        stats.strip_dirs()
        stats.save("tests/c1.pstats", type="pstat")
        fsc_pid = pstat_id(utils.find_stat_by_name(stats, "c"))

        # merge saved stats and check pstats values are correct
        import pstats
        p = pstats.Stats(
            'tests/a1.pstats', 'tests/a2.pstats', 'tests/b1.pstats',
            'tests/c1.pstats'
        )
        p.strip_dirs()
        # ct = ttot, tt = tsub
        (cc, nc, tt, ct, callers) = p.stats[fsa_pid]
        self.assertEqual(cc, nc, 2)
        self.assertEqual(tt, 20)
        self.assertEqual(ct, 24)
        (cc, nc, tt, ct, callers) = p.stats[fsd_pid]
        self.assertEqual(cc, nc, 3)
        self.assertEqual(tt, 6)
        self.assertEqual(ct, 6)
        self.assertEqual(len(callers), 2)
        (cc, nc, tt, ct) = callers[fsa_pid]
        self.assertEqual(cc, nc, 2)
        self.assertEqual(tt, 4)
        self.assertEqual(ct, 4)
        (cc, nc, tt, ct) = callers[fsb_pid]
        self.assertEqual(cc, nc, 1)
        self.assertEqual(tt, 2)
        self.assertEqual(ct, 2)

    def test_merge_stats(self):
        _timings = {
            "a_1": 15,
            "b_1": 14,
            "c_1": 12,
            "d_1": 10,
            "e_1": 9,
            "f_1": 7,
            "g_1": 6,
            "h_1": 5,
            "i_1": 1
        }
        _yappi._set_test_timings(_timings)

        def a():
            b()

        def b():
            c()

        def c():
            d()

        def d():
            e()

        def e():
            f()

        def f():
            g()

        def g():
            h()

        def h():
            i()

        def i():
            pass

        yappi.start()
        a()
        a()
        yappi.stop()
        stats = yappi.get_func_stats()
        self.assertRaises(
            NotImplementedError, stats.save, "", "INVALID_SAVE_TYPE"
        )
        stats.save("tests/ystats2.ys")
        yappi.clear_stats()
        _yappi._set_test_timings(_timings)
        yappi.start()
        a()
        stats = yappi.get_func_stats().add("tests/ystats2.ys")
        fsa = utils.find_stat_by_name(stats, "a")
        fsb = utils.find_stat_by_name(stats, "b")
        fsc = utils.find_stat_by_name(stats, "c")
        fsd = utils.find_stat_by_name(stats, "d")
        fse = utils.find_stat_by_name(stats, "e")
        fsf = utils.find_stat_by_name(stats, "f")
        fsg = utils.find_stat_by_name(stats, "g")
        fsh = utils.find_stat_by_name(stats, "h")
        fsi = utils.find_stat_by_name(stats, "i")
        self.assertEqual(fsa.ttot, 45)
        self.assertEqual(fsa.ncall, 3)
        self.assertEqual(fsa.nactualcall, 3)
        self.assertEqual(fsa.tsub, 3)
        self.assertEqual(fsa.children[fsb].ttot, fsb.ttot)
        self.assertEqual(fsa.children[fsb].tsub, fsb.tsub)
        self.assertEqual(fsb.children[fsc].ttot, fsc.ttot)
        self.assertEqual(fsb.children[fsc].tsub, fsc.tsub)
        self.assertEqual(fsc.tsub, 6)
        self.assertEqual(fsc.children[fsd].ttot, fsd.ttot)
        self.assertEqual(fsc.children[fsd].tsub, fsd.tsub)
        self.assertEqual(fsd.children[fse].ttot, fse.ttot)
        self.assertEqual(fsd.children[fse].tsub, fse.tsub)
        self.assertEqual(fse.children[fsf].ttot, fsf.ttot)
        self.assertEqual(fse.children[fsf].tsub, fsf.tsub)
        self.assertEqual(fsf.children[fsg].ttot, fsg.ttot)
        self.assertEqual(fsf.children[fsg].tsub, fsg.tsub)
        self.assertEqual(fsg.ttot, 18)
        self.assertEqual(fsg.tsub, 3)
        self.assertEqual(fsg.children[fsh].ttot, fsh.ttot)
        self.assertEqual(fsg.children[fsh].tsub, fsh.tsub)
        self.assertEqual(fsh.ttot, 15)
        self.assertEqual(fsh.tsub, 12)
        self.assertEqual(fsh.tavg, 5)
        self.assertEqual(fsh.children[fsi].ttot, fsi.ttot)
        self.assertEqual(fsh.children[fsi].tsub, fsi.tsub)
        #stats.debug_print()

    def test_merge_multithreaded_stats(self):
        import _yappi
        timings = {"a_1": 2, "b_1": 1}
        _yappi._set_test_timings(timings)

        def a():
            pass

        def b():
            pass

        yappi.start()
        t = threading.Thread(target=a)
        t.start()
        t.join()
        t = threading.Thread(target=b)
        t.start()
        t.join()
        yappi.get_func_stats().save("tests/ystats1.ys")
        yappi.clear_stats()
        _yappi._set_test_timings(timings)
        self.assertEqual(len(yappi.get_func_stats()), 0)
        self.assertEqual(len(yappi.get_thread_stats()), 1)
        t = threading.Thread(target=a)
        t.start()
        t.join()

        self.assertEqual(_yappi._get_start_flags()["profile_builtins"], 0)
        self.assertEqual(_yappi._get_start_flags()["profile_multicontext"], 1)
        yappi.get_func_stats().save("tests/ystats2.ys")

        stats = yappi.YFuncStats([
            "tests/ystats1.ys",
            "tests/ystats2.ys",
        ])
        fsa = utils.find_stat_by_name(stats, "a")
        fsb = utils.find_stat_by_name(stats, "b")
        self.assertEqual(fsa.ncall, 2)
        self.assertEqual(fsb.ncall, 1)
        self.assertEqual(fsa.tsub, fsa.ttot, 4)
        self.assertEqual(fsb.tsub, fsb.ttot, 1)

    def test_merge_load_different_clock_types(self):
        yappi.start(builtins=True)

        def a():
            b()

        def b():
            c()

        def c():
            pass

        t = threading.Thread(target=a)
        t.start()
        t.join()
        yappi.get_func_stats().sort("name", "asc").save("tests/ystats1.ys")
        yappi.stop()
        yappi.clear_stats()
        yappi.start(builtins=False)
        t = threading.Thread(target=a)
        t.start()
        t.join()
        yappi.get_func_stats().save("tests/ystats2.ys")
        yappi.stop()
        self.assertRaises(_yappi.error, yappi.set_clock_type, "wall")
        yappi.clear_stats()
        yappi.set_clock_type("wall")
        yappi.start()
        t = threading.Thread(target=a)
        t.start()
        t.join()
        yappi.get_func_stats().save("tests/ystats3.ys")
        self.assertRaises(
            yappi.YappiError,
            yappi.YFuncStats().add("tests/ystats1.ys").add, "tests/ystats3.ys"
        )
        stats = yappi.YFuncStats(["tests/ystats1.ys",
                                  "tests/ystats2.ys"]).sort("name")
        fsa = utils.find_stat_by_name(stats, "a")
        fsb = utils.find_stat_by_name(stats, "b")
        fsc = utils.find_stat_by_name(stats, "c")
        self.assertEqual(fsa.ncall, 2)
        self.assertEqual(fsa.ncall, fsb.ncall, fsc.ncall)

    def test_merge_aabab_aabbc(self):
        _timings = {
            "a_1": 15,
            "a_2": 14,
            "b_1": 12,
            "a_3": 10,
            "b_2": 9,
            "c_1": 4
        }
        _yappi._set_test_timings(_timings)

        def a():
            if self._ncall == 1:
                self._ncall += 1
                a()
            elif self._ncall == 5:
                self._ncall += 1
                a()
            else:
                b()

        def b():
            if self._ncall == 2:
                self._ncall += 1
                a()
            elif self._ncall == 6:
                self._ncall += 1
                b()
            elif self._ncall == 7:
                c()
            else:
                return

        def c():
            pass

        self._ncall = 1
        stats = utils.run_and_get_func_stats(a, )
        stats.save("tests/ystats1.ys")
        yappi.clear_stats()
        _yappi._set_test_timings(_timings)
        #stats.print_all()

        self._ncall = 5
        stats = utils.run_and_get_func_stats(a, )
        stats.save("tests/ystats2.ys")

        #stats.print_all()

        def a():  # same name but another function(code object)
            pass

        yappi.start()
        a()
        stats = yappi.get_func_stats().add(
            ["tests/ystats1.ys", "tests/ystats2.ys"]
        )
        #stats.print_all()
        self.assertEqual(len(stats), 4)

        fsa = None
        for stat in stats:
            if stat.name == "a" and stat.ttot == 45:
                fsa = stat
                break
        self.assertTrue(fsa is not None)

        self.assertEqual(fsa.ncall, 7)
        self.assertEqual(fsa.nactualcall, 3)
        self.assertEqual(fsa.ttot, 45)
        self.assertEqual(fsa.tsub, 10)
        fsb = utils.find_stat_by_name(stats, "b")
        fsc = utils.find_stat_by_name(stats, "c")
        self.assertEqual(fsb.ncall, 6)
        self.assertEqual(fsb.nactualcall, 3)
        self.assertEqual(fsb.ttot, 36)
        self.assertEqual(fsb.tsub, 27)
        self.assertEqual(fsb.tavg, 6)
        self.assertEqual(fsc.ttot, 8)
        self.assertEqual(fsc.tsub, 8)
        self.assertEqual(fsc.tavg, 4)
        self.assertEqual(fsc.nactualcall, fsc.ncall, 2)


class MultithreadedScenarios(utils.YappiUnitTestCase):

    def test_issue_32(self):
        '''
        Start yappi from different thread and we get Internal Error(15) as 
        the current_ctx_id() called while enumerating the threads in start() 
        and as it does not swap to the enumerated ThreadState* the THreadState_GetDict()
        returns wrong object and thus sets an invalid id for the _ctx structure.

        When this issue happens multiple Threads have same tid as the internal ts_ptr
        will be same for different contexts. So, let's see if that happens
        '''

        def foo():
            time.sleep(0.2)

        def bar():
            time.sleep(0.1)

        def thread_func():
            yappi.set_clock_type("wall")
            yappi.start()

            bar()

        t = threading.Thread(target=thread_func)
        t.start()
        t.join()

        foo()

        yappi.stop()

        thread_ids = set()
        for tstat in yappi.get_thread_stats():
            self.assertTrue(tstat.tid not in thread_ids)
            thread_ids.add(tstat.tid)

    def test_subsequent_profile(self):
        WORKER_COUNT = 5

        def a():
            pass

        def b():
            pass

        def c():
            pass

        _timings = {
            "a_1": 3,
            "b_1": 2,
            "c_1": 1,
        }

        yappi.start()

        def g():
            pass

        g()
        yappi.stop()
        yappi.clear_stats()
        _yappi._set_test_timings(_timings)
        yappi.start()

        _dummy = []
        for i in range(WORKER_COUNT):
            t = threading.Thread(target=a)
            t.start()
            t.join()
        for i in range(WORKER_COUNT):
            t = threading.Thread(target=b)
            t.start()
            _dummy.append(t)
            t.join()
        for i in range(WORKER_COUNT):
            t = threading.Thread(target=a)
            t.start()
            t.join()
        for i in range(WORKER_COUNT):
            t = threading.Thread(target=c)
            t.start()
            t.join()
        yappi.stop()
        yappi.start()

        def f():
            pass

        f()
        stats = yappi.get_func_stats()
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        fsc = utils.find_stat_by_name(stats, 'c')
        self.assertEqual(fsa.ncall, 10)
        self.assertEqual(fsb.ncall, 5)
        self.assertEqual(fsc.ncall, 5)
        self.assertEqual(fsa.ttot, fsa.tsub, 30)
        self.assertEqual(fsb.ttot, fsb.tsub, 10)
        self.assertEqual(fsc.ttot, fsc.tsub, 5)

        # MACOSx optimizes by only creating one worker thread
        self.assertTrue(len(yappi.get_thread_stats()) >= 2)

    def test_basic(self):
        yappi.set_clock_type('wall')

        def dummy():
            pass

        def a():
            time.sleep(0.2)

        class Worker1(threading.Thread):

            def a(self):
                time.sleep(0.3)

            def run(self):
                self.a()

        yappi.start(builtins=False, profile_threads=True)

        c = Worker1()
        c.start()
        c.join()
        a()
        stats = yappi.get_func_stats()
        fsa1 = utils.find_stat_by_name(stats, 'Worker1.a')
        fsa2 = utils.find_stat_by_name(stats, 'a')
        self.assertTrue(fsa1 is not None)
        self.assertTrue(fsa2 is not None)
        self.assertTrue(fsa1.ttot > 0.2)
        self.assertTrue(fsa2.ttot > 0.1)
        tstats = yappi.get_thread_stats()
        self.assertEqual(len(tstats), 2)
        tsa = utils.find_stat_by_name(tstats, 'Worker1')
        tsm = utils.find_stat_by_name(tstats, '_MainThread')
        dummy()  # call dummy to force ctx name to be retrieved again.
        self.assertTrue(tsa is not None)
        # TODO: I put dummy() to fix below, remove the comments after a while.
        self.assertTrue( # FIX: I see this fails sometimes?
            tsm is not None,
            f"Could not find \"_MainThread\". Found: {', '.join(utils.get_stat_names(tstats))}")

    def test_ctx_stats(self):
        from threading import Thread
        DUMMY_WORKER_COUNT = 5
        yappi.start()

        class DummyThread(Thread):
            pass

        def dummy():
            pass

        def dummy_worker():
            pass

        for i in range(DUMMY_WORKER_COUNT):
            t = DummyThread(target=dummy_worker)
            t.start()
            t.join()
        yappi.stop()
        stats = yappi.get_thread_stats()
        tsa = utils.find_stat_by_name(stats, "DummyThread")
        self.assertTrue(tsa is not None)
        yappi.clear_stats()
        time.sleep(1.0)
        _timings = {
            "a_1": 6,
            "b_1": 5,
            "c_1": 3,
            "d_1": 1,
            "a_2": 4,
            "b_2": 3,
            "c_2": 2,
            "d_2": 1
        }
        _yappi._set_test_timings(_timings)

        class Thread1(Thread):
            pass

        class Thread2(Thread):
            pass

        def a():
            b()

        def b():
            c()

        def c():
            d()

        def d():
            time.sleep(0.6)

        yappi.set_clock_type("wall")
        yappi.start()
        t1 = Thread1(target=a)
        t1.start()
        t2 = Thread2(target=a)
        t2.start()
        t1.join()
        t2.join()
        stats = yappi.get_thread_stats()

        # the fist clear_stats clears the context table?
        tsa = utils.find_stat_by_name(stats, "DummyThread")
        self.assertTrue(tsa is None)

        tst1 = utils.find_stat_by_name(stats, "Thread1")
        tst2 = utils.find_stat_by_name(stats, "Thread2")
        tsmain = utils.find_stat_by_name(stats, "_MainThread")
        dummy()  # call dummy to force ctx name to be retrieved again.
        self.assertTrue(len(stats) == 3)
        self.assertTrue(tst1 is not None)
        self.assertTrue(tst2 is not None)
        # TODO: I put dummy() to fix below, remove the comments after a while.
        self.assertTrue( # FIX: I see this fails sometimes
            tsmain is not None,
            f"Could not find \"_MainThread\". Found: {', '.join(utils.get_stat_names(stats))}")
        self.assertTrue(1.0 > tst2.ttot >= 0.5)
        self.assertTrue(1.0 > tst1.ttot >= 0.5)

        # test sorting of the ctx stats
        stats = stats.sort("totaltime", "desc")
        prev_stat = None
        for stat in stats:
            if prev_stat:
                self.assertTrue(prev_stat.ttot >= stat.ttot)
            prev_stat = stat
        stats = stats.sort("totaltime", "asc")
        prev_stat = None
        for stat in stats:
            if prev_stat:
                self.assertTrue(prev_stat.ttot <= stat.ttot)
            prev_stat = stat
        stats = stats.sort("schedcount", "desc")
        prev_stat = None
        for stat in stats:
            if prev_stat:
                self.assertTrue(prev_stat.sched_count >= stat.sched_count)
            prev_stat = stat
        stats = stats.sort("name", "desc")
        prev_stat = None
        for stat in stats:
            if prev_stat:
                self.assertTrue(prev_stat.name.lower() >= stat.name.lower())
            prev_stat = stat
        self.assertRaises(
            yappi.YappiError, stats.sort, "invalid_thread_sorttype_arg"
        )
        self.assertRaises(
            yappi.YappiError, stats.sort, "invalid_thread_sortorder_arg"
        )

    def test_ctx_stats_cpu(self):

        def get_thread_name():
            try:
                return threading.current_thread().name
            except AttributeError:
                return "Anonymous"

        def burn_cpu(sec):
            t0 = yappi.get_clock_time()
            elapsed = 0
            while (elapsed < sec):
                for _ in range(1000):
                    pass
                elapsed = yappi.get_clock_time() - t0

        def test():

            ts = []
            for i in (0.01, 0.05, 0.1):
                t = threading.Thread(target=burn_cpu, args=(i, ))
                t.name = f"burn_cpu-{str(i)}"
                t.start()
                ts.append(t)
            for t in ts:
                t.join()

        yappi.set_clock_type("cpu")
        yappi.set_context_name_callback(get_thread_name)

        yappi.start()

        test()

        yappi.stop()

        tstats = yappi.get_thread_stats()
        r1 = '''
        burn_cpu-0.1      3      123145356058624  0.100105  8
        burn_cpu-0.05     2      123145361313792  0.050149  8
        burn_cpu-0.01     1      123145356058624  0.010127  2
        MainThread        0      4321620864       0.001632  6
        '''
        self.assert_ctx_stats_almost_equal(r1, tstats)

    def test_producer_consumer_with_queues(self):
        # we currently just stress yappi, no functionality test is done here.
        yappi.start()
        from queue import Queue
        from threading import Thread
        WORKER_THREAD_COUNT = 50
        WORK_ITEM_COUNT = 2000

        def worker():
            while True:
                item = q.get()
                # do the work with item
                q.task_done()

        q = Queue()
        for i in range(WORKER_THREAD_COUNT):
            t = Thread(target=worker)
            t.daemon = True
            t.start()

        for item in range(WORK_ITEM_COUNT):
            q.put(item)
        q.join()  # block until all tasks are done
        #yappi.get_func_stats().sort("callcount").print_all()
        yappi.stop()

    def test_temporary_lock_waiting(self):
        yappi.start()
        _lock = threading.Lock()

        def worker():
            _lock.acquire()
            try:
                time.sleep(1.0)
            finally:
                _lock.release()

        t1 = threading.Thread(target=worker)
        t2 = threading.Thread(target=worker)
        t1.start()
        t2.start()
        t1.join()
        t2.join()
        #yappi.get_func_stats().sort("callcount").print_all()
        yappi.stop()

    @unittest.skipIf(os.name != "posix", "requires Posix compliant OS")
    def test_signals_with_blocking_calls(self):
        import signal, os, time

        # just to verify if signal is handled correctly and stats/yappi are not corrupted.
        def handler(signum, frame):
            raise Exception("Signal handler executed!")

        yappi.start()
        signal.signal(signal.SIGALRM, handler)
        signal.alarm(1)
        self.assertRaises(Exception, time.sleep, 2)
        stats = yappi.get_func_stats()
        fsh = utils.find_stat_by_name(stats, "handler")
        self.assertTrue(fsh is not None)

    def test_concurrent_futures(self):
        yappi.start()
        from concurrent.futures import ThreadPoolExecutor
        with ThreadPoolExecutor(max_workers=5) as executor:
            f = executor.submit(pow, 5, 2)
            self.assertEqual(f.result(), 25)
        time.sleep(1.0)
        yappi.stop()

    def test_barrier(self):
        yappi.start()
        b = threading.Barrier(2, timeout=1)

        def worker():
            try:
                b.wait()
            except threading.BrokenBarrierError:
                pass
            except Exception:
                raise Exception("BrokenBarrierError not raised")

        t1 = threading.Thread(target=worker)
        t1.start()
        #b.wait()
        t1.join()
        yappi.stop()


class NonRecursiveFunctions(utils.YappiUnitTestCase):

    def test_abcd(self):
        _timings = {"a_1": 6, "b_1": 5, "c_1": 3, "d_1": 1}
        _yappi._set_test_timings(_timings)

        def a():
            b()

        def b():
            c()

        def c():
            d()

        def d():
            pass

        stats = utils.run_and_get_func_stats(a)
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        fsc = utils.find_stat_by_name(stats, 'c')
        fsd = utils.find_stat_by_name(stats, 'd')
        cfsab = fsa.children[fsb]
        cfsbc = fsb.children[fsc]
        cfscd = fsc.children[fsd]

        self.assertEqual(fsa.ttot, 6)
        self.assertEqual(fsa.tsub, 1)
        self.assertEqual(fsb.ttot, 5)
        self.assertEqual(fsb.tsub, 2)
        self.assertEqual(fsc.ttot, 3)
        self.assertEqual(fsc.tsub, 2)
        self.assertEqual(fsd.ttot, 1)
        self.assertEqual(fsd.tsub, 1)
        self.assertEqual(cfsab.ttot, 5)
        self.assertEqual(cfsab.tsub, 2)
        self.assertEqual(cfsbc.ttot, 3)
        self.assertEqual(cfsbc.tsub, 2)
        self.assertEqual(cfscd.ttot, 1)
        self.assertEqual(cfscd.tsub, 1)

    def test_stop_in_middle(self):
        _timings = {"a_1": 6, "b_1": 4}
        _yappi._set_test_timings(_timings)

        def a():
            b()
            yappi.stop()

        def b():
            time.sleep(0.2)

        yappi.start()
        a()
        stats = yappi.get_func_stats()
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')

        self.assertEqual(fsa.ncall, 1)
        self.assertEqual(fsa.nactualcall, 0)
        self.assertEqual(fsa.ttot, 0)  # no call_leave called
        self.assertEqual(fsa.tsub, 0)  # no call_leave called
        self.assertEqual(fsb.ttot, 4)


class RecursiveFunctions(utils.YappiUnitTestCase):

    def test_fibonacci(self):

        def fib(n):
            if n > 1:
                return fib(n - 1) + fib(n - 2)
            else:
                return n

        stats = utils.run_and_get_func_stats(fib, 22)
        fs = utils.find_stat_by_name(stats, 'fib')
        self.assertEqual(fs.ncall, 57313)
        self.assertEqual(fs.ttot, fs.tsub)

    def test_abcadc(self):
        _timings = {
            "a_1": 20,
            "b_1": 19,
            "c_1": 17,
            "a_2": 13,
            "d_1": 12,
            "c_2": 10,
            "a_3": 5
        }
        _yappi._set_test_timings(_timings)

        def a(n):
            if n == 3:
                return
            if n == 1 + 1:
                d(n)
            else:
                b(n)

        def b(n):
            c(n)

        def c(n):
            a(n + 1)

        def d(n):
            c(n)

        stats = utils.run_and_get_func_stats(a, 1)
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        fsc = utils.find_stat_by_name(stats, 'c')
        fsd = utils.find_stat_by_name(stats, 'd')
        self.assertEqual(fsa.ncall, 3)
        self.assertEqual(fsa.nactualcall, 1)
        self.assertEqual(fsa.ttot, 20)
        self.assertEqual(fsa.tsub, 7)
        self.assertEqual(fsb.ttot, 19)
        self.assertEqual(fsb.tsub, 2)
        self.assertEqual(fsc.ttot, 17)
        self.assertEqual(fsc.tsub, 9)
        self.assertEqual(fsd.ttot, 12)
        self.assertEqual(fsd.tsub, 2)
        cfsca = fsc.children[fsa]
        self.assertEqual(cfsca.nactualcall, 0)
        self.assertEqual(cfsca.ncall, 2)
        self.assertEqual(cfsca.ttot, 13)
        self.assertEqual(cfsca.tsub, 6)

    def test_aaaa(self):
        _timings = {"d_1": 9, "d_2": 7, "d_3": 3, "d_4": 2}
        _yappi._set_test_timings(_timings)

        def d(n):
            if n == 3:
                return
            d(n + 1)

        stats = utils.run_and_get_func_stats(d, 0)
        fsd = utils.find_stat_by_name(stats, 'd')
        self.assertEqual(fsd.ncall, 4)
        self.assertEqual(fsd.nactualcall, 1)
        self.assertEqual(fsd.ttot, 9)
        self.assertEqual(fsd.tsub, 9)
        cfsdd = fsd.children[fsd]
        self.assertEqual(cfsdd.ttot, 7)
        self.assertEqual(cfsdd.tsub, 7)
        self.assertEqual(cfsdd.ncall, 3)
        self.assertEqual(cfsdd.nactualcall, 0)

    def test_abcabc(self):
        _timings = {
            "a_1": 20,
            "b_1": 19,
            "c_1": 17,
            "a_2": 13,
            "b_2": 11,
            "c_2": 9,
            "a_3": 6
        }
        _yappi._set_test_timings(_timings)

        def a(n):
            if n == 3:
                return
            else:
                b(n)

        def b(n):
            c(n)

        def c(n):
            a(n + 1)

        stats = utils.run_and_get_func_stats(a, 1)
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        fsc = utils.find_stat_by_name(stats, 'c')
        self.assertEqual(fsa.ncall, 3)
        self.assertEqual(fsa.nactualcall, 1)
        self.assertEqual(fsa.ttot, 20)
        self.assertEqual(fsa.tsub, 9)
        self.assertEqual(fsb.ttot, 19)
        self.assertEqual(fsb.tsub, 4)
        self.assertEqual(fsc.ttot, 17)
        self.assertEqual(fsc.tsub, 7)
        cfsab = fsa.children[fsb]
        cfsbc = fsb.children[fsc]
        cfsca = fsc.children[fsa]
        self.assertEqual(cfsab.ttot, 19)
        self.assertEqual(cfsab.tsub, 4)
        self.assertEqual(cfsbc.ttot, 17)
        self.assertEqual(cfsbc.tsub, 7)
        self.assertEqual(cfsca.ttot, 13)
        self.assertEqual(cfsca.tsub, 8)

    def test_abcbca(self):
        _timings = {"a_1": 10, "b_1": 9, "c_1": 7, "b_2": 4, "c_2": 2, "a_2": 1}
        _yappi._set_test_timings(_timings)
        self._ncall = 1

        def a():
            if self._ncall == 1:
                b()
            else:
                return

        def b():
            c()

        def c():
            if self._ncall == 1:
                self._ncall += 1
                b()
            else:
                a()

        stats = utils.run_and_get_func_stats(a)
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        fsc = utils.find_stat_by_name(stats, 'c')
        cfsab = fsa.children[fsb]
        cfsbc = fsb.children[fsc]
        cfsca = fsc.children[fsa]
        self.assertEqual(fsa.ttot, 10)
        self.assertEqual(fsa.tsub, 2)
        self.assertEqual(fsb.ttot, 9)
        self.assertEqual(fsb.tsub, 4)
        self.assertEqual(fsc.ttot, 7)
        self.assertEqual(fsc.tsub, 4)
        self.assertEqual(cfsab.ttot, 9)
        self.assertEqual(cfsab.tsub, 2)
        self.assertEqual(cfsbc.ttot, 7)
        self.assertEqual(cfsbc.tsub, 4)
        self.assertEqual(cfsca.ttot, 1)
        self.assertEqual(cfsca.tsub, 1)
        self.assertEqual(cfsca.ncall, 1)
        self.assertEqual(cfsca.nactualcall, 0)

    def test_aabccb(self):
        _timings = {
            "a_1": 13,
            "a_2": 11,
            "b_1": 9,
            "c_1": 5,
            "c_2": 3,
            "b_2": 1
        }
        _yappi._set_test_timings(_timings)
        self._ncall = 1

        def a():
            if self._ncall == 1:
                self._ncall += 1
                a()
            else:
                b()

        def b():
            if self._ncall == 3:
                return
            else:
                c()

        def c():
            if self._ncall == 2:
                self._ncall += 1
                c()
            else:
                b()

        stats = utils.run_and_get_func_stats(a)
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        fsc = utils.find_stat_by_name(stats, 'c')
        cfsaa = fsa.children[fsa.index]
        cfsab = fsa.children[fsb]
        cfsbc = fsb.children[fsc.full_name]
        cfscc = fsc.children[fsc]
        cfscb = fsc.children[fsb]
        self.assertEqual(fsb.ttot, 9)
        self.assertEqual(fsb.tsub, 5)
        self.assertEqual(cfsbc.ttot, 5)
        self.assertEqual(cfsbc.tsub, 2)
        self.assertEqual(fsa.ttot, 13)
        self.assertEqual(fsa.tsub, 4)
        self.assertEqual(cfsab.ttot, 9)
        self.assertEqual(cfsab.tsub, 4)
        self.assertEqual(cfsaa.ttot, 11)
        self.assertEqual(cfsaa.tsub, 2)
        self.assertEqual(fsc.ttot, 5)
        self.assertEqual(fsc.tsub, 4)

    def test_abaa(self):
        _timings = {"a_1": 13, "b_1": 10, "a_2": 9, "a_3": 5}
        _yappi._set_test_timings(_timings)

        self._ncall = 1

        def a():
            if self._ncall == 1:
                b()
            elif self._ncall == 2:
                self._ncall += 1
                a()
            else:
                return

        def b():
            self._ncall += 1
            a()

        stats = utils.run_and_get_func_stats(a)
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        cfsaa = fsa.children[fsa]
        cfsba = fsb.children[fsa]
        self.assertEqual(fsb.ttot, 10)
        self.assertEqual(fsb.tsub, 1)
        self.assertEqual(fsa.ttot, 13)
        self.assertEqual(fsa.tsub, 12)
        self.assertEqual(cfsaa.ttot, 5)
        self.assertEqual(cfsaa.tsub, 5)
        self.assertEqual(cfsba.ttot, 9)
        self.assertEqual(cfsba.tsub, 4)

    def test_aabb(self):
        _timings = {"a_1": 13, "a_2": 10, "b_1": 9, "b_2": 5}
        _yappi._set_test_timings(_timings)

        self._ncall = 1

        def a():
            if self._ncall == 1:
                self._ncall += 1
                a()
            elif self._ncall == 2:
                b()
            else:
                return

        def b():
            if self._ncall == 2:
                self._ncall += 1
                b()
            else:
                return

        stats = utils.run_and_get_func_stats(a)
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        cfsaa = fsa.children[fsa]
        cfsab = fsa.children[fsb]
        cfsbb = fsb.children[fsb]
        self.assertEqual(fsa.ttot, 13)
        self.assertEqual(fsa.tsub, 4)
        self.assertEqual(fsb.ttot, 9)
        self.assertEqual(fsb.tsub, 9)
        self.assertEqual(cfsaa.ttot, 10)
        self.assertEqual(cfsaa.tsub, 1)
        self.assertEqual(cfsab.ttot, 9)
        self.assertEqual(cfsab.tsub, 4)
        self.assertEqual(cfsbb.ttot, 5)
        self.assertEqual(cfsbb.tsub, 5)

    def test_abbb(self):
        _timings = {"a_1": 13, "b_1": 10, "b_2": 6, "b_3": 1}
        _yappi._set_test_timings(_timings)

        self._ncall = 1

        def a():
            if self._ncall == 1:
                b()

        def b():
            if self._ncall == 3:
                return
            self._ncall += 1
            b()

        stats = utils.run_and_get_func_stats(a)
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        cfsab = fsa.children[fsb]
        cfsbb = fsb.children[fsb]
        self.assertEqual(fsa.ttot, 13)
        self.assertEqual(fsa.tsub, 3)
        self.assertEqual(fsb.ttot, 10)
        self.assertEqual(fsb.tsub, 10)
        self.assertEqual(fsb.ncall, 3)
        self.assertEqual(fsb.nactualcall, 1)
        self.assertEqual(cfsab.ttot, 10)
        self.assertEqual(cfsab.tsub, 4)
        self.assertEqual(cfsbb.ttot, 6)
        self.assertEqual(cfsbb.tsub, 6)
        self.assertEqual(cfsbb.nactualcall, 0)
        self.assertEqual(cfsbb.ncall, 2)

    def test_aaab(self):
        _timings = {"a_1": 13, "a_2": 10, "a_3": 6, "b_1": 1}
        _yappi._set_test_timings(_timings)

        self._ncall = 1

        def a():
            if self._ncall == 3:
                b()
                return
            self._ncall += 1
            a()

        def b():
            return

        stats = utils.run_and_get_func_stats(a)
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        cfsaa = fsa.children[fsa]
        cfsab = fsa.children[fsb]
        self.assertEqual(fsa.ttot, 13)
        self.assertEqual(fsa.tsub, 12)
        self.assertEqual(fsb.ttot, 1)
        self.assertEqual(fsb.tsub, 1)
        self.assertEqual(cfsaa.ttot, 10)
        self.assertEqual(cfsaa.tsub, 9)
        self.assertEqual(cfsab.ttot, 1)
        self.assertEqual(cfsab.tsub, 1)

    def test_abab(self):
        _timings = {"a_1": 13, "b_1": 10, "a_2": 6, "b_2": 1}
        _yappi._set_test_timings(_timings)

        self._ncall = 1

        def a():
            b()

        def b():
            if self._ncall == 2:
                return
            self._ncall += 1
            a()

        stats = utils.run_and_get_func_stats(a)
        fsa = utils.find_stat_by_name(stats, 'a')
        fsb = utils.find_stat_by_name(stats, 'b')
        cfsab = fsa.children[fsb]
        cfsba = fsb.children[fsa]
        self.assertEqual(fsa.ttot, 13)
        self.assertEqual(fsa.tsub, 8)
        self.assertEqual(fsb.ttot, 10)
        self.assertEqual(fsb.tsub, 5)
        self.assertEqual(cfsab.ttot, 10)
        self.assertEqual(cfsab.tsub, 5)
        self.assertEqual(cfsab.ncall, 2)
        self.assertEqual(cfsab.nactualcall, 1)
        self.assertEqual(cfsba.ttot, 6)
        self.assertEqual(cfsba.tsub, 5)


if __name__ == '__main__':
    #     import sys;sys.argv = ['', 'BasicUsage.test_run_as_script']
    #     import sys;sys.argv = ['', 'MultithreadedScenarios.test_subsequent_profile']
    unittest.main()