""" Test the bdb module.

    A test defines a list of tuples that may be seen as paired tuples, each
    pair being defined by 'expect_tuple, set_tuple' as follows:

        ([event, [lineno[, co_name[, eargs]]]]), (set_type, [sargs])

    * 'expect_tuple' describes the expected current state of the Bdb instance.
      It may be the empty tuple and no check is done in that case.
    * 'set_tuple' defines the set_*() method to be invoked when the Bdb
      instance reaches this state.

    Example of an 'expect_tuple, set_tuple' pair:

        ('line', 2, 'tfunc_main'), ('step', )

    Definitions of the members of the 'expect_tuple':
        event:
            Name of the trace event. The set methods that do not give back
            control to the tracer [1] do not trigger a tracer event and in
            that case the next 'event' may be 'None' by convention, its value
            is not checked.
            [1] Methods that trigger a trace event are set_step(), set_next(),
            set_return(), set_until() and set_continue().
        lineno:
            Line number. Line numbers are relative to the start of the
            function when tracing a function in the test_bdb module (i.e. this
            module).
        co_name:
            Name of the function being currently traced.
        eargs:
            A tuple:
            * On an 'exception' event the tuple holds a class object, the
              current exception must be an instance of this class.
            * On a 'line' event, the tuple holds a dictionary and a list. The
              dictionary maps each breakpoint number that has been hit on this
              line to its hits count. The list holds the list of breakpoint
              number temporaries that are being deleted.

    Definitions of the members of the 'set_tuple':
        set_type:
            The type of the set method to be invoked. This may
            be the type of one of the Bdb set methods: 'step', 'next',
            'until', 'return', 'continue', 'break', 'quit', or the type of one
            of the set methods added by test_bdb.Bdb: 'ignore', 'enable',
            'disable', 'clear', 'up', 'down'.
        sargs:
            The arguments of the set method if any, packed in a tuple.
"""

import bdb as _bdb
import sys
import os
import unittest
import textwrap
import importlib
import linecache
from contextlib import contextmanager
from itertools import islice, repeat
import test.support

class BdbException(Exception): pass
class BdbError(BdbException): """Error raised by the Bdb instance."""
class BdbSyntaxError(BdbException): """Syntax error in the test case."""
class BdbNotExpectedError(BdbException): """Unexpected result."""

# When 'dry_run' is set to true, expect tuples are ignored and the actual
# state of the tracer is printed after running each set_*() method of the test
# case. The full list of breakpoints and their attributes is also printed
# after each 'line' event where a breakpoint has been hit.
dry_run = 0

def reset_Breakpoint():
    _bdb.Breakpoint.next = 1
    _bdb.Breakpoint.bplist = {}
    _bdb.Breakpoint.bpbynumber = [None]

def info_breakpoints():
    bp_list = [bp for  bp in _bdb.Breakpoint.bpbynumber if bp]
    if not bp_list:
        return ''

    header_added = False
    for bp in bp_list:
        if not header_added:
            info = 'BpNum Temp Enb Hits Ignore Where\n'
            header_added = True

        disp = 'yes ' if bp.temporary else 'no  '
        enab = 'yes' if bp.enabled else 'no '
        info += ('%-5d %s %s %-4d %-6d at %s:%d' %
                    (bp.number, disp, enab, bp.hits, bp.ignore,
                     os.path.basename(bp.file), bp.line))
        if bp.cond:
            info += '\n\tstop only if %s' % (bp.cond,)
        info += '\n'
    return info

class Bdb(_bdb.Bdb):
    """Extend Bdb to enhance test coverage."""

    def trace_dispatch(self, frame, event, arg):
        self.currentbp = None
        return super().trace_dispatch(frame, event, arg)

    def set_break(self, filename, lineno, temporary=False, cond=None,
                  funcname=None):
        if isinstance(funcname, str):
            if filename == __file__:
                globals_ = globals()
            else:
                module = importlib.import_module(filename[:-3])
                globals_ = module.__dict__
            func = eval(funcname, globals_)
            code = func.__code__
            filename = code.co_filename
            lineno = code.co_firstlineno
            funcname = code.co_name

        res = super().set_break(filename, lineno, temporary=temporary,
                                 cond=cond, funcname=funcname)
        if isinstance(res, str):
            raise BdbError(res)
        return res

    def get_stack(self, f, t):
        self.stack, self.index = super().get_stack(f, t)
        self.frame = self.stack[self.index][0]
        return self.stack, self.index

    def set_ignore(self, bpnum):
        """Increment the ignore count of Breakpoint number 'bpnum'."""
        bp = self.get_bpbynumber(bpnum)
        bp.ignore += 1

    def set_enable(self, bpnum):
        bp = self.get_bpbynumber(bpnum)
        bp.enabled = True

    def set_disable(self, bpnum):
        bp = self.get_bpbynumber(bpnum)
        bp.enabled = False

    def set_clear(self, fname, lineno):
        err = self.clear_break(fname, lineno)
        if err:
            raise BdbError(err)

    def set_up(self):
        """Move up in the frame stack."""
        if not self.index:
            raise BdbError('Oldest frame')
        self.index -= 1
        self.frame = self.stack[self.index][0]

    def set_down(self):
        """Move down in the frame stack."""
        if self.index + 1 == len(self.stack):
            raise BdbError('Newest frame')
        self.index += 1
        self.frame = self.stack[self.index][0]

class Tracer(Bdb):
    """A tracer for testing the bdb module."""

    def __init__(self, expect_set, skip=None, dry_run=False, test_case=None):
        super().__init__(skip=skip)
        self.expect_set = expect_set
        self.dry_run = dry_run
        self.header = ('Dry-run results for %s:' % test_case if
                       test_case is not None else None)
        self.init_test()

    def init_test(self):
        self.cur_except = None
        self.expect_set_no = 0
        self.breakpoint_hits = None
        self.expected_list = list(islice(self.expect_set, 0, None, 2))
        self.set_list = list(islice(self.expect_set, 1, None, 2))

    def trace_dispatch(self, frame, event, arg):
        # On an 'exception' event, call_exc_trace() in Python/ceval.c discards
        # a BdbException raised by the Tracer instance, so we raise it on the
        # next trace_dispatch() call that occurs unless the set_quit() or
        # set_continue() method has been invoked on the 'exception' event.
        if self.cur_except is not None:
            raise self.cur_except

        if event == 'exception':
            try:
                res = super().trace_dispatch(frame, event, arg)
                return res
            except BdbException as e:
                self.cur_except = e
                return self.trace_dispatch
        else:
            return super().trace_dispatch(frame, event, arg)

    def user_call(self, frame, argument_list):
        # Adopt the same behavior as pdb and, as a side effect, skip also the
        # first 'call' event when the Tracer is started with Tracer.runcall()
        # which may be possibly considered as a bug.
        if not self.stop_here(frame):
            return
        self.process_event('call', frame, argument_list)
        self.next_set_method()

    def user_line(self, frame):
        self.process_event('line', frame)

        if self.dry_run and self.breakpoint_hits:
            info = info_breakpoints().strip('\n')
            # Indent each line.
            for line in info.split('\n'):
                print('  ' + line)
        self.delete_temporaries()
        self.breakpoint_hits = None

        self.next_set_method()

    def user_return(self, frame, return_value):
        self.process_event('return', frame, return_value)
        self.next_set_method()

    def user_exception(self, frame, exc_info):
        self.exc_info = exc_info
        self.process_event('exception', frame)
        self.next_set_method()

    def do_clear(self, arg):
        # The temporary breakpoints are deleted in user_line().
        bp_list = [self.currentbp]
        self.breakpoint_hits = (bp_list, bp_list)

    def delete_temporaries(self):
        if self.breakpoint_hits:
            for n in self.breakpoint_hits[1]:
                self.clear_bpbynumber(n)

    def pop_next(self):
        self.expect_set_no += 1
        try:
            self.expect = self.expected_list.pop(0)
        except IndexError:
            raise BdbNotExpectedError(
                'expect_set list exhausted, cannot pop item %d' %
                self.expect_set_no)
        self.set_tuple = self.set_list.pop(0)

    def process_event(self, event, frame, *args):
        # Call get_stack() to enable walking the stack with set_up() and
        # set_down().
        tb = None
        if event == 'exception':
            tb = self.exc_info[2]
        self.get_stack(frame, tb)

        # A breakpoint has been hit and it is not a temporary.
        if self.currentbp is not None and not self.breakpoint_hits:
            bp_list = [self.currentbp]
            self.breakpoint_hits = (bp_list, [])

        # Pop next event.
        self.event= event
        self.pop_next()
        if self.dry_run:
            self.print_state(self.header)
            return

        # Validate the expected results.
        if self.expect:
            self.check_equal(self.expect[0], event, 'Wrong event type')
            self.check_lno_name()

        if event in ('call', 'return'):
            self.check_expect_max_size(3)
        elif len(self.expect) > 3:
            if event == 'line':
                bps, temporaries = self.expect[3]
                bpnums = sorted(bps.keys())
                if not self.breakpoint_hits:
                    self.raise_not_expected(
                        'No breakpoints hit at expect_set item %d' %
                        self.expect_set_no)
                self.check_equal(bpnums, self.breakpoint_hits[0],
                    'Breakpoint numbers do not match')
                self.check_equal([bps[n] for n in bpnums],
                    [self.get_bpbynumber(n).hits for
                        n in self.breakpoint_hits[0]],
                    'Wrong breakpoint hit count')
                self.check_equal(sorted(temporaries), self.breakpoint_hits[1],
                    'Wrong temporary breakpoints')

            elif event == 'exception':
                if not isinstance(self.exc_info[1], self.expect[3]):
                    self.raise_not_expected(
                        "Wrong exception at expect_set item %d, got '%s'" %
                        (self.expect_set_no, self.exc_info))

    def check_equal(self, expected, result, msg):
        if expected == result:
            return
        self.raise_not_expected("%s at expect_set item %d, got '%s'" %
                                (msg, self.expect_set_no, result))

    def check_lno_name(self):
        """Check the line number and function co_name."""
        s = len(self.expect)
        if s > 1:
            lineno = self.lno_abs2rel()
            self.check_equal(self.expect[1], lineno, 'Wrong line number')
        if s > 2:
            self.check_equal(self.expect[2], self.frame.f_code.co_name,
                                                'Wrong function name')

    def check_expect_max_size(self, size):
        if len(self.expect) > size:
            raise BdbSyntaxError('Invalid size of the %s expect tuple: %s' %
                                 (self.event, self.expect))

    def lno_abs2rel(self):
        fname = self.canonic(self.frame.f_code.co_filename)
        lineno = self.frame.f_lineno
        return ((lineno - self.frame.f_code.co_firstlineno + 1)
            if fname == self.canonic(__file__) else lineno)

    def lno_rel2abs(self, fname, lineno):
        return (self.frame.f_code.co_firstlineno + lineno - 1
            if (lineno and self.canonic(fname) == self.canonic(__file__))
            else lineno)

    def get_state(self):
        lineno = self.lno_abs2rel()
        co_name = self.frame.f_code.co_name
        state = "('%s', %d, '%s'" % (self.event, lineno, co_name)
        if self.breakpoint_hits:
            bps = '{'
            for n in self.breakpoint_hits[0]:
                if bps != '{':
                    bps += ', '
                bps += '%s: %s' % (n, self.get_bpbynumber(n).hits)
            bps += '}'
            bps = '(' + bps + ', ' + str(self.breakpoint_hits[1]) + ')'
            state += ', ' + bps
        elif self.event == 'exception':