pybik 2.1-1 / usr / lib / pybik / pybiktest / runner.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-

#  Copyright © 2013-2015  B. Clausius <barcc@gmx.de>
#
#  This program is free software: you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation, either version 3 of the License, or
#  (at your option) any later version.
#
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program.  If not, see <http://www.gnu.org/licenses/>.



import sys, os, re
import random
from collections import namedtuple, OrderedDict
from contextlib import suppress
import tempfile
import time

from PyQt5.QtCore import Qt, QTimer
from PyQt5.QtWidgets import QAction
from PyQt5.QtTest import QTest

from . import utils
from pybiklib.settings import settings


class Quit (Exception): pass    # pylint: disable=C0321


class Logger:
    def __init__(self):
        self.lineno = None
        self.result = None
        
    def log(self, *args, **kwargs):
        if self.lineno is not None:
            args = ('line {}:'.format(self.lineno),) + args
        print(*args, **kwargs)
            
    def logf(self, format, *args):
        self.log(format.format(*args))
        
    def error(self, message, *args):
        self.log(message, *args, file=sys.stderr)
        message = message.rstrip(':')
        if message not in self.result:
            self.result[message] = 1
        else:
            self.result[message] += 1
        
    def errorf(self, message, format, *args):
        self.error(message, format.format(*args))
        
logger = Logger()
log = logger.log
logf = logger.logf
log_error = logger.error
logf_error = logger.errorf


class NamedObject:     # pylint: disable=R0903
    def __init__(self, name):
        self.name = name
    def __repr__(self):
        return self.name
matchall = NamedObject('matchall')
nomatch = NamedObject('nomatch')


def mkState(fields):
    _State = namedtuple('_State', fields)
    class State(_State):   # pylint: disable=W0232
        __slots__ = ()
        def tostr(self, other=None):
            def field_tostr(i):
                if self[i] is matchall:
                    return ''
                elif other is None or self[i] != other[i]:
                    return '{}={!r}, '.format(self.fields[i], self[i])
                else:
                    return ' ' * (len(self.fields[i]) + len(repr(self[i])) + 3)
            return ''.join([field_tostr(i) for i in range(len(self))]).rstrip(', ')
            
        fields = _State._fields
        asdict = _State._asdict
        replace = _State._replace
    State.default = State(**{f: matchall for f in State.fields})
    return State
    
    
class StateInfo:   # pylint: disable=R0903
    def __init__(self):
        self.untested_tnames = None
        self.islimit = None
        
    def __str__(self):
        return 'StateInfo: untested_t: #{}, islimit: {}'.format(
                len(self.untested_tnames), self.islimit)
                
        
class StatesInfo (dict):
    __slots__ = ()
    
    def __init__(self):
        dict.__init__(self)
        
    def istested(self, *args):
        if len(args) == 0:
            for s in self.values():
                if s.untested_tnames and s.islimit:
                    return False
            return True
        elif len(args) == 1:
            state = args[0]
            return state in self and not (self[state].untested_tnames and self[state].islimit)
        else:
            raise TypeError('istested expected at most 1 argument, got %s' % len(args))
            
    def reset(self, untested_tnames):
        for v in self.values():
            v.untested_tnames = untested_tnames[:]
            
    def get_unreached(self):
        return [s for s, si in self.items() if si.untested_tnames and si.islimit]
        
        
class Transition:
    Func = namedtuple('Func', 'name fname args func')
    
    def __init__(self, name, func=None):
        if name:
            fname, args = self._parse_name(name)
            self.funcs = [self.Func(name, fname, args, func)]
        else:
            self.funcs = []
        self.exprs_src = []
        self.exprs = {}
        self.chains = {}
        
    @property
    def name(self):
        return self.funcs[0].name if self.funcs else ''
        
    def appendfunc(self, name):
        assert self.funcs
        fname, args = self._parse_name(name)
        self.funcs.append(self.Func(name, fname, args, None))
        
    def setfunc(self, i, func):
        assert self.funcs[i].func is None
        self.funcs[i] = self.funcs[i]._replace(func=func)
        
    @staticmethod
    def _parse_name(name):
        fname, *args = name.split(' ', 1)
        args = args[0] if args else None
        return fname, args
        
        
class TestData:
    def __init__(self):
        self.constants = {}
        self.transitions = {}
        self.limits = []
        self.stateinfos = StatesInfo()
        self.path_to_untested = []
        
    def islimit(self, state):
        state = state.asdict()
        for limit in self.limits:
            try:
                if not eval(limit, self.constants, state):
                    return False
            except Exception as e:  # pylint: disable=W0703
                log_error('error in limit:', e)
        return True
        
    def reset(self):
        self.stateinfos.reset(list(self.transitions.keys()))
        
    def add_stateinfo(self, state):
        try:
            return self.stateinfos[state]
        except KeyError:
            stateinfo = dict.setdefault(self.stateinfos, state, StateInfo())
            stateinfo.untested_tnames = list(self.transitions.keys())
            stateinfo.islimit = self.islimit(state)
            return stateinfo
            
    def update_transition(self, transition, state, chain):
        self.add_stateinfo(state)
        target = chain[-1]
        if target is not None and target not in transition.chains:
            transition.chains[target] = [None] * len(transition.funcs)
            self.add_stateinfo(target)
            
    @staticmethod
    def _path_to_list(path):
        result = []
        while path is not None:
            item, path = path
            result.append(item)
        return result
        
    def get_path_to_untested(self, state):
        paths = [[(state, None), None]]
        reachable = set()
        while paths:
            path = paths.pop(0)
            state = path[0][0]
            stateinfo = self.stateinfos[state]
            reachable.add(state)
            if stateinfo.islimit and stateinfo.untested_tnames:
                return self._path_to_list(path)
            for name, transition in self.transitions.items():
                target = transition.chains[state][-1]
                if not stateinfo.islimit or self.stateinfos[target].islimit:
                    if target not in reachable:
                        paths.append([(target, name), path])
        return None
        
    def get_random_transition(self, state):
        if not self.path_to_untested:
            stateinfo = self.stateinfos[state]
            while stateinfo.untested_tnames:
                name = stateinfo.untested_tnames.pop(random.randrange(len(stateinfo.untested_tnames)))
                chain = self.transitions[name].chains.get(state, None)
                target = None if chain is None else chain[-1]
                if target is None or self.stateinfos[target].islimit:
                    return name
            self.path_to_untested = self.get_path_to_untested(state)
            if self.path_to_untested is None:
                raise Quit('End of test:')
            target, name = self.path_to_untested.pop()
            assert name is None
            assert target == state
        target, name = self.path_to_untested.pop()
        assert name is not None
        return name
        
        
class Result (OrderedDict):
    __slots__ = ('_perf_counter', '_process_time', '_showtime')
    
    def __init__(self, showtime):
        super().__init__()
        
        self._dict = dict(visited=0, states=0, transitions=0)
        self._showtime = showtime
        
    errors = property(lambda self: sum(self.values()))
    
    def __enter__(self):
        self._perf_counter = -time.perf_counter()
        self._process_time = -time.process_time()
        return self
        
    def __exit__(self, *unused_exc):
        self._perf_counter += time.perf_counter()
        self._process_time += time.process_time()
        return False
        
    @staticmethod
    def _fmt_dtime(val):
        fmt_sec = lambda val, fmt: format(val, fmt).rstrip('0').rstrip('.')
        if val < 60:
            return '{}s'.format(fmt_sec(val, '.2f'))
        s = fmt_sec(val % 60, '.1f')
        val = int(val // 60)
        if val < 60:
            return '{}m {}s'.format(val, s)
        m = val % 60
        val //= 60
        if val < 24:
            return '{}h {}m {}s'.format(val, m, s)
        h = val % 60
        val //= 24
        return '{}d {}h {}m {}s'.format(val, h, m, s)
        
    def fmt_dtime(self):
        return 'time: {}, {}'.format(self._fmt_dtime(self._perf_counter), self._fmt_dtime(self._process_time))
        
    def __str__(self):
        summary = '\n  visited: {0.visited}, states: {0.states}, transitions: {0.transitions}'.format(self)
        errors = self.errors
        if errors:
            summary += '\n  errors: {}'.format(errors)
            for k, v in self.items():
                summary += '\n    {} {}'.format(v, k)
        if self._showtime:
            summary += '\n  {}'.format(self.fmt_dtime())
        return summary
        
    def __getattr__(self, key):
        if key.startswith('_'):
            raise AttributeError()
        return self._dict[key]
        
    def __setattr__(self, key, value):
        if key.startswith('_'):
            OrderedDict.__setattr__(self, key, value)
        else:
            self._dict[key] = value
            
    def __iadd__(self, other):
        for k, v in other._dict.items():    # pylint: disable=W0212
            self._dict[k] += v
        for k, v in other.items():
            self[k] = self.get(k, 0) + v
        return self
        
    add = __iadd__
        
        
class TestReader:
    class ParseArgs:
        __slots__ = ('transition', 'initial', 'chain')
        
    def __init__(self, filename):
        self.current_testfile = filename
        self.constantstrs = []
        self.fields = OrderedDict()
        self.initial = None
        self.testdata = TestData()
        self.conditions_src = []
        self.conditions_code = []
        self.conditions_needed = []
        
    def isvalid(self, state, report=False):
        state = state.asdict()
        for i, condition in enumerate(self.conditions_code):
            try:
                if not eval(condition, self.testdata.constants, state):
                    if report:
                        logf_error('condition failed', '({}): {}', i+1, self.conditions_src[i])
                    self.conditions_needed[i] = True
                    return False
            except Exception as e:  # pylint: disable=W0703
                logf_error('error in condition', '({}): {}: {}', i+1, e, self.conditions_src[i])
        return True
        
    def _token_generic(self, pattern, line, *args):
        match = re.match(pattern, line.rstrip())
        if match is None:
            return False, args
        groups = match.groups()
        for arg, group in zip(args, groups):
            if arg is not None and arg != group:
                return False, groups
        return True, groups
        
    def token_keyvalue(self, line):
        success, (key, value) = self._token_generic(r'^(?:([\w-]+):|End\.)\s*(.*)$', line, None, None)
        return success, key, value
        
    def token_indent(self, line, indent):
        success, (indent, value) = self._token_generic(
                    r'^(\s*)(.+)$', line, indent, None)
        return success, value
        
    def token_indent2(self, line):
        success, (indent, value) = self._token_generic(
                    r'^(\s*)(\S.*)$', line, None, None)
        assert success
        return indent
        
    def token_assignvalue(self, expr):
        success, (field, value) = self._token_generic(r'^(\w+)\s*=\s*(.*)$', expr, None, None)
        return success, field, value
        
    def parse_file(self):
        structure = {
                # func:                 key,            required, repeat, multiln, children, [endhook]
                None:                   (None,          False,    False,  False,   [
                                                 self.parse_constants, self.parse_fields,
                                                 self.parse_conditions, self.parse_limits,
                                                 self.parse_initial, self.parse_transition,
                                                 self.parse_error_end,
                                                ]),
                self.parse_constants:   ('Constants',   False,    False,  True,    []),
                self.parse_fields:      ('Fields',      True,     False,  True,    [], self.parse_fields_end),
                self.parse_conditions:  ('Conditions',  False,    False,  True,    []),
                self.parse_limits:      ('Limits',      False,    False,  True,    []),
                self.parse_initial:     ('Initial',     True,     False,  True,    [self.parse_state]),
                self.parse_transition:  ('Transition',  False,    True,   True,    [self.parse_field_expr, self.parse_state]),
                self.parse_field_expr:  ('Expression',  False,    True,   False,   []),
                self.parse_state:       ('State',       False,    True,   True,    [], self.parse_state_end),
                self.parse_error_end:   (None,          False,    False,  False,   []),
        }
        valid_keys = [k for k, *unused in structure.values()]
        def _parse_indent(parentindent, line):
            indent = self.token_indent2(line)
            if parentindent is None:
                if indent != '':
                    log_error('unexpected indent')
                    return None
            else:
                if indent == parentindent:
                    return False
                if not indent.startswith(parentindent):
                    if parentindent.startswith(indent):
                        return False
                    log_error('non-matching indent')
                    return None
            return indent
        def _parse_indent_next(parentindent, line):
            indent = self.token_indent2(line)
            if parentindent == indent or parentindent.startswith(indent):
                return indent
            if indent.startswith(parentindent):
                log_error('unexpected indent')
            else:
                log_error('non-matching indent')
            return None
        def _parse_block(blocks, parentindent):
            nonlocal line
            indent = _parse_indent(parentindent, line)
            while indent is None:
                line = yield
                indent = _parse_indent(parentindent, line)
            if indent is False:
                return
            for func in blocks:
                key, required, repeat, multiline, children, *extra = structure[func]
                endhook = extra[0] if extra else lambda: None
                while True:
                    success, expr = self.token_indent(line, indent)
                    if not success:
                        break
                    success, lkey, value = self.token_keyvalue(expr)
                    if not success or lkey != key:
                        if lkey not in valid_keys:
                            log_error('unknown statement', repr(lkey))
                            line = yield
                            continue
                        if required:
                            func(None)
                            endhook()
                        break
                    if not func(value):
                        break
                    line = yield
                    if multiline:
                        while True:
                            multilineindent = _parse_indent(indent, line)
                            while multilineindent is None:
                                line = yield
                                multilineindent = _parse_indent(indent, line)
                            success, expr = self.token_indent(line, multilineindent)
                            if not success:
                                break
                            success, lkey, value = self.token_keyvalue(expr)
                            if success:
                                break
                            if not func(expr, first=False):
                                break
                            line = yield
                    yield from _parse_block(children, indent)
                    endhook()
                    if not repeat:
                        break
            assert parentindent is not None
            indent = _parse_indent_next(parentindent, line)
            while indent is None:
                line = yield
                indent = _parse_indent_next(parentindent, line)
        key, required, repeat, multiline, blocks = structure[None]
        self.parse_args = self.ParseArgs()
        try:
            line = yield
            yield from _parse_block(blocks, None)
            assert False, 'Should not be reached'
        finally:
            # GeneratorExit
            del self.parse_args
            
    def parse_constants(self, value, first=True):
        if not first:
            success, cname, cexpr = self.token_assignvalue(value)
            try:
                self.testdata.constants[cname] = eval(cexpr, self.testdata.constants)
            except SyntaxError:
                log_error('error parsing constant:', cname)
            else:
                self.constantstrs.append((cname, cexpr))
        return True
        
    def parse_fields(self, value, first=True):
        if first:
            if value is None:
                log_error('missing statement:', 'Fields')
                for f in sorted(utils.field_functions.keys()):
                    self.fields[f] = []
            return True
        else:
            success, field, fvalue = self.token_assignvalue(value)
            if not success:
                log_error('error parsing:', value)
            elif field not in list(utils.field_functions.keys()):
                log_error('unknown field:', field)
            else:
                if field in self.testdata.constants:
                    log_error('field overwrites constant:', field)
                    del self.testdata.constants[field]
                self.fields[field] = eval(fvalue, self.testdata.constants)
        return success
        
    def parse_fields_end(self):
        self.State = mkState(self.fields)
        
    def parse_conditions(self, value, first=True):
        if not first:
            self.conditions_src.append(value)
            try:
                code = compile(value, '<condition>', 'eval')
            except Exception as e:  # pylint: disable=W0703
                log_error('error in condition:', e)
                code = compile('True', '<condition>', 'eval')
            self.conditions_code.append(code)
            self.conditions_needed.append(False)
        return True
        
    def parse_limits(self, value, first=True):
        if not first:
            self.testdata.limits.append(value)
        return True
        
    def _parse_state_value(self, default, line, exprs=None):
        try:
            state = eval('dict({})'.format(line), self.testdata.constants)
        except SyntaxError:
            logf_error('error parsing state:', '{!r}', line)
            state = None
        else:
            exprs = list((exprs or {}).keys())
            for field in list(state.keys()):
                if field not in self.State.fields:
                    log_error('unselected field:', field)
                    del state[field]
                elif field in exprs:
                    logf_error('state contains expression field', '{}: {}', field, line)
            state = default and default.replace(**state)
        return state
        
    def parse_initial(self, value, first=True):
        if first:
            if value is None:
                log_error('missing statement:', 'Initial')
                value = ''
            transition = Transition(value)
            self.initial = transition
            self.parse_args.transition = transition
            self.parse_args.initial = True
        else:
            self.parse_args.transition.appendfunc(value)
        return True
        
    def parse_transition(self, value, first=True):
        if first:
            transition = Transition(value)
            if value in self.testdata.transitions:
                logf_error('duplicate transition:', '{!r}', value)
            else:
                self.testdata.transitions[value] = transition
            self.parse_args.transition = transition
            self.parse_args.initial = False
        else:
            self.parse_args.transition.appendfunc(value)
        return True
        
    def parse_field_expr(self, value):
        success, field, expr = self.token_assignvalue(value)
        if not success:
            log_error('error parsing:', value)
        else:
            transition = self.parse_args.transition
            transition.exprs_src.append((field, expr))
            if field in transition.exprs:
                logf_error('duplicate expression', 'for {}: {!r}', field, expr)
            elif field not in self.State.fields:
                log_error('unknown expression field:', field)
            else:
                try:
                    expr = compile(expr, '<expression>', 'eval')
                except Exception as e:
                    log_error('error in expression:', e)
                else:
                    transition.exprs[field] = expr
        return success
        
    def _replace_target(self, exprs, dstate, target, field, value):
        target_value = getattr(target, field)
        if field in exprs:
            try:
                value = eval(exprs[field], self.testdata.constants, dstate)
            except Exception as e:  # pylint: disable=W0703
                logf_error('error parsing expression', '{!r}: {}', exprs[field], e)
        if target_value is matchall:
            if value is matchall:
                value = dstate[field]
            target = target.replace(**{field: value})
        else:
            if field in exprs:
                if value == target_value:
                    log_error('target contains expression field:', field, 'value:', target_value)
                else:
                    log_error('target contains expression field:', field, 'value:', target_value,
                                'expected:', value)
        return target
        
    def _expand_state(self, fields, state):
        if state is None:
            return
        if len(fields) == 0:
            yield state
            return
            
        stack = [None for f in state.fields]
        i = 0
        dstate = state.asdict()
        
        while True:
            field = state.fields[i]
            value = getattr(state, field)
            values = iter(fields[field] if value is matchall else (value,))
            while True:
                try:
                    value = next(values)
                except StopIteration:
                    if i == 0:
                        return
                    i -= 1
                    values = stack[i]
                    field = state.fields[i]
                else:
                    if i == len(stack) - 1:
                        dstate[field] = value
                        yield self.State(**dstate)
                    else:
                        stack[i] = values
                        i += 1
                        dstate[field] = value
                        break
                
    def _expand_target(self, state, target, exprs):
        assert state is not None and target is not None, (state, target)
        dstate = state.asdict()
        for field, value in dstate.items():
            target = self._replace_target(exprs, dstate, target, field, value)
        return target
        
    def parse_state(self, value, first=True):
        if first:
            state = self._parse_state_value(self.State.default, value)
            self.parse_args.chain = [state]
        else:
            len_chain_max = len(self.parse_args.transition.funcs) + 1
            if len(self.parse_args.chain) == len_chain_max:
                log_error('too many states:', 'not more than', len_chain_max, 'possible')
            else:
                if value == '.':
                    value = ''
                self.parse_args.chain.append(self._parse_state_value(self.State.default, value, self.parse_args.transition.exprs))
        return True
        
    def parse_state_end(self):
        transition = self.parse_args.transition
        state, *chain = self.parse_args.chain
        while len(chain) < len(transition.funcs):
            chain.append(self.State.default)
        assert len(chain) == len(transition.funcs)
        target = chain[-1] if chain else state
        for state in self._expand_state(self.fields, state):
            if state in transition.chains and transition.chains[state][-1] is not None:
                log_error('duplicate state:', state.tostr())
            elif self.isvalid(state):
                chain_ = []
                state_ = state
                for target in chain:
                    target = self._expand_target(state_, target, transition.exprs)
                    if not self.isvalid(target, report=True):
                        log_error('invalid target state:', target.tostr())
                        target = None
                    chain_.append(target)
                    state_ = target
                transition.chains[state] = chain_
                if not self.parse_args.initial:
                    self.testdata.update_transition(transition, state, chain_)
                    
    def parse_error_end(self, value):
        if value.rstrip() != '':
            logf_error('error parsing:', '{!r}', value)
        return True     # success in case of an error just means skipping the line
        
    def read_test(self):
        try:
            parse_file = self.parse_file()
            next(parse_file)
            with open(self.current_testfile, 'rt', encoding='utf-8') as testfile:
                if self.current_testfile.endswith('.py'):
                    _locals = {}
                    exec(testfile.read(), {}, _locals)
                    testfile = _locals['lines']()
                for logger.lineno, line in enumerate(testfile):
                    if line.lstrip().startswith('#'):
                        continue
                    parse_file.send(line)
        finally:
            parse_file.send('End.')
            parse_file.close()
            logger.lineno = None
            for i, needed in enumerate(self.conditions_needed):
                if not needed:
                    logf_error('condition not needed:', '({}): {}', i+1, self.conditions_src[i])
                    
    def _ignore_fields(self, chains, igntype, exprs=None):
        result = {}
        for i, field in enumerate(self.State.fields):
            target_value = nomatch
            for state, chain in chains.items():
                target = chain[-1]
                if igntype == 'unchanged':
                    if state[i] != target[i]:
                        result = {}
                        break
                    state = state.replace(**{field: matchall})
                    target = target.replace(**{field: matchall})
                elif igntype == 'sametarget':
                    if target_value in {nomatch, target[i]}:
                        state = state.replace(**{field: matchall})
                        target_value = target[i]
                    else:
                        result = {}
                        break
                elif igntype == 'expressions':
                    if field in exprs:
                        target = target.replace(**{field: matchall})
                else:
                    assert False, igntype
                if state in result and result[state][-1] != target:
                    result = {}
                    break
                chain_ = chain[:]
                chain_[-1] = target
                result[state] = chain_
            else:
                chains = result
                result = {}
        return chains
        
    def write_test(self):
        with open(self.current_testfile, 'wt', encoding='utf-8') as testfile:
            testfile.write('Constants:\n')
            for name, value in self.constantstrs:
                print(' ', name, '=', value, file=testfile)
            testfile.write('Fields:\n')
            for field, values in self.fields.items():
                print(' ', field, '=', values, file=testfile)
            testfile.write('Conditions:\n')
            for condition in self.conditions_src:
                print(' ', condition, file=testfile)
            testfile.write('Limits:\n')
            for limit in self.testdata.limits:
                print(' ', limit, file=testfile)
            transition = self.initial
            testfile.write('Initial: %s\n' % transition.name)
            chains = list(transition.chains.items())
            #assert len(chains) == 1
            for state, chain in chains:
                testfile.write('  State: {}\n'.format(state.tostr()))
                targetstrs = []
                for target in chain:
                    targetstr = target.tostr(state)
                    if not targetstr.strip():
                        targetstr = '.'
                    targetstrs.append(targetstr)
                while targetstrs and targetstrs[-1] == '.':
                    del targetstrs[-1]
                for targetstr in targetstrs:
                    testfile.write('         {}\n'.format(targetstr))
            for name, transition in sorted(self.testdata.transitions.items()):
                testfile.write('Transition: %s\n' % name)
                chains = {s:c for s, c in transition.chains.items() if self.testdata.stateinfos[s].islimit}
                chains = self._ignore_fields(chains, 'expressions', transition.exprs)
                chains = self._ignore_fields(chains, 'unchanged')
                chains = self._ignore_fields(chains, 'sametarget')
                if len(transition.funcs) > 1:
                    for f in transition.funcs[1:]:
                        testfile.write('            {}\n'.format(f.name))
                for field, expr in sorted(transition.exprs_src):
                    testfile.write('  Expression: {} = {}\n'.format(field, expr))
                for state, chain in sorted(chains.items()):
                    testfile.write('  State: {}\n'.format(state.tostr()))
                    targetstrs = []
                    for target in chain:
                        targetstr = target.tostr(state)
                        if not targetstr.strip():
                            targetstr = '.'
                        targetstrs.append(targetstr)
                    while targetstrs and targetstrs[-1] == '.':
                        del targetstrs[-1]
                    for targetstr in targetstrs:
                        testfile.write('         {}\n'.format(targetstr))
                    
            
class TestRunner:
    _instance = None
    
    def __init__(self, testdata_dir, test, test_args):
        assert self._instance is None
        self.current_test = test
        self.write_mode = 'no'
        self.log_widgets = False
        for a in test_args:
            if a.startswith('write-n'):
                self.write_mode = 'no'
            elif a.startswith('write-y'):
                self.write_mode = 'yes'
            elif a.startswith('write-e'):
                self.write_mode = 'error'
            elif a == 'log-widgets':
                self.log_widgets = True
        filename = os.path.join(testdata_dir, test)
        self.reader = TestReader(filename)
        self.testdata = self.reader.testdata
        self.widgets = {}
        self.known_widget_functions = []
        self.field_functions = []
        
    def run(self, main_window):
        self.main_window = main_window
        self.running = True
        settings.draw.speed = 120
        QTimer.singleShot(0, self.loop)
        
    def get_state(self):
        state = {field: func() for field, func in self.field_functions}
        return self.State(**state)
        
    def find_qobjects(self, root_widget, write_template=False):
        def log_obj(indent, msg, name, obj):
            if self.log_widgets:
                logf('{}{}: {} ({})', '  '*indent, msg, name, obj.__class__.__name__)
            
        objects = [(root_widget, 0)]
        while objects:
            obj, indent = objects.pop()
            name = obj.objectName()
            if name in self.widgets:
                log_obj(indent, 'kwnobj', name, obj)
            elif name and not name.startswith('qt_'):
                self.widgets[name] = obj
                if isinstance(obj, QAction):
                    if self.reader.initial is not None:
                        transition = self.reader.initial
                        if transition.name == name:
                            transition.setfunc(0, obj.trigger)
                    if name in self.testdata.transitions:
                        transition = self.testdata.transitions[name]
                        transition.setfunc(0, obj.trigger)
                    elif write_template:
                        self.testdata.transitions[name] = Transition(name, obj.trigger)
                log_obj(indent, 'object', name, obj)
            else:
                log_obj(indent, 'intobj', name, obj)
            for child in reversed(obj.children()):
                objects.append((child, indent+1))
                
    def update_field_functions(self):
        self.field_functions = []
        mk_func = lambda wname, func: lambda: func(self.widgets[wname])
        mk_default = lambda default: lambda: default
        for field, (wname, default, func) in utils.field_functions.items():
            if field in self.State.fields:
                if wname in self.widgets:
                    self.field_functions.append((field, mk_func(wname, func)))
                else:
                    self.field_functions.append((field, mk_default(default)))
        
    def update_transition_functions(self, write_template=False):
        def mk_transition(func, widgets, *args):
            if len(widgets) == 1:
                widgets = widgets[0]
            return lambda: func(widgets, *args)
            
        def parse_transition_func(func, widgets, tfunc):
            if tfunc.args is None:
                return mk_transition(func, widgets)
            try:
                args = eval(tfunc.args, {'Qt': Qt}, self.testdata.constants)
            except Exception as e:  # pylint: disable=W0703
                logf_error('error parsing expression', '{!r}: {}', func.args, e)
                func = lambda: None
            else:
                if type(args) is not tuple:
                    args = (args,)
                func = mk_transition(func, widgets, *args)
            return func
            
        for wnames, fname, func in utils.widget_functions:
            if fname in self.known_widget_functions:
                continue
            if not all((n in self.widgets) for n in wnames):
                continue
            self.known_widget_functions.append(fname)
            widgets = [self.widgets[n] for n in wnames]
            initial = ([] if self.reader.initial is None or not self.reader.initial.funcs
                          else [(self.reader.initial.name, self.reader.initial)])
            for tname, transition in initial + list(self.testdata.transitions.items()):
                tfunc = transition.funcs[0]
                assert tname == tfunc.name
                if tfunc.fname == fname:
                    tfunc = parse_transition_func(func, widgets, tfunc)
                    transition.setfunc(0, tfunc)
                for i, chainfunc in enumerate(transition.funcs[1:]):
                    if transition.funcs[i+1].func is not None:
                        continue
                    cfname = chainfunc.fname
                    for cwnames, _cfname, cfunc in utils.widget_functions:
                        if _cfname == cfname:
                            cwidgets = [self.widgets[n] for n in cwnames]
                            cfunc = parse_transition_func(cfunc, cwidgets, chainfunc)
                            break
                    else:
                        if cfname in self.widgets and isinstance(self.widgets[cfname], QAction):
                            cwidgets = [self.widgets[cfname]]
                            cfunc = cwidgets[0].trigger
                        else:
                            assert False, '%s not found in utils.widget_functions and actions' % cfname
                    transition.setfunc(i+1, cfunc)
            if write_template:
                assert fname not in self.testdata.transitions
                self.testdata.transitions[fname] = Transition(fname, mk_transition(func, widgets))
                
    def init_test(self):
        try:
            self.reader.read_test()
        except OSError as e:
            log('Error reading test data file:', e)
        finally:
            self.State = self.reader.State
            self.fields = self.reader.fields
        assert list(self.fields.keys()) == list(self.State.fields), (list(self.fields.keys()), self.State.fields)
        write_template = False
        if not self.testdata.transitions:
            log_error('empty test')
            write_template = True
            
        # introspect ui
        self.find_qobjects(self.main_window, write_template)
        self.update_field_functions()
        self.update_transition_functions(write_template)
        
        self.current_state = self.get_state()
            
        self.testdata.reset()
        if write_template:
            raise Quit('Template created:')
        
    def check_initial(self):
        transition = self.reader.initial
        if transition is None:
            transition = self.reader.initial = Transition('')
        if len(transition.chains) > 1:
            logf_error('ambiguous initial transition:', '\n  {}: {} states',
                            transition.name, len(transition.chains))
        for state in list(transition.chains.keys()):
            if state != self.current_state:
                logf_error('wrong initial state:', '\n     found: {}\n  expected: {}',
                            self.current_state.tostr(), state)
                del transition.chains[state]
        if len(transition.chains) == 0:
            logf_error('missing initial state:', '{}', self.current_state)
            transition.chains[self.current_state] = [None for f in transition.funcs]
                    
    def check_transition(self, transition, chain, i, initial=False):
        target = chain[i]
        current_state = self.get_state()
        if target != current_state:
            if target is None:
                logf_error('unknown transition:', '\n  {}: {} -> {}',
                            transition.name, self.current_state.tostr(), current_state.tostr())
            else:
                logf_error('wrong target', 'for {}: {}\n     found: -> {}\n  expected: -> {}',
                            transition.name, self.current_state.tostr(), current_state.tostr(), target.tostr())
            chain[i] = current_state
            if not initial and i+1 == len(chain):
                self.testdata.update_transition(transition, self.current_state, chain)
        self.current_state = current_state
        
    def check_result(self):
        for name, transition in list(self.testdata.transitions.items()):
            if transition.funcs[0].func is None:
                log_error('unused transition:', name)
                del self.testdata.transitions[name]
        all_states = list(self.testdata.stateinfos.keys())
        all_transitions = [(name, state) for name, transition in self.testdata.transitions.items()
                                            for state in transition.chains.keys()]
        logger.result.states = len(all_states)    # pylint: disable=W0201
        logger.result.transitions = len(all_transitions)  # pylint: disable=W0201
        unreached = self.testdata.stateinfos.get_unreached()
        for state in unreached:
            log_error('unreached', state.tostr())
        field_values = {}
        for name, transition in self.testdata.transitions.items():
            chains = transition.chains
            for state, chain in list(chains.items()):
                target = chain[-1]
                if state in unreached or not self.testdata.stateinfos[state].islimit or target is None:
                    del chains[state]
                    continue
                for field, svalue, tvalue in zip(self.State.fields, state, target):
                    field_values.setdefault(field, set()).update([svalue, tvalue])
        for field, values in self.fields.items():
            if field in field_values:
                if set(values) != field_values[field]:
                    log_error('changed field values:', field)
                    values[:] = sorted(field_values[field])
            else:
                log_error('unused field', field)
                
    def step(self, transition, initial=False):
        chain = transition.chains.get(self.current_state, None)
        if chain is None:
            logf_error('unknown chain:', '\n  {}: {}',
                        transition.name, self.current_state.tostr())
            chain = [None] * len(transition.funcs)
            transition.chains[self.current_state] = chain
        for i, func in enumerate(transition.funcs):
            yield from self._step(func)
            self.check_transition(transition, chain, i, initial)
        
    def _step(self, func):
        postfunc = func.func()
        if postfunc == 'find_qobjects':
            self.find_qobjects(self.main_window)
            self.update_field_functions()
            self.update_transition_functions()
        logger.result.visited += 1    # pylint: disable=E1101
        yield
        while self.main_window.is_animating():
            yield
            
    def loop(self):
        if not QTest.qWaitForWindowExposed(self.main_window, timeout=5000):
            log_error('Wait for window timed out')
            self.running = False
            return
        for unused in self._iloop():
            if not self.main_window.isVisible():
                log_error('Unexpected end of test')
                self.running = False
                return
            QTest.qWait(10)
        # without this line sometimes there is a segfault in the next window
        self.main_window.deleteLater()
            
    def _iloop(self):
        self.current_state = name = target = None  # for except statement
        try:
            log('Initializing:', self.current_test)
            self.init_test()
            yield
            log('Running:', self.current_test)
            self.check_initial()
            assert self.reader.initial is not None
            yield from self.step(self.reader.initial, True)
            if self.current_state not in self.testdata.stateinfos:
                log_error('unknown state:', self.current_state)
                stateinfo = self.testdata.add_stateinfo(self.current_state)
                if not stateinfo.islimit:
                    log_error('state not in limit:', self.current_state)
            while True:
                name = self.testdata.get_random_transition(self.current_state)
                transition = self.testdata.transitions[name]
                yield from self.step(transition)
        except Quit as e:
            self.check_result()
            if self.write_mode == 'yes' or logger.result.errors and self.write_mode == 'error':
                self.reader.write_test()
                log('test data file written')
            log(e, self.current_test)
        except Exception:
            self.check_result()
            logf('exception in {}: {} -> {}', name,
                    self.current_state and self.current_state.tostr(),
                    target and target.tostr())
            sys.excepthook(*sys.exc_info())
        finally:
            if self.running:
                self.running = False
                self.main_window.close()
        
    @classmethod
    def wrap(cls, testdata_dir, tests, test_args):
        showtime = 'notime' not in test_args
        with Result(showtime) as result:
            temp_dir = tempfile.mkdtemp(prefix='pybiktest-')
            results = []
            cnt_tests = 0
            for test in tests:
                if '=' in test:
                    test, repeat_cnt = test.split('=', 1)
                    repeat_cnt = int(repeat_cnt)
                else:
                    repeat_cnt = 1
                with Result(showtime) as cnt_result:
                    results.append((test, cnt_result))
                    cnt_tests += repeat_cnt
                    for repeat_idx in range(repeat_cnt):
                        with Result(showtime) as logger.result:
                            if repeat_cnt > 1:
                                logf('Test: {} {}/{}', test, repeat_idx+1, repeat_cnt)
                            else:
                                log('Test:', test)
                            instance = cls(testdata_dir, test, test_args)
                            settings.reset()
                            instance.settings_file = tempfile.mktemp(prefix='settings.conf-', dir=temp_dir)
                            instance.games_file = tempfile.mktemp(prefix='games-', dir=temp_dir)
                            yield instance
                            if instance.running:
                                log_error('Unexpected end of testrunner')
                        log('Result:', test, logger.result)
                        log('')
                        os.remove(instance.settings_file)
                        del instance.settings_file
                        os.remove(instance.games_file)
                        del instance.games_file
                        result += logger.result
                        cnt_result.add(logger.result)
            with suppress(OSError):
                os.rmdir(temp_dir)
                
        for test, testresult in results:
            fmt = 'Test {0}: {1} errors' if testresult.errors else 'Test {0}: success'
            if showtime:
                fmt += ', {2}'
                dtime = testresult.fmt_dtime()
                logf(fmt, test, testresult.errors, dtime)
            else:
                logf(fmt, test, testresult.errors)
        if cnt_tests > 1:
            log('')
            logf('Summary ({}): {}', cnt_tests, result)