#!/usr/bin/python3 # -*- coding: utf-8 -*- # Copyright © 2013-2015 B. Clausius # # 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 . 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, '', 'eval') except Exception as e: # pylint: disable=W0703 log_error('error in condition:', e) code = compile('True', '', '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, '', '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)